diff --git a/tests/test_tui_gateway_ws.py b/tests/test_tui_gateway_ws.py index 096f5a4c2c61..834935982c42 100644 --- a/tests/test_tui_gateway_ws.py +++ b/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) diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 14d6dbc2f2fa..11046700a3a0 100644 --- a/tui_gateway/server.py +++ b/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()