fix(prism): fail loud on unexpected combined-mode worker exit

In combined mode an unexpected background worker-loop exit (raise OR clean
return, but NOT CancelledError) now logs CRITICAL AND raises SIGTERM so Swarm
restarts the single combined service, instead of leaving a live API with a dead
eval drainer (silent outage). A cancelled task on shutdown stays a silent no-op.
The lifespan shutdown now consumes finished tasks via gather(return_exceptions),
eliminating the cosmetic double-log when a task fails just before shutdown. The
standalone prism-worker CLI (run_worker, resilient=False) is unchanged.

Author: echobt

src/prism_challenge/sdk/app_factory.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import logging
+import signal
 from collections.abc import Awaitable, Callable, Coroutine, Sequence
 from contextlib import asynccontextmanager
 from time import time
@@ -26,11 +27,13 @@ async def close(self) -> None: ...
 
 
 def _log_unexpected_background_exit(task: asyncio.Task[None]) -> None:
-    """Surface an UNEXPECTED background-task exit loudly (never silently swallowed).
+    """Fail loud on an UNEXPECTED background-task exit (never silently swallowed).
 
-    A task cancelled during shutdown is expected (no log). A task that finishes on its own while
-    the service is still up -- with or without an exception -- means the drainer died under a live
-    API (a silent eval outage), so it is logged CRITICAL.
+    A task cancelled during shutdown is expected -> silent no-op. Any other completion --
+    whether it RAISED or RETURNED normally -- means the sole eval-queue drainer died while the
+    co-hosted API keeps serving ``/health`` 200 (a silent eval outage), so we log CRITICAL AND
+    raise SIGTERM so uvicorn runs graceful shutdown and Swarm restarts the single combined
+    service (mirrors agent-challenge's ``_handle_worker_task_done``).
     """
     if task.cancelled():
         return
@@ -39,6 +42,7 @@ def _log_unexpected_background_exit(task: asyncio.Task[None]) -> None:
         _logger.critical("background task exited unexpectedly", exc_info=exc)
     else:
         _logger.critical("background task exited unexpectedly without error")
+    signal.raise_signal(signal.SIGTERM)
 
 
 def create_challenge_app(
@@ -62,13 +66,12 @@ async def lifespan(app: FastAPI):
         finally:
             for task in tasks:
                 task.cancel()
-            for task in tasks:
-                try:
-                    await task
-                except asyncio.CancelledError:
-                    pass
-                except Exception:
-                    _logger.exception("background task crashed during shutdown")
+            if tasks:
+                # gather(return_exceptions=True) awaits every task and consumes its outcome
+                # (CancelledError on clean shutdown, or an error already surfaced by the
+                # done-callback) without re-raising -- so a task that failed just before shutdown
+                # is not logged a second time.
+                await asyncio.gather(*tasks, return_exceptions=True)
             await database.close()
 
     app = FastAPI(title=settings.name, version=settings.version, lifespan=lifespan)

tests/test_combined_mode.py

@@ -2,11 +2,13 @@
 
 import asyncio
 import logging
+import signal
 from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import AsyncMock
 
 import pytest
+from fastapi import APIRouter
 from fastapi.testclient import TestClient
 
 from prism_challenge import worker as worker_module
@@ -26,6 +28,10 @@ def _settings(tmp_path: Path, **overrides: object) -> PrismSettings:
     )
 
 
+async def _empty_weights() -> dict[str, float]:
+    return {}
+
+
 def test_combined_mode_default_is_off() -> None:
     settings = PrismSettings(shared_token="x")
     assert settings.combined_mode is False
@@ -167,9 +173,12 @@ async def fake_sleep(_seconds: float) -> None:
     assert process_next.await_count == 1
 
 
-async def test_log_unexpected_background_exit_logs_on_error(
-    caplog: pytest.LogCaptureFixture,
+async def test_log_unexpected_background_exit_logs_and_signals_on_error(
+    caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch
 ) -> None:
+    raised: list[signal.Signals] = []
+    monkeypatch.setattr(signal, "raise_signal", lambda sig: raised.append(sig))
+
     async def boom() -> None:
         raise RuntimeError("worker died")
 
@@ -182,11 +191,15 @@ async def boom() -> None:
 
     assert any(r.levelno == logging.CRITICAL for r in caplog.records)
     assert any("exited unexpectedly" in r.message for r in caplog.records)
+    assert raised == [signal.SIGTERM]
 
 
-async def test_log_unexpected_background_exit_logs_on_clean_return(
-    caplog: pytest.LogCaptureFixture,
+async def test_log_unexpected_background_exit_logs_and_signals_on_clean_return(
+    caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch
 ) -> None:
+    raised: list[signal.Signals] = []
+    monkeypatch.setattr(signal, "raise_signal", lambda sig: raised.append(sig))
+
     async def clean() -> None:
         return None
 
@@ -197,11 +210,15 @@ async def clean() -> None:
         app_factory._log_unexpected_background_exit(task)
 
     assert any("without error" in r.message for r in caplog.records)
+    assert raised == [signal.SIGTERM]
 
 
 async def test_log_unexpected_background_exit_ignores_cancelled(
-    caplog: pytest.LogCaptureFixture,
+    caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch
 ) -> None:
+    raised: list[signal.Signals] = []
+    monkeypatch.setattr(signal, "raise_signal", lambda sig: raised.append(sig))
+
     async def sleeper() -> None:
         await asyncio.sleep(3600)
 
@@ -215,3 +232,44 @@ async def sleeper() -> None:
         app_factory._log_unexpected_background_exit(task)
 
     assert not [r for r in caplog.records if r.levelno == logging.CRITICAL]
+    assert raised == []
+
+
+def test_combined_shutdown_no_double_log_when_task_fails(
+    caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A background task that dies just before shutdown is surfaced exactly once.
+
+    The done-callback logs CRITICAL + raises SIGTERM; the lifespan shutdown then consumes the
+    already-failed task without re-logging it (no cosmetic double-log).
+    """
+    raised: list[signal.Signals] = []
+    monkeypatch.setattr(signal, "raise_signal", lambda sig: raised.append(sig))
+
+    class _DB:
+        async def init(self) -> None:
+            return None
+
+        async def close(self) -> None:
+            return None
+
+    async def _boom(app: object) -> None:
+        raise RuntimeError("drainer died")
+
+    app = app_factory.create_challenge_app(
+        settings=PrismSettings(shared_token="x"),
+        database=_DB(),
+        public_router=APIRouter(),
+        get_weights_fn=_empty_weights,
+        background_tasks=(_boom,),
+    )
+
+    with caplog.at_level(logging.CRITICAL, logger="prism.sdk.app_factory"):
+        with TestClient(app) as client:
+            assert client.get("/health").status_code == 200
+
+    critical = [r for r in caplog.records if r.levelno == logging.CRITICAL]
+    assert len(critical) == 1
+    assert "exited unexpectedly" in critical[0].message
+    assert "crashed during shutdown" not in caplog.text
+    assert raised == [signal.SIGTERM]