diff --git a/tests/test_server_authz.py b/tests/test_server_authz.py index 549ce2e0..0dc2dd87 100644 --- a/tests/test_server_authz.py +++ b/tests/test_server_authz.py @@ -1807,6 +1807,41 @@ class TestCompactCommandDispatch: assert not zombie.is_alive() wait_until(lambda: self._drain_idle(ws), timeout=8.0) + def test_force_cancel_runs_abandon_machinery_before_idle_emission(self, app_client): + """The force branch runs the session's abandon machinery BEFORE + clearing ownership and emitting idle. Subscribers on the IDLE + fan-out (the coordinator idle observer's operator-Stop gate, + the wake watcher's queue read) use the latch and the demote to + tell an operator-forced IDLE from a turn reaching idle under + its own power — and the thread force-cancel abandons is stuck + by definition, so its own exception handler cannot be relied + on to have run first.""" + client, mgr = app_client + ws_id = self._create_ws(client) + ws = mgr.get(ws_id) + assert ws is not None + gate = threading.Event() + ws.session.compact_gate = gate + client.post( + "/v1/api/command", + json={"command": "/compact", "ws_id": ws_id}, + headers=_auth("user-1"), + ) + zombie = ws.worker_thread + calls: list[str] = [] + ws.session._drain_pending_advisories = lambda: calls.append("abandon") + ws.ui.on_state_change = lambda s: calls.append(f"state:{s}") + resp = client.post( + f"/v1/api/workstreams/{ws_id}/cancel", + json={"force": True}, + headers=_auth("user-1"), + ) + assert resp.status_code == 200 + assert calls == ["abandon", "state:idle"] + gate.set() + zombie.join(timeout=5) + assert not zombie.is_alive() + def test_ws_close_mid_window_drops_pending_and_drain_exits(self, app_client): """A workstream closed with deferred sends outstanding drops them (documented at-most-once contract) and the drain task retires diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py index 3f0e1b8b..ff0e2438 100644 --- a/turnstone/core/session_routes.py +++ b/turnstone/core/session_routes.py @@ -1400,6 +1400,28 @@ def make_cancel_handler( # escape hatch for an already-wedged session, not a # routine path. Revisit if commands ever gain generation # discipline. + # + # Abandon machinery FIRST — before the ownership clear + # and the idle emission below. The latch and the queue + # demote are how subscribers on the IDLE fan-out (the + # idle observer's operator-Stop gate, the wake watcher) + # tell this operator-forced IDLE apart from a turn + # reaching idle under its own power. The worker's own + # exception handler normally runs this, but the thread + # force-cancel abandons is stuck by definition and may + # not reach that handler for minutes — emitting IDLE + # first let advisories fire and wakes spawn against an + # operator who had just pressed Stop at its hardest. + # The abandoned thread re-running the drain at its + # eventual death is idempotent. + try: + session._drain_pending_advisories() + except Exception: + log.debug( + "ws.cancel.abandon_latch_failed ws=%s", + ws_id[:8], + exc_info=True, + ) with ws._lock: ws.worker_thread = None ws._worker_running = False