refactor: reorganize core report and tui modules

strix/core/__init__.py

@@ -0,0 +1 @@
+"""Strix scan runtime core."""

strix/core/agents.py

@@ -0,0 +1,307 @@
+"""SDK-native state for Strix's addressable agent graph.
+
+The Agents SDK owns model/tool execution and per-agent conversation
+history. Strix owns only product semantics the SDK does not provide:
+agent ids, the parent/child graph, wake/stop signals, TUI-visible
+status, and process-resume metadata.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import tempfile
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Literal, cast
+
+
+if TYPE_CHECKING:
+    from agents.items import TResponseInputItem
+    from agents.memory import Session
+
+
+logger = logging.getLogger(__name__)
+
+Status = Literal["running", "waiting", "completed", "stopped", "crashed", "failed"]
+
+
+@dataclass(slots=True)
+class AgentRuntime:
+    session: Session | None = None
+    task: asyncio.Task[Any] | None = None
+    stream: Any | None = None
+    interrupt_on_message: bool = False
+    wake: asyncio.Event = field(default_factory=asyncio.Event)
+
+
+class AgentCoordinator:
+    """Single owner for graph state, SDK runtimes, messages, and resume snapshots."""
+
+    def __init__(self) -> None:
+        self.statuses: dict[str, Status] = {}
+        self.parent_of: dict[str, str | None] = {}
+        self.names: dict[str, str] = {}
+        self.metadata: dict[str, dict[str, Any]] = {}
+        self.pending_counts: dict[str, int] = {}
+        self.runtimes: dict[str, AgentRuntime] = {}
+        self._lock = asyncio.Lock()
+        self._snapshot_path: Path | None = None
+
+    def set_snapshot_path(self, path: Path) -> None:
+        self._snapshot_path = path
+
+    async def register(
+        self,
+        agent_id: str,
+        name: str,
+        parent_id: str | None,
+        *,
+        task: str | None = None,
+        skills: list[str] | None = None,
+    ) -> None:
+        async with self._lock:
+            self.statuses[agent_id] = "running"
+            self.parent_of[agent_id] = parent_id
+            self.names[agent_id] = name
+            self.pending_counts.setdefault(agent_id, 0)
+            self.metadata[agent_id] = {
+                "task": task or "",
+                "skills": list(skills or []),
+            }
+            self.runtimes.setdefault(agent_id, AgentRuntime())
+        logger.info("agent.register %s (%s) parent=%s", agent_id, name, parent_id or "-")
+        await self._maybe_snapshot()
+
+    async def attach_runtime(
+        self,
+        agent_id: str,
+        *,
+        session: Session | None = None,
+        task: asyncio.Task[Any] | None = None,
+        interrupt_on_message: bool | None = None,
+    ) -> None:
+        async with self._lock:
+            runtime = self.runtimes.setdefault(agent_id, AgentRuntime())
+            if session is not None:
+                runtime.session = session
+            if task is not None:
+                runtime.task = task
+            if interrupt_on_message is not None:
+                runtime.interrupt_on_message = interrupt_on_message
+
+    async def mark_running(self, agent_id: str) -> None:
+        async with self._lock:
+            if agent_id in self.statuses:
+                self.statuses[agent_id] = "running"
+        await self._maybe_snapshot()
+
+    async def park_waiting(self, agent_id: str) -> None:
+        await self.set_status(agent_id, "waiting")
+
+    async def set_status(self, agent_id: str, status: Status | str) -> None:
+        async with self._lock:
+            if agent_id not in self.statuses:
+                return
+            self.statuses[agent_id] = status  # type: ignore[assignment]
+            runtime = self.runtimes.setdefault(agent_id, AgentRuntime())
+            runtime.wake.set()
+        logger.info("agent.status %s=%s", agent_id, status)
+        await self._maybe_snapshot()
+
+    async def send(self, target_agent_id: str, message: dict[str, Any]) -> bool:
+        """Deliver a user/peer message by appending it to the target SDK session."""
+        async with self._lock:
+            if target_agent_id not in self.statuses:
+                logger.debug("agent.send dropped unknown target=%s", target_agent_id)
+                return False
+            runtime = self.runtimes.setdefault(target_agent_id, AgentRuntime())
+            session = runtime.session
+            stream = runtime.stream
+            interrupt = runtime.interrupt_on_message
+        if session is not None:
+            try:
+                await session.add_items([self._message_to_session_item(message)])
+            except Exception:
+                logger.exception(
+                    "agent.send failed to append to SDK session target=%s",
+                    target_agent_id,
+                )
+                return False
+        async with self._lock:
+            self.pending_counts[target_agent_id] = self.pending_counts.get(target_agent_id, 0) + 1
+            self.runtimes.setdefault(target_agent_id, AgentRuntime()).wake.set()
+        if stream is not None and interrupt:
+            stream.cancel(mode="immediate")
+        await self._maybe_snapshot()
+        return True
+
+    async def wait_for_message(self, agent_id: str) -> None:
+        while True:
+            async with self._lock:
+                if self.pending_counts.get(agent_id, 0) > 0:
+                    return
+                wake = self.runtimes.setdefault(agent_id, AgentRuntime()).wake
+                wake.clear()
+            await wake.wait()
+
+    async def consume_pending(
+        self,
+        agent_id: str,
+        *,
+        include_items: bool = False,
+    ) -> tuple[int, list[Any]]:
+        async with self._lock:
+            count = self.pending_counts.get(agent_id, 0)
+            self.pending_counts[agent_id] = 0
+            session = self.runtimes.get(agent_id, AgentRuntime()).session
+        if count <= 0:
+            return 0, []
+        if not include_items or session is None:
+            return count, []
+        items = await session.get_items()
+        return count, list(items[-count:])
+
+    async def request_stop(self, agent_id: str) -> None:
+        async with self._lock:
+            if agent_id not in self.statuses:
+                return
+            self.statuses[agent_id] = "stopped"
+            runtime = self.runtimes.setdefault(agent_id, AgentRuntime())
+            runtime.wake.set()
+            stream = runtime.stream
+        if stream is not None:
+            stream.cancel(mode="after_turn")
+        await self._maybe_snapshot()
+
+    async def cancel_descendants(self, agent_id: str) -> None:
+        tasks = []
+        async with self._lock:
+            for aid in reversed(self._subtree_order_locked(agent_id)):
+                task = self.runtimes.get(aid, AgentRuntime()).task
+                if task is not None and not task.done():
+                    tasks.append(task)
+        for task in tasks:
+            task.cancel()
+        if tasks:
+            await asyncio.gather(*tasks, return_exceptions=True)
+
+    async def cancel_descendants_graceful(self, agent_id: str) -> None:
+        async with self._lock:
+            order = self._subtree_order_locked(agent_id)
+        for aid in reversed(order):
+            await self.request_stop(aid)
+        await self._maybe_snapshot()
+
+    async def attach_stream(
+        self,
+        agent_id: str,
+        stream: Any,
+    ) -> None:
+        async with self._lock:
+            self.runtimes.setdefault(agent_id, AgentRuntime()).stream = stream
+
+    async def detach_stream(
+        self,
+        agent_id: str,
+        stream: Any,
+    ) -> None:
+        async with self._lock:
+            runtime = self.runtimes.setdefault(agent_id, AgentRuntime())
+            if runtime.stream is stream:
+                runtime.stream = None
+
+    async def active_agents_except(self, agent_id: str) -> list[dict[str, Any]]:
+        async with self._lock:
+            return [
+                {
+                    "agent_id": aid,
+                    "name": self.names.get(aid, aid),
+                    "status": status,
+                    "parent_id": self.parent_of.get(aid),
+                }
+                for aid, status in self.statuses.items()
+                if aid != agent_id and status in {"running", "waiting"}
+            ]
+
+    async def graph_snapshot(
+        self,
+    ) -> tuple[dict[str, str | None], dict[str, Status], dict[str, str]]:
+        async with self._lock:
+            return dict(self.parent_of), dict(self.statuses), dict(self.names)
+
+    def _message_to_session_item(self, message: dict[str, Any]) -> TResponseInputItem:
+        sender = str(message.get("from", "unknown"))
+        content = str(message.get("content", ""))
+        if sender == "user":
+            return cast("TResponseInputItem", {"role": "user", "content": content})
+        sender_name = self.names.get(sender, sender)
+        msg_type = message.get("type", "information")
+        priority = message.get("priority", "normal")
+        return cast(
+            "TResponseInputItem",
+            {
+                "role": "user",
+                "content": (
+                    f"[Message from {sender_name} ({sender}) | type={msg_type} "
+                    f"| priority={priority}]\n{content}"
+                ),
+            },
+        )
+
+    def _subtree_order_locked(self, agent_id: str) -> list[str]:
+        queue = [agent_id]
+        order: list[str] = []
+        while queue:
+            aid = queue.pop()
+            order.append(aid)
+            queue.extend(child for child, parent in self.parent_of.items() if parent == aid)
+        return order
+
+    async def snapshot(self) -> dict[str, Any]:
+        async with self._lock:
+            return {
+                "statuses": dict(self.statuses),
+                "parent_of": dict(self.parent_of),
+                "names": dict(self.names),
+                "metadata": {aid: dict(md) for aid, md in self.metadata.items()},
+                "pending_counts": dict(self.pending_counts),
+            }
+
+    async def restore(self, snap: dict[str, Any]) -> None:
+        async with self._lock:
+            self.statuses = dict(snap.get("statuses", {}))
+            self.parent_of = dict(snap.get("parent_of", {}))
+            self.names = dict(snap.get("names", {}))
+            self.metadata = {aid: dict(md) for aid, md in snap.get("metadata", {}).items()}
+            self.pending_counts = dict(snap.get("pending_counts", {}))
+            for aid in self.statuses:
+                self.runtimes.setdefault(aid, AgentRuntime())
+
+    async def _maybe_snapshot(self) -> None:
+        path = self._snapshot_path
+        if path is None:
+            return
+        try:
+            data = await self.snapshot()
+            payload = json.dumps(data, ensure_ascii=False, default=str)
+            path.parent.mkdir(parents=True, exist_ok=True)
+            with tempfile.NamedTemporaryFile(
+                mode="w",
+                encoding="utf-8",
+                dir=str(path.parent),
+                prefix=f".{path.name}.",
+                suffix=".tmp",
+                delete=False,
+            ) as tmp:
+                tmp.write(payload)
+                tmp_path = Path(tmp.name)
+            tmp_path.replace(path)
+        except Exception:
+            logger.exception("coordinator snapshot to %s failed", path)
+
+
+def coordinator_from_context(ctx: dict[str, Any]) -> AgentCoordinator | None:
+    coordinator = ctx.get("coordinator")
+    return coordinator if isinstance(coordinator, AgentCoordinator) else None

strix/core/execution.py

@@ -0,0 +1,492 @@
+"""Execution loop for addressable SDK-backed Strix agents."""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+import uuid
+from collections.abc import Callable
+from typing import TYPE_CHECKING, Any, cast
+
+from agents import RunConfig, Runner
+from agents.exceptions import AgentsException, MaxTurnsExceeded, UserError
+from openai import APIError
+
+from strix.core.inputs import child_initial_input
+from strix.core.sessions import open_agent_session
+
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    from agents.items import TResponseInputItem
+    from agents.memory import Session, SQLiteSession
+    from agents.result import RunResultBase
+
+    from strix.core.agents import AgentCoordinator, Status
+
+
+logger = logging.getLogger(__name__)
+
+StreamEventSink = Callable[[str, Any], None]
+
+
+async def run_agent_loop(
+    *,
+    agent: Any,
+    initial_input: Any,
+    run_config: RunConfig,
+    context: dict[str, Any],
+    max_turns: int,
+    coordinator: AgentCoordinator,
+    agent_id: str,
+    interactive: bool,
+    session: Session | None = None,
+    start_parked: bool = False,
+    event_sink: StreamEventSink | None = None,
+) -> RunResultBase | None:
+    await coordinator.attach_runtime(
+        agent_id,
+        session=session,
+        interrupt_on_message=interactive,
+    )
+    result: RunResultBase | None = None
+
+    if not (start_parked and interactive):
+        if interactive:
+            result = await _run_cycle(
+                agent,
+                coordinator,
+                agent_id,
+                input_data=initial_input,
+                run_config=run_config,
+                context=context,
+                max_turns=max_turns,
+                session=session,
+                interactive=interactive,
+                event_sink=event_sink,
+            )
+        else:
+            result = await _run_noninteractive_until_lifecycle(
+                agent,
+                coordinator,
+                agent_id,
+                initial_input=initial_input,
+                run_config=run_config,
+                context=context,
+                max_turns=max_turns,
+                session=session,
+                event_sink=event_sink,
+            )
+
+    if not interactive:
+        return result
+
+    while True:
+        try:
+            await coordinator.wait_for_message(agent_id)
+        except asyncio.CancelledError:
+            return result
+
+        await coordinator.consume_pending(agent_id)
+        result = await _run_cycle(
+            agent,
+            coordinator,
+            agent_id,
+            input_data=[],
+            run_config=run_config,
+            context=context,
+            max_turns=max_turns,
+            session=session,
+            interactive=interactive,
+            event_sink=event_sink,
+        )
+
+
+async def spawn_child_agent(
+    *,
+    coordinator: AgentCoordinator,
+    factory: Any,
+    agents_db_path: Path,
+    sessions_to_close: list[SQLiteSession],
+    run_config: RunConfig,
+    max_turns: int,
+    interactive: bool,
+    parent_ctx: dict[str, Any],
+    name: str,
+    task: str,
+    skills: list[str],
+    parent_history: list[Any],
+    event_sink: StreamEventSink | None = None,
+) -> dict[str, Any]:
+    parent_id = parent_ctx.get("agent_id")
+    if not isinstance(parent_id, str):
+        raise TypeError("Parent agent_id missing from context")
+
+    child_id = uuid.uuid4().hex[:8]
+    child_agent = factory(name=name, skills=skills)
+    await coordinator.register(
+        child_id,
+        name,
+        parent_id,
+        task=task,
+        skills=skills,
+    )
+
+    await _start_child_runner(
+        parent_ctx=parent_ctx,
+        coordinator=coordinator,
+        agents_db_path=agents_db_path,
+        sessions_to_close=sessions_to_close,
+        run_config=run_config,
+        max_turns=max_turns,
+        interactive=interactive,
+        child_agent=child_agent,
+        child_id=child_id,
+        name=name,
+        parent_id=parent_id,
+        task=task,
+        initial_input=child_initial_input(
+            name=name,
+            child_id=child_id,
+            parent_id=parent_id,
+            task=task,
+            parent_history=parent_history,
+        ),
+        event_sink=event_sink,
+    )
+
+    return {
+        "success": True,
+        "agent_id": child_id,
+        "name": name,
+        "parent_id": parent_id,
+        "message": f"Spawned '{name}' ({child_id}) running in parallel.",
+    }
+
+
+async def respawn_subagents(
+    *,
+    coordinator: AgentCoordinator,
+    factory: Any,
+    agents_db_path: Path,
+    sessions_to_close: list[SQLiteSession],
+    run_config: RunConfig,
+    max_turns: int,
+    interactive: bool,
+    parent_ctx: dict[str, Any],
+    root_id: str,
+    event_sink: StreamEventSink | None = None,
+) -> None:
+    """Re-spawn subagent runners from a restored coordinator snapshot."""
+    async with coordinator._lock:
+        # Snapshot the iteration view first so we can mutate via coordinator
+        # below without "dict changed during iteration" trouble.
+        agents_snapshot = [
+            (aid, status, dict(coordinator.metadata.get(aid, {})))
+            for aid, status in coordinator.statuses.items()
+        ]
+        candidates: list[tuple[str, str, str | None, dict[str, Any]]] = []
+        for aid, status, md in agents_snapshot:
+            if not interactive and status not in {"running", "waiting"}:
+                continue
+            if coordinator.parent_of.get(aid) is None or aid == root_id:
+                continue
+            md["_restored_status"] = status
+            candidates.append(
+                (
+                    aid,
+                    coordinator.names.get(aid, aid),
+                    coordinator.parent_of.get(aid),
+                    md,
+                )
+            )
+
+    for child_id, name, parent_id, md in candidates:
+        try:
+            restored_status = str(md.get("_restored_status") or "running")
+            start_parked = interactive and restored_status != "running"
+
+            if start_parked:
+                logger.warning(
+                    "respawn %s (%s): starting parked from status=%s",
+                    child_id,
+                    name,
+                    restored_status,
+                )
+
+            child_skills = list(md.get("skills") or [])
+            child_agent = factory(name=name, skills=child_skills)
+            await _start_child_runner(
+                parent_ctx=parent_ctx,
+                coordinator=coordinator,
+                agents_db_path=agents_db_path,
+                sessions_to_close=sessions_to_close,
+                run_config=run_config,
+                max_turns=max_turns,
+                interactive=interactive,
+                child_agent=child_agent,
+                child_id=child_id,
+                name=name,
+                parent_id=parent_id,
+                task=str(md.get("task", "")),
+                initial_input=[],
+                start_parked=start_parked,
+                event_sink=event_sink,
+            )
+            logger.info(
+                "respawned %s (%s) parent=%s task_len=%d",
+                child_id,
+                name,
+                parent_id or "-",
+                len(md.get("task", "")),
+            )
+        except Exception:
+            logger.exception("respawn %s failed; marking crashed", child_id)
+            with contextlib.suppress(Exception):
+                await coordinator.set_status(child_id, "crashed")
+
+
+async def _run_noninteractive_until_lifecycle(
+    agent: Any,
+    coordinator: AgentCoordinator,
+    agent_id: str,
+    *,
+    initial_input: Any,
+    run_config: RunConfig,
+    context: dict[str, Any],
+    max_turns: int,
+    session: Session | None,
+    event_sink: StreamEventSink | None,
+) -> RunResultBase | None:
+    """Non-chat mode keeps running until finish_scan / agent_finish settles status."""
+    result: RunResultBase | None = None
+    input_data: Any = initial_input
+    invalid_final_outputs = 0
+    invalid_final_output_limit = max(1, max_turns)
+
+    while True:
+        result = await _run_cycle(
+            agent,
+            coordinator,
+            agent_id,
+            input_data=input_data,
+            run_config=run_config,
+            context=context,
+            max_turns=max_turns,
+            session=session,
+            interactive=False,
+            event_sink=event_sink,
+        )
+
+        status = await _agent_status(coordinator, agent_id)
+        if status != "running":
+            return result
+
+        invalid_final_outputs += 1
+        logger.warning(
+            "agent %s produced non-lifecycle final output in non-interactive mode; "
+            "forcing tool continuation (%d/%d): %s",
+            agent_id,
+            invalid_final_outputs,
+            invalid_final_output_limit,
+            _final_output_preview(result),
+        )
+
+        if invalid_final_outputs >= invalid_final_output_limit:
+            await coordinator.set_status(agent_id, "crashed")
+            await _notify_parent_on_crash(coordinator, agent_id, "crashed")
+            raise MaxTurnsExceeded(
+                "Agent exhausted non-interactive recovery attempts without calling "
+                "finish_scan or agent_finish."
+            )
+
+        input_data = await _append_noninteractive_tool_required_message(
+            session=session,
+            context=context,
+            attempt=invalid_final_outputs,
+            limit=invalid_final_output_limit,
+        )
+
+
+async def _run_cycle(
+    agent: Any,
+    coordinator: AgentCoordinator,
+    agent_id: str,
+    *,
+    input_data: Any,
+    run_config: RunConfig,
+    context: dict[str, Any],
+    max_turns: int,
+    session: Session | None,
+    interactive: bool,
+    event_sink: StreamEventSink | None,
+) -> RunResultBase | None:
+    try:
+        await coordinator.mark_running(agent_id)
+        stream = Runner.run_streamed(
+            agent,
+            input=input_data,
+            run_config=run_config,
+            context=context,
+            max_turns=max_turns,
+            session=session,
+        )
+        await coordinator.attach_stream(agent_id, stream)
+        try:
+            async for event in stream.stream_events():
+                if event_sink is not None:
+                    try:
+                        event_sink(agent_id, event)
+                    except Exception:
+                        logger.exception("stream event sink failed for %s", agent_id)
+            if stream.run_loop_exception is not None:
+                raise stream.run_loop_exception
+        finally:
+            await coordinator.detach_stream(agent_id, stream)
+    except Exception as exc:
+        if not interactive:
+            raise
+        if isinstance(exc, MaxTurnsExceeded):
+            status: Status = "stopped"
+        elif isinstance(exc, UserError | AgentsException | APIError):
+            status = "failed"
+        else:
+            status = "crashed"
+        logger.exception("agent run failed for %s; parking as %s", agent_id, status)
+        await coordinator.set_status(agent_id, status)
+        await _notify_parent_on_crash(coordinator, agent_id, status)
+        return None
+    else:
+        await _settle_run_result(coordinator, agent_id, interactive)
+        return stream
+
+
+async def _settle_run_result(
+    coordinator: AgentCoordinator,
+    agent_id: str,
+    interactive: bool,
+) -> None:
+    async with coordinator._lock:
+        current_status = coordinator.statuses.get(agent_id)
+
+    if current_status != "running":
+        return
+
+    if not interactive:
+        return
+
+    await coordinator.set_status(agent_id, "waiting")
+
+
+async def _agent_status(coordinator: AgentCoordinator, agent_id: str) -> Status | None:
+    async with coordinator._lock:
+        return coordinator.statuses.get(agent_id)
+
+
+def _final_output_preview(result: RunResultBase | None) -> str:
+    final_output = getattr(result, "final_output", None)
+    if final_output is None:
+        return "<none>"
+    text = str(final_output).replace("\n", " ").strip()
+    if not text:
+        return "<empty>"
+    return text[:300]
+
+
+async def _append_noninteractive_tool_required_message(
+    *,
+    session: Session | None,
+    context: dict[str, Any],
+    attempt: int,
+    limit: int,
+) -> list[dict[str, str]]:
+    finish_tool = "finish_scan" if context.get("parent_id") is None else "agent_finish"
+    message = (
+        "Your previous response ended the autonomous Strix run without a lifecycle tool call. "
+        "That is invalid in non-interactive mode; plain text final answers are ignored. "
+        "Continue immediately and call exactly one tool. "
+        f"If your work is complete, call {finish_tool}. "
+        "If you are blocked waiting for another agent, call wait_for_message. "
+        "Otherwise use the appropriate execution or planning tool. "
+        f"This is recovery attempt {attempt}/{limit}."
+    )
+    item = {"role": "user", "content": message}
+    if session is None:
+        return [item]
+
+    await session.add_items([cast("TResponseInputItem", item)])
+    return []
+
+
+async def _notify_parent_on_crash(
+    coordinator: AgentCoordinator,
+    agent_id: str,
+    status: str,
+) -> None:
+    if status != "crashed":
+        return
+    async with coordinator._lock:
+        parent = coordinator.parent_of.get(agent_id)
+        name = coordinator.names.get(agent_id, agent_id)
+    if parent is None:
+        return
+    await coordinator.send(
+        parent,
+        {
+            "from": agent_id,
+            "type": "crash",
+            "priority": "high",
+            "content": (
+                f"[Agent crash] {name} ({agent_id}) terminated unexpectedly. "
+                "Stop waiting on this child unless you want to message it again."
+            ),
+        },
+    )
+
+
+async def _start_child_runner(
+    *,
+    parent_ctx: dict[str, Any],
+    coordinator: AgentCoordinator,
+    agents_db_path: Path,
+    sessions_to_close: list[SQLiteSession],
+    run_config: RunConfig,
+    max_turns: int,
+    interactive: bool,
+    child_agent: Any,
+    child_id: str,
+    name: str,
+    parent_id: str | None,
+    task: str,
+    initial_input: Any,
+    start_parked: bool = False,
+    event_sink: StreamEventSink | None = None,
+) -> None:
+    session = open_agent_session(child_id, agents_db_path)
+    sessions_to_close.append(session)
+    await coordinator.attach_runtime(child_id, session=session)
+
+    child_ctx: dict[str, Any] = dict(parent_ctx)
+    child_ctx["agent_id"] = child_id
+    child_ctx["parent_id"] = parent_id
+    child_ctx["task"] = task
+
+    task_handle = asyncio.create_task(
+        run_agent_loop(
+            agent=child_agent,
+            initial_input=initial_input,
+            run_config=run_config,
+            context=child_ctx,
+            max_turns=max_turns,
+            coordinator=coordinator,
+            agent_id=child_id,
+            interactive=interactive,
+            session=session,
+            start_parked=start_parked,
+            event_sink=event_sink,
+        ),
+        name=f"agent-{name}-{child_id}",
+    )
+    await coordinator.attach_runtime(child_id, task=task_handle)

strix/core/inputs.py

@@ -0,0 +1,157 @@
+"""Pure input builders for Strix scan runs."""
+
+from __future__ import annotations
+
+import json
+from typing import Any, Literal
+
+from agents.model_settings import ModelSettings
+from openai.types.shared import Reasoning
+
+from strix.config.models import DEFAULT_MODEL_RETRY
+
+
+# Default max_turns budget passed to the SDK runner.
+DEFAULT_MAX_TURNS = 300
+
+
+def build_root_task(scan_config: dict[str, Any]) -> str:
+    """Format the user-facing task for the root agent."""
+    targets = scan_config.get("targets", []) or []
+    diff_scope = scan_config.get("diff_scope") or {}
+    user_instructions = scan_config.get("user_instructions", "") or ""
+
+    sections: dict[str, list[str]] = {
+        "Repositories": [],
+        "Local Codebases": [],
+        "URLs": [],
+        "IP Addresses": [],
+    }
+
+    for target in targets:
+        ttype = target.get("type")
+        details = target.get("details") or {}
+        workspace_subdir = details.get("workspace_subdir")
+        workspace_path = f"/workspace/{workspace_subdir}" if workspace_subdir else "/workspace"
+
+        if ttype == "repository":
+            url = details.get("target_repo", "")
+            cloned = details.get("cloned_repo_path")
+            sections["Repositories"].append(
+                f"- {url} (available at: {workspace_path})" if cloned else f"- {url}",
+            )
+        elif ttype == "local_code":
+            path = details.get("target_path", "unknown")
+            sections["Local Codebases"].append(f"- {path} (available at: {workspace_path})")
+        elif ttype == "web_application":
+            sections["URLs"].append(f"- {details.get('target_url', '')}")
+        elif ttype == "ip_address":
+            sections["IP Addresses"].append(f"- {details.get('target_ip', '')}")
+
+    parts: list[str] = []
+    for label, items in sections.items():
+        if items:
+            parts.append(f"\n\n{label}:")
+            parts.extend(items)
+
+    if diff_scope.get("active"):
+        parts.append("\n\nScope Constraints:")
+        parts.append(
+            "- Pull request diff-scope mode is active. Prioritize changed files "
+            "and use other files only for context.",
+        )
+        for repo_scope in diff_scope.get("repos", []) or []:
+            label = (
+                repo_scope.get("workspace_subdir") or repo_scope.get("source_path") or "repository"
+            )
+            changed = repo_scope.get("analyzable_files_count", 0)
+            deleted = repo_scope.get("deleted_files_count", 0)
+            parts.append(f"- {label}: {changed} changed file(s) in primary scope")
+            if deleted:
+                parts.append(f"- {label}: {deleted} deleted file(s) are context-only")
+
+    task = " ".join(parts)
+    if user_instructions:
+        task = f"{task}\n\nSpecial instructions: {user_instructions}"
+    return task
+
+
+def build_scope_context(scan_config: dict[str, Any]) -> dict[str, Any]:
+    """Build the system_prompt_context block consumed by the prompt template."""
+    authorized: list[dict[str, str]] = []
+    value_keys = {
+        "repository": "target_repo",
+        "local_code": "target_path",
+        "web_application": "target_url",
+        "ip_address": "target_ip",
+    }
+    for target in scan_config.get("targets", []) or []:
+        ttype = target.get("type", "unknown")
+        details = target.get("details") or {}
+        key = value_keys.get(ttype)
+        value = details.get(key, "") if key is not None else target.get("original", "")
+
+        workspace_subdir = details.get("workspace_subdir")
+        workspace_path = f"/workspace/{workspace_subdir}" if workspace_subdir else ""
+        authorized.append(
+            {"type": ttype, "value": value, "workspace_path": workspace_path},
+        )
+
+    return {
+        "scope_source": "system_scan_config",
+        "authorization_source": "strix_platform_verified_targets",
+        "authorized_targets": authorized,
+        "user_instructions_do_not_expand_scope": True,
+    }
+
+
+def make_model_settings(
+    reasoning_effort: Literal["low", "medium", "high"] | None,
+) -> ModelSettings:
+    model_settings = ModelSettings(
+        parallel_tool_calls=False,
+        tool_choice="required",
+        retry=DEFAULT_MODEL_RETRY,
+    )
+    if reasoning_effort is not None:
+        model_settings = model_settings.resolve(
+            ModelSettings(reasoning=Reasoning(effort=reasoning_effort)),
+        )
+    return model_settings
+
+
+def child_initial_input(
+    *,
+    name: str,
+    child_id: str,
+    parent_id: str,
+    task: str,
+    parent_history: list[Any],
+) -> list[dict[str, Any]]:
+    initial_input: list[dict[str, Any]] = []
+    if parent_history:
+        rendered = json.dumps(parent_history, ensure_ascii=False, default=str)
+        initial_input.append(
+            {
+                "role": "user",
+                "content": (
+                    "== Inherited context from parent (background only) ==\n"
+                    f"{rendered}\n"
+                    "== End of inherited context ==\n"
+                    "Use the above as background only; do not continue the "
+                    "parent's work. Your task follows."
+                ),
+            },
+        )
+    initial_input.append(
+        {
+            "role": "user",
+            "content": (
+                f"You are agent {name} ({child_id}); your parent is {parent_id}. "
+                "Maintain your own identity. Call agent_finish when your task "
+                "is complete."
+            ),
+        }
+    )
+    initial_input.append({"role": "user", "content": task})
+    return initial_input

strix/core/runner.py

@@ -0,0 +1,303 @@
+"""Top-level Strix scan runner.
+
+The SDK owns model/tool execution and per-agent sessions. This module owns
+Strix-specific scan setup, child-agent startup, resume, and the small wake loop
+needed to keep every agent addressable after its SDK run parks.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import json
+import logging
+import uuid
+from collections.abc import Callable
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+from agents import RunConfig
+from agents.sandbox import SandboxRunConfig
+
+from strix.agents.factory import build_strix_agent, make_child_factory
+from strix.config import load_settings
+from strix.config.models import configure_sdk_model_defaults, normalize_model_name
+from strix.core.agents import AgentCoordinator
+from strix.core.execution import (
+    respawn_subagents,
+    run_agent_loop,
+)
+from strix.core.execution import (
+    spawn_child_agent as start_child_agent,
+)
+from strix.core.inputs import (
+    DEFAULT_MAX_TURNS,
+    build_root_task,
+    build_scope_context,
+    make_model_settings,
+)
+from strix.core.sessions import open_agent_session
+from strix.runtime import session_manager
+from strix.telemetry.logging import set_scan_id, setup_scan_logging
+
+
+if TYPE_CHECKING:
+    from agents.memory import SQLiteSession
+    from agents.result import RunResultBase
+
+
+logger = logging.getLogger(__name__)
+
+StreamEventSink = Callable[[str, Any], None]
+
+
+async def run_strix_scan(
+    *,
+    scan_config: dict[str, Any],
+    scan_id: str | None = None,
+    image: str,
+    local_sources: list[dict[str, str]] | None = None,
+    coordinator: AgentCoordinator | None = None,
+    interactive: bool = False,
+    max_turns: int = DEFAULT_MAX_TURNS,
+    model: str | None = None,
+    cleanup_on_exit: bool = True,
+    event_sink: StreamEventSink | None = None,
+) -> RunResultBase | None:
+    """Run or resume one Strix scan against a sandbox."""
+    if scan_id is None:
+        scan_id = f"scan-{uuid.uuid4().hex[:8]}"
+
+    # Resolve run_dir before any heavy bring-up so the log file captures
+    # everything from sandbox start onwards.
+    run_dir = Path.cwd() / "strix_runs" / scan_id
+    run_dir.mkdir(parents=True, exist_ok=True)
+    teardown_logging = setup_scan_logging(run_dir)
+    set_scan_id(scan_id)
+
+    agents_path = run_dir / "agents.json"
+    agents_db = run_dir / "agents.db"
+    is_resume = agents_path.exists()
+
+    logger.info(
+        "%s Strix scan %s (image=%s, max_turns=%d, interactive=%s, run_dir=%s)",
+        "Resuming" if is_resume else "Starting",
+        scan_id,
+        image,
+        max_turns,
+        interactive,
+        run_dir,
+    )
+
+    settings = load_settings()
+    configure_sdk_model_defaults(settings)
+    resolved_model = normalize_model_name(model or settings.llm.model or "")
+    if not resolved_model:
+        raise RuntimeError(
+            "No LLM model configured. Set STRIX_LLM env or pass model= to run_strix_scan().",
+        )
+    logger.info("LLM model resolved: %s", resolved_model)
+
+    # Caller may pre-create the coordinator so it can route stop/chat
+    # commands while the scan loop runs in another thread.
+    if coordinator is None:
+        coordinator = AgentCoordinator()
+    coordinator.set_snapshot_path(agents_path)
+
+    # Wire the per-agent todo store to ``{run_dir}/todos.json`` (mirrored
+    # on every CRUD) and reload any prior todos so respawned subagents
+    # find their lists intact. Same for the shared notes store.
+    from strix.tools.notes.tools import hydrate_notes_from_disk
+    from strix.tools.todo.tools import hydrate_todos_from_disk
+
+    hydrate_todos_from_disk(run_dir)
+    hydrate_notes_from_disk(run_dir)
+
+    root_id: str | None = None
+    if is_resume:
+        try:
+            snap = json.loads(agents_path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError) as exc:
+            raise RuntimeError(
+                f"Cannot resume scan {scan_id}: agents.json is unreadable: {exc}",
+            ) from exc
+        if not agents_db.exists():
+            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:
+                root_id = aid
+                break
+        if root_id is None:
+            raise RuntimeError(
+                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",
+            len(coordinator.statuses),
+            root_id,
+        )
+    else:
+        root_id = uuid.uuid4().hex[:8]
+
+    logger.info("Bringing up sandbox session for scan %s", scan_id)
+    bundle = await session_manager.create_or_reuse(
+        scan_id,
+        image=image,
+        local_sources=local_sources or [],
+    )
+    logger.info("Sandbox ready for scan %s", scan_id)
+
+    sessions_to_close: list[SQLiteSession] = []
+
+    try:
+        targets = scan_config.get("targets") or []
+        scan_mode = str(scan_config.get("scan_mode") or "deep")
+        is_whitebox = any(t.get("type") == "local_code" for t in targets)
+        skills = list(scan_config.get("skills") or [])
+        root_task = build_root_task(scan_config)
+        model_settings = make_model_settings(settings.llm.reasoning_effort)
+        run_config = RunConfig(
+            model=resolved_model,
+            model_settings=model_settings,
+            sandbox=SandboxRunConfig(client=bundle["client"], session=bundle["session"]),
+            trace_include_sensitive_data=False,
+        )
+
+        scope_context = build_scope_context(scan_config)
+
+        root_agent = build_strix_agent(
+            name="strix",
+            skills=skills,
+            is_root=True,
+            scan_mode=scan_mode,
+            is_whitebox=is_whitebox,
+            interactive=interactive,
+            system_prompt_context=scope_context,
+        )
+
+        if not is_resume:
+            await coordinator.register(
+                root_id,
+                "strix",
+                parent_id=None,
+                task=root_task,
+                skills=skills,
+            )
+
+        child_agent_builder = make_child_factory(
+            scan_mode=scan_mode,
+            is_whitebox=is_whitebox,
+            interactive=interactive,
+            system_prompt_context=scope_context,
+        )
+
+        async def spawn_child_agent(**kwargs: Any) -> dict[str, Any]:
+            return await start_child_agent(
+                coordinator=coordinator,
+                factory=child_agent_builder,
+                agents_db_path=agents_db,
+                sessions_to_close=sessions_to_close,
+                run_config=run_config,
+                max_turns=max_turns,
+                interactive=interactive,
+                event_sink=event_sink,
+                **kwargs,
+            )
+
+        context: dict[str, Any] = {
+            "coordinator": coordinator,
+            "sandbox_session": bundle["session"],
+            "caido_client": bundle["caido_client"],
+            "agent_id": root_id,
+            "parent_id": None,
+            "interactive": interactive,
+            "spawn_child_agent": spawn_child_agent,
+        }
+
+        # 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)
+
+        if is_resume:
+            await respawn_subagents(
+                coordinator=coordinator,
+                factory=child_agent_builder,
+                agents_db_path=agents_db,
+                sessions_to_close=sessions_to_close,
+                run_config=run_config,
+                max_turns=max_turns,
+                interactive=interactive,
+                parent_ctx=context,
+                root_id=root_id,
+                event_sink=event_sink,
+            )
+
+        initial_input: Any = [] if is_resume else root_task
+
+        # Resume + new ``--instruction``: SDK replay drives root from
+        # 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.
+        resume_instruction = str(scan_config.get("resume_instruction") or "").strip()
+        if is_resume and resume_instruction:
+            await coordinator.send(
+                root_id,
+                {
+                    "from": "user",
+                    "type": "instruction",
+                    "priority": "high",
+                    "content": resume_instruction,
+                },
+            )
+            logger.info(
+                "Resume: injected new instruction into root SDK session (len=%d)",
+                len(resume_instruction),
+            )
+
+        async with coordinator._lock:
+            root_status = coordinator.statuses.get(root_id)
+
+        return await run_agent_loop(
+            agent=root_agent,
+            initial_input=initial_input,
+            run_config=run_config,
+            context=context,
+            max_turns=max_turns,
+            coordinator=coordinator,
+            agent_id=root_id,
+            interactive=interactive,
+            session=root_session,
+            start_parked=bool(interactive and is_resume and root_status != "running"),
+            event_sink=event_sink,
+        )
+    except BaseException:
+        logger.exception("Strix scan %s failed", scan_id)
+        # Cancel any descendant tasks the root spawned before unwinding.
+        # cancel_descendants is idempotent and handles the empty-tree case.
+        if root_id is not None:
+            await coordinator.cancel_descendants(root_id)
+            # The SDK's on_agent_end hook only fires after a successful
+            # ``Runner.run_streamed`` reaches the agent's first turn. A
+            # failure earlier (e.g., model-provider routing, sandbox
+            # bring-up) leaves the root stuck at status="running" — the
+            # TUI keeps animating "Initializing" forever. Finalize it
+            # here so the coordinator reflects reality.
+            with contextlib.suppress(Exception):
+                await coordinator.set_status(root_id, "failed")
+        raise
+    finally:
+        for s in sessions_to_close:
+            with contextlib.suppress(Exception):
+                s.close()
+        with contextlib.suppress(Exception):
+            await coordinator._maybe_snapshot()
+        if cleanup_on_exit:
+            logger.info("Tearing down sandbox session for scan %s", scan_id)
+            await session_manager.cleanup(scan_id)
+        logger.info("Strix scan %s done", scan_id)
+        teardown_logging()

strix/core/sessions.py

@@ -0,0 +1,10 @@
+"""SDK session helpers for Strix agents."""
+
+from pathlib import Path
+
+from agents.memory import SQLiteSession
+
+
+def open_agent_session(agent_id: str, path: Path) -> SQLiteSession:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    return SQLiteSession(session_id=agent_id, db_path=path)