fix(tui-gateway): clear abandoned orphan interrupt claims

tests/tui_gateway/test_ws_orphan_races.py

@@ -100,6 +100,43 @@ def test_obsolete_orphan_cannot_replace_new_detachment(monkeypatch, phase):
     assert timers == [old, newest]
 
 
+def test_orphan_interrupt_claim_clears_when_session_leaves_detached_state(monkeypatch):
+    timers = []
+
+    class Timer:
+        def __init__(self, _delay, callback):
+            self.callback = callback
+            timers.append(self)
+
+        def start(self):
+            pass
+
+        def cancel(self):
+            pass
+
+    sid = "writer-rebound"
+    session = dict(transport=server._detached_ws_transport, running=True)
+    monkeypatch.setattr(server, "_sessions", {sid: session})
+    monkeypatch.setattr(server, "_pending_ws_reaps", {})
+    monkeypatch.setattr(server.threading, "Timer", Timer)
+    monkeypatch.setattr(server, "_WS_ORPHAN_REAP_GRACE_S", 20)
+    monkeypatch.setattr(server, "_WS_ORPHAN_ACTIVITY_STALE_S", 0)
+    monkeypatch.setattr(server, "_session_has_active_delegations", lambda *a: False)
+    monkeypatch.setattr(server, "_interrupt_session_turn", lambda *a, **kw: False)
+
+    server._schedule_ws_orphan_reap(sid)
+    timers[0].callback()
+    assert session["_client_gone_interrupt_requested"]
+
+    session["transport"] = object()
+    timers[1].callback()
+
+    assert "_client_gone_interrupt_requested" not in session
+    assert "_client_gone_interrupt_polls" not in session
+    assert server._reattach_refusal(1, sid, session) is None
+    assert sid not in server._pending_ws_reaps
+
+
 @pytest.mark.parametrize("path", ["unpersisted", "reuse", "eager", "activate", "prompt"])
 @pytest.mark.parametrize("claim", ["already_claimed", "wins_lock", "retired"])
 def test_reconnect_cannot_cross_orphan_interrupt_claim(monkeypatch, path, claim):

tui_gateway/session_lifecycle.py

@@ -519,7 +519,16 @@ def _schedule_ws_orphan_reap(
             if _pending_ws_reaps.get(sid) is not timer:
                 return
             current = _sessions.get(sid)
-            if current is None or not _ws_session_is_detached(current):
+            if current is None:
+                _pending_ws_reaps.pop(sid, None)
+                return
+            if not _ws_session_is_detached(current):
+                # This Timer is abandoning the interrupt claim because another
+                # writer moved the live record off the detached transport.
+                # Do not leave reattach RPCs fenced with 4009, or let this
+                # generation's settlement polls shorten a later detachment.
+                current.pop("_client_gone_interrupt_requested", None)
+                current.pop("_client_gone_interrupt_polls", None)
                 _pending_ws_reaps.pop(sid, None)
                 return
             if _session_has_active_delegations(sid, current):