"""Pre-warmed, supervised background workers — the agent pool. WHY THIS EXISTS (2026-07-14, Ahmed's call after three failed fixes): the old delegation path ran workers as background Task-tool subagents INSIDE the brain's CLI session. That broke in three ways at once: 1. SLOW START — a "spawn" was a full coordinator turn streaming the whole worker prompt into a Task call before the subagent even booted (10-20s). 2. BLIND — the SDK stream is only read while a brain turn is open, so a worker finishing/dying/stalling while the room was quiet sat unread: HUD chip stuck "running" forever, no report, Ahmed waiting on nothing. 3. HONOUR-SYSTEM REPORTS — a worker only reported by remembering to append to control/report.txt; a crashed or stuck worker wrote nothing, and nothing supervised it. The pool fixes all three by making workers SEPARATE ClaudeSDKClient sessions owned by the engine, exactly like Claude Code's own harness: - PRE-WARMED: WARM_SPARES (default 1) idle sessions sit fully connected (CLI subprocess + MCP servers booted). dispatch() hands the prompt to a warm spare INSTANTLY, and the moment a spare is taken a replacement starts warming — there is always one ready (Ahmed's N+1 rule). - SUPERVISED: every running worker has a dedicated reader task consuming its stream the whole time. Every tool call emits task_progress (live HUD chip note), a no-activity watchdog reports a stall LOUDLY instead of letting Ahmed wait forever, a crash/death reports LOUDLY, and the final result is delivered through report_sink (the engine inbox rail) the moment it lands — no report.txt honour system. - KILLABLE: kill() disconnects the CLI subprocess immediately and clears the HUD chip in the same breath — no zombie UI. The warm spare means capacity is instantly back. Tiers map to models exactly like the old _AGENTS: worker=sonnet (the only pre-warmed tier — it's the default), heavy=opus, genius=fable (cold-started on demand; a few seconds slower, which is fine for rare heavy jobs). Env knobs: WARM_SPARES (1), AGENT_MAX (4 running), WORKER_STALL_S (300, 0=off), AGENT_POOL=0 disables the whole pool (llm.py then restores the old Task-tool path). Wiring: main.py sets pool.report_sink = app.inbox.put_nowait and calls pool.start() at boot / pool.stop() at shutdown; llm.py registers build_server() as the "agents" MCP server on the BRAIN session only (workers never get it — no recursive fan-out). """ from __future__ import annotations import asyncio import os import time from voice import task_ledger from voice.events import emit _TIER_MODELS = {"worker": "sonnet", "heavy": "opus", "genius": "fable"} WARM_SPARES = int(os.environ.get("WARM_SPARES", "1")) AGENT_MAX = int(os.environ.get("AGENT_MAX", "4")) STALL_S = float(os.environ.get("WORKER_STALL_S", "300")) # reader poll granularity — how often the watchdog wakes to check for a stall _POLL_S = 20.0 def _log(line: str) -> None: """Same trail as llm.py's _log_agent — one lifecycle log for everything.""" try: from voice.llm import _log_agent _log_agent(line) except Exception: # noqa: BLE001 pass def _checkpoint(rec) -> None: """Drop a milestone into jarvis-memory when a task ends, so "what happened to X?" works days later from any device — not just this process's lifetime. Fire-and-forget (never blocks the pool); skips trivial no-op tasks and only checkpoints once per task.""" if not rec or rec.get("checkpointed"): return # skip empties: a task that did nothing and said nothing isn't worth a memory if not (rec.get("result") or rec.get("error") or rec.get("steps")): return try: from voice import task_ledger as _tl from voice import memory_control if not memory_control.enabled(): return text = "Background task — " + _tl.summarize(rec) memory_control.fire_save(text, kind="task") _tl.mark_checkpointed(rec.get("id", "")) except Exception as e: # noqa: BLE001 — a checkpoint must never break the pool _log(f"POOL checkpoint skipped: {str(e)[:120]}") def _tool_note(block) -> str: """One short human line for a tool call — what the HUD chip shows live.""" name = getattr(block, "name", "") or "tool" args = getattr(block, "input", None) or {} hint = "" try: if name == "Bash": hint = str(args.get("command", ""))[:48] elif name in ("Read", "Write", "Edit"): hint = os.path.basename(str(args.get("file_path", "")))[:40] elif name == "WebSearch": hint = str(args.get("query", ""))[:40] elif name == "WebFetch": hint = str(args.get("url", ""))[:40] else: # first stringy arg value, briefly for v in args.values(): if isinstance(v, str) and v.strip(): hint = v.strip()[:40] break except Exception: # noqa: BLE001 hint = "" name = name.split("__")[-1] # mcp__desktop__Click -> Click return f"{name}: {hint}" if hint else name class _Worker: def __init__(self, wid: str, tier: str, title: str, client) -> None: self.id = wid self.tier = tier self.title = title self.client = client self.reader: asyncio.Task | None = None self.started = time.monotonic() self.last_activity = time.monotonic() self.last_note = "starting" self.stall_warned = False self.ledger_id = "" # durable task_ledger record for this task def age(self) -> str: s = int(time.monotonic() - self.started) return f"{s // 60}m{s % 60:02d}s" if s >= 60 else f"{s}s" def idle(self) -> float: return time.monotonic() - self.last_activity class AgentPool: """Owns every background worker session: warm spares + running tasks.""" def __init__(self) -> None: self._spares: list = [] # connected idle clients (worker tier) self._warming = 0 # spares currently connecting self._running: dict[str, _Worker] = {} self._seq = 0 self._started = False self.report_sink = None # main.py: app.inbox.put_nowait # ------------------------------------------------------------------ # lifecycle # ------------------------------------------------------------------ def start(self) -> None: """Begin keeping WARM_SPARES sessions warm (call on the event loop).""" self._started = True self._ensure_warm() async def stop(self) -> None: self._started = False for w in list(self._running.values()): await self._terminate(w, announce=False) for client in self._spares: try: await client.disconnect() except Exception: # noqa: BLE001 pass self._spares.clear() def _ensure_warm(self) -> None: """Top the spare pool back up to WARM_SPARES, asynchronously.""" if not self._started: return want = WARM_SPARES - len(self._spares) - self._warming for _ in range(max(0, want)): self._warming += 1 asyncio.get_running_loop().create_task(self._warm_one()) async def _warm_one(self) -> None: try: client = await self._connect("worker") self._spares.append(client) _log(f"POOL warm spare ready (spares={len(self._spares)})") except Exception as e: # noqa: BLE001 # no hot retry loop — the next dispatch/ensure_warm tries again _log(f"POOL warm FAILED: {str(e)[:200]}") print(f" [agent pool: warm-up failed: {str(e)[:120]}]") finally: self._warming -= 1 async def _connect(self, tier: str): from claude_agent_sdk import ClaudeSDKClient from voice.llm import worker_options client = ClaudeSDKClient(options=worker_options(tier)) await client.connect() return client # ------------------------------------------------------------------ # dispatch / kill / status (called from the brain's MCP tools) # ------------------------------------------------------------------ async def dispatch(self, task: str, title: str, tier: str = "worker") -> str: if tier not in _TIER_MODELS: tier = "worker" if len(self._running) >= AGENT_MAX: names = ", ".join(f"'{w.title}'" for w in self._running.values()) return (f"REFUSED: {AGENT_MAX} agents already running ({names}). " f"Finish or kill one first (agent_kill).") self._seq += 1 wid = f"w{self._seq}" warm = False if tier == "worker" and self._spares: client = self._spares.pop(0) warm = True else: try: client = await self._connect(tier) except Exception as e: # noqa: BLE001 _log(f"POOL {wid} connect FAILED: {str(e)[:200]}") return (f"FAILED to start a {tier} agent: {str(e)[:150]}. " f"Tell Ahmed plainly; do not pretend it started.") w = _Worker(wid, tier, title[:80], client) self._running[wid] = w # durable trail: the task is on the books the instant it's dispatched, # so a crash/restart mid-task leaves a record (recovered as # "interrupted" on next boot) and "did you do X?" is always answerable. w.ledger_id = task_ledger.create(wid, w.title, task, tier) try: await client.query(task) except Exception as e: # noqa: BLE001 self._running.pop(wid, None) try: await client.disconnect() except Exception: # noqa: BLE001 pass task_ledger.fail(w.ledger_id, f"never started: {str(e)[:150]}") _log(f"POOL {wid} query FAILED: {str(e)[:200]}") return f"FAILED to hand the task over: {str(e)[:150]}." w.reader = asyncio.get_running_loop().create_task(self._read(w)) emit("task_started", desc=w.title, agent=tier, id=wid) _log(f"POOL {wid} DISPATCHED [{tier}{' warm' if warm else ' cold'}] " f"title={w.title!r}") self._ensure_warm() # replace the spare we just spent return (f"Agent {wid} started ({tier}, " f"{'pre-warmed — already working' if warm else 'fresh session'}). " f"It reports back automatically when done; agent_status checks " f"on it; agent_kill stops it.") async def kill(self, target: str = "latest") -> str: """target: a worker id ('w3'), 'latest', 'all', or a title fragment.""" victims: list[_Worker] = [] t = (target or "latest").strip().lower() if t == "all": victims = list(self._running.values()) elif t == "latest": if self._running: victims = [list(self._running.values())[-1]] elif t in self._running: victims = [self._running[t]] else: victims = [w for w in self._running.values() if t in w.title.lower()] if not victims: return ("No matching agent running." + (f" Running: {self._status_short()}" if self._running else "")) names = [] for w in victims: await self._terminate(w, announce=False, status="killed") names.append(f"{w.id} '{w.title}'") _log(f"POOL {w.id} KILLED by request") self._ensure_warm() return "Killed: " + "; ".join(names) + ". A warm spare is ready." def status(self) -> str: if not self._running: spare = len(self._spares) return (f"No agents running. {spare} warm spare" f"{'s' if spare != 1 else ''} ready.") lines = [] for w in self._running.values(): idle = int(w.idle()) state = f"last activity {idle}s ago — {w.last_note}" if STALL_S and idle > STALL_S: state = (f"POSSIBLY STUCK — no activity for {idle // 60}m " f"(last: {w.last_note})") lines.append(f"{w.id} [{w.tier}] '{w.title}' — running {w.age()}, " f"{state}") lines.append(f"({len(self._spares)} warm spare ready)") return "\n".join(lines) def _status_short(self) -> str: return ", ".join(f"{w.id} '{w.title}'" for w in self._running.values()) # ------------------------------------------------------------------ # supervision — the always-on reader that the old design never had # ------------------------------------------------------------------ async def _read(self, w: _Worker) -> None: """Consume the worker's whole stream: progress out, stall watchdog on, result/death delivered loudly. This is what makes agents visible.""" from claude_agent_sdk import (AssistantMessage, ResultMessage, TextBlock, ToolResultBlock, UserMessage, ToolUseBlock) report = "" failed = "" last_text = "" tool_errors: list[str] = [] # tool failures seen — never swallow these pending: asyncio.Future | None = None try: stream = w.client.receive_response().__aiter__() while True: # NEVER asyncio.wait_for() here. wait_for CANCELS the awaited # coroutine on timeout, and cancelling __anext__() DESTROYS the # async generator for good — the next call raises # StopAsyncIteration, so the pool declared a perfectly healthy # worker "done with no result", tore it down, and closed its # stdin. The CLI routes every in-process MCP tool over that same # stdin, so the still-running worker's gmail/asana calls then all # failed with "Stream closed" (built-ins like Bash survived — they # don't use the control channel). That was the silent-worker-death # bug. asyncio.wait() times out WITHOUT cancelling: the watchdog # ticks and the stream lives on. if pending is None: pending = asyncio.ensure_future(stream.__anext__()) done, _ = await asyncio.wait({pending}, timeout=_POLL_S) if not done: # watchdog tick — no message yet, the read is still in flight if (STALL_S and not w.stall_warned and w.idle() > STALL_S): w.stall_warned = True mins = int(w.idle() // 60) _log(f"POOL {w.id} STALL warning ({mins}m idle)") emit("task_stalled", id=w.id, desc=w.title) self._report( f"Heads-up: the agent working on '{w.title}' has " f"gone quiet — no activity for {mins} minutes " f"(last: {w.last_note}). It may be stuck; Ahmed " f"can say the word to kill it or leave it.") continue try: msg = pending.result() except StopAsyncIteration: break finally: pending = None w.last_activity = time.monotonic() w.stall_warned = False if isinstance(msg, AssistantMessage): for block in msg.content: if isinstance(block, ToolUseBlock): w.last_note = _tool_note(block) emit("task_progress", id=w.id, note=w.last_note) task_ledger.step(w.ledger_id, w.last_note) _log(f"POOL {w.id} … {w.last_note}") elif isinstance(block, TextBlock) and block.text.strip(): last_text = block.text.strip() elif isinstance(msg, UserMessage): # A FAILING TOOL is the thing that used to sink an agent # silently ("Stream closed" on every MCP call, then it gave # up with no summary). Record them so an empty finish still # tells Ahmed WHAT broke instead of "no summary". content = getattr(msg, "content", None) or [] if isinstance(content, list): for block in content: if (isinstance(block, ToolResultBlock) and getattr(block, "is_error", False)): c = block.content txt = c if isinstance(c, str) else str(c) tool_errors.append(txt.strip()[:120]) _log(f"POOL {w.id} TOOL-ERROR {txt.strip()[:120]}") elif isinstance(msg, ResultMessage): if getattr(msg, "is_error", False): failed = (getattr(msg, "result", None) or "the session ended with an error") else: report = getattr(msg, "result", None) or last_text break except asyncio.CancelledError: if pending is not None and not pending.done(): pending.cancel() # killed — don't leak the in-flight read return # _terminate handles teardown + chip except Exception as e: # noqa: BLE001 failed = str(e)[:200] finally: if pending is not None and not pending.done(): pending.cancel() # -- delivery ------------------------------------------------------ # An agent that ends with NOTHING to say has not "finished" — it fell # over. Report it as the failure it is, naming the tool errors that # sank it, so Ahmed hears the truth and a fresh agent is relaunched # (never "done, but no summary", never quietly nothing). if not failed and not report and not last_text: failed = (f"it ended without doing the job or saying anything" + (f" — every tool call failed: {tool_errors[0]}" if tool_errors else f" (last thing it did: {w.last_note})")) if failed: _log(f"POOL {w.id} FAILED after {w.age()}: {failed[:200]}") rec = task_ledger.fail(w.ledger_id, failed) _checkpoint(rec) await self._terminate(w, announce=False, status="failed") errs = (f" Tool errors seen: {'; '.join(tool_errors[:3])}." if tool_errors else "") self._report(f"The agent working on '{w.title}' FAILED after " f"{w.age()}: {failed[:300]}.{errs} Tell Ahmed plainly " f"what broke, then relaunch a fresh agent for it — do " f"NOT do the task yourself.") else: _log(f"POOL {w.id} DONE after {w.age()}: {(report or last_text)[:200]!r}") rec = task_ledger.finish(w.ledger_id, report or last_text) _checkpoint(rec) await self._terminate(w, announce=False, status="completed") self._report(f"Finished '{w.title}' ({w.age()}): " + (report or last_text)[:1200]) self._ensure_warm() def _report(self, text: str) -> None: """Deliver to the engine inbox rail (announced when Ahmed is quiet).""" if self.report_sink is not None: try: self.report_sink(text) return except Exception: # noqa: BLE001 pass print(f"TASK {text}") async def _terminate(self, w: _Worker, announce: bool, status: str = "stopped") -> None: """Tear a worker down NOW: reader cancelled, CLI subprocess gone, HUD chip cleared in the same breath — never a zombie chip again.""" self._running.pop(w.id, None) # ledger: kill/stop mark the trail (no-op if _read already finalized it # as done/failed — mark() never downgrades a terminal status). task_ledger.mark(w.ledger_id, status) emit("task_done", id=w.id, status=status) if w.reader is not None and not w.reader.done(): current = None try: current = asyncio.current_task() except Exception: # noqa: BLE001 pass if w.reader is not current: w.reader.cancel() try: await w.client.disconnect() except Exception: # noqa: BLE001 pass if announce: self._report(f"Stopped the agent working on '{w.title}'.") # Module-level singleton — llm.py registers its tools; main.py wires + starts. pool = AgentPool() def enabled() -> bool: return os.environ.get("AGENT_POOL", "1") != "0" def build_server(): """In-process MCP server exposing the pool to the BRAIN session only. Explicit JSON schemas (dict-shorthand marks every param required — the Asana-era bug); `task` and `title` required, the rest optional.""" from claude_agent_sdk import tool, create_sdk_mcp_server def _text(msg: str) -> dict: return {"content": [{"type": "text", "text": msg}]} @tool("dispatch", "Hand a task to a pre-warmed background agent — THE way to delegate " "(the old Task tool is gone). Starts instantly and reports back " "automatically when done; you stay free to talk. `task` = the FULL " "self-contained job description (the agent can't hear the " "conversation — include every detail, path and constraint). `title` " "= a few words for the HUD chip ('scrape coffee shops'). `tier`: " "worker (default, Sonnet, pre-warmed = instant) | heavy (Opus, for " "genuinely complex/long jobs) | genius (Fable, hardest problems or " "when Ahmed asks for fable) — heavy/genius start fresh (a few " "seconds).", {"type": "object", "properties": { "task": {"type": "string", "description": "Full self-contained task prompt"}, "title": {"type": "string", "description": "2-5 word label for the HUD"}, "tier": {"type": "string", "enum": ["worker", "heavy", "genius"], "description": "worker (default) | heavy | genius"}, }, "required": ["task", "title"]}) async def dispatch(args: dict) -> dict: task = str(args.get("task", "")).strip() title = str(args.get("title", "")).strip() or task[:60] if not task: return _text("dispatch failed: empty task.") out = await pool.dispatch(task, title, str(args.get("tier", "worker"))) return _text(out) @tool("agent_status", "Live status of every background agent: what it's doing right now, " "how long it's been running, and whether it looks stuck. Use when " "Ahmed asks how a task is going.", {"type": "object", "properties": {}}) async def agent_status(args: dict) -> dict: return _text(pool.status()) @tool("agent_kill", "Stop a background agent immediately (its HUD chip clears at once). " "`target`: an agent id from agent_status ('w3'), 'latest' (default), " "'all', or a fragment of the task title.", {"type": "object", "properties": {"target": {"type": "string", "description": "'w3' | 'latest' | 'all' | title text"}}, "required": []}) async def agent_kill(args: dict) -> dict: return _text(await pool.kill(str(args.get("target", "latest")))) @tool("task_recall", "Look up what happened to a PAST or running background task from the " "durable ledger — the answer to 'did you do X?' / 'what happened to " "X?' / 'is that done yet?'. Survives restarts and crashes, unlike " "agent_status (which only sees agents alive right now). ALWAYS call " "this before telling Ahmed you don't know the state of something he " "asked you to do. `query` = a few words from the task (e.g. 'email " "Elie', 'coffee shops Khobar'); empty = the most recent tasks. " "Returns each match's status (done/running/failed/killed/interrupted), " "when, how long it took, and its result or the error.", {"type": "object", "properties": {"query": {"type": "string", "description": "words from the task; " "empty = most recent"}}, "required": []}) async def task_recall(args: dict) -> dict: return _text(task_ledger.recall_text(str(args.get("query", "")).strip())) return create_sdk_mcp_server( name="agents", version="1.0.0", tools=[dispatch, agent_status, agent_kill, task_recall], )