From dc03f1f4edb2f868d95099cdcb445175d0d3fdab Mon Sep 17 00:00:00 2001 From: 0xallam Date: Sun, 26 Apr 2026 09:30:13 -0700 Subject: [PATCH] Use shared agent persistence files --- AGENT_ORCHESTRATION_REFACTOR_PLAN.md | 6 +- strix/interface/main.py | 12 ++-- strix/orchestration/coordinator.py | 14 +---- strix/orchestration/scan.py | 85 +++++++++++----------------- strix/telemetry/tracer.py | 56 +++++++++--------- strix/tools/agents_graph/tools.py | 21 ++----- 6 files changed, 75 insertions(+), 119 deletions(-) diff --git a/AGENT_ORCHESTRATION_REFACTOR_PLAN.md b/AGENT_ORCHESTRATION_REFACTOR_PLAN.md index 22052b3..32c7af6 100644 --- a/AGENT_ORCHESTRATION_REFACTOR_PLAN.md +++ b/AGENT_ORCHESTRATION_REFACTOR_PLAN.md @@ -31,8 +31,8 @@ Removed custom orchestration modules: - `hooks.py` - `run_loop.py` -`bus.json` remains only as the backward-compatible snapshot filename for old runs and tracer -hydration. It is no longer backed by `AgentMessageBus`. +`agents.json` is the only graph/status snapshot. `agents.db` is the only SDK session database; +each agent uses its own SDK `session_id` inside that shared database. ## Lifecycle Semantics @@ -65,5 +65,5 @@ Unknown agent id is the only invalid message target. There is no routing-closed - Interactive `wait_for_message` parks and returns control to the continuation loop. - Non-interactive `wait_for_message` returns the newly appended session message content. - `finish_scan` blocks while child agents are active. -- Resume rebuilds parked child runners from `bus.json` plus per-agent session DBs. +- Resume rebuilds parked child runners from `agents.json` plus the shared `agents.db`. - Graceful stop works for both active-stream and parked agents. diff --git a/strix/interface/main.py b/strix/interface/main.py index 63fea43..b500b5a 100644 --- a/strix/interface/main.py +++ b/strix/interface/main.py @@ -394,7 +394,7 @@ Examples: help=( "Resume a prior scan by its run name (the dir under ./strix_runs/). " "Picks up the root + every non-terminal subagent's full LLM history " - "and bus topology. Skips fresh run-name generation." + "and agent topology. Skips fresh run-name generation." ), ) @@ -417,7 +417,7 @@ Examples: # Capture before ``_load_resume_state`` overrides — used by the resume # path in ``run_strix_scan`` to decide whether to inject the new - # instruction into the root's bus inbox after session replay. + # instruction into the root's SDK session after replay. args.user_explicit_instruction = args.instruction if args.resume else None if args.resume: @@ -427,11 +427,11 @@ Examples: "the prior run left off, including the original target list." ) _load_resume_state(args, parser) - bus_path = Path("strix_runs") / args.resume / "bus.json" - if not bus_path.exists(): + agents_path = Path("strix_runs") / args.resume / "agents.json" + if not agents_path.exists(): parser.error( - f"--resume {args.resume}: missing {bus_path}. The run was " - f"persisted but never reached its first bus snapshot — " + f"--resume {args.resume}: missing {agents_path}. The run was " + f"persisted but never reached its first agent snapshot — " f"there's nothing to resume from. Pick a fresh --run-name " f"or remove --resume to start over with the same targets." ) diff --git a/strix/orchestration/coordinator.py b/strix/orchestration/coordinator.py index c382d46..a6f96cb 100644 --- a/strix/orchestration/coordinator.py +++ b/strix/orchestration/coordinator.py @@ -34,7 +34,7 @@ logger = logging.getLogger(__name__) Status = Literal["running", "waiting", "completed", "stopped", "crashed", "failed", "llm_failed"] ACTIVE_AGENT_STATUSES = {"running", "waiting", "llm_failed"} -_SNAPSHOT_VERSION = 2 +_SNAPSHOT_VERSION = 3 _WAITING_TIMEOUT_SUBAGENT = 300.0 _TIMEOUT_RESUME_MESSAGE = "Waiting timeout reached. Resuming execution." @@ -78,7 +78,6 @@ class AgentCoordinator: is_whitebox: bool = False, scan_mode: str = "deep", diff_scope: dict[str, Any] | None = None, - session_path: str | Path | None = None, ) -> None: async with self._lock: self.statuses[agent_id] = "running" @@ -93,7 +92,6 @@ class AgentCoordinator: "is_whitebox": bool(is_whitebox), "scan_mode": scan_mode, "diff_scope": diff_scope, - "session_path": str(session_path) if session_path is not None else "", } self.runtimes.setdefault(agent_id, AgentRuntime()) logger.info("agent.register %s (%s) parent=%s", agent_id, name, parent_id or "-") @@ -362,8 +360,6 @@ class AgentCoordinator: aid: [dict(msg) for msg in msgs] for aid, msgs in self.queued_messages.items() }, "stats_live": {aid: dict(s) for aid, s in self.stats_live.items()}, - # Kept for old tracer hydration shape. - "stats_completed": {}, } async def restore(self, snap: dict[str, Any]) -> None: @@ -378,15 +374,7 @@ class AgentCoordinator: aid: [dict(msg) for msg in msgs] for aid, msgs in snap.get("queued_messages", {}).items() } - for aid in snap.get("stopping", []): - if aid in self.statuses: - self.statuses[aid] = "stopped" self.stats_live = {aid: dict(s) for aid, s in snap.get("stats_live", {}).items()} - for aid, msgs in snap.get("inboxes", {}).items(): - # Legacy bus snapshots used inboxes. Preserve them as queued - # messages so attaching the SDK session makes them visible. - self.pending_counts[aid] = max(self.pending_counts.get(aid, 0), len(msgs)) - self.queued_messages.setdefault(aid, []).extend(dict(msg) for msg in msgs) for aid in self.statuses: self.runtimes.setdefault(aid, AgentRuntime()) diff --git a/strix/orchestration/scan.py b/strix/orchestration/scan.py index f2c261a..d8d9755 100644 --- a/strix/orchestration/scan.py +++ b/strix/orchestration/scan.py @@ -1,23 +1,23 @@ """Top-level scan entry point with auto-resume. 1. Build (or take from caller) the per-scan ``AgentCoordinator``. -2. Wire a snapshot path so every lifecycle event auto-persists ``bus.json``. +2. Wire a snapshot path so lifecycle events auto-persist ``agents.json``. 3. Acquire an advisory file lock so a second ``strix`` process can't run on the same ``scan_id`` concurrently. -4. **Resume detection**: if ``{run_dir}/bus.json`` already exists, restore +4. **Resume detection**: if ``{run_dir}/agents.json`` already exists, restore the coordinator, hydrate the tracer, reuse the persisted ``root_id`` instead of generating a fresh one, and respawn every non-terminal subagent - from its per-child ``SQLiteSession`` before starting the root. + from the shared SDK ``agents.db`` before starting the root. 5. Bring up (or reuse) a sandbox session for ``scan_id``. 6. Build the root ``Agent`` + child factory. -7. Open root ``SQLiteSession`` at the same path so the SDK replays prior +7. Open root ``SQLiteSession`` in ``agents.db`` so the SDK replays prior turns on resume. 8. Call ``Runner.run`` (via ``run_with_continuation``). 9. ``finally``: close every per-agent session, take a final snapshot, tear down the sandbox, release the lock. -Resume is **always on**: there is no flag — presence of ``bus.json`` is -the trigger. Fresh runs simply have no ``bus.json`` to begin with. +Resume is **always on**: there is no flag — presence of ``agents.json`` is +the trigger. Fresh runs simply have no ``agents.json`` to begin with. """ from __future__ import annotations @@ -229,10 +229,9 @@ async def run_strix_scan( teardown_logging = setup_scan_logging(run_dir) set_scan_id(scan_id) - bus_path = run_dir / "bus.json" - is_resume = bus_path.exists() - sessions_dir = run_dir / "sessions" - sessions_dir.mkdir(parents=True, exist_ok=True) + agents_path = run_dir / "agents.json" + agents_db = run_dir / "agents.db" + is_resume = agents_path.exists() lock_handle = _acquire_run_lock(run_dir) @@ -258,7 +257,7 @@ async def run_strix_scan( # commands while the scan loop runs in another thread. if coordinator is None: coordinator = AgentCoordinator() - coordinator.set_snapshot_path(bus_path) + coordinator.set_snapshot_path(agents_path) if tracer is not None and hasattr(tracer, "hydrate_from_run_dir"): tracer.hydrate_from_run_dir() @@ -275,12 +274,17 @@ async def run_strix_scan( root_id: str | None = None if is_resume: try: - snap = json.loads(bus_path.read_text(encoding="utf-8")) + snap = json.loads(agents_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: _release_run_lock(lock_handle) raise RuntimeError( - f"Cannot resume scan {scan_id}: bus.json is unreadable: {exc}", + f"Cannot resume scan {scan_id}: agents.json is unreadable: {exc}", ) from exc + if not agents_db.exists(): + _release_run_lock(lock_handle) + raise RuntimeError( + f"Cannot resume scan {scan_id}: missing SDK session database at {agents_db}", + ) await coordinator.restore(snap) for aid, parent in coordinator.parent_of.items(): if parent is None: @@ -289,7 +293,7 @@ async def run_strix_scan( if root_id is None: _release_run_lock(lock_handle) raise RuntimeError( - f"Cannot resume scan {scan_id}: bus.json has no root agent (parent=None)", + f"Cannot resume scan {scan_id}: agents.json has no root agent (parent=None)", ) logger.info( "Resume: restored coordinator with %d agent(s); root=%s", @@ -368,6 +372,7 @@ async def run_strix_scan( "diff_scope": diff_scope, "run_id": run_id, "agent_factory": agent_factory, + "agents_db_path": agents_db, "_sessions_to_close": sessions_to_close, } @@ -395,7 +400,7 @@ async def run_strix_scan( if is_resume: await _respawn_subagents( coordinator=coordinator, - sessions_dir=sessions_dir, + agents_db_path=agents_db, factory=agent_factory, parent_ctx=context, resolved_model=resolved_model, @@ -404,18 +409,16 @@ async def run_strix_scan( sessions_to_close=sessions_to_close, ) - # Root SDK session — same path on fresh + resume so SDK replay - # picks up prior turns automatically when ``initial_input`` is - # an empty list. - session_db = run_dir / "session.db" - root_session = SQLiteSession(session_id=scan_id, db_path=session_db) + # All agents share one SQLite database; SDK session_id separates + # each agent's conversation inside that database. + root_session = open_agent_session(root_id, agents_db) sessions_to_close.append(root_session) await coordinator.attach_runtime(root_id, session=root_session) initial_input: Any = [] if is_resume else _build_root_task(scan_config) # Resume + new ``--instruction``: SDK replay drives root from - # session.db with ``initial_input=[]``, so a brand-new instruction + # agents.db with ``initial_input=[]``, so a brand-new instruction # passed on the resume CLI would otherwise be silently ignored. # Inject it as a fresh user message in root's SDK session; the # next run cycle will replay it with the rest of the session. @@ -431,7 +434,7 @@ async def run_strix_scan( }, ) logger.info( - "Resume: injected new instruction into root inbox (len=%d)", + "Resume: injected new instruction into root SDK session (len=%d)", len(resume_instruction), ) @@ -489,7 +492,7 @@ async def run_strix_scan( async def _respawn_subagents( *, coordinator: AgentCoordinator, - sessions_dir: Path, + agents_db_path: Path, factory: Any, parent_ctx: dict[str, Any], resolved_model: str, @@ -499,14 +502,12 @@ async def _respawn_subagents( ) -> None: """Re-spawn subagent runners from a restored coordinator snapshot. - Each child gets its own :class:`SQLiteSession` reopened at - ``sessions_dir/.db`` so the SDK replays its prior - conversation. Per-child failure (missing/corrupt session DB, - factory raising) finalizes that child as ``crashed`` and continues. - Interactive mode respawns every registered child as a parked runner - unless it was actively running before the crash. That keeps - completed/stopped/crashed/failed agents addressable: a later message - wakes the SDK session instead of being dropped into a dead inbox. + Each child gets its own SDK ``session_id`` inside the shared + ``agents.db`` so the SDK replays its prior conversation. Interactive + mode respawns every registered child as a parked runner unless it was + actively running before the crash. That keeps completed/stopped/ + crashed/failed agents addressable: a later message wakes the SDK + session instead of being dropped into a dead inbox. """ interactive = bool(parent_ctx.get("interactive", False)) async with coordinator._lock: @@ -534,26 +535,6 @@ async def _respawn_subagents( for child_id, name, parent_id, md in candidates: try: - session_path = sessions_dir / f"{child_id}.db" - if not session_path.exists(): - if interactive: - logger.warning( - "respawn %s (%s): session db missing at %s; starting parked with empty session", - child_id, - name, - session_path, - ) - else: - logger.warning( - "respawn %s (%s): session db missing at %s — marking crashed", - child_id, - name, - session_path, - ) - await coordinator.set_status(child_id, "crashed") - continue - session_path.parent.mkdir(parents=True, exist_ok=True) - restored_status = str(md.get("_restored_status") or "running") start_parked = interactive and restored_status != "running" @@ -565,7 +546,7 @@ async def _respawn_subagents( restored_status, ) - child_session = open_agent_session(child_id, session_path) + child_session = open_agent_session(child_id, agents_db_path) sessions_to_close.append(child_session) await coordinator.attach_runtime(child_id, session=child_session) diff --git a/strix/telemetry/tracer.py b/strix/telemetry/tracer.py index 72eb589..f6dbde2 100644 --- a/strix/telemetry/tracer.py +++ b/strix/telemetry/tracer.py @@ -167,29 +167,29 @@ class Tracer: json_path, ) - bus_path = run_dir / "bus.json" - if bus_path.exists(): + agents_path = run_dir / "agents.json" + if agents_path.exists(): try: - bus_data = json.loads(bus_path.read_text(encoding="utf-8")) + agents_data = json.loads(agents_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): - # Caller will surface this same corruption via ``bus.restore``; + # Caller will surface this same corruption via coordinator restore; # no need to fail twice. Skip the agents/stats hydrate path. - bus_data = None - if isinstance(bus_data, dict): - self._hydrate_agents_tree(bus_data) - self._hydrate_llm_stats(bus_data) + agents_data = None + if isinstance(agents_data, dict): + self._hydrate_agents_tree(agents_data) + self._hydrate_llm_stats(agents_data) - def _hydrate_agents_tree(self, bus_data: dict[str, Any]) -> None: - """Populate ``self.agents`` from a bus snapshot. + def _hydrate_agents_tree(self, agents_data: dict[str, Any]) -> None: + """Populate ``self.agents`` from the coordinator snapshot. Without this, the TUI tree on resume would only show agents currently running (mirrored by ``on_agent_start``); completed / crashed / stopped children from the prior run would be invisible - even though the bus knows about them. + even though the coordinator knows about them. """ - statuses = bus_data.get("statuses") or {} - names = bus_data.get("names") or {} - parent_of = bus_data.get("parent_of") or {} + statuses = agents_data.get("statuses") or {} + names = agents_data.get("names") or {} + parent_of = agents_data.get("parent_of") or {} if not isinstance(statuses, dict): return timestamp = self.start_time @@ -206,29 +206,25 @@ class Tracer: } logger.info("tracer hydrated %d agent(s) into tree", len(self.agents)) - def _hydrate_llm_stats(self, bus_data: dict[str, Any]) -> None: - """Seed ``self._llm_stats`` from the bus snapshot's per-agent counters. + def _hydrate_llm_stats(self, agents_data: dict[str, Any]) -> None: + """Seed ``self._llm_stats`` from the coordinator snapshot's counters. - Aggregates ``stats_live + stats_completed`` so the resumed scan's - TUI footer shows cumulative tokens / requests across the prior - run plus whatever the resume adds, instead of resetting to zero. + This keeps the resumed scan's TUI footer cumulative instead of + resetting to zero. """ totals = {"input_tokens": 0, "output_tokens": 0, "cached_tokens": 0, "requests": 0} - for bucket_key in ("stats_live", "stats_completed"): - bucket = bus_data.get(bucket_key) or {} - if not isinstance(bucket, dict): - continue + bucket = agents_data.get("stats_live") or {} + if isinstance(bucket, dict): for entry in bucket.values(): - if not isinstance(entry, dict): - continue - totals["input_tokens"] += int(entry.get("in", 0) or 0) - totals["output_tokens"] += int(entry.get("out", 0) or 0) - totals["cached_tokens"] += int(entry.get("cached", 0) or 0) - totals["requests"] += int(entry.get("calls", 0) or 0) + if isinstance(entry, dict): + totals["input_tokens"] += int(entry.get("in", 0) or 0) + totals["output_tokens"] += int(entry.get("out", 0) or 0) + totals["cached_tokens"] += int(entry.get("cached", 0) or 0) + totals["requests"] += int(entry.get("calls", 0) or 0) for k, v in totals.items(): self._llm_stats[k] = v logger.info( - "tracer hydrated llm stats from bus (in=%d out=%d cached=%d requests=%d)", + "tracer hydrated llm stats from agents snapshot (in=%d out=%d cached=%d requests=%d)", totals["input_tokens"], totals["output_tokens"], totals["cached_tokens"], diff --git a/strix/tools/agents_graph/tools.py b/strix/tools/agents_graph/tools.py index e17a1ae..38527c2 100644 --- a/strix/tools/agents_graph/tools.py +++ b/strix/tools/agents_graph/tools.py @@ -492,13 +492,6 @@ async def create_agent( default=str, ) - tracer = inner.get("tracer") - if tracer is not None and hasattr(tracer, "get_run_dir"): - child_session_path = tracer.get_run_dir() / "sessions" / f"{child_id}.db" - else: - run_id = inner.get("run_id") or "default" - child_session_path = Path.cwd() / "strix_runs" / str(run_id) / "sessions" / f"{child_id}.db" - await coordinator.register( child_id, name, @@ -508,7 +501,6 @@ async def create_agent( is_whitebox=bool(inner.get("is_whitebox", False)), scan_mode=str(inner.get("scan_mode", "deep")), diff_scope=inner.get("diff_scope"), - session_path=child_session_path, ) # ``ctx.turn_input`` carries the parent's full conversation up to and @@ -569,13 +561,12 @@ async def create_agent( "_sessions_to_close": inner.get("_sessions_to_close", []), } - # Per-child SQLiteSession at ``{run_dir}/sessions/{child_id}.db`` so - # this subagent's full conversation survives a process restart and - # can be replayed by the SDK on resume. Path is derived from the - # tracer's run_dir (root-side construction in - # :func:`run_strix_scan`); fall back to ``./strix_runs/{run_id}/`` if - # the tracer is absent (unit-test path). - child_session = open_agent_session(child_id, child_session_path) + # Every agent gets its own SDK session_id inside the shared agents.db. + agents_db_path = inner.get("agents_db_path") + if not isinstance(agents_db_path, Path): + run_id = inner.get("run_id") or "default" + agents_db_path = Path.cwd() / "strix_runs" / str(run_id) / "agents.db" + child_session = open_agent_session(child_id, agents_db_path) sessions_list = child_ctx.get("_sessions_to_close") if isinstance(sessions_list, list): sessions_list.append(child_session)