"""Durable task ledger — the trail every background task leaves behind. WHY THIS EXISTS (2026-07-16, Ahmed's call): a background worker used to live only in RAM. If Jarvis crashed or restarted mid-task, the task — and any finding it had produced — vanished silently. And when Ahmed later asked "did you do X?" / "what happened to X?", the master had nothing durable to consult, so it either guessed or shrugged. Both are wrong. This ledger records EVERY dispatched task to disk the moment it starts, updates it tool-by-tool as it runs, and finalizes it on done/fail/kill — so: 1. RECOVERY — on boot, any task still marked "running" is stale (the process that owned it died); we relabel it "interrupted" so the trail shows exactly where things stopped. 2. RECALL — the master can query this ledger ("did you send that email?") and answer the TRUE state (done at 4:12 / still running / failed because the Gmail call errored / interrupted by a restart) instead of guessing. 3. CROSS-DEVICE MEMORY — on completion the pool also drops a milestone checkpoint into jarvis-memory (fire_save), so "what happened to X?" works days later from any device, not just this process's lifetime. Storage is a single JSON file (control/task-ledger.json) rewritten atomically on every change — the task volume is a handful per hour, so this is trivially cheap and dead simple to reason about. Kept to the last MAX_TASKS records. The module is self-contained: it imports nothing from the rest of the voice package at import time (memory_control is lazy-imported inside the checkpoint helper), so it can be used from the pool, the engine, or a plain thread. """ from __future__ import annotations import json import os import threading import time import uuid from datetime import datetime, timezone from pathlib import Path PROJECT = Path(__file__).resolve().parent.parent LEDGER = PROJECT / "control" / "task-ledger.json" MAX_TASKS = 200 # keep the last N tasks; older ones roll off MAX_STEPS = 24 # keep the last N tool-steps per task (head+tail matters) # terminal states — a task in one of these is finished and mark() won't move it _TERMINAL = {"done", "failed", "killed", "interrupted"} _lock = threading.Lock() _tasks: list[dict] = [] # in-memory mirror, oldest → newest _loaded = False def _now() -> str: return datetime.now(timezone.utc).isoformat(timespec="seconds") def _persist_locked() -> None: """Write the ledger atomically. Caller holds _lock.""" try: LEDGER.parent.mkdir(exist_ok=True) tmp = LEDGER.with_suffix(".json.tmp") tmp.write_text(json.dumps(_tasks[-MAX_TASKS:], ensure_ascii=False, indent=0)) os.replace(tmp, LEDGER) except Exception as e: # noqa: BLE001 — persistence must never crash a task print(f" [task-ledger: write failed: {str(e)[:120]}]") def load() -> None: """Read the ledger from disk into memory. Call once at boot.""" global _tasks, _loaded with _lock: if _loaded: return try: if LEDGER.is_file(): data = json.loads(LEDGER.read_text() or "[]") if isinstance(data, list): _tasks = data[-MAX_TASKS:] except Exception as e: # noqa: BLE001 print(f" [task-ledger: load failed: {str(e)[:120]}]") _tasks = [] _loaded = True def recover_stale() -> list[dict]: """Boot recovery: any task still 'running'/'queued' from a PRIOR process can't be alive now (we just started), so mark it 'interrupted'. Returns the records that were interrupted so the caller can proactively tell Ahmed.""" interrupted: list[dict] = [] with _lock: for t in _tasks: if t.get("status") in ("running", "queued"): t["status"] = "interrupted" t["ended"] = _now() t["error"] = "interrupted by a restart/crash before it finished" interrupted.append(dict(t)) if interrupted: _persist_locked() return interrupted def create(worker: str, title: str, task: str, tier: str = "worker") -> str: """Register a freshly-dispatched task. Returns its durable ledger id.""" tid = "t_" + uuid.uuid4().hex[:10] rec = { "id": tid, "worker": worker, "title": (title or "").strip()[:120], "task": (task or "").strip()[:800], "tier": tier, "status": "running", "steps": [], "result": "", "error": "", "created": _now(), "started": _now(), "ended": "", "checkpointed": False, } with _lock: _tasks.append(rec) # trim in memory too so the mirror can't grow unbounded if len(_tasks) > MAX_TASKS: del _tasks[:-MAX_TASKS] _persist_locked() return tid def _find_locked(tid: str) -> dict | None: for t in reversed(_tasks): if t.get("id") == tid: return t return None def step(tid: str, note: str) -> None: """Append a live tool-step note (what the worker is doing right now).""" if not tid or not note: return with _lock: t = _find_locked(tid) if t is None or t.get("status") in _TERMINAL: return t.setdefault("steps", []).append({"t": _now(), "note": note[:160]}) if len(t["steps"]) > MAX_STEPS: # keep the first 4 (how it started) + the most recent tail t["steps"] = t["steps"][:4] + t["steps"][-(MAX_STEPS - 4):] t["last_step"] = note[:160] _persist_locked() def finish(tid: str, result: str) -> dict | None: """Mark a task done with its result. Returns the record (for checkpointing).""" with _lock: t = _find_locked(tid) if t is None: return None t["status"] = "done" t["result"] = (result or "").strip()[:1500] t["ended"] = _now() _persist_locked() return dict(t) def fail(tid: str, error: str) -> dict | None: """Mark a task failed with the reason. Returns the record.""" with _lock: t = _find_locked(tid) if t is None: return None t["status"] = "failed" t["error"] = (error or "").strip()[:800] t["ended"] = _now() _persist_locked() return dict(t) def mark(tid: str, status: str) -> None: """Set a NON-result terminal status (killed/interrupted). Never downgrades a task that already finished (done/failed) — those keep their real outcome. Accepts the pool's status words and normalizes them.""" norm = {"killed": "killed", "stopped": "interrupted", "interrupted": "interrupted", "completed": "done", "failed": "failed"}.get(status, status) with _lock: t = _find_locked(tid) if t is None or t.get("status") in _TERMINAL: return t["status"] = norm t["ended"] = _now() if norm == "killed": t["error"] = "killed on request" elif norm == "interrupted": t["error"] = "interrupted before it finished" _persist_locked() def mark_checkpointed(tid: str) -> None: with _lock: t = _find_locked(tid) if t is not None: t["checkpointed"] = True _persist_locked() # -------------------------------------------------------------------------- # recall — what the master reads to answer "did you do X?" / "what happened…" # -------------------------------------------------------------------------- def get(tid: str) -> dict | None: with _lock: t = _find_locked(tid) return dict(t) if t else None def recent(n: int = 12) -> list[dict]: with _lock: return [dict(t) for t in _tasks[-n:]][::-1] # newest first def open_tasks() -> list[dict]: with _lock: return [dict(t) for t in _tasks if t.get("status") == "running"][::-1] def find(query: str, limit: int = 8) -> list[dict]: """Case-insensitive substring search over title/task/result/steps, newest first. Empty query returns the most recent tasks.""" q = (query or "").strip().lower() with _lock: pool = list(_tasks) if not q: return [dict(t) for t in pool[-limit:]][::-1] hits = [] for t in reversed(pool): hay = " ".join([ t.get("title", ""), t.get("task", ""), t.get("result", ""), t.get("error", ""), " ".join(s.get("note", "") for s in t.get("steps", [])), ]).lower() if q in hay: hits.append(dict(t)) if len(hits) >= limit: break return hits def _age(rec: dict) -> str: try: start = datetime.fromisoformat(rec.get("started") or rec.get("created")) end = (datetime.fromisoformat(rec["ended"]) if rec.get("ended") else datetime.now(timezone.utc)) s = int((end - start).total_seconds()) return f"{s // 60}m{s % 60:02d}s" if s >= 60 else f"{s}s" except Exception: # noqa: BLE001 return "?" def summarize(rec: dict) -> str: """One human line describing a task's outcome — for recall + checkpoints.""" st = rec.get("status", "?") title = rec.get("title", "task") when = (rec.get("ended") or rec.get("started") or "")[:16].replace("T", " ") if st == "done": body = rec.get("result", "").strip().splitlines()[0][:200] if rec.get("result") else "finished" return f"'{title}' — DONE ({_age(rec)}, ended {when}): {body}" if st == "running": return (f"'{title}' — STILL RUNNING ({_age(rec)} so far; last step: " f"{rec.get('last_step', '…')})") if st == "failed": return f"'{title}' — FAILED ({when}): {rec.get('error', 'unknown error')}" if st == "killed": return f"'{title}' — KILLED on request ({when})" if st == "interrupted": return f"'{title}' — INTERRUPTED by a restart before finishing ({when})" return f"'{title}' — {st}" def recall_text(query: str = "", limit: int = 8) -> str: """A ready-to-read block answering 'did you do X?' from the ledger.""" hits = find(query, limit=limit) if not hits: return ("No matching task found in the ledger" + (f" for '{query}'." if query else ".")) return "\n".join("- " + summarize(h) for h in hits)