Spaces:
Running
Running
File size: 2,841 Bytes
a31f556 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 | """
modules/context_carry.py - Phase 7 context carry across restarts
Stores:
- last mode
- last few user/assistant interactions (compact)
- last active tasks snapshot
Used to:
- rebuild a small "resume" hint for the brain
- show continuity after a crash/reboot
"""
from __future__ import annotations
import json
import os
import time
from datetime import datetime
from config import DATA_DIR
CTX_FILE = os.path.join(DATA_DIR, "session_context.json")
def _load() -> dict:
if not os.path.exists(CTX_FILE):
return {}
try:
with open(CTX_FILE, "r") as f:
data = json.load(f)
if isinstance(data, dict):
return data
except Exception:
pass
return {}
def _save(data: dict) -> None:
os.makedirs(os.path.dirname(CTX_FILE), exist_ok=True)
try:
with open(CTX_FILE, "w") as f:
json.dump(data, f, indent=2)
except Exception:
pass
def snapshot(mode: str | None = None) -> None:
"""
Save a compact snapshot for resuming context.
"""
data = _load()
data["time"] = time.time()
data["iso"] = datetime.now().isoformat()
if mode:
data["mode"] = mode
# last interactions (small)
try:
from modules.memory import INTERACTION_FILE
with open(INTERACTION_FILE, "r") as f:
interactions = json.load(f)
if not isinstance(interactions, list):
interactions = []
compact = interactions[-12:]
data["recent"] = [
{"role": it.get("role", ""), "text": (it.get("text", "")[:180]), "time": it.get("time", 0)}
for it in compact
]
except Exception:
pass
# tasks snapshot
try:
from modules.memory import get_unfinished_tasks
tasks = get_unfinished_tasks()
data["tasks"] = tasks[:10]
except Exception:
pass
_save(data)
def resume_block(max_lines: int = 12) -> str:
"""
Return a short block injected into prompts for continuity.
"""
data = _load()
if not data:
return ""
lines: list[str] = []
ts = data.get("iso") or ""
if ts:
lines.append(f"[SESSION CONTEXT] Last snapshot: {ts}")
mode = data.get("mode")
if mode:
lines.append(f"[LAST MODE] {mode}")
tasks = data.get("tasks") or []
if tasks:
lines.append("[PENDING] " + "; ".join(tasks[:3]))
recent = data.get("recent") or []
if recent:
lines.append("[RECENT]")
for it in recent[-8:]:
r = (it.get("role") or "").upper()
t = it.get("text") or ""
if r and t:
lines.append(f"{r}: {t}")
return "\n".join(lines[:max_lines])
|