From e15c4e3415e04cba252c732b347858062ac09721 Mon Sep 17 00:00:00 2001 From: Renn F Date: Sun, 28 Jun 2026 18:09:47 +0200 Subject: [PATCH] [F071] abort non-blocking intake/secretary spawn on mid-spawn shutdown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The non-blocking spawn (start_intake_session / start_secretary_session) schedules _spawn_intake_container_guarded / _spawn_secretary_container_guarded via _schedule_bg. Those run docker run and only register in _instances at the END. If shutdown arrived between docker run and the registration line, the container was started but the orchestrator had no handle — stop() iterates only _instances, so the container was orphaned (leaked, manual docker rm). Worse, the F070 drain could let the spawn coroutine complete the registration AFTER stop() already iterated _instances, landing a live container into a shutting-down registry nothing tears down. Add a post-docker-run shutdown guard in _spawn_intake_container and _spawn_secretary_container: re-check self._running after _run_container_cmd returns; if the orchestrator began shutting down, remove the just-started container (by its deterministic name) and raise _SpawnAbortedDuringShutdown WITHOUT registering. The guarded wrappers catch that BEFORE except Exception and close the live relay silently (shutdown is not a user-facing failure, no error pushed to the SSE stream). The F070 stop() drain awaits the bg spawn coroutine, so the abort surfaces cleanly. TOCTOU-safe: between the _running check and the _instances assignment there is no await (config + instance construction are sync), so once the check passes, registration completes before the event loop can interleave stop(). The normal running path is unchanged (sanity tests pin it). --- roboco/runtime/orchestrator.py | 44 +++++ tests/unit/runtime/test_intake_spawn.py | 98 +++++++++++ .../runtime/test_secretary_spawn_shutdown.py | 165 ++++++++++++++++++ 3 files changed, 307 insertions(+) create mode 100644 tests/unit/runtime/test_secretary_spawn_shutdown.py diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index 7a80b83c..e82da362 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -744,6 +744,20 @@ class AgentReadinessError(Exception): """ +class _SpawnAbortedDuringShutdown(Exception): + """Raised when a non-blocking intake/secretary spawn completes ``docker run`` + after the orchestrator began shutting down. + + The raiser has already removed the just-started container (so it isn't + orphaned); the guarded wrapper catches this BEFORE its generic + ``except Exception`` and closes the live relay silently — shutdown is not a + user-facing failure, so no error is pushed to the SSE stream. The F070 + ``stop()`` drain awaits the bg spawn coroutine, so this surfaces cleanly + instead of the registration landing a live container into a registry that + ``stop()`` has already finished iterating. + """ + + class AgentOrchestrator: """ Manages Claude Code containers for all agents. @@ -3421,6 +3435,12 @@ class AgentOrchestrator: project_ids=project_ids, initial_message=initial_message, ) + except _SpawnAbortedDuringShutdown: + # Shutdown began mid-spawn; the just-started container was already + # removed by the raiser. Close the relay silently — shutdown is not a + # user-facing failure, so no error is pushed to the SSE stream. + get_live_registry().close(session_id) + return except Exception as exc: logger.error( "Intake container spawn failed", session_id=session_id, error=str(exc) @@ -3500,6 +3520,16 @@ class AgentOrchestrator: ) container_id = await self._run_container_cmd(cmd) + # Shutdown may have begun while this (non-blocking) spawn was in flight + # — the bg coroutine runs concurrently with stop(). If so, remove the + # just-started container and abort WITHOUT registering: stop()'s + # _instances iteration has already run (or is running), so a registration + # now would land a live container nothing tears down (the orphan). The + # stop() drain awaits this coroutine, so the abort surfaces cleanly. + if not self._running: + await self._remove_container(container_name) + raise _SpawnAbortedDuringShutdown(INTAKE_AGENT_ID) + config = AgentConfig( agent_id=INTAKE_AGENT_ID, blueprint_path=prompt_path, @@ -3595,6 +3625,12 @@ class AgentOrchestrator: await self._spawn_secretary_container( session_id, initial_message=initial_message ) + except _SpawnAbortedDuringShutdown: + # Shutdown began mid-spawn; the just-started container was already + # removed by the raiser. Close the relay silently — shutdown is not + # a user-facing failure, so no error is pushed to the SSE stream. + get_live_registry().close(session_id) + return except Exception as exc: logger.error( "Secretary container spawn failed", @@ -3666,6 +3702,14 @@ class AgentOrchestrator: ) container_id = await self._run_container_cmd(cmd) + # Shutdown may have begun while this (non-blocking) spawn was in flight + # — see the matching guard in _spawn_intake_container. Remove the + # just-started container and abort WITHOUT registering, so it isn't + # orphaned by a stop() that has already iterated _instances. + if not self._running: + await self._remove_container(container_name) + raise _SpawnAbortedDuringShutdown(SECRETARY_AGENT_ID) + config = AgentConfig( agent_id=SECRETARY_AGENT_ID, blueprint_path=prompt_path, diff --git a/tests/unit/runtime/test_intake_spawn.py b/tests/unit/runtime/test_intake_spawn.py index 88d49be8..b8668f07 100644 --- a/tests/unit/runtime/test_intake_spawn.py +++ b/tests/unit/runtime/test_intake_spawn.py @@ -32,6 +32,11 @@ def _make_minimal_orchestrator() -> AgentOrchestrator: orch = AgentOrchestrator.__new__(AgentOrchestrator) orch._instances = {} orch._bg_tasks = set() + # A minimal orchestrator is a RUNNING one. The non-blocking spawn path + # reads ``self._running`` after docker run to detect a mid-spawn shutdown + # (F071); without this the post-docker-run guard would AttributeError on + # the constructor-skipped instance. + orch._running = True return orch @@ -506,3 +511,96 @@ class TestDeliverWhenReady: await orch._deliver_when_ready("sess-y", "hi", attempts=5, delay=0) assert attempts["n"] == succeed_on # stopped as soon as delivery succeeded + + +# --------------------------------------------------------------------------- +# F071 — non-blocking intake spawn must not orphan a container if shutdown +# arrives between ``docker run`` and the _instances registration. The guarded +# wrapper runs concurrently with stop(); without a post-docker-run shutdown +# check, the just-started container is never recorded in _instances (which +# stop() already iterated) so nothing tears it down — a leaked container. +# --------------------------------------------------------------------------- + + +class TestSpawnIntakeShutdownNoOrphan: + @pytest.mark.asyncio + async def test_shutdown_mid_spawn_removes_container_and_skips_registration( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """docker run completes, THEN the orchestrator begins shutting down + (``_running`` flips to False) before the registration line. The just- + started container must be removed and NOT registered — otherwise it is + orphaned (live, untracked by stop()).""" + orch = _make_minimal_orchestrator() + run_calls: list[list[str]] = [] + _wire_spawn_mocks(monkeypatch, orch, run_calls) + removed: list[str] = [] + + async def _remove(name: str) -> None: + removed.append(name) + + async def _run(cmd: list[str]) -> str: + run_calls.append(cmd) + # Shutdown arrives AFTER docker run started the container but BEFORE + # the registration line runs. + orch._running = False + return "containerid0123456789" + + monkeypatch.setattr(orch, "_run_container_cmd", _run) + monkeypatch.setattr(orch, "_remove_container", _remove) + + registry = prompter_live.get_live_registry() + pushed: list[tuple[str, dict[str, Any]]] = [] + closed: list[str] = [] + monkeypatch.setattr(registry, "push", lambda sid, ev: pushed.append((sid, ev))) + monkeypatch.setattr(registry, "close", closed.append) + registry.open("sess-orphan", INTAKE_AGENT_ID) + + await orch._spawn_intake_container_guarded( + "sess-orphan", + project_slug="roboco", + product_id=None, + initial_message=None, + ) + + # The just-started container was removed by name (not orphaned). Two + # removes: the pre-spawn reap of any stale container, then the + # post-docker-run shutdown guard reaping the just-started one. Without + # the guard there is only ONE remove (the pre-spawn reap) and the + # just-started container is orphaned — so asserting two proves the guard + # ran. + assert removed == [ + f"roboco-agent-{INTAKE_AGENT_ID}", + f"roboco-agent-{INTAKE_AGENT_ID}", + ] + # No instance registered — stop()'s _instances iteration has already + # run, so a registration now would land a live container nothing stops. + assert INTAKE_AGENT_ID not in orch._instances + # Shutdown is not a user-facing failure: the relay closes silently, + # no error pushed to the SSE stream. + assert pushed == [] + assert closed == ["sess-orphan"] + + @pytest.mark.asyncio + async def test_running_spawn_registers_normally( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Sanity: when the orchestrator stays running, the spawn registers the + instance as before — the shutdown guard does not fire on a healthy spawn.""" + orch = _make_minimal_orchestrator() + run_calls: list[list[str]] = [] + _wire_spawn_mocks(monkeypatch, orch, run_calls) + # _wire_spawn_mocks' _remove_container is a no-op; override to record. + removed: list[str] = [] + + async def _remove(name: str) -> None: + removed.append(name) + + monkeypatch.setattr(orch, "_remove_container", _remove) + + instance = await orch.spawn_intake_session("sess-ok", project_slug="roboco") + + assert orch._instances[INTAKE_AGENT_ID] is instance + # The pre-spawn reap remove is the only remove call (the shutdown guard + # did NOT remove the just-started container). + assert removed == [f"roboco-agent-{INTAKE_AGENT_ID}"] diff --git a/tests/unit/runtime/test_secretary_spawn_shutdown.py b/tests/unit/runtime/test_secretary_spawn_shutdown.py new file mode 100644 index 00000000..e46b0d31 --- /dev/null +++ b/tests/unit/runtime/test_secretary_spawn_shutdown.py @@ -0,0 +1,165 @@ +"""F071 — the Secretary non-blocking spawn (``start_secretary_session`` → +``_schedule_bg(_spawn_secretary_container_guarded)``) runs ``docker run`` and +only registers the instance in ``_instances`` at the END. If shutdown arrives +between ``docker run`` and the registration line, the container is started but +the orchestrator has no handle to it — ``stop()`` iterates only ``_instances``, +so the container is orphaned (leaked, must be cleaned up with ``docker rm``). +Worse, the F070 drain can let the spawn coroutine COMPLETE the registration +AFTER ``stop()`` already iterated ``_instances``, landing a live container into +a shutting-down registry that nothing tears down. + +The fix: after ``docker run`` returns the container id, re-check ``self._running`` +and, if the orchestrator began shutting down, remove the just-started container +and abort WITHOUT registering. The guarded wrapper closes the relay silently +(shutdown is not a user-facing failure). +""" + +from __future__ import annotations + +from pathlib import Path +from types import SimpleNamespace +from typing import Any +from unittest.mock import patch +from uuid import UUID + +import pytest +from roboco.runtime.orchestrator import ( + SECRETARY_AGENT_ID, + AgentOrchestrator, +) +from roboco.services import prompter_live + + +def _make_orchestrator() -> AgentOrchestrator: + """AgentOrchestrator with constructor I/O skipped; a RUNNING minimal one.""" + with patch.object(AgentOrchestrator, "__init__", return_value=None): + orch = AgentOrchestrator.__new__(AgentOrchestrator) + orch._instances = {} + orch._bg_tasks = set() + orch._running = True + return orch + + +def _wire_secretary_spawn_mocks( + monkeypatch: pytest.MonkeyPatch, + orch: AgentOrchestrator, + removed: list[str], + *, + flip_running_on_run: bool, +) -> None: + """Patch every external boundary _spawn_secretary_container touches.""" + + async def _noop(*_a: Any, **_k: Any) -> None: + return None + + async def _route(_aid: str) -> Any: + return SimpleNamespace( + provider_type=SimpleNamespace(value="anthropic"), + model_name="opus", + base_url=None, + auth_token=None, + ) + + async def _run(_cmd: list[str]) -> str: + if flip_running_on_run: + # Shutdown arrives AFTER docker run started the container but BEFORE + # the registration line runs. + orch._running = False + return "containerid0123456789" + + async def _remove(name: str) -> None: + removed.append(name) + + monkeypatch.setattr( + orch, + "_generate_composed_prompt", + lambda *_a, **_k: Path("/tmp/secretary-prompt.md"), + ) + monkeypatch.setattr(orch, "_resolve_agent_route", _route) + monkeypatch.setattr(orch, "_ensure_agent_image", _noop) + monkeypatch.setattr(orch, "_remove_container", _remove) + monkeypatch.setattr(orch, "_run_container_cmd", _run) + monkeypatch.setattr( + orch, + "_resolve_secretary_host_paths", + lambda: {"claude": "/h/.claude", "prompt": "/h/p.md"}, + ) + monkeypatch.setattr(orch, "_record_spawn_session", _noop) + monkeypatch.setattr(orch, "_fire_audit", lambda **_k: None) + # issue_agent_token + AGENTS are imported INSIDE _spawn_secretary_container; + # patch them at their source so the spec construction doesn't touch crypto / + # the real agent table. + monkeypatch.setattr( + "roboco.agents_config.issue_agent_token", lambda *_a, **_k: "tok" + ) + monkeypatch.setattr( + "roboco.foundation.identity.AGENTS", + { + SECRETARY_AGENT_ID: SimpleNamespace( + uuid=UUID("00000000-0000-0000-0000-000000000001") + ) + }, + ) + + +@pytest.fixture(autouse=True) +def _fresh_registry() -> None: + """Isolate the process-wide live registry per test.""" + prev = prompter_live._RegistryHolder.instance + prompter_live._RegistryHolder.instance = prompter_live.PrompterLiveRegistry() + yield + prompter_live._RegistryHolder.instance = prev + + +@pytest.mark.asyncio +async def test_shutdown_mid_spawn_removes_container_and_skips_registration( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """docker run completes, THEN ``_running`` flips to False before the + registration line. The just-started Secretary container must be removed and + NOT registered — otherwise it is orphaned (live, untracked by stop()).""" + orch = _make_orchestrator() + removed: list[str] = [] + _wire_secretary_spawn_mocks(monkeypatch, orch, removed, flip_running_on_run=True) + + registry = prompter_live.get_live_registry() + pushed: list[tuple[str, dict[str, Any]]] = [] + closed: list[str] = [] + monkeypatch.setattr(registry, "push", lambda sid, ev: pushed.append((sid, ev))) + monkeypatch.setattr(registry, "close", closed.append) + registry.open("sess-sec-orphan", SECRETARY_AGENT_ID) + + await orch._spawn_secretary_container_guarded( + "sess-sec-orphan", initial_message=None + ) + + # Two removes: the pre-spawn reap of any stale container, then the + # post-docker-run shutdown guard reaping the just-started one. Without the + # guard there is only ONE remove and the just-started container is orphaned. + assert removed == [ + f"roboco-agent-{SECRETARY_AGENT_ID}", + f"roboco-agent-{SECRETARY_AGENT_ID}", + ] + # No instance registered — stop()'s _instances iteration has already run. + assert SECRETARY_AGENT_ID not in orch._instances + # Shutdown is not a user-facing failure: relay closes silently, no error. + assert pushed == [] + assert closed == ["sess-sec-orphan"] + + +@pytest.mark.asyncio +async def test_running_spawn_registers_normally( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Sanity: when the orchestrator stays running, the Secretary spawn + registers the instance as before — the shutdown guard does not fire.""" + orch = _make_orchestrator() + removed: list[str] = [] + _wire_secretary_spawn_mocks(monkeypatch, orch, removed, flip_running_on_run=False) + + instance = await orch.spawn_secretary_session("sess-sec-ok", initial_message=None) + + assert orch._instances[SECRETARY_AGENT_ID] is instance + # Only the pre-spawn reap remove — the shutdown guard did NOT remove the + # just-started container (the orchestrator stayed running). + assert removed == [f"roboco-agent-{SECRETARY_AGENT_ID}"]