Merge PR #7749 (contributor head bbca954c5c)

tests/browser_conversation_lifecycle.py

@@ -39,6 +39,7 @@ SCENARIO = os.environ.get("LIFECYCLE_SCENARIO", "normal").strip() or "normal"
 TOOL_NAME = "read_file"
 TOOL_ID = "lifecycle-tool-1"
 TEST_BITE = os.environ.get("LIFECYCLE_TEST_BITE", "").strip()
+_FRAME_PROOF_BITES = {"settle-worklog-frame-proof", "sidebar-idle-missing-terminal"}
 GATEWAY_ACTIVITY_TIMEOUT = 60.0
 ANCHOR_SCENE_PERSIST_TIMEOUT = 60.0
 ANCHOR_SCENE_PROJECTION_TIMEOUT = 10_000
@@ -267,8 +268,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 +296,50 @@ 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
+    if TEST_BITE != "sidebar-idle-missing-terminal":
+        (barrier_dir / "release").touch()
+
+
 class DeterministicGateway:
     """A localhost-only Gateway Runs server with test-controlled phase gates."""
 
@@ -733,11 +787,12 @@ def main() -> int:
         "drop-anchor-persistence",
         "drop-terminal-anchor-row",
         "settle-worklog-frame-proof",
+        "sidebar-idle-missing-terminal",
     }:
         raise ValueError(
             f"Unsupported LIFECYCLE_TEST_BITE {TEST_BITE!r}; "
             "expected one of '', 'drop-anchor-persistence', "
-            "'drop-terminal-anchor-row', 'settle-worklog-frame-proof'"
+            "'drop-terminal-anchor-row', 'settle-worklog-frame-proof', 'sidebar-idle-missing-terminal'"
         )
     if TEST_BITE == "drop-terminal-anchor-row" and scenario != "terminal-error":
         raise ValueError(
@@ -782,6 +837,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 in _FRAME_PROOF_BITES 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
@@ -873,7 +934,7 @@ def main() -> int:
         # must never expose an expanded empty Worklog shell or an activity-less
         # gap.  Sample after rAF callbacks for a frame have drained; DOM mutations
         # within one task or earlier rAF callbacks are not user-visible frames.
-        if TEST_BITE == "settle-worklog-frame-proof":
+        if TEST_BITE in _FRAME_PROOF_BITES:
             page.evaluate(
                 """() => {
                   const samples = [];
@@ -941,12 +1002,17 @@ 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)""",
                 arg=FINAL_TEXT,
-                timeout=15000,
+                timeout=11000 if TEST_BITE == "sidebar-idle-missing-terminal" else 15000,
             )
+            if TEST_BITE == "sidebar-idle-missing-terminal":
+                assert not (terminal_barrier_dir / "release").exists(), "terminal frame was unexpectedly released"
+                print("OK  missing terminal: canonical snapshot settled the still-OPEN transport before terminal delivery")
         else:
             gateway.release_terminal.set()
             page.wait_for_function(
@@ -958,7 +1024,7 @@ def main() -> int:
             )
         session_id = page.evaluate("S.session && S.session.session_id")
         assert session_id, "active session id missing after settlement"
-        if TEST_BITE == "settle-worklog-frame-proof":
+        if TEST_BITE in _FRAME_PROOF_BITES:
             page.wait_for_timeout(100)
             settle_proof = page.evaluate(
                 "window.__lifecycleSettleFrameProof && window.__lifecycleSettleFrameProof.stop()"
@@ -1176,6 +1242,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_idle_stream_recovery.py

@@ -0,0 +1,173 @@
+"""A sidebar-idle hint must retain the live handoff without waiting forever.
+
+Drive the production scheduling helper with a fake clock and a deferred snapshot
+boundary. The browser lifecycle gate separately covers the real HTTP restore.
+"""
+
+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):
+    sessions = (ROOT / "static/sessions.js").read_text(encoding="utf-8")
+    messages = (ROOT / "static/messages.js").read_text(encoding="utf-8")
+    functions = "\n".join(
+        f"function {name}({params}) {{" + _function_body(sessions, f"function {name}(") + "}"
+        for name, params in [
+            ("_isServerIdleSessionRow", "s"),
+            ("_hasOwnedOpenLiveStream", "sid"),
+            ("_reconcileActiveSessionIdleStateFromList", "serverRows"),
+        ]
+    )
+    if "function _bindSidebarIdleRecovery(" in messages:
+        functions += "\nfunction _bindSidebarIdleRecovery(live) {" + _function_body(
+            messages, "function _bindSidebarIdleRecovery("
+        ) + "}"
+    script = r"""
+const assert=require('assert');
+const activeSid='current',streamId='turn-1';
+let _terminalStateReached=false,_streamFinalized=false;
+let _pendingStreamEndRecovery=CASE==='stream-end-pending';
+const S={session:{session_id:activeSid},busy:true,activeStreamId:streamId};
+const INFLIGHT={current:{streamId}},original=INFLIGHT.current;
+const timers=[],listeners={},calls=[],statusCalls=[],failures=[];
+function setTimeout(fn,delay){const t={fn,delay,cancelled:false};timers.push(t);return t;}
+function clearTimeout(t){if(t)t.cancelled=true;}
+const source={readyState:1,addEventListener(name,fn){(listeners[name]??=[]).push(fn);}};
+const live={streamId,source},LIVE_STREAMS={current:live};
+let resolveSnapshot;
+async function api(path,options){
+  statusCalls.push({path,options});
+  if(CASE==='probe-error') throw new Error('status unavailable');
+  return {active:CASE==='runtime-active'};
+}
+async function _restoreSettledSession(source,options){
+  calls.push({source,options});
+  return new Promise(resolve=>{resolveSnapshot=resolve;});
+}
+function _handleStreamError(s){failures.push(s);S.busy=false;S.activeStreamId=null;}
+FUNCTIONS
+if(typeof _bindSidebarIdleRecovery==='function')_bindSidebarIdleRecovery(live);
+const idle=[{session_id:activeSid,is_streaming:false,active_stream_id:null}];
+(async()=>{
+  _reconcileActiveSessionIdleStateFromList(idle);
+  _reconcileActiveSessionIdleStateFromList(idle);
+  if(CASE==='stream-end-pending'){assert.strictEqual(timers.length,0,'existing stream-end recovery must remain the sole owner');console.log('ok');return;}
+  assert.strictEqual(timers.length,1,'an OPEN transport must schedule one bounded snapshot recovery');
+  assert.ok(timers[0].delay>0&&timers[0].delay<=2000);
+  assert.strictEqual(S.busy,true,'the first idle hint must not destroy the live handoff');
+  const mode=CASE;
+  if(mode==='cancel-rearm'){
+    live.cancelIdleRecovery();
+    _reconcileActiveSessionIdleStateFromList(idle);
+    assert.strictEqual(timers.length,2);
+    await timers[0].fn();
+    assert.strictEqual(timers[1].cancelled,false,'a queued old callback cannot cancel a replacement ticket');
+    const replacement=timers[1].fn();
+    await Promise.resolve();
+    assert.strictEqual(calls.length,1);
+    resolveSnapshot('active');await replacement;
+    console.log('ok');return;
+  }
+  if(mode==='cancel-before'){
+    live.cancelIdleRecovery();
+    assert.strictEqual(timers[0].cancelled,true);
+    console.log('ok');return;
+  }
+  if(mode==='replace-before')LIVE_STREAMS.current={streamId,source:{readyState:1}};
+  const running=timers[0].fn();
+  if(mode==='replace-before'){
+    await running;assert.strictEqual(calls.length,0);console.log('ok');return;
+  }
+  await Promise.resolve();
+  assert.strictEqual(statusCalls.length,1);
+  assert.ok(statusCalls[0].path.includes('/api/chat/stream/status?stream_id=turn-1'));
+  assert.deepStrictEqual(statusCalls[0].options,{timeoutMs:8000,retries:0,timeoutToast:false});
+  if(mode==='runtime-active'){
+    await running;
+    assert.strictEqual(calls.length,0,'active runtime must not fall through to session restore');
+    assert.strictEqual(failures.length,0);
+    assert.strictEqual(S.busy,true);
+    assert.strictEqual(INFLIGHT.current,original);
+    console.log('ok');return;
+  }
+  if(mode==='probe-error'){
+    await running;
+    assert.strictEqual(calls.length,0,'failed runtime probe must not trust idle session flags');
+    assert.strictEqual(failures.length,1);
+    console.log('ok');return;
+  }
+  assert.strictEqual(calls.length,1);
+  _reconcileActiveSessionIdleStateFromList(idle);
+  assert.strictEqual(timers.length,1,'repeated idle hints must not restart a pending deadline');
+  const owns=calls[0].options.isCurrent;
+  assert.strictEqual(owns(),true);
+  assert.deepStrictEqual(calls[0].options.requestOptions,{timeoutMs:8000,retries:0,timeoutToast:false});
+  if(mode==='replace-during')LIVE_STREAMS.current={streamId,source:{readyState:1}};
+  if(mode==='new-turn')INFLIGHT.current={streamId:null};
+  if(mode==='navigate')S.session={session_id:'other'};
+  if(mode==='done'){
+    for(const fn of listeners.done||[])fn();
+    _streamFinalized=true;S.busy=false;S.activeStreamId=null;
+  }
+  if(['replace-during','new-turn','navigate','done'].includes(mode))assert.strictEqual(owns(),false);
+  resolveSnapshot(mode==='active'?'active':mode==='restored'?'restored':'error');
+  await running;
+  assert.strictEqual(failures.length,mode==='error'?1:0);
+  assert.strictEqual(owns(),false,'settled requests must lose authority');
+  if(mode==='active'){
+    assert.strictEqual(S.busy,true);assert.strictEqual(INFLIGHT.current,original);
+  }
+  console.log('ok');
+})().catch(e=>{console.error(e.stack||e);process.exit(1);});
+""".replace("FUNCTIONS", functions).replace("CASE", json.dumps(case))
+    node = shutil.which("node")
+    if not node:
+        pytest.skip("node required")
+    result = subprocess.run([node, "-e", script], capture_output=True, text=True, timeout=20)
+    assert result.returncode == 0, result.stdout + result.stderr
+
+
+@pytest.mark.parametrize("case", [
+    "error", "active", "restored", "runtime-active", "probe-error", "cancel-before", "replace-before",
+    "replace-during", "new-turn", "navigate", "done", "stream-end-pending", "cancel-rearm",
+])
+def test_sidebar_idle_open_stream_has_bounded_owner_safe_recovery(case):
+    _run(case)
+
+
+def test_snapshot_restore_rechecks_request_owner_after_await():
+    source = (ROOT / "static/messages.js").read_text(encoding="utf-8")
+    body = _function_body(source, "async function _restoreSettledSession(")
+    script = r"""
+const assert=require('assert');
+const activeSid='current',streamId='turn-1';
+let current=true,resolveSnapshot;
+const S={session:{session_id:activeSid},activeStreamId:streamId};
+let _streamFinalized=false;
+function _isActiveSession(){return true;}
+function _closeSource(){throw Error('stale restore must not close a replacement');}
+function api(){return new Promise(resolve=>{resolveSnapshot=resolve;});}
+async function _restoreSettledSession(source,options=null){BODY}
+(async()=>{
+ const pending=_restoreSettledSession({}, {status:true,isCurrent:()=>current});
+ current=false;
+ resolveSnapshot({session:{session_id:activeSid,active_stream_id:'newer-turn',messages:[]}});
+ assert.strictEqual(await pending,'stale','request authority must be rechecked before using the response');
+ assert.strictEqual(S.activeStreamId,streamId);
+ console.log('ok');
+})().catch(e=>{console.error(e.stack||e);process.exit(1);});
+""".replace("BODY", body)
+    node = shutil.which("node")
+    if not node:
+        pytest.skip("node required")
+    result = subprocess.run([node, "-e", script], capture_output=True, text=True, timeout=20)
+    assert result.returncode == 0, result.stdout + result.stderr

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

tests/test_webui_gateway_chat_backend.py

@@ -4,6 +4,7 @@ from email.message import Message
 import json
 from pathlib import Path
 import re
+import threading
 import time
 import urllib.error
 
@@ -858,6 +859,104 @@ def test_gateway_chat_worker_emits_goal_continue_for_goal_related_turn(tmp_path,
     assert saved.messages[-1]["content"] == "goal reply"
 
 
+def test_gateway_goal_eval_keeps_runtime_active_after_success_writeback_before_done(
+    tmp_path, monkeypatch
+):
+    """Persisted idle flags can precede terminal emission while STREAMS is still active."""
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir()
+    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json")
+    monkeypatch.setattr(models, "SESSIONS", OrderedDict())
+
+    class FakeResponse:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+        def __iter__(self):
+            yield b'data: {"choices":[{"delta":{"content":"goal reply"}}]}\n\n'
+            yield b"data: [DONE]\n\n"
+
+    monkeypatch.setenv("HERMES_WEBUI_GATEWAY_BASE_URL", "http://gateway.local")
+    monkeypatch.setattr(
+        gateway_chat.urllib.request, "urlopen", lambda req, timeout=0: FakeResponse()
+    )
+
+    from api import goals as webui_goals
+
+    goal_entered = threading.Event()
+    release_goal = threading.Event()
+    observed = {}
+
+    monkeypatch.setattr(webui_goals, "has_active_goal", lambda *args, **kwargs: True)
+
+    def blocked_goal_eval(*args, **kwargs):
+        saved = models.get_session(s.session_id)
+        observed["active_stream_id"] = saved.active_stream_id
+        observed["pending_user_message"] = saved.pending_user_message
+        observed["stream_registered"] = stream_id in STREAMS
+        goal_entered.set()
+        assert release_goal.wait(timeout=10)
+        return {"should_continue": False, "message": "Goal checked"}
+
+    monkeypatch.setattr(webui_goals, "evaluate_goal_after_turn", blocked_goal_eval)
+
+    s = new_session()
+    stream_id = "stream-gateway-goal-runtime-window"
+    s.active_stream_id = stream_id
+    s.pending_user_message = "finish it"
+    s.pending_attachments = []
+    s.pending_started_at = 123
+    s.save()
+    channel = create_stream_channel()
+    subscriber = channel.subscribe()
+    STREAMS[stream_id] = channel
+
+    worker = threading.Thread(
+        target=gateway_chat._run_gateway_chat_streaming,
+        args=(
+            s.session_id,
+            "finish it",
+            "test-model",
+            str(tmp_path),
+            stream_id,
+            [],
+        ),
+        kwargs={"goal_related": True},
+        daemon=True,
+    )
+    worker.start()
+    assert goal_entered.wait(timeout=10), "worker did not reach post-writeback goal evaluation"
+
+    # This is the exact authority split the browser recovery must respect:
+    # persisted session metadata already looks idle, while the runtime stream
+    # (and therefore /api/chat/stream/status) is still active and can emit
+    # goal / goal_continue / done / stream_end.
+    assert observed == {
+        "active_stream_id": None,
+        "pending_user_message": None,
+        "stream_registered": True,
+    }
+    assert stream_id in STREAMS
+    assert worker.is_alive()
+    assert not any(item[0] == "done" for item in list(subscriber.queue))
+
+    release_goal.set()
+    worker.join(timeout=10)
+    assert not worker.is_alive()
+    events = []
+    while not subscriber.empty():
+        events.append(subscriber.get_nowait())
+    names = [item[0] for item in events]
+    assert "goal" in names
+    assert "done" in names
+    assert "stream_end" in names
+    assert names.index("goal") < names.index("done") < names.index("stream_end")
+
+
 def test_gateway_chat_worker_skips_goal_judge_for_non_goal_turn(tmp_path, monkeypatch):
     session_dir = tmp_path / "sessions"
     session_dir.mkdir()