server: own final session sweep

docs/development/07-transports-and-production.md

@@ -186,6 +186,9 @@ task identity.
 HTTP site and runner cleanup stages use a third named cohort plus a stage-keyed
 retry ledger, preventing a later `stop()` from racing a second cleanup call
 against an operation that survived the prior hard deadline.
+The final `SessionManager.stop_all(force=True)` sweep is named and scope-owned
+as well, so a cancellation-resistant Session remains attached to the server's
+cleanup owner after the hard deadline records an incomplete stop.
 
 Read [`server/transports.py`](../../src/easycat/server/transports.py) for
 `CapacityGate` and hard-timeout helpers, and

src/easycat/server/voice_server.py

@@ -81,6 +81,8 @@
 _WS_HANDLER_COHORT = "voice-server-ws-handler"
 _LISTENER_CLEANUP_TASK = "voice_server_listener_cleanup"
 _LISTENER_CLEANUP_COHORT = "voice-server-listener-cleanup"
+_SESSION_SWEEP_TASK = "voice_server_session_sweep"
+_SESSION_SWEEP_COHORT = "voice-server-session-sweep"
 
 
 class VoiceServer:
@@ -112,6 +114,14 @@ def __init__(
         # Bare session registry only: add/remove/stop_all/connection. Capacity
         # and draining are NOT attributed to it (it has neither).
         self._manager: SessionManager[int] = SessionManager()
+        self._session_sweep_task_scope = RuntimeTaskScope(
+            owner_label="voice-server-session-sweep",
+            member_name=_SESSION_SWEEP_TASK,
+            cohort=_SESSION_SWEEP_COHORT,
+            logger=logger,
+            failure_message="VoiceServer SessionManager sweep task failed",
+            drop_if_closed=False,
+        )
 
         # Shared capacity + draining collaborator (the M5 lift). It owns the
         # reservation counter, the active-connection set, and the draining flag
@@ -605,7 +615,11 @@ async def _stop_unlocked(self, *, force: bool = False) -> None:
         # retries it after the handler has unwound. Bound the sweep with
         # ``force_shutdown_timeout_s`` so a force-stop that never returns cannot
         # block server teardown.
-        sweep_task = asyncio.create_task(self._manager.stop_all(force=True))
+        sweep_task = self._session_sweep_task_scope.create_task(
+            self._manager.stop_all(force=True),
+            task_name="easycat-voice-server-session-sweep",
+        )
+        assert sweep_task is not None
         sweep_succeeded, swept = await self._attempt_cleanup(
             "SessionManager hard sweep",
             _await_with_hard_timeout(
@@ -620,6 +634,7 @@ async def _stop_unlocked(self, *, force: bool = False) -> None:
                 report=sweep_task.result() if swept else None,
                 cleanup_errors=cleanup_errors,
             )
+        await self._session_sweep_task_scope.release_standalone_if_empty()
 
         # Keep session/resource references when any cleanup stage failed so a
         # later stop can retry them. The gate itself is always reset to a

tests/ratchets/source-baseline.json

@@ -1,12 +1,12 @@
 {
   "version": 1,
-  "rationale": "VoiceServer cancellation-resistant listener cleanup now uses named RuntimeTaskScope ownership.",
+  "rationale": "VoiceServer final SessionManager hard sweeps now use named RuntimeTaskScope ownership.",
   "counts": {
     "cancelled_error_handler": 100,
     "epoch_field": 17,
     "gather_return_exceptions": 22,
     "module_task_set": 1,
-    "raw_task_spawn": 30,
+    "raw_task_spawn": 29,
     "shield_loop": 4,
     "task_cancelling": 51
   },
@@ -165,7 +165,6 @@
     "raw_task_spawn\tserver/transports.py\t_await_with_hard_timeout\tasyncio.ensure_future\t2197525e3944cdd4\t0",
     "raw_task_spawn\tserver/transports.py\t_safe_await\tasyncio.ensure_future\t2197525e3944cdd4\t0",
     "raw_task_spawn\tserver/transports.py\tclose_websocket_connections\tasyncio.ensure_future\td0b144f338429d6b\t0",
-    "raw_task_spawn\tserver/voice_server.py\tVoiceServer._stop_unlocked\tasyncio.create_task\t17a66ec385918dba\t0",
     "raw_task_spawn\tsession/_journal_sink.py\tSessionJournalSink.append_record_async\tasyncio.create_task\t933bcec6345d252c\t0",
     "raw_task_spawn\tsession/_session.py\tSession._cut_off_turn_playback\tasyncio.create_task\t002fb7c8f4960eaf\t0",
     "raw_task_spawn\tsession/_session.py\tSession._finish_interrupted_start\tasyncio.create_task\tc3c0859252009636\t0",

tests/server/test_voice_server_lifecycle.py

@@ -650,6 +650,7 @@ async def stop_site() -> None:
 
 async def test_failed_session_hard_sweep_blocks_restart_and_retains_retry_ownership(
     caplog: pytest.LogCaptureFixture,
+    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     class FailingSession(_FakeSession):
         def __init__(self) -> None:
@@ -671,6 +672,16 @@ async def stop(self, *, force: bool = False) -> None:
     server._active_session_objs[key] = session
     assert server._gate.try_acquire()
     server._gate.track(key)
+    sweep_task_names: list[str] = []
+    stop_all = server._manager.stop_all
+
+    async def named_stop_all(*, force: bool = False) -> object:
+        current = asyncio.current_task()
+        assert current is not None
+        sweep_task_names.append(current.get_name())
+        return await stop_all(force=force)
+
+    monkeypatch.setattr(server._manager, "stop_all", named_stop_all)
 
     with caplog.at_level(logging.ERROR), pytest.raises(RuntimeError, match="retained 1 session"):
         await server.stop(force=True)
@@ -679,6 +690,8 @@ async def stop(self, *, force: bool = False) -> None:
     assert "session teardown failed" in caplog.text
     assert server._manager.get(key) is session
     assert server._active_session_objs == {key: session}
+    assert sweep_task_names == ["easycat-voice-server-session-sweep"]
+    assert server._session_sweep_task_scope.tasks() == ()
     assert isinstance(server._lifecycle_cleanup_error, RuntimeError)
     with pytest.raises(RuntimeError, match="previous teardown cleanup is incomplete"):
         await server.start()
@@ -692,6 +705,47 @@ async def stop(self, *, force: bool = False) -> None:
     assert server._lifecycle_cleanup_error is None
 
 
+async def test_hard_sweep_timeout_keeps_named_task_owned_until_settlement(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    server = _idle_server(
+        enable_websocket=False,
+        enable_webrtc=False,
+        force_shutdown_timeout_s=0.01,
+    )
+    server._started = True
+    sweep_started = asyncio.Event()
+    cancellation_seen = asyncio.Event()
+    release_sweep = asyncio.Event()
+
+    async def stop_all(*, force: bool = False) -> object:
+        assert force is True
+        sweep_started.set()
+        while not release_sweep.is_set():
+            try:
+                await release_sweep.wait()
+            except asyncio.CancelledError:
+                cancellation_seen.set()
+        return object()
+
+    monkeypatch.setattr(server._manager, "stop_all", stop_all)
+
+    with pytest.raises(RuntimeError, match="SessionManager.stop_all did not finish"):
+        await asyncio.wait_for(server.stop(force=True), timeout=0.5)
+
+    assert sweep_started.is_set()
+    assert cancellation_seen.is_set()
+    tasks = server._session_sweep_task_scope.tasks()
+    assert len(tasks) == 1
+    sweep_task = tasks[0]
+    assert sweep_task.get_name() == "easycat-voice-server-session-sweep"
+
+    release_sweep.set()
+    await asyncio.wait_for(sweep_task, timeout=0.5)
+    await server._session_sweep_task_scope.release_standalone_if_empty()
+    assert server._session_sweep_task_scope.tasks() == ()
+
+
 async def test_cancelled_stop_publishes_retryable_stopped_state_before_reraising() -> None:
     server = _idle_server(enable_websocket=False, enable_webrtc=False)
     site_stop_started = asyncio.Event()