fix(gateway): stop a resuming viewer stealing a running session's stream

tests/test_tui_gateway_ws.py

@@ -228,3 +228,47 @@ async def send_second():
     asyncio.run(scenario())
 
 
+
+
+def _rebind_session(running, current_transport):
+    return {"running": running, "transport": current_transport}
+
+
+def test_resume_does_not_steal_a_running_sessions_transport():
+    """A second viewer resuming a session must not take over its live stream.
+
+    Resume is also how another window / tile / client PEEKS at a session. When
+    it rebound unconditionally, a turn already streaming to viewer A suddenly
+    emitted to viewer B, and A watched its own answer stop mid-sentence.
+    """
+    viewer_a = object()
+    viewer_b = object()
+
+    assert not server._resume_may_rebind_transport(_rebind_session(True, viewer_a), viewer_b)
+
+
+def test_resume_rebinds_an_idle_session():
+    """With no turn in flight there is nothing to interrupt."""
+    viewer_a = object()
+    viewer_b = object()
+
+    assert server._resume_may_rebind_transport(_rebind_session(False, viewer_a), viewer_b)
+
+
+def test_resume_rebinds_a_running_session_whose_viewer_disconnected():
+    """A detached session has no live reader, so the resuming client wins."""
+    viewer_b = object()
+
+    detached = _rebind_session(True, server._detached_ws_transport)
+    assert server._resume_may_rebind_transport(detached, viewer_b)
+
+    on_stdio = _rebind_session(True, server._stdio_transport)
+    assert server._resume_may_rebind_transport(on_stdio, viewer_b)
+
+
+def test_resume_is_idempotent_for_the_same_viewer():
+    """The viewer that already owns the stream may always re-assert it."""
+    viewer_a = object()
+
+    assert server._resume_may_rebind_transport(_rebind_session(True, viewer_a), viewer_a)
+    assert server._resume_may_rebind_transport(_rebind_session(True, None), viewer_a)

tui_gateway/server.py

@@ -8163,6 +8163,33 @@ def _live_visible_history(session: dict, db, in_memory_fallback: list[dict]) ->
     return in_memory_fallback
 
 
+def _resume_may_rebind_transport(session: dict, transport: Transport) -> bool:
+    """Whether a ``session.resume`` may point this session at *transport*.
+
+    Resume is also how a second viewer PEEKS at a session (another window, a
+    tile, a mobile client). Rebinding unconditionally meant that peek stole the
+    stream: a turn already running for viewer A suddenly emitted to viewer B,
+    and A watched its own answer stop mid-sentence.
+
+    So a RUNNING session keeps the transport it is streaming to, unless that
+    transport is gone — a disconnect points detached sessions at
+    ``_detached_ws_transport`` (see ``_reap_or_detach_sessions_for_transport``),
+    and a session parked there has no live reader, so the resuming client is
+    strictly better than nothing. An idle session always rebinds: with no turn
+    in flight there is nothing to interrupt, and the next prompt would rebind
+    anyway.
+    """
+    current = session.get("transport")
+
+    if current is transport or current is None:
+        return True
+
+    if not session.get("running"):
+        return True
+
+    return current is _detached_ws_transport or current is _stdio_transport
+
+
 def _live_session_payload(
     sid: str,
     session: dict,
@@ -8175,7 +8202,7 @@ def _live_session_payload(
     with session["history_lock"]:
         if cols is not None:
             session["cols"] = cols
-        if transport is not None:
+        if transport is not None and _resume_may_rebind_transport(session, transport):
             session["transport"] = transport
         if touch:
             session["last_active"] = time.time()