From fc01bee39c01e70eb0b44857f87d8f3856091bb1 Mon Sep 17 00:00:00 2001 From: Renn F Date: Sun, 28 Jun 2026 23:10:47 +0200 Subject: [PATCH] [F117] stop the orchestrator in lifespan shutdown BEFORE closing the DB MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The lifespan shutdown closed OptimalService + the DB, and only THEN did bootstrap's finally block call orchestrator.stop() — so stop() ran with the DB already closed. stop() drains fire-and-forget _bg_tasks writes (respawn_tracker upserts, audit-log rows) and stop_agent finalizes work sessions / agent state, all needing the DB still open; closing it first silently dropped those final writes (the durable PM-respawn counter's last few strikes, the metrics-bearing audit trail tail). Move orchestrator.stop() into the lifespan shutdown path, BEFORE close_optimal_service + close_db, guarded by a new get_orchestrator_or_none() safe accessor (no crash when no orchestrator is wired — tests, skip_orchestrator). bootstrap's finally-block stop() becomes an idempotent safety net: stop() gains a _stopped flag (getattr-guarded so __new__- constructed test instances still stop) so the double-call is a clean no-op, not a re-stop of already-stopped agents / re-drain of an empty bg set. --- roboco/api/app.py | 20 ++++++- roboco/api/deps.py | 16 ++++++ roboco/runtime/orchestrator.py | 13 +++++ tests/unit/api/test_app.py | 57 +++++++++++++++++++ .../test_orchestrator_shutdown_drain.py | 26 +++++++++ 5 files changed, 131 insertions(+), 1 deletion(-) diff --git a/roboco/api/app.py b/roboco/api/app.py index a9a0a179..6cbf8415 100644 --- a/roboco/api/app.py +++ b/roboco/api/app.py @@ -11,7 +11,7 @@ from contextlib import asynccontextmanager from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware -from roboco.api.deps import _auth_required +from roboco.api.deps import _auth_required, get_orchestrator_or_none from roboco.api.middleware import setup_middleware from roboco.api.routes.a2a import router as a2a_router from roboco.api.routes.a2a import wellknown_router as a2a_wellknown_router @@ -167,6 +167,24 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]: if _AppServices.transcription: await _AppServices.transcription.stop() + # Stop the orchestrator BEFORE closing the DB / OptimalService. stop() + # cancels the background loops, stops the agents (finalizing work sessions + # + agent state via DB writes), and drains fire-and-forget bg writes + # (respawn_tracker upserts, audit-log rows) — all needing the DB still + # open. Closing the DB first silently dropped those final writes (the old + # order, where only bootstrap's finally block called stop() AFTER lifespan + # had already closed the DB). Best-effort: a stop error must not block the + # resource teardown below. No-op when no orchestrator is wired (tests, + # skip_orchestrator). bootstrap's finally block re-calls stop() as a safety + # net; stop() is idempotent (guarded by the _stopped flag) so the second + # call is a no-op. + orchestrator = get_orchestrator_or_none() + if orchestrator is not None: + try: + await orchestrator.stop() + except Exception as e: + logger.warning("Orchestrator stop failed during shutdown", error=str(e)) + # Close Phase 3 services await close_optimal_service() diff --git a/roboco/api/deps.py b/roboco/api/deps.py index 020a59b6..5f0cb59c 100644 --- a/roboco/api/deps.py +++ b/roboco/api/deps.py @@ -93,6 +93,11 @@ def set_orchestrator(orchestrator: AgentOrchestrator) -> None: _ServiceHolder.orchestrator = orchestrator +def clear_orchestrator() -> None: + """Clear the global orchestrator instance (test teardown / re-init).""" + _ServiceHolder.orchestrator = None + + def get_orchestrator() -> AgentOrchestrator: """Get the global orchestrator instance.""" if _ServiceHolder.orchestrator is None: @@ -103,6 +108,17 @@ def get_orchestrator() -> AgentOrchestrator: return _ServiceHolder.orchestrator +def get_orchestrator_or_none() -> AgentOrchestrator | None: + """The global orchestrator instance, or ``None`` if not set. + + Used by the lifespan shutdown path, which must stop the orchestrator + BEFORE closing the DB (orchestrator.stop() drains fire-and-forget DB + writes and finalizes agent state) but must not crash when the app is run + without a bootstrap-set orchestrator (e.g. tests, ``skip_orchestrator``). + """ + return _ServiceHolder.orchestrator + + OrchestratorDep = Annotated[AgentOrchestrator, Depends(get_orchestrator)] diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index 0be4e8b5..03efd40c 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -844,6 +844,10 @@ class AgentOrchestrator: # instead of waiting for the next 30-second tick. self._dispatch_wake: asyncio.Event = asyncio.Event() self._running = False + # Set True once stop() completes — makes the (lifespan + bootstrap + # safety-net) double-call a clean no-op instead of re-stopping already + # stopped agents / re-draining an empty bg-task set. + self._stopped = False self._lock = asyncio.Lock() # Serializes CEO supersede calls so a double-click can't pass the # find_supersede_umbrella dedup check twice and cut two branches / @@ -1023,6 +1027,14 @@ class AgentOrchestrator: async def stop(self) -> None: """Stop the orchestrator and all agents.""" + if getattr(self, "_stopped", False): + # Idempotent: the lifespan shutdown path stops the orchestrator + # before closing the DB, and bootstrap's finally block re-calls + # stop() as a safety net. The second call must be a no-op, not a + # re-stop of already-stopped agents. ``getattr`` so a ``__new__``- + # constructed instance (unit-test pattern) without ``__init__`` is + # still stoppable. + return self._running = False # Cancel every background loop, then stop the agents. @@ -1057,6 +1069,7 @@ class AgentOrchestrator: # stuck task can't hang shutdown — it is cancelled past the deadline. await self._drain_bg_tasks() + self._stopped = True logger.info("Orchestrator stopped") async def _ensure_agent_image(self, agent_id: str | None = None) -> None: diff --git a/tests/unit/api/test_app.py b/tests/unit/api/test_app.py index 31b13952..c1d7784d 100644 --- a/tests/unit/api/test_app.py +++ b/tests/unit/api/test_app.py @@ -16,6 +16,7 @@ import pytest from fastapi import FastAPI from roboco.api.app import app as default_app from roboco.api.app import create_app, lifespan +from roboco.api.deps import clear_orchestrator, set_orchestrator if TYPE_CHECKING: from collections.abc import AsyncIterator @@ -144,6 +145,62 @@ async def test_lifespan_startup_and_shutdown_happy_path() -> None: transcription_mock.stop.assert_awaited_once() +@pytest.mark.asyncio +async def test_lifespan_stops_orchestrator_before_closing_db_and_optimal() -> None: + """F117: orchestrator.stop() must run BEFORE close_optimal_service / close_db + on shutdown. stop() drains fire-and-forget DB writes (respawn_tracker + upserts, audit-log rows) and stop_agent finalizes work sessions / agent + state — all needing the DB still open. Closing the DB first (the old order, + where only bootstrap's finally called stop() after lifespan had already + closed the DB) silently dropped those final writes.""" + order: list[str] = [] + + def _record(label: str) -> AsyncMock: + async def _fn() -> None: + order.append(label) + + return AsyncMock(side_effect=_fn) + + transcription_mock = MagicMock() + transcription_mock.start = AsyncMock() + transcription_mock.stop = AsyncMock() + orchestrator_mock = MagicMock() + orchestrator_mock.stop = _record("orchestrator.stop") + + set_orchestrator(orchestrator_mock) + try: + with ( + patch("roboco.api.app.init_db", new=AsyncMock()), + patch("roboco.api.app.close_db", new=_record("close_db")), + patch( + "roboco.api.app.close_optimal_service", + new=_record("close_optimal_service"), + ), + patch( + "roboco.api.app.TranscriptionService", return_value=transcription_mock + ), + patch("roboco.api.app.ExtractionService"), + patch("roboco.api.app.ExtractionPipeline"), + patch( + "roboco.api.app.get_optimal_service", + new=AsyncMock(return_value=MagicMock()), + ), + ): + app = create_app() + async with lifespan(app): + pass + finally: + # Clear the global so it doesn't leak into other tests. + clear_orchestrator() + + # orchestrator.stop() ran, and it ran BEFORE close_optimal_service + close_db. + assert "orchestrator.stop" in order + assert order.index("orchestrator.stop") < order.index("close_optimal_service") + assert order.index("orchestrator.stop") < order.index("close_db") + # The DB is still the last thing closed (innermost resource). + assert order.index("close_optimal_service") < order.index("close_db") + + @pytest.mark.asyncio async def test_lifespan_handles_optimal_init_failure_gracefully() -> None: """Optimal-service init failure → app.state.optimal=None, no raise.""" diff --git a/tests/unit/runtime/test_orchestrator_shutdown_drain.py b/tests/unit/runtime/test_orchestrator_shutdown_drain.py index 628037a4..2320f996 100644 --- a/tests/unit/runtime/test_orchestrator_shutdown_drain.py +++ b/tests/unit/runtime/test_orchestrator_shutdown_drain.py @@ -148,3 +148,29 @@ async def test_stop_failing_agent_does_not_skip_drain() -> None: await asyncio.wait_for(orch.stop(), timeout=5.0) assert ran == [True], "failing stop_agent skipped the drain (data lost)" + + +@pytest.mark.asyncio +async def test_stop_is_idempotent_double_call_is_noop() -> None: + """F117: stop() is idempotent. The lifespan shutdown path now stops the + orchestrator before closing the DB, and bootstrap's finally block re-calls + stop() as a safety net. The second call must be a clean no-op — not a + re-drain, not a re-stop of already-stopped agents — guarded by ``_stopped``.""" + orch = _make_orchestrator() + real_drain = orch._drain_bg_tasks + drain_calls = 0 + + async def counting_drain() -> None: + nonlocal drain_calls + drain_calls += 1 + await real_drain() + + orch._drain_bg_tasks = counting_drain + + await orch.stop() + assert drain_calls == 1, "first stop() drained the bg tasks" + assert orch._stopped is True + + await orch.stop() # safety-net double-call (lifespan already stopped it) + assert drain_calls == 1, "second stop() must not re-drain (idempotent no-op)" + assert orch._stopped is True