fix(tui-gateway): measure reap sleep by clock divergence, re-arm via scheduler

Detect the host sleep by the divergence between the wall clock and the
monotonic clock since arming (the monotonic clock does not advance
while asleep), instead of an absolute monotonic deadline — a timer
that fires without elapsed wait time (tests, spurious dispatch) shows
zero divergence and reaps normally. Re-arm through
_schedule_ws_orphan_reap with the fired timer as the expected one so
the fresh closure's identity guard matches the entry it installs.

tests/tui_gateway/test_ws_orphan_races.py

@@ -294,12 +294,24 @@ def test_ws_orphan_reap_rearms_after_system_sleep(monkeypatch):
 
     sid = "slept-through-grace"
     reaped = []
-    clock = {"now": 1000.0}
+    # Two clocks. monotonic advances only while the host is awake; wall
+    # advances through sleep too. Their divergence is the host sleep time.
+    clocks = {"monotonic": 1000.0, "wall": 5000.0}
+
+    def sleep_host(seconds):
+        clocks["wall"] += seconds  # the wall clock runs during sleep
+        clocks["monotonic"] += 0.0  # mach_absolute_time does not
+
+    def stay_awake(seconds):
+        clocks["wall"] += seconds
+        clocks["monotonic"] += seconds
+
     session = dict(transport=server._detached_ws_transport, running=False)
     monkeypatch.setattr(server, "_sessions", {sid: session})
     monkeypatch.setattr(server, "_pending_ws_reaps", {})
     monkeypatch.setattr(server.threading, "Timer", Timer)
-    monkeypatch.setattr(server.time, "monotonic", lambda: clock["now"])
+    monkeypatch.setattr(server.time, "monotonic", lambda: clocks["monotonic"])
+    monkeypatch.setattr(server.time, "time", lambda: clocks["wall"])
     monkeypatch.setattr(server, "_WS_ORPHAN_REAP_GRACE_S", 20)
     monkeypatch.setattr(server, "_pop_session_by_id", lambda s: reaped.append(s) or session)
 
@@ -309,8 +321,9 @@ def test_ws_orphan_reap_rearms_after_system_sleep(monkeypatch):
 
     # Host sleeps 2s into the grace: the wall-clock wait expires during the
     # sleep and the timer fires at wake with only 2s of awake time elapsed —
-    # spared, re-armed for the remaining 18s.
-    clock["now"] += 2.0
+    # spared, re-armed for the remaining 18s of AWAKE grace.
+    stay_awake(2.0)
+    sleep_host(40.0)
     timers[0].callback()
     assert reaped == []
     assert len(timers) == 2
@@ -318,7 +331,7 @@ def test_ws_orphan_reap_rearms_after_system_sleep(monkeypatch):
     assert server._pending_ws_reaps[sid] is timers[1]
 
     # The re-armed timer runs its full remainder awake: reap proceeds.
-    clock["now"] += 18.0
+    stay_awake(18.0)
     timers[1].callback()
     assert reaped == [sid]
 
@@ -345,17 +358,20 @@ def test_ws_orphan_reap_rearm_spares_post_wake_reconnect(monkeypatch):
 
     sid = "woke-and-reconnected"
     reaped = []
-    clock = {"now": 1000.0}
+    clocks = {"monotonic": 1000.0, "wall": 5000.0}
     session = dict(transport=server._detached_ws_transport, running=False)
     monkeypatch.setattr(server, "_sessions", {sid: session})
     monkeypatch.setattr(server, "_pending_ws_reaps", {})
     monkeypatch.setattr(server.threading, "Timer", Timer)
-    monkeypatch.setattr(server.time, "monotonic", lambda: clock["now"])
+    monkeypatch.setattr(server.time, "monotonic", lambda: clocks["monotonic"])
+    monkeypatch.setattr(server.time, "time", lambda: clocks["wall"])
     monkeypatch.setattr(server, "_WS_ORPHAN_REAP_GRACE_S", 20)
     monkeypatch.setattr(server, "_pop_session_by_id", lambda s: reaped.append(s) or session)
 
     server._schedule_ws_orphan_reap(sid)
-    clock["now"] += 2.0
+    clocks["monotonic"] += 2.0
+    clocks["wall"] += 2.0
+    clocks["wall"] += 40.0  # host sleeps through the wall-clock wait
     timers[0].callback()
     assert len(timers) == 2  # slept through the wait — re-armed
 
@@ -363,6 +379,7 @@ def test_ws_orphan_reap_rearm_spares_post_wake_reconnect(monkeypatch):
     # the re-armed remainder elapses: the reap is a no-op and the chain stops.
     server._cancel_ws_orphan_reap(sid)
     session["transport"] = LiveTransport()
-    clock["now"] += 18.0
+    clocks["monotonic"] += 18.0
+    clocks["wall"] += 18.0
     timers[1].callback()
     assert reaped == []

tui_gateway/server.py

@@ -132,10 +132,10 @@ def _resolve_ws_orphan_reap_grace() -> float:
 
 
 _WS_ORPHAN_REAP_GRACE_S = _resolve_ws_orphan_reap_grace()
-# If the reap timer fires with more than this much of the grace still unelapsed on the
-# monotonic clock, the host slept through the wall-clock wait — re-arm for the remainder
-# instead of reaping (#44183). Big enough to ignore timer jitter and wall-clock NTP
-# nudges, small relative to any real sleep.
+# A reap Timer whose wall-clock wait outlasted its awake-time (monotonic) wait by more than
+# this fired early because the host slept through it (#44183); re-arm for the remaining
+# awake grace instead of reaping. Large enough to ignore timer jitter and NTP slew, small
+# relative to any real sleep.
 _WS_ORPHAN_REAP_SLEEP_SLACK_S = 0.5
 # A detached RUNNING turn is interrupted only once its activity clock (API waits, stream tokens, tool
 # heartbeats) idled this long; 600s = the turn-liveness watchdog so "wedged" means the same. 0 disables.

tui_gateway/session_lifecycle.py

@@ -755,32 +755,36 @@ def _schedule_ws_orphan_reap(
 
     The grace is measured in AWAKE (monotonic) time: ``threading.Timer``'s wait elapses in wall-clock time on
     platforms without a monotonic condvar (macOS lacks ``pthread_condattr_setclock``), so a system sleep makes
-    the timer fire "early" in awake-time terms. Without the monotonic deadline check below, closing a laptop lid
-    for longer than the grace reaped the parked session at the instant of wake — before the Desktop's WS
-    reconnect or ``session.resume`` could re-bind a transport — so every sleep/wake cycle 404'd the open chat
-    (#44183)."""
+    the timer fire "early" in awake-time terms. Without the sleep check below, closing a laptop lid for longer
+    than the grace reaped the parked session at the instant of wake — before the Desktop's WS reconnect or
+    ``session.resume`` could re-bind a transport — so every sleep/wake cycle 404'd the open chat (#44183)."""
     if _WS_ORPHAN_REAP_GRACE_S <= 0:
         return
-    # time.monotonic() (mach_absolute_time / CLOCK_MONOTONIC) does not advance while the
-    # host is asleep, so this deadline measures awake time only.
-    deadline = time.monotonic() + (_WS_ORPHAN_REAP_GRACE_S if delay_s is None else max(0.0, delay_s))
+    grace_s = _WS_ORPHAN_REAP_GRACE_S if delay_s is None else max(0.0, delay_s)
+    # Sample both clocks at arm time: time.monotonic() (mach_absolute_time / CLOCK_MONOTONIC)
+    # does not advance while the host is asleep, so the divergence between the two clocks'
+    # elapsed times at fire time is exactly the time the host spent asleep.
+    armed_monotonic = time.monotonic()
+    armed_wall = time.time()
 
     def _reap() -> None:
-        # The wall-clock timer fired. If the monotonic (awake-time) clock says the grace
-        # hasn't actually elapsed — the host slept through the wait — re-arm for the
-        # remainder so the Desktop reconnect gets its full grace of awake time. The
-        # slack keeps ordinary timer jitter and wall-clock NTP nudges from re-arming
-        # a legitimately-expired reap.
-        remaining = deadline - time.monotonic()
-        if remaining > _WS_ORPHAN_REAP_SLEEP_SLACK_S:
-            with _sessions_lock:
-                if _pending_ws_reaps.get(sid) is not timer:
-                    return
-                rearm = threading.Timer(remaining, _reap)
-                rearm.daemon = True
-                _pending_ws_reaps[sid] = rearm
-            rearm.start()
-            return
+        # The timer fired. If more wall-clock than monotonic time elapsed, the host
+        # slept through the wait: the grace has NOT been granted in awake time, so
+        # re-arm for the remaining awake grace instead of reaping. The slack keeps
+        # ordinary timer jitter and NTP slew from re-arming a legitimately-expired
+        # reap, and a fired-without-elapsed timer (tests, spurious dispatch) shows
+        # zero divergence and reaps normally.
+        slept_s = (time.time() - armed_wall) - (time.monotonic() - armed_monotonic)
+        if slept_s > _WS_ORPHAN_REAP_SLEEP_SLACK_S:
+            rearm_delay = max(0.0, grace_s - (time.monotonic() - armed_monotonic))
+            if rearm_delay <= 0:
+                pass  # no awake grace left — fall through and reap
+            else:
+                # Re-arm through the public scheduler with THIS timer as the expected
+                # one: the fresh closure's identity guard then matches the entry it
+                # installs, so the awake-remainder fire proceeds to the real reap.
+                _schedule_ws_orphan_reap(sid, delay_s=rearm_delay, _expected_timer=timer)
+                return
         # Serialize the re-check against session.resume (rebinds under _session_resume_lock). Claim teardown by popping
         # under both locks, then release the resume lock before slow finalization. Order: resume_lock -> sessions_lock.
         reschedule_delay = interrupt_session = session = None