Use shared agent persistence files

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.

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."
             )

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())
 

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/<child_id>.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)
 

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"],

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)