fix: preserve terminal handoff across idle sidebar refresh

(cherry picked from commit 85f1c2d34b62b0ade85f51ed2cf8c1cf56f5ab9e)

tests/browser_conversation_lifecycle.py

@@ -267,8 +267,17 @@ def _start_webui_server(repo_root: Path, env: dict, artifact_dir: Path):
         suffix = "" if attempts == 1 else f"-attempt-{attempt + 1}"
         log_path = artifact_dir / f"server{suffix}.log"
         log = log_path.open("w", encoding="utf-8")
+        command = [sys.executable, str(repo_root / "server.py")]
+        barrier_dir = env.get("HERMES_TEST_TERMINAL_BARRIER_DIR")
+        if barrier_dir:
+            command = [
+                sys.executable,
+                str(repo_root / "tests" / "sidebar_terminal_barrier_server.py"),
+                str(repo_root),
+                barrier_dir,
+            ]
         proc = subprocess.Popen(
-            [sys.executable, str(repo_root / "server.py")],
+            command,
             cwd=repo_root,
             env=run_env,
             stdout=log,
@@ -286,6 +295,49 @@ def _start_webui_server(repo_root: Path, env: dict, artifact_dir: Path):
     raise RuntimeError(f"WebUI server did not become healthy{detail}")
 
 
+def _exercise_sidebar_terminal_handoff(page, barrier_dir: Path, artifact_dir: Path):
+    """Order the real idle-list response before the real terminal SSE write."""
+    deadline = time.monotonic() + 10
+    while not (barrier_dir / "ready").exists():
+        if time.monotonic() >= deadline:
+            raise AssertionError("terminal HTTP barrier was not reached")
+        page.wait_for_timeout(10)
+    observation = page.evaluate(
+        """async () => {
+          const sid = S.session.session_id;
+          const stream = S.activeStreamId;
+          const inflight = INFLIGHT[sid];
+          const source = LIVE_STREAMS[sid] && LIVE_STREAMS[sid].source;
+          const registries = window._liveAnchorRegistries;
+          const registry = registries && registries.get(stream);
+          await renderSessionList();
+          await new Promise(resolve => requestAnimationFrame(() => requestAnimationFrame(resolve)));
+          const live = document.querySelector('#liveAssistantTurn');
+          return {
+            serverIdle: _allSessions.some(row => row.session_id === sid && _isServerIdleSessionRow(row)),
+            paneRetained: S.busy === true && S.activeStreamId === stream,
+            inflightRetained: !!inflight && INFLIGHT[sid] === inflight,
+            sourceRetained: !!source && source.readyState === EventSource.OPEN &&
+              LIVE_STREAMS[sid] && LIVE_STREAMS[sid].source === source,
+            registryRetained: !!registry && registries.get(stream) === registry,
+            liveRows: live ? live.querySelectorAll('[data-anchor-scene-row="1"]').length : 0,
+          };
+        }"""
+    )
+    (artifact_dir / "sidebar-before-terminal.json").write_text(
+        json.dumps(observation, indent=2), encoding="utf-8"
+    )
+    page.screenshot(path=str(artifact_dir / "sidebar-before-terminal.png"), full_page=True)
+    for invariant in ("serverIdle", "paneRetained", "inflightRetained", "sourceRetained", "registryRetained"):
+        assert observation[invariant] is True, {
+            "message": "idle sidebar response revoked the live terminal handoff",
+            "invariant": invariant,
+            "observation": observation,
+        }
+    assert observation["liveRows"] > 0, observation
+    (barrier_dir / "release").touch()
+
+
 class DeterministicGateway:
     """A localhost-only Gateway Runs server with test-controlled phase gates."""
 
@@ -782,6 +834,12 @@ def main() -> int:
         "NO_PROXY": "127.0.0.1,localhost",
         "no_proxy": "127.0.0.1,localhost",
     })
+    env.pop("HERMES_TEST_TERMINAL_BARRIER_DIR", None)
+    terminal_barrier_dir = None
+    if TEST_BITE == "settle-worklog-frame-proof" and scenario == "normal":
+        terminal_barrier_dir = state_dir / "terminal-barrier"
+        terminal_barrier_dir.mkdir()
+        env["HERMES_TEST_TERMINAL_BARRIER_DIR"] = str(terminal_barrier_dir)
     proc = None
     log = None
     log_path = None
@@ -941,6 +999,8 @@ def main() -> int:
                 timeout=10000,
             )
             gateway.release_terminal.set()
+            if terminal_barrier_dir is not None:
+                _exercise_sidebar_terminal_handoff(page, terminal_barrier_dir, artifact_dir)
             page.wait_for_function(
                 """text => typeof S !== 'undefined' && S.busy === false && !S.activeStreamId &&
                   ((document.querySelector('#msgInner') || {}).innerText || '').includes(text)""",
@@ -1176,6 +1236,8 @@ def main() -> int:
         exit_code = 1
         return 1
     finally:
+        if terminal_barrier_dir is not None:
+            (terminal_barrier_dir / "release").touch()
         gateway.close()
         if browser is not None:
             browser.close()

tests/sidebar_terminal_barrier_server.py

@@ -0,0 +1,40 @@
+"""Test-only server for the public settle-frame gate's cross-channel race.
+
+Hold the actual HTTP terminal frame after the worker has finished. The browser
+keeps its native, open EventSource while /api/sessions independently reports idle.
+No browser state, renderer, persistence result, or terminal payload is mocked.
+"""
+
+from pathlib import Path
+import sys
+import time
+
+
+def main():
+    if len(sys.argv) != 3:
+        raise SystemExit("expected repository root and isolated barrier directory")
+    repo_root, barrier_dir = (Path(value) for value in sys.argv[1:])
+    sys.path.insert(0, str(repo_root))
+    import server
+    from api import routes
+
+    ready = barrier_dir / "ready"
+    release = barrier_dir / "release"
+    original = routes._sse_with_id
+
+    def terminal_barrier(handler, event, data, event_id=None):
+        if event == "done" and handler.path.startswith("/api/chat/stream?"):
+            ready.touch()
+            deadline = time.monotonic() + 15
+            while not release.exists():
+                if time.monotonic() >= deadline:
+                    raise TimeoutError("test did not release the terminal HTTP frame")
+                time.sleep(0.005)
+        return original(handler, event, data, event_id)
+
+    routes._sse_with_id = terminal_barrier
+    server.main()
+
+
+if __name__ == "__main__":
+    main()

tests/test_issue2454_active_session_spinner.py

@@ -183,6 +183,7 @@ def test_optimistic_merge_preserves_server_running_state_when_local_cache_is_sta
     optimistic_body = _function_body(SESSIONS_SRC, "function _isOptimisticFirstTurnSessionRow(")
     keep_body = _function_body(SESSIONS_SRC, "function _shouldKeepLocalOnlyOptimisticSessionRow(")
     drop_body = _function_body(SESSIONS_SRC, "function _dropStaleOptimisticSessionRow(")
+    owner_body = _function_body(SESSIONS_SRC, "function _hasOwnedOpenLiveStream(")
     merge_body = _function_body(SESSIONS_SRC, "function _mergeOptimisticFirstTurnSessions(")
 
     script = f"""
@@ -206,6 +207,7 @@ function _isServerIdleSessionRow(s) {{{helper_body}}}
 function _isSessionLocallyStreaming(s) {{{local_body}}}
 function _isOptimisticFirstTurnSessionRow(s) {{{optimistic_body}}}
 function _shouldKeepLocalOnlyOptimisticSessionRow(local) {{{keep_body}}}
+function _hasOwnedOpenLiveStream(sid) {{{owner_body}}}
 function _dropStaleOptimisticSessionRow(sid) {{{drop_body}}}
 function _mergeOptimisticFirstTurnSessions(fetchedSessions) {{{merge_body}}}
 const merged = _mergeOptimisticFirstTurnSessions([{{

tests/test_issue4766_sidebar_source_pushdown.py

@@ -536,7 +536,7 @@ def test_source_filtered_cache_preserves_hidden_bucket_runtime_state():
     remember_source_fn = _extract_function(src, "_rememberSessionListSource")
     remember_streaming_fn = _extract_function(src, "_rememberRenderedStreamingState")
     remember_snapshot_fn = _extract_function(src, "_rememberRenderedSessionSnapshot")
-    purge_fn = _extract_function(src, "_purgeStaleInflightEntries")
+    purge_fn = _extract_function(src, "_hasOwnedOpenLiveStream") + "\n" + _extract_function(src, "_purgeStaleInflightEntries")
     mark_fn = _extract_function(src, "_markPollingCompletionUnreadTransitions")
     script = f"""
 global._allSessions = [{{
@@ -654,7 +654,7 @@ def test_session_list_response_omits_bucket_counts_when_missing(monkeypatch):
 @pytest.mark.skipif(NODE is None, reason="node not on PATH")
 def test_scope_mismatch_error_path_respects_sidebar_source():
     src = SESSIONS_JS.read_text(encoding="utf-8")
-    purge_fn = _extract_function(src, "_purgeStaleInflightEntries")
+    purge_fn = _extract_function(src, "_hasOwnedOpenLiveStream") + "\n" + _extract_function(src, "_purgeStaleInflightEntries")
     clear_fn = _extract_function(src, "_clearSessionSourceTabCounts")
     requested_source_fn = _extract_function(src, "_requestedSessionSidebarSource")
     exclude_hidden_fn = _extract_function(src, "_sessionListExcludeHiddenEnabled")

tests/test_sidebar_terminal_handoff.py

@@ -0,0 +1,96 @@
+"""Sidebar metadata must not retire a still-open, exact chat transport.
+
+The live SSE and /api/sessions responses are independently scheduled. An idle
+sidebar row can arrive before the terminal frame carrying the Anchor scene.
+"""
+
+import json
+from pathlib import Path
+import shutil
+import subprocess
+
+import pytest
+
+from tests.test_issue2454_active_session_spinner import _function_body
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def _run_case(operation, transport, *, absent=False, close_after=False):
+    source = (ROOT / "static/sessions.js").read_text(encoding="utf-8")
+    names = ["_isServerIdleSessionRow", "_reconcileActiveSessionIdleStateFromList", "_purgeStaleInflightEntries", "_dropStaleOptimisticSessionRow"]
+    if "function _hasOwnedOpenLiveStream(" in source:
+        names.append("_hasOwnedOpenLiveStream")
+    parameters = {
+        "_isServerIdleSessionRow": "s",
+        "_reconcileActiveSessionIdleStateFromList": "serverRows",
+        "_hasOwnedOpenLiveStream": "sid",
+        "_dropStaleOptimisticSessionRow": "sid",
+    }
+    functions = "\n".join(
+        f"function {name}({parameters.get(name, '')}) {{"
+        + _function_body(source, f"function {name}(") + "}"
+        for name in names
+    )
+    script = """
+const S={session:{session_id:'current',active_stream_id:'turn-1'},busy:true,activeStreamId:'turn-1'};
+const INFLIGHT={current:{streamId:'turn-1',messages:[{role:'assistant',content:'work so far'}]}};
+const original=INFLIGHT.current;
+const LIVE_STREAMS={current:TRANSPORT};
+const _sessionStreamingById=new Map();
+const calls=[];
+const idle={session_id:'current',is_streaming:false,active_stream_id:null,pending_user_message:null};
+const _allSessions=ABSENT?[]:[idle];
+function clearInflightState(){calls.push('clear');}
+function _forgetObservedStreamingSession(){}
+function hideApprovalCard(){}
+function hideLiveRunStatus(){}
+function clearLiveToolCards(){calls.push('clear-dom');}
+function updateSendBtn(){}
+function _scheduleActiveSessionIdleReload(){calls.push('reload');}
+FUNCTIONS
+function run(){if(OPERATION==='idle')_reconcileActiveSessionIdleStateFromList([idle]);else if(OPERATION==='optimistic')_dropStaleOptimisticSessionRow('current');else _purgeStaleInflightEntries();}
+run();
+const beforeClose={busy:S.busy,stream:S.activeStreamId,retained:INFLIGHT.current===original,calls:[...calls]};
+if(CLOSE_AFTER){LIVE_STREAMS.current.source.readyState=2;run();}
+console.log(JSON.stringify({beforeClose,afterClose:{retained:!!INFLIGHT.current,busy:S.busy}}));
+"""
+    script = script.replace("TRANSPORT", json.dumps(transport)).replace("ABSENT", json.dumps(absent))
+    script = script.replace("FUNCTIONS", functions).replace("OPERATION", json.dumps(operation))
+    script = script.replace("CLOSE_AFTER", json.dumps(close_after))
+    node = shutil.which("node")
+    if not node:
+        pytest.skip("node required")
+    result = subprocess.run([node, "-e", script], text=True, capture_output=True, timeout=20)
+    assert result.returncode == 0, result.stderr
+    return json.loads(result.stdout)
+
+
+@pytest.mark.parametrize("operation", ["idle", "purge", "optimistic"])
+def test_current_open_transport_keeps_terminal_handoff_and_recovers_after_close(operation):
+    result = _run_case(operation, {"streamId": "turn-1", "source": {"readyState": 1}}, close_after=True)
+    assert result["beforeClose"] == {"busy": True, "stream": "turn-1", "retained": True, "calls": []}
+    assert result["afterClose"]["retained"] is False
+    if operation == "idle":
+        assert result["afterClose"]["busy"] is False
+
+
+@pytest.mark.parametrize("operation", ["idle", "purge", "optimistic"])
+@pytest.mark.parametrize("transport", [
+    None,
+    {"streamId": "turn-1", "source": None},
+    {"streamId": "turn-1", "source": {"readyState": 0}},
+    {"streamId": "turn-1", "source": {"readyState": 2}},
+    {"streamId": "older-turn", "source": {"readyState": 1}},
+])
+def test_missing_closed_connecting_or_wrong_transport_does_not_block_idle_recovery(operation, transport):
+    result = _run_case(operation, transport)
+    assert result["beforeClose"]["retained"] is False
+    if operation == "idle":
+        assert result["beforeClose"]["busy"] is False
+        assert "reload" in result["beforeClose"]["calls"]
+
+
+def test_current_open_transport_survives_a_filtered_sidebar_list():
+    result = _run_case("purge", {"streamId": "turn-1", "source": {"readyState": 1}}, absent=True)
+    assert result["beforeClose"]["retained"] is True