fix: pre-v1-style lifecycle resilience — mailbox delivery, uniform revival, unexitable runner, waiting timeout, broader retries, crash-safe identity (#923)

2026-08-16 17:27:26 +02:00 · 2026-08-01 11:17:08 -07:00
parent a9deb84260
commit 5602bc23ca
8 changed files with 413 additions and 145 deletions
@@ -32,6 +32,8 @@ class AgentRuntime:
    stream: Any | None = None
    interrupt_on_message: bool = False
    wake: asyncio.Event = field(default_factory=asyncio.Event)
+    mailbox: list[dict[str, Any]] = field(default_factory=list)
+    user_wake_required: bool = False


 class AgentCoordinator:
@@ -179,6 +181,7 @@ class AgentCoordinator:
            if agent_id in self.statuses:
                self.statuses[agent_id] = "running"
                self.errors.pop(agent_id, None)
+                self.runtimes.setdefault(agent_id, AgentRuntime()).user_wake_required = False
        await self._maybe_snapshot()

    async def park_waiting(self, agent_id: str) -> None:
@@ -196,6 +199,7 @@ class AgentCoordinator:
            elif status == "running":
                self.errors.pop(agent_id, None)
            runtime = self.runtimes.setdefault(agent_id, AgentRuntime())
+            runtime.user_wake_required = status in {"failed", "crashed"}
            runtime.wake.set()
        logger.info("agent.status %s=%s", agent_id, status)
        await self._maybe_snapshot()
@@ -203,49 +207,47 @@ class AgentCoordinator:
    async def send(
        self, target_agent_id: str, message: dict[str, Any], *, interrupt: bool = True
    ) -> bool:
-        """Deliver a user/peer message by appending it to the target SDK session."""
-        if message.get("from") == "user" and self._budget_paused:
+        """Queue a user/peer message in the target's mailbox and wake it."""
+        from_user = message.get("from") == "user"
+        if from_user and self._budget_paused:
            await self.resume_from_budget_pause(exclude=target_agent_id)
        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
+            runtime.mailbox.append(dict(message))
+            self.pending_counts[target_agent_id] = self.pending_counts.get(target_agent_id, 0) + 1
+            if from_user:
+                runtime.user_wake_required = False
+            runtime.wake.set()
            stream = runtime.stream
            interrupt_on_message = runtime.interrupt_on_message
-        if session is None:
-            logger.warning(
-                "agent.send dropped target=%s because its SDK session is not attached",
-                target_agent_id,
-            )
-            return False
-        try:
-            async with session_write_lock(session):
-                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 and interrupt_on_message:
            stream.cancel(mode="immediate")
        await self._maybe_snapshot()
        return True

-    async def wait_for_message(self, agent_id: str) -> None:
+    async def wait_for_message(self, agent_id: str, *, timeout: float | None = None) -> bool:
+        """Wait until a message is ready for ``agent_id``; False on ``timeout``."""
        while True:
            async with self._lock:
+                runtime = self.runtimes.setdefault(agent_id, AgentRuntime())
                reserve_exit = self._reserve_stopped and self.parent_of.get(agent_id) is not None
-                if self._budget_stopped or reserve_exit or self.pending_counts.get(agent_id, 0) > 0:
-                    return
-                wake = self.runtimes.setdefault(agent_id, AgentRuntime()).wake
+                pending_ready = (
+                    self.pending_counts.get(agent_id, 0) > 0 and not runtime.user_wake_required
+                )
+                if self._budget_stopped or reserve_exit or pending_ready:
+                    return True
+                wake = runtime.wake
                wake.clear()
-            await wake.wait()
+            if timeout is None:
+                await wake.wait()
+            else:
+                try:
+                    await asyncio.wait_for(wake.wait(), timeout)
+                except TimeoutError:
+                    return False

    async def consume_pending(
        self,
@@ -253,17 +255,38 @@ class AgentCoordinator:
        *,
        include_items: bool = False,
    ) -> tuple[int, list[Any]]:
+        """Drain the agent's mailbox into its own SDK session."""
        async with self._lock:
-            count = self.pending_counts.get(agent_id, 0)
+            runtime = self.runtimes.setdefault(agent_id, AgentRuntime())
+            queued = list(runtime.mailbox)
+            runtime.mailbox.clear()
+            count = max(self.pending_counts.get(agent_id, 0), len(queued))
            self.pending_counts[agent_id] = 0
-            session = self.runtimes.get(agent_id, AgentRuntime()).session
+            session = runtime.session
        if count <= 0:
            return 0, []
+        items = [self._message_to_session_item(m) for m in queued]
+        if items:
+            if session is None:
+                logger.warning(
+                    "agent %s has no SDK session attached; %d queued messages were not persisted",
+                    agent_id,
+                    len(items),
+                )
+            else:
+                try:
+                    async with session_write_lock(session):
+                        await session.add_items(items)
+                except Exception:
+                    logger.exception(
+                        "failed to append %d queued messages to the session of %s",
+                        len(items),
+                        agent_id,
+                    )
        await self._maybe_snapshot()
-        if not include_items or session is None:
+        if not include_items:
            return count, []
-        items = await session.get_items()
-        return count, list(items[-count:])
+        return count, items

    async def request_stop(self, agent_id: str) -> None:
        async with self._lock:
@@ -374,6 +397,11 @@ class AgentCoordinator:
                "names": dict(self.names),
                "metadata": {aid: dict(md) for aid, md in self.metadata.items()},
                "pending_counts": dict(self.pending_counts),
+                "mailboxes": {
+                    aid: [dict(m) for m in runtime.mailbox]
+                    for aid, runtime in self.runtimes.items()
+                    if runtime.mailbox
+                },
                "errors": dict(self.errors),
                "budget_stopped": self._budget_stopped,
                "reserve_stopped": self._reserve_stopped,
@@ -388,6 +416,12 @@ class AgentCoordinator:
            self.metadata = {aid: dict(md) for aid, md in snap.get("metadata", {}).items()}
            self.pending_counts = dict(snap.get("pending_counts", {}))
            self.errors = dict(snap.get("errors", {}))
+            mailboxes = snap.get("mailboxes", {})
+            if isinstance(mailboxes, dict):
+                for aid, msgs in mailboxes.items():
+                    if isinstance(msgs, list):
+                        runtime = self.runtimes.setdefault(aid, AgentRuntime())
+                        runtime.mailbox = [dict(m) for m in msgs if isinstance(m, dict)]
            self._budget_stopped = bool(snap.get("budget_stopped", False))
            self._reserve_stopped = bool(snap.get("reserve_stopped", False))
            self._budget_paused = bool(snap.get("budget_paused", False))
@@ -9,6 +9,7 @@ import uuid
 from collections.abc import Callable
 from typing import TYPE_CHECKING, Any, cast

+import litellm
 from agents import RunConfig, Runner
 from agents.exceptions import AgentsException, MaxTurnsExceeded, UserError
 from agents.sandbox.errors import ExecTransportError
@@ -16,9 +17,7 @@ from docker import errors as docker_errors  # type: ignore[import-untyped, unuse
 from openai import (
    APIConnectionError,
    APIError,
-    APIStatusError,
    APITimeoutError,
-    RateLimitError,
 )

 from strix.config import codex
@@ -31,6 +30,8 @@ from strix.core.inputs import child_initial_input
 from strix.core.sessions import (
    enforce_image_budget,
    open_agent_session,
+    replace_session_items,
+    seed_initial_input,
    strip_all_images_from_session,
 )
 from strix.llm.compaction import is_context_overflow, maybe_compact
@@ -106,15 +107,9 @@ async def _compact_session(
    )


-_GUARDRAIL_PARK_ERROR = (
-    "Blocked by the model's content guardrail (flagged as a possible cybersecurity risk). "
-    "Set STRIX_LLM to a model that isn't blocked and resume the scan to continue."
-)
-
-_TRANSIENT_MODEL_STATUS_CODES = frozenset({408, 500, 502, 503, 504})
-_MAX_TRANSIENT_MODEL_RETRIES = 4
+_MAX_TRANSIENT_MODEL_RETRIES = 5
 _TRANSIENT_MODEL_RETRY_BASE_DELAY_S = 2.0
-_TRANSIENT_MODEL_RETRY_MAX_DELAY_S = 30.0
+_TRANSIENT_MODEL_RETRY_MAX_DELAY_S = 90.0


 def _model_error_status_code(exc: BaseException) -> int | None:
@@ -123,15 +118,16 @@ def _model_error_status_code(exc: BaseException) -> int | None:


 def _is_transient_model_error(exc: BaseException) -> bool:
-    if isinstance(exc, RateLimitError):
+    if codex.is_content_guardrail_error(exc):
        return False
-    if isinstance(exc, APITimeoutError | APIConnectionError):
+    if isinstance(
+        exc, APITimeoutError | APIConnectionError | TimeoutError | ConnectionError | OSError
+    ):
        return True
-    if isinstance(exc, APIStatusError):
-        return exc.status_code in _TRANSIENT_MODEL_STATUS_CODES
-    if isinstance(exc, APIError):
-        return _model_error_status_code(exc) is None
-    return False
+    code = _model_error_status_code(exc)
+    if code is not None:
+        return bool(litellm._should_retry(code))
+    return isinstance(exc, APIError)


 def _transient_model_retry_delay(attempt: int) -> float:
@@ -139,6 +135,40 @@ def _transient_model_retry_delay(attempt: int) -> float:
    return min(delay, _TRANSIENT_MODEL_RETRY_MAX_DELAY_S)


+async def _salvage_stream_to_session(
+    session: Session,
+    pre_run_items: list[Any],
+    stream: Any,
+    agent_id: str,
+) -> None:
+    """Persist a crashed run's full history so a revived agent loses no context."""
+    if stream is None:
+        return
+    try:
+        replay = list(stream.to_input_list())
+    except Exception:
+        logger.exception("could not build salvage history for %s", agent_id)
+        return
+    desired = list(pre_run_items) + replay
+    if len(desired) <= len(pre_run_items):
+        return
+    try:
+        await replace_session_items(session, desired)
+    except Exception:
+        logger.exception("salvaging crashed run history failed for %s", agent_id)
+
+
+async def _seed_and_prepare_first_input(
+    session: Session | None, initial_input: Any, *, start_parked: bool
+) -> Any:
+    """Persist the opening input up front so it survives a first-turn crash."""
+    if initial_input and session is not None and not start_parked:
+        with contextlib.suppress(Exception):
+            if await seed_initial_input(session, initial_input):
+                return []
+    return initial_input
+
+
 async def run_agent_loop(
    *,
    agent: Any,
@@ -161,6 +191,10 @@ async def run_agent_loop(
    )
    result: RunResultBase | None = None

+    first_cycle_input = await _seed_and_prepare_first_input(
+        session, initial_input, start_parked=start_parked
+    )
+
    budget_stopped = coordinator.budget_stopped
    reserve_stopped = coordinator.reserve_stopped
    if budget_stopped:
@@ -176,16 +210,15 @@ async def run_agent_loop(
    if not (start_parked and interactive):
        if interactive:
            with contextlib.suppress(BudgetPausedError):
-                result = await _run_cycle(
+                result = await _run_cycle_parked(
                    agent,
                    coordinator,
                    agent_id,
-                    input_data=initial_input,
+                    input_data=first_cycle_input,
                    run_config=run_config,
                    context=context,
                    max_turns=max_turns,
                    session=session,
-                    interactive=interactive,
                    event_sink=event_sink,
                    hooks=hooks,
                )
@@ -194,7 +227,7 @@ async def run_agent_loop(
                agent,
                coordinator,
                agent_id,
-                initial_input=initial_input,
+                initial_input=first_cycle_input,
                run_config=run_config,
                context=context,
                max_turns=max_turns,
@@ -207,8 +240,9 @@ async def run_agent_loop(
        return result

    while True:
+        timeout = await _plain_waiting_timeout(coordinator, agent_id, context)
        try:
-            await coordinator.wait_for_message(agent_id)
+            woke = await coordinator.wait_for_message(agent_id, timeout=timeout)
        except asyncio.CancelledError:
            return result

@@ -220,9 +254,21 @@ async def run_agent_loop(
            await coordinator.set_status(agent_id, "stopped")
            raise SubagentBudgetReservedError("scan reached the sub-agent budget reserve")

+        if not woke:
+            logger.info("agent %s reached its waiting timeout; auto-resuming", agent_id)
+            await coordinator.send(
+                agent_id,
+                {
+                    "from": "system",
+                    "type": "auto_resume",
+                    "content": "Waiting timeout reached. Resuming execution.",
+                },
+                interrupt=False,
+            )
+
        await coordinator.consume_pending(agent_id)
        with contextlib.suppress(BudgetPausedError):
-            result = await _run_cycle(
+            result = await _run_cycle_parked(
                agent,
                coordinator,
                agent_id,
@@ -231,7 +277,6 @@ async def run_agent_loop(
                context=context,
                max_turns=max_turns,
                session=session,
-                interactive=interactive,
                event_sink=event_sink,
                hooks=hooks,
            )
@@ -327,7 +372,6 @@ async def respawn_subagents(
            if coordinator.parent_of.get(aid) is None or aid == root_id:
                continue
            md["_restored_status"] = status
-            md["_restored_error"] = coordinator.errors.get(aid)
            candidates.append(
                (
                    aid,
@@ -340,8 +384,7 @@ async def respawn_subagents(
    for child_id, name, parent_id, md in candidates:
        try:
            restored_status = str(md.get("_restored_status") or "running")
-            recoverable_park = restored_status == "waiting" and bool(md.get("_restored_error"))
-            start_parked = interactive and restored_status != "running" and not recoverable_park
+            start_parked = interactive and restored_status != "running"

            if start_parked:
                logger.warning(
@@ -456,6 +499,64 @@ async def _run_noninteractive_until_lifecycle(
        )


+_WAITING_AUTO_RESUME_TIMEOUT_S = 600.0
+
+
+async def _plain_waiting_timeout(
+    coordinator: AgentCoordinator,
+    agent_id: str,
+    context: dict[str, Any],
+) -> float | None:
+    """Auto-resume timeout for a plainly-waiting subagent; None waits forever."""
+    if context.get("parent_id") is None:
+        return None
+    async with coordinator._lock:
+        status = coordinator.statuses.get(agent_id)
+        has_error = agent_id in coordinator.errors
+        runtime = coordinator.runtimes.get(agent_id)
+        gated = runtime.user_wake_required if runtime is not None else False
+    if status == "waiting" and not has_error and not gated:
+        return _WAITING_AUTO_RESUME_TIMEOUT_S
+    return None
+
+
+async def _run_cycle_parked(
+    agent: Any,
+    coordinator: AgentCoordinator,
+    agent_id: str,
+    *,
+    input_data: Any,
+    run_config: RunConfig,
+    context: dict[str, Any],
+    max_turns: int,
+    session: Session | None,
+    event_sink: StreamEventSink | None,
+    hooks: RunHooks[dict[str, Any]] | None,
+) -> RunResultBase | None:
+    """Interactive run cycle that parks on any error instead of killing the runner."""
+    try:
+        return await _run_cycle(
+            agent,
+            coordinator,
+            agent_id,
+            input_data=input_data,
+            run_config=run_config,
+            context=context,
+            max_turns=max_turns,
+            session=session,
+            interactive=True,
+            event_sink=event_sink,
+            hooks=hooks,
+        )
+    except (BudgetExceededError, BudgetPausedError, SubagentBudgetReservedError):
+        raise
+    except Exception as exc:
+        logger.exception("error escaped the run cycle for %s; parking as failed", agent_id)
+        await coordinator.set_status(agent_id, "failed", error=str(exc) or type(exc).__name__)
+        await _notify_parent_on_terminal(coordinator, agent_id, "failed")
+        return None
+
+
 async def _run_cycle(  # noqa: PLR0912, PLR0915
    agent: Any,
    coordinator: AgentCoordinator,
@@ -474,6 +575,8 @@ async def _run_cycle(  # noqa: PLR0912, PLR0915
    compactions = 0
    model_retries = 0
    while True:
+        stream: Any = None
+        pre_run_items: list[Any] = []
        try:
            await coordinator.mark_running(agent_id)
            if session is not None:
@@ -487,6 +590,8 @@ async def _run_cycle(  # noqa: PLR0912, PLR0915
                    await _compact_session(agent, session, run_config, force=False)
                except Exception:
                    logger.exception("proactive compaction failed for %s", agent_id)
+                with contextlib.suppress(Exception):
+                    pre_run_items = list(await session.get_items())
            stream = Runner.run_streamed(
                agent,
                input=input_data,
@@ -599,10 +704,8 @@ async def _run_cycle(  # noqa: PLR0912, PLR0915
                if session is not None:
                    input_data = []
                continue
-            if codex.is_content_guardrail_error(exc):
-                return await _handle_content_guardrail(
-                    coordinator, agent_id, exc, interactive=interactive
-                )
+            if session is not None:
+                await _salvage_stream_to_session(session, pre_run_items, stream, agent_id)
            if isinstance(exc, ProviderRefusalError):
                logger.warning("agent %s refused by the model provider: %s", agent_id, exc)
                await coordinator.set_status(agent_id, "failed", error=str(exc))
@@ -622,23 +725,7 @@ async def _run_cycle(  # noqa: PLR0912, PLR0915
            return None
        else:
            await _settle_run_result(coordinator, agent_id, interactive)
-            return stream
-
-
-async def _handle_content_guardrail(
-    coordinator: AgentCoordinator,
-    agent_id: str,
-    exc: BaseException,
-    *,
-    interactive: bool,
-) -> RunResultBase | None:
-    logger.warning("agent %s blocked by the model's content guardrail: %s", agent_id, exc)
-    if interactive:
-        await coordinator.set_status(agent_id, "waiting", error=_GUARDRAIL_PARK_ERROR)
-        return None
-    await coordinator.set_status(agent_id, "failed", error=_GUARDRAIL_PARK_ERROR)
-    await _notify_parent_on_terminal(coordinator, agent_id, "failed")
-    return None
+            return cast("RunResultBase | None", stream)


 async def _settle_run_result(
@@ -377,12 +377,6 @@ async def run_strix_scan(

        async with coordinator._lock:
            root_status = coordinator.statuses.get(root_id)
-            root_error = coordinator.errors.get(root_id)
-
-        root_recoverable_park = root_status == "waiting" and bool(root_error)
-        root_start_parked = bool(
-            interactive and is_resume and root_status != "running" and not root_recoverable_park
-        )

        result = await run_agent_loop(
            agent=root_agent,
@@ -394,7 +388,7 @@ async def run_strix_scan(
            agent_id=root_id,
            interactive=interactive,
            session=root_session,
-            start_parked=root_start_parked,
+            start_parked=bool(interactive and is_resume and root_status != "running"),
            event_sink=event_sink,
            hooks=hooks,
        )
@@ -7,6 +7,7 @@ import logging
 from typing import TYPE_CHECKING, Any, cast
 from weakref import WeakKeyDictionary

+from agents.items import ItemHelpers
 from agents.memory import SQLiteSession


@@ -26,6 +27,18 @@ def open_agent_session(agent_id: str, path: Path) -> SQLiteSession:
    return SQLiteSession(session_id=agent_id, db_path=path)


+async def seed_initial_input(session: Session, initial_input: Any) -> bool:
+    """Commit an agent's opening identity/task input before its first run cycle."""
+    items = ItemHelpers.input_to_new_input_list(initial_input)
+    if not items:
+        return False
+    async with session_write_lock(session):
+        if await session.get_items():
+            return False
+        await session.add_items(items)
+    return True
+
+
 _IMAGE_REJECTED_TEXT = "[image rejected by the model]"
 _IMAGE_ELIDED_TEXT = "[older screenshot elided to bound context memory]"
 _INHERITED_IMAGE_TEXT = "[screenshot omitted from inherited context]"
@@ -1041,9 +1041,9 @@ class StrixTUIApp(App):  # type: ignore[misc]
                            name=names.get(agent_id, agent_id),
                            parent_id=parent_of.get(agent_id),
                            status=status,
-                            error_message=error,
+                            error_message=error or "",
                        )
-                        if status in {"failed", "crashed"} and error:
+                        if error:
                            if agent_id not in self._error_noted_agents:
                                self._error_noted_agents.add(agent_id)
                                self.live_view.record_agent_error(agent_id, error)
@@ -1293,6 +1293,10 @@ class StrixTUIApp(App):  # type: ignore[misc]
                text.append("Send a message to continue", style="dim")
                keymap = keymap_styled([("ctrl-q", "quit")])
            else:
+                error_msg = agent_data.get("error_message") or ""
+                if error_msg:
+                    text.append(error_msg, style="red")
+                    text.append(" \u00b7 ", style="dim")
                text.append("Send message to resume", style="dim")
            return (text, keymap, False)

@@ -82,7 +82,7 @@ class TuiLiveView:
            current["parent_id"] = parent_id
        if status is not None:
            current["status"] = status
-        if error_message:
+        if error_message is not None:
            current["error_message"] = error_message
        current["updated_at"] = now

@@ -5,7 +5,7 @@ from __future__ import annotations
 import asyncio
 import contextlib
 import json
-from typing import Any
+from typing import Any, cast
 from unittest.mock import MagicMock

 import pytest
@@ -14,18 +14,19 @@ from agents.memory import SQLiteSession
 from agents.tool_context import ToolContext
 from openai.types.responses import ResponseOutputMessage, ResponseOutputRefusal

-from strix.config import codex
 from strix.core import execution
 from strix.core.agents import AgentCoordinator
 from strix.core.execution import (
-    _handle_content_guardrail,
    _notify_parent_on_terminal,
    _notify_root_on_budget_reserve,
-    respawn_subagents,
 )
+from strix.core.sessions import seed_initial_input
 from strix.tools.finish.tool import finish_scan


+_NO_STREAM_EVENTS: list[Any] = []
+
+
 class _StructuredRefusalStream:
    def __init__(self, refusal: str) -> None:
        self.run_loop_exception: BaseException | None = None
@@ -43,8 +44,8 @@ class _StructuredRefusalStream:
        ]

    async def stream_events(self) -> Any:
-        if False:
-            yield None
+        for event in _NO_STREAM_EVENTS:
+            yield event

    def cancel(self, mode: str = "immediate") -> None:  # noqa: ARG002
        return
@@ -529,44 +530,152 @@ async def test_terminal_notice_does_not_cancel_parent_stream(tmp_path: Any) -> N


@pytest.mark.asyncio
-async def test_guardrail_interactive_parks_agent_wakeable(tmp_path: Any) -> None:
+async def test_send_queues_without_session_and_drains_on_consume(tmp_path: Any) -> None:
    coordinator = AgentCoordinator()
    await coordinator.register("root", "strix", parent_id=None)
-    await coordinator.register("child", "recon", parent_id="root")
-    exc = codex.CodexContentGuardrailError("chatgpt/gpt-5.6-sol")

-    result = await _handle_content_guardrail(coordinator, "child", exc, interactive=True)
+    assert await coordinator.send("root", {"from": "user", "content": "hello"}) is True
+    assert coordinator.pending_counts["root"] == 1

-    assert result is None
-    assert coordinator.statuses["child"] == "waiting"
-    assert "STRIX_LLM" in coordinator.errors["child"]
+    session = SQLiteSession("root", tmp_path / "agents.db")
+    await coordinator.attach_runtime("root", session=session)

-    waiter = asyncio.create_task(coordinator.wait_for_message("child"))
-    await asyncio.sleep(0)
-    assert not waiter.done()
-    session = SQLiteSession("child", tmp_path / "agents.db")
-    await coordinator.attach_runtime("child", session=session)
-    await coordinator.send("child", {"from": "user", "content": "switched model, resume"})
-    await asyncio.wait_for(waiter, timeout=1.0)
+    count, items = await coordinator.consume_pending("root", include_items=True)
+    assert count == 1
+    assert items[0]["content"] == "hello"
+    stored = await session.get_items()
+    last = cast("dict[str, Any]", stored[-1])
+    assert last["content"] == "hello"
    session.close()


@pytest.mark.asyncio
-async def test_guardrail_noninteractive_fails_only_blocked_agent(tmp_path: Any) -> None:
+async def test_error_parked_agent_only_released_by_user_message(tmp_path: Any) -> None:
    coordinator = AgentCoordinator()
    await coordinator.register("root", "strix", parent_id=None)
    await coordinator.register("child", "recon", parent_id="root")
-    session = SQLiteSession("root", tmp_path / "agents.db")
-    await coordinator.attach_runtime("root", session=session)
-    exc = codex.CodexContentGuardrailError("chatgpt/gpt-5.6-sol")
+    session = SQLiteSession("child", tmp_path / "agents.db")
+    await coordinator.attach_runtime("child", session=session)
+    await coordinator.set_status("child", "crashed", error="boom")

-    result = await _handle_content_guardrail(coordinator, "child", exc, interactive=False)
+    await coordinator.send("child", {"from": "root", "content": "peer nudge"})
+    waiter = asyncio.create_task(coordinator.wait_for_message("child"))
+    await asyncio.sleep(0.05)
+    assert not waiter.done()
+
+    await coordinator.send("child", {"from": "user", "content": "wake up"})
+    assert await asyncio.wait_for(waiter, timeout=1.0) is True
+
+    count, items = await coordinator.consume_pending("child", include_items=True)
+    assert count == 2
+    assert items[0]["content"].endswith("peer nudge")
+    assert items[1]["content"] == "wake up"
+    session.close()
+
+
+@pytest.mark.asyncio
+async def test_wait_for_message_timeout_returns_false() -> None:
+    coordinator = AgentCoordinator()
+    await coordinator.register("child", "recon", parent_id="root")
+
+    assert await coordinator.wait_for_message("child", timeout=0.05) is False
+
+
+@pytest.mark.asyncio
+async def test_snapshot_round_trip_preserves_mailboxes() -> None:
+    coordinator = AgentCoordinator()
+    await coordinator.register("root", "strix", parent_id=None)
+    await coordinator.send("root", {"from": "user", "content": "queued"})
+
+    snap = await coordinator.snapshot()
+    restored = AgentCoordinator()
+    await restored.restore(snap)
+
+    assert restored.pending_counts["root"] == 1
+    assert restored.runtimes["root"].mailbox == [{"from": "user", "content": "queued"}]
+
+
+@pytest.mark.asyncio
+async def test_run_cycle_parked_parks_instead_of_raising(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    async def _boom(*_args: Any, **_kwargs: Any) -> Any:
+        raise RuntimeError("unexpected explosion")
+
+    monkeypatch.setattr(execution, "_run_cycle", _boom)
+
+    coordinator = AgentCoordinator()
+    await coordinator.register("root", "strix", parent_id=None)
+
+    result = await execution._run_cycle_parked(
+        object(),
+        coordinator,
+        "root",
+        input_data=[],
+        run_config=None,  # type: ignore[arg-type]
+        context={},
+        max_turns=5,
+        session=None,
+        event_sink=None,
+        hooks=None,
+    )

    assert result is None
-    assert coordinator.statuses["child"] == "failed"
-    assert "STRIX_LLM" in coordinator.errors["child"]
-    assert coordinator.statuses["root"] == "running"
-    assert coordinator.pending_counts.get("root", 0) > 0
+    assert coordinator.statuses["root"] == "failed"
+    assert coordinator.errors["root"] == "unexpected explosion"
+
+
+class _SalvageStream:
+    def __init__(self, replay: list[dict[str, Any]]) -> None:
+        self._replay = replay
+
+    def to_input_list(self) -> list[dict[str, Any]]:
+        return self._replay
+
+
+@pytest.mark.asyncio
+async def test_salvage_stream_to_session_preserves_full_history(tmp_path: Any) -> None:
+    session = SQLiteSession("child", tmp_path / "agents.db")
+    await session.add_items([{"role": "user", "content": "identity + task"}])
+    pre_run = list(await session.get_items())
+
+    # A crash mid-run: the stream produced two turns the SDK never committed.
+    stream = _SalvageStream(
+        [
+            {"role": "assistant", "content": "recon turn 1"},
+            {"role": "assistant", "content": "recon turn 2"},
+        ]
+    )
+    await execution._salvage_stream_to_session(session, pre_run, stream, "child")
+
+    stored = [cast("dict[str, Any]", i) for i in await session.get_items()]
+    assert [i["content"] for i in stored] == [
+        "identity + task",
+        "recon turn 1",
+        "recon turn 2",
+    ]
+
+    # A crash with nothing new to salvage leaves the session untouched.
+    await execution._salvage_stream_to_session(
+        session, list(await session.get_items()), _SalvageStream([]), "child"
+    )
+    assert len(await session.get_items()) == 3
+    session.close()
+
+
+@pytest.mark.asyncio
+async def test_seed_initial_input_persists_and_is_idempotent(tmp_path: Any) -> None:
+    session = SQLiteSession("child", tmp_path / "agents.db")
+    identity = [{"role": "user", "content": "You are agent recon (abc); do X."}]
+
+    assert await seed_initial_input(session, identity) is True
+    assert len(await session.get_items()) == 1
+
+    # A populated session is left untouched (no duplicate identity message).
+    assert await seed_initial_input(session, identity) is False
+    assert len(await session.get_items()) == 1
+
+    assert await seed_initial_input(session, []) is False
    session.close()


@@ -576,7 +685,9 @@ async def test_structured_provider_refusal_fails_interactive_agent(
 ) -> None:
    refusal = "This request was blocked under the provider's usage policy."
    stream = _StructuredRefusalStream(refusal)
-    monkeypatch.setattr(execution.Runner, "run_streamed", lambda *_args, **_kwargs: stream)
+    monkeypatch.setattr(
+        "strix.core.execution.Runner.run_streamed", lambda *_args, **_kwargs: stream
+    )
    coordinator = AgentCoordinator()
    await coordinator.register("root", "strix", parent_id=None)

@@ -606,7 +717,9 @@ async def test_structured_provider_refusal_fails_noninteractive_child(
 ) -> None:
    refusal = "This request was blocked under the provider's usage policy."
    stream = _StructuredRefusalStream(refusal)
-    monkeypatch.setattr(execution.Runner, "run_streamed", lambda *_args, **_kwargs: stream)
+    monkeypatch.setattr(
+        "strix.core.execution.Runner.run_streamed", lambda *_args, **_kwargs: stream
+    )
    coordinator = AgentCoordinator()
    await coordinator.register("root", "strix", parent_id=None)
    await coordinator.register("child", "recon", parent_id="root")
@@ -635,34 +748,40 @@ async def test_structured_provider_refusal_fails_noninteractive_child(


@pytest.mark.asyncio
-async def test_resume_revives_guardrail_parked_child_but_not_plain_waiting(
+async def test_run_agent_loop_seeds_identity_before_first_cycle(
    tmp_path: Any, monkeypatch: pytest.MonkeyPatch
 ) -> None:
    coordinator = AgentCoordinator()
-    await coordinator.register("root", "strix", parent_id=None)
-    await coordinator.register("blocked", "recon", parent_id="root")
-    await coordinator.register("peer_waiter", "recon", parent_id="root")
-    await coordinator.set_status("blocked", "waiting", error="STRIX_LLM guardrail")
-    await coordinator.set_status("peer_waiter", "waiting")
+    await coordinator.register("child", "recon", parent_id="root")
+    session = SQLiteSession("child", tmp_path / "agents.db")

-    parked: dict[str, bool] = {}
+    captured: dict[str, Any] = {}

-    async def _fake_start_child_runner(**kwargs: Any) -> None:
-        parked[kwargs["child_id"]] = bool(kwargs["start_parked"])
+    async def _crash_first_turn(*_args: Any, **kwargs: Any) -> Any:
+        captured["input_data"] = kwargs.get("input_data")
+        captured["items_at_start"] = await session.get_items()
+        raise RuntimeError("first-turn crash")

-    monkeypatch.setattr(execution, "_start_child_runner", _fake_start_child_runner)
+    monkeypatch.setattr(execution, "_run_cycle", _crash_first_turn)

-    await respawn_subagents(
-        coordinator=coordinator,
-        factory=lambda **_kwargs: object(),
-        agents_db_path=tmp_path / "agents.db",
-        sessions_to_close=[],
-        run_config=MagicMock(),
-        max_turns=10,
-        interactive=True,
-        parent_ctx={"agent_id": "root", "parent_id": None},
-        root_id="root",
-    )
+    identity = [{"role": "user", "content": "You are agent recon (abc); maintain your identity."}]
+    with pytest.raises(RuntimeError, match="first-turn crash"):
+        await execution.run_agent_loop(
+            agent=object(),
+            initial_input=identity,
+            run_config=None,  # type: ignore[arg-type]
+            context={"agent_id": "child", "parent_id": "root"},
+            max_turns=5,
+            coordinator=coordinator,
+            agent_id="child",
+            interactive=False,
+            session=session,
+        )

-    assert parked["blocked"] is False
-    assert parked["peer_waiter"] is True
+    # The first cycle ran with an empty input against the pre-seeded session.
+    assert captured["input_data"] == []
+    assert captured["items_at_start"]
+    # The identity/task survives the first-turn crash, so a revival can resume it.
+    stored = await session.get_items()
+    assert any("recon" in str(cast("dict[str, Any]", i).get("content", "")) for i in stored)
+    session.close()
@@ -15,6 +15,7 @@ from openai import (
    RateLimitError,
 )

+from strix.config import codex
 from strix.core import execution
 from strix.core.agents import AgentCoordinator

@@ -55,11 +56,27 @@ def test_server_errors_are_transient() -> None:
        assert execution._is_transient_model_error(_status_error(status)) is True


-def test_rate_limit_is_not_retried_here() -> None:
+def test_rate_limit_is_retried() -> None:
    rate_limited = RateLimitError(
        "slow down", response=httpx.Response(429, request=_request()), body=None
    )
-    assert execution._is_transient_model_error(rate_limited) is False
+    assert execution._is_transient_model_error(rate_limited) is True
+
+
+def test_dns_and_connection_errors_are_transient() -> None:
+    assert execution._is_transient_model_error(OSError("nodename nor servname provided")) is True
+    assert execution._is_transient_model_error(ConnectionError("reset")) is True
+    assert execution._is_transient_model_error(TimeoutError("timed out")) is True
+
+
+def test_content_guardrail_is_not_retried() -> None:
+    guardrail = APIError(
+        "This content was flagged for possible cybersecurity risk",
+        _request(),
+        body=None,
+    )
+    assert codex.is_content_guardrail_error(guardrail) is True
+    assert execution._is_transient_model_error(guardrail) is False


 def test_client_errors_are_not_transient() -> None: