fix(session): force-cancel runs the abandon machinery before emitting idle

The force branch cleared worker ownership and emitted idle from the
route thread, while the abandon latch and the queue demote ran only
in the stuck worker's own exception handler — a thread force-cancel
abandons precisely because it is not making progress. Subscribers on
the IDLE fan-out therefore saw an operator-forced idle with the latch
unset: the idle observer's operator-Stop gate did not suppress
advice, and wake-eligible entries survived un-demoted, so a nudge
wake could resume a workstream seconds after the operator forced it
to stop. The route now runs the session's abandon machinery first;
the abandoned thread re-running it at its eventual death is
idempotent.

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

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