From bbca954c5c8b78d09d761300a0f3adc1f3183080 Mon Sep 17 00:00:00 2001 From: Frank Song Date: Sun, 20 Sep 2026 21:11:55 +0800 Subject: [PATCH] fix: gate idle recovery on exact runtime status (cherry picked from commit 35597715018e598bf24f0b708cbd32b039533a99) --- docs/pr-media/6469-sidebar-handoff/README.md | 23 +++-- docs/rfcs/live-to-final-assistant-replies.md | 10 +- static/messages.js | 24 +++++ tests/test_sidebar_idle_stream_recovery.py | 28 +++++- tests/test_webui_gateway_chat_backend.py | 99 ++++++++++++++++++++ 5 files changed, 172 insertions(+), 12 deletions(-) diff --git a/docs/pr-media/6469-sidebar-handoff/README.md b/docs/pr-media/6469-sidebar-handoff/README.md index 341e6a5f..d458093d 100644 --- a/docs/pr-media/6469-sidebar-handoff/README.md +++ b/docs/pr-media/6469-sidebar-handoff/README.md @@ -59,13 +59,16 @@ the existing stale recovery path. The OPEN-only guard at `85f1c2d34b62b0ade85f51ed2cf8c1cf56f5ab9e` could wait forever when the terminal frame never arrived. The follow-up keeps the live -handoff for 1.5 seconds, then uses the existing canonical session restore with -an eight-second request timeout and no retries. Repeated sidebar hints cannot -restart the deadline or create concurrent requests. An active server snapshot -preserves the run; a failed or timed-out request uses existing interrupted-stream -recovery rather than claiming success. Normal terminal delivery, transport -replacement/close, navigation, and a newer optimistic turn invalidate the old -request. Already-running stream-end recovery remains the sole recovery owner. +handoff for 1.5 seconds, then probes the exact stream through +`/api/chat/stream/status` with an eight-second timeout and no retries. +Repeated sidebar hints cannot restart the deadline or create concurrent +requests. `active:true` keeps the OPEN owner intact even when the persisted +session fields already look idle; only explicit runtime inactivity permits the +canonical session restore. A failed or malformed runtime probe uses the +existing interrupted-stream path rather than claiming successful settlement. +Normal terminal delivery, transport replacement/close, navigation, and a newer +optimistic turn invalidate the old request. Already-running stream-end recovery +remains the sole recovery owner. ```sh LIFECYCLE_SCENARIO=normal LIFECYCLE_TEST_BITE=sidebar-idle-missing-terminal \ @@ -80,3 +83,9 @@ The patched implementation restores the final answer and passes the unchanged painted-frame, semantic activity, persistence and hard-reload checks before the barrier is released. The cloud workflow runs both delayed- and missing-terminal rows. No live provider or production state is used. + +A production Gateway regression also blocks post-turn goal evaluation after +the success transcript has already cleared `active_stream_id` and pending +fields. During that interval the worker and exact entry in `STREAMS` remain +live and no `done` event has been emitted, proving why persisted session flags +cannot authorize sidebar settlement on their own. diff --git a/docs/rfcs/live-to-final-assistant-replies.md b/docs/rfcs/live-to-final-assistant-replies.md index 10599623..60e762c6 100644 --- a/docs/rfcs/live-to-final-assistant-replies.md +++ b/docs/rfcs/live-to-final-assistant-replies.md @@ -476,9 +476,13 @@ Expected behavior: Missing, connecting, closed, or mismatched transports keep the normal stale state recovery path; a cached stream ID or busy flag is insufficient. OPEN is not proof that a terminal frame will arrive: an idle hint must also - schedule bounded canonical snapshot recovery without erasing the live scene. - A newer turn, replacement source, or received terminal event invalidates that - request; a still-active server snapshot is not permission to terminate work. + schedule bounded recovery without erasing the live scene. Before reading a + settled session snapshot, that recovery must check the exact stream's runtime + status. A runtime-active stream retains ownership even when persisted + `active_stream_id` and pending fields are already clear; only explicit + runtime inactivity permits canonical snapshot settlement. Probe failure is a + degraded/interrupted state, not evidence of successful completion. A newer + turn, replacement source, or received terminal event invalidates the request. - Background completion, cancellation, or failure should be represented without stealing the visible pane from the user. - Session switching should not erase pending live context, in-flight snapshots, diff --git a/static/messages.js b/static/messages.js index 9423fb55..45cc458f 100644 --- a/static/messages.js +++ b/static/messages.js @@ -5771,6 +5771,30 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ return; } try{ + let runtimeStatus=null; + try{ + runtimeStatus=await api( + `/api/chat/stream/status?stream_id=${encodeURIComponent(streamId)}`, + {timeoutMs:8000,retries:0,timeoutToast:false} + ); + }catch(_){ + // Runtime ownership is authoritative here. If the probe itself + // fails, do not infer completion from already-cleared session + // fields; surface the existing interrupted-stream path instead. + if(isCurrent()) _handleStreamError(live.source); + return; + } + if(!isCurrent()) return; + if(!runtimeStatus||typeof runtimeStatus.active!=='boolean'){ + _handleStreamError(live.source); + return; + } + // Gateway success writeback clears persisted active/pending fields + // before post-turn goal evaluation and terminal event emission. + // STREAMS-backed status therefore owns this decision: an exact active + // runtime must keep its OPEN browser handoff even if the sidebar row + // already looks idle. + if(runtimeStatus.active) return; const status=await _restoreSettledSession(live.source,{ status:true, isCurrent, diff --git a/tests/test_sidebar_idle_stream_recovery.py b/tests/test_sidebar_idle_stream_recovery.py index eea5abd0..7de907fa 100644 --- a/tests/test_sidebar_idle_stream_recovery.py +++ b/tests/test_sidebar_idle_stream_recovery.py @@ -38,12 +38,17 @@ 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=[],failures=[]; +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;}); @@ -67,6 +72,7 @@ const idle=[{session_id:activeSid,is_streaming:false,active_stream_id:null}]; 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; @@ -81,6 +87,24 @@ const idle=[{session_id:activeSid,is_streaming:false,active_stream_id:null}]; 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'); @@ -113,7 +137,7 @@ const idle=[{session_id:activeSid,is_streaming:false,active_stream_id:null}]; @pytest.mark.parametrize("case", [ - "error", "active", "restored", "cancel-before", "replace-before", + "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): diff --git a/tests/test_webui_gateway_chat_backend.py b/tests/test_webui_gateway_chat_backend.py index 0e7d2830..226fee6c 100644 --- a/tests/test_webui_gateway_chat_backend.py +++ b/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()