diff --git a/static/messages.js b/static/messages.js index eaa96574..ef113d84 100644 --- a/static/messages.js +++ b/static/messages.js @@ -2061,6 +2061,20 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ msg.provider_details='The only assistant text returned for this turn was the internal preserved-task-list compression marker, so the WebUI replaced it with an explicit error instead of rendering the marker as a model response.'; return true; } + function _isTerminalStreamErrorMarkerMessage(message){ + return message&&message.role==='assistant'&&typeof message.content==='string'&& + message.content.startsWith('**Connection interrupted:** The browser lost the live SSE connection before the response finished.'); + } + function _appendTerminalStreamErrorMarkerIfMissing(messages){ + if(!Array.isArray(messages)) return; + while(messages.length && _isTerminalStreamErrorMarkerMessage(messages[messages.length-1])){ + messages.pop(); + } + messages.push({ + role:'assistant', + content:'**Connection interrupted:** The browser lost the live SSE connection before the response finished. If the worker completed, reopening this session should restore the settled transcript.', + }); + } function _setActivePaneIdleIfOwner(){ if(_isActiveSession()||!S.session||!INFLIGHT[S.session.session_id]){ setBusy(false); @@ -2351,7 +2365,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ }catch(_){ if(_deferStreamErrorIfOffline()||_pageHiddenForStreamError()) return; } - if(await _restoreSettledSession(source)) return; + if(await _restoreSettledSession(source, {preserveVisibleOnShorterTerminalSnapshot:true})) return; if(_deferStreamErrorIfOffline()||_pageHiddenForStreamError()) return; _flushReasoningToAnchor(); _scheduleAnchorRegistryCleanup(120000); @@ -5453,7 +5467,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ } if(isRecoveryControlMessage){ (async()=>{ - if(await _restoreSettledSession(source)) return; + if(await _restoreSettledSession(source, {preserveVisibleOnShorterTerminalSnapshot:true})) return; if(S.session&&S.session.session_id===activeSid){ S.messages=_filterRecoveryControlMessages(S.messages||[]); _markSessionViewed(activeSid, S.messages.length); @@ -5549,7 +5563,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ }catch(_){ if(_deferStreamErrorIfOffline()) return; } - if(await _restoreSettledSession(source)) return; + if(await _restoreSettledSession(source, {preserveVisibleOnShorterTerminalSnapshot:true})) return; if(_deferStreamErrorIfOffline()) return; if(_deferStreamErrorIfPageHidden(source)) return; const nextDelay=_retryDelays[attempt+1]; @@ -5565,7 +5579,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ setTimeout(()=>{void _probeReconnect(0);},_retryDelays[0]); return; } - if(await _restoreSettledSession(source)) return; + if(await _restoreSettledSession(source, {preserveVisibleOnShorterTerminalSnapshot:true})) return; if(_deferStreamErrorIfOffline()) return; if(_deferStreamErrorIfPageHidden(source)) return; _flushReasoningToAnchor(); @@ -5695,6 +5709,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ async function _restoreSettledSession(source, options=null){ const returnStatus=!!(options&&options.status); + const preserveVisibleOnShorterTerminalSnapshot=!!(options&&options.preserveVisibleOnShorterTerminalSnapshot); if(_isActiveSession() && S.activeStreamId!==streamId){ _closeSource(source); return returnStatus?'stale':false; @@ -5730,9 +5745,16 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ clearLiveToolCards();if(!assistantText)removeThinking(); S.session=session; const _nextMsgs3018=(session.messages||[]).filter(m=>m&&m.role); - _attachProjectedAnchorSceneToLastAssistant(_nextMsgs3018); - S.messages=_carryForwardEphemeralTurnFields(S.messages||[], _nextMsgs3018); - S.messages=_filterRecoveryControlMessages(S.messages || []); + const _currentMessages=Array.isArray(S.messages)?S.messages:[]; + const _stagedMessages=_carryForwardEphemeralTurnFields(_currentMessages, _nextMsgs3018); + const _preserveCurrentTranscript=preserveVisibleOnShorterTerminalSnapshot&& + _stagedMessages.length>0&& + _stagedMessages.length<_currentMessages.length; + const _nextMessageKeys=new Set(_stagedMessages.map(_messageIdentityKey).filter(Boolean)); + const _tailMessages=_currentMessages.filter(m=>!_nextMessageKeys.has(_messageIdentityKey(m))); + const _resolvedMessages=_preserveCurrentTranscript?[..._stagedMessages,..._tailMessages]:_stagedMessages; + S.messages=_filterRecoveryControlMessages(_resolvedMessages || []); + _attachProjectedAnchorSceneToLastAssistant(S.messages); if(typeof _hydrateTodosFromSession==='function') _hydrateTodosFromSession(S.session); if(S.session&&S.session.session_id){ try{localStorage.setItem('hermes-webui-session',S.session.session_id);}catch(_){} @@ -5799,7 +5821,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ session_id:activeSid, },null); clearLiveToolCards();if(!assistantText)removeThinking(); - S.messages.push({role:'assistant',content:'**Connection interrupted:** The browser lost the live SSE connection before the response finished. If the worker completed, reopening this session should restore the settled transcript.'}); + _appendTerminalStreamErrorMarkerIfMissing(S.messages); _attachProjectedAnchorSceneToLastAssistant(S.messages); renderMessages({preserveScroll:true}); _markSessionViewed(activeSid, S.messages.length); diff --git a/tests/test_issue5224_terminal_error_transcript_preserve.py b/tests/test_issue5224_terminal_error_transcript_preserve.py new file mode 100644 index 00000000..6219fd10 --- /dev/null +++ b/tests/test_issue5224_terminal_error_transcript_preserve.py @@ -0,0 +1,328 @@ +"""Regression tests for #5224: preserve terminal-visible transcript on recovery.""" + +import json +import subprocess +import shutil +from pathlib import Path + +import pytest + + +REPO_ROOT = Path(__file__).parent.parent.resolve() +MESSAGES_JS = REPO_ROOT / "static" / "messages.js" +NODE = shutil.which("node") + +pytestmark = pytest.mark.skipif(NODE is None, reason="node is required to execute message runtime tests") + + +_DRIVER = r""" +const fs = require('fs'); + +const src = fs.readFileSync(process.argv[2], 'utf8'); +const scenario = JSON.parse(process.argv[3] || '{}'); + +function extractFunction(source, name) { + const markers = [`async function ${name}(`, `function ${name}(`]; + let start = -1; + for (const marker of markers) { + start = source.indexOf(marker); + if (start >= 0) break; + } + if (start < 0) { + throw new Error(`missing ${name}`); + } + let i = source.indexOf('{', start); + if (i < 0) { + throw new Error(`missing function body for ${name}`); + } + let depth = 1; + i++; + while (i < source.length) { + const ch = source[i]; + if (ch === '{') depth++; + if (ch === '}') { + depth--; + if (depth === 0) return source.slice(start, i + 1); + } + i++; + } + throw new Error(`unterminated function body for ${name}`); +} + +function extractFunctionByName(name) { + return extractFunction(src, name); +} + +function installRuntimeHelpers() { + const helpers = [ + "_isMarkerOnlyAssistantMessage", + "_streamRecoveryControlMessageText", + "_streamRecoveryControlMessage", + "_filterRecoveryControlMessages", + "_replaceMarkerOnlyAssistantWithStreamError", + "_messageIdentityKey", + "_carryForwardEphemeralTurnFields", + "_isTerminalStreamErrorMarkerMessage", + "_appendTerminalStreamErrorMarkerIfMissing", + "_restoreSettledSession", + "_handleStreamError", + ]; + for (const helper of helpers) { + const body = extractFunctionByName(helper); + const factory = new Function(`${body}; return ${helper};`); + globalThis[helper] = factory(); + } +} + +function buildRuntime() { + const activeSid = scenario.activeSid || 'session-5224'; + const streamId = scenario.streamId || 'stream-5224'; + const calls = []; + globalThis.activeSid = activeSid; + globalThis.streamId = streamId; + globalThis.assistantText = false; + globalThis.S = JSON.parse(JSON.stringify(scenario.state || {})); + if (!globalThis.S.session) { + globalThis.S.session = { session_id: activeSid }; + } + if (!Object.prototype.hasOwnProperty.call(globalThis.S, 'activeStreamId')) { + globalThis.S.activeStreamId = streamId; + } + globalThis.INFLIGHT = {}; + globalThis._EPHEMERAL_TURN_FIELDS = [ + '_turnUsage', + '_turnDuration', + '_turnTps', + '_gatewayRouting', + '_statusCard', + '_anchor_stream_id', + '_anchor_activity_scene', + ]; + globalThis._isActiveSession = () => scenario.isActiveSession !== false; + globalThis._isSessionCurrentPane = () => scenario.isSessionCurrentPane !== false; + globalThis._isSessionActivelyViewed = () => !!scenario.isSessionActivelyViewed; + globalThis._closeSource = () => calls.push('closeSource'); + globalThis._clearStreamEndRecovery = () => calls.push('clearStreamEndRecovery'); + globalThis._clearOwnerInflightState = () => calls.push('clearOwnerInflight'); + globalThis.clearLiveToolCards = () => calls.push('clearLiveToolCards'); + globalThis.removeThinking = () => calls.push('removeThinking'); + globalThis._flushReasoningToAnchor = () => calls.push('flushReasoning'); + globalThis._applyToAnchor = () => calls.push('applyToAnchor'); + globalThis._attachProjectedAnchorSceneToLastAssistant = () => calls.push('attachProjected'); + globalThis._hydrateTodosFromSession = () => calls.push('hydrateTodos'); + globalThis._scheduleAnchorRegistryCleanup = () => calls.push('scheduleAnchorRegistryCleanup'); + globalThis._smdEndParser = () => calls.push('smdEndParser'); + globalThis._markSessionCompletionUnread = () => calls.push('markCompletionUnread'); + globalThis._markSessionViewed = () => calls.push('markSessionViewed'); + globalThis.localStorage = { + setItem: () => calls.push('setLocalStorageItem'), + getItem: () => null, + removeItem: () => calls.push('removeLocalStorageItem'), + }; + globalThis._setActiveSessionUrl = () => calls.push('setActiveSessionUrl'); + globalThis.showToast = () => calls.push('showToast'); + globalThis._clearApprovalForOwner = () => calls.push('clearApprovalForOwner'); + globalThis._clearClarifyForOwner = () => calls.push('clearClarifyForOwner'); + globalThis._streamFadeCleanupReduceMotionListener = () => calls.push('streamFadeCleanup'); + globalThis._cancelAnimationFramePendingStreamRender = () => calls.push('cancelRaf'); + globalThis.finalizeThinkingCard = () => calls.push('finalizeThinkingCard'); + globalThis.syncTopbar = () => calls.push('syncTopbar'); + globalThis.renderMessages = () => calls.push('renderMessages'); + globalThis.renderSessionList = () => calls.push('renderSessionList'); + globalThis._setActivePaneIdleIfOwner = () => calls.push('setActivePaneIdle'); + globalThis.setBusy = () => calls.push('setBusy'); + globalThis.setComposerStatus = () => calls.push('setComposerStatus'); + globalThis.setStatus = () => calls.push('setStatus'); + globalThis._messageRenderableMessageCount = () => scenario.messageRenderableCount || 50; + globalThis._currentMessageRenderWindowSize = () => scenario.currentWindowSize || 12; + globalThis._messageRenderWindowSize = 20; + globalThis._streamFinalized = !!scenario.streamFinalized; + globalThis._persistTimer = null; + globalThis.api = async () => scenario.apiPayload || { session: null }; + globalThis.msgContent = undefined; + globalThis._isPreservedCompressionTaskListMarkerOnlyText = () => false; + return calls; +} + +(async () => { + installRuntimeHelpers(); + const calls = buildRuntime(); + if (scenario.action === 'restore_shorter_terminal') { + const status = await _restoreSettledSession({}, { + status: true, + preserveVisibleOnShorterTerminalSnapshot: true, + }); + const messages = Array.isArray(S.messages) ? S.messages : []; + const terminalMarkerCount = messages.filter(_isTerminalStreamErrorMarkerMessage).length; + console.log(JSON.stringify({ + action: scenario.action, + status, + messages: messages.map((m) => ({ role: m.role, content: m.content })), + terminalMarkerCount, + calls, + })); + return; + } + + if (scenario.action === 'restore_fuller_terminal') { + const status = await _restoreSettledSession({}, { + status: true, + preserveVisibleOnShorterTerminalSnapshot: true, + }); + const messages = Array.isArray(S.messages) ? S.messages : []; + const terminalMarkerCount = messages.filter(_isTerminalStreamErrorMarkerMessage).length; + console.log(JSON.stringify({ + action: scenario.action, + status, + messages: messages.map((m) => ({ role: m.role, content: m.content })), + terminalMarkerCount, + calls, + })); + return; + } + + if (scenario.action === 'terminal_marker_idempotent') { + const messages = JSON.parse(JSON.stringify(scenario.messages || [])); + _appendTerminalStreamErrorMarkerIfMissing(messages); + _appendTerminalStreamErrorMarkerIfMissing(messages); + const terminalMarkerCount = messages.filter(_isTerminalStreamErrorMarkerMessage).length; + console.log(JSON.stringify({ + action: scenario.action, + messages: messages.map((m) => ({ role: m.role, content: m.content })), + terminalMarkerCount, + calls, + })); + return; + } + + throw new Error(`unknown action: ${scenario.action}`); +})().catch((err) => { + console.error(err && err.stack ? err.stack : String(err)); + process.exit(1); +}); +""" + + +@pytest.fixture(scope="module") +def driver_path(tmp_path_factory): + driver = tmp_path_factory.mktemp("issue5224_driver") / "driver.js" + driver.write_text(_DRIVER, encoding="utf-8") + return str(driver) + + +def _run_scenario(driver_path: str, scenario: dict) -> dict: + command = [NODE, driver_path, str(MESSAGES_JS), json.dumps(scenario)] + result = subprocess.run( + command, + capture_output=True, + text=True, + encoding="utf-8", + check=False, + timeout=30, + ) + if result.returncode != 0: + raise RuntimeError(f"node driver failed: {result.stderr}") + return json.loads(result.stdout.strip()) + + +def test_terminal_error_restore_preserves_visible_transcript_when_server_snapshot_shorter(driver_path): + """Terminal recovery must keep richer visible transcript when restored snapshot is shorter.""" + outcome = _run_scenario(driver_path, { + "action": "restore_shorter_terminal", + "state": { + "session": {"session_id": "session-5224", "message_count": 5}, + "messages": [ + {"role": "user", "content": "Question about data?", "_ts": "u1"}, + {"role": "assistant", "content": "Visible assistant answer segment one", "_ts": "a1"}, + {"role": "assistant", "content": "Visible assistant answer segment two", "_ts": "a2"}, + {"role": "assistant", "content": "**Connection interrupted:** The browser lost the live SSE connection before the response finished.", "_ts": "err"}, + ], + "activeStreamId": "stream-5224", + }, + "apiPayload": { + "session": { + "session_id": "session-5224", + "active_stream_id": None, + "pending_user_message": None, + "messages": [ + {"role": "user", "content": "Question about data?", "_ts": "u1"}, + {"role": "assistant", "content": "Visible assistant answer segment one", "_ts": "a1"}, + ], + }, + }, + "activeSid": "session-5224", + "streamId": "stream-5224", + "isActiveSession": True, + "isSessionCurrentPane": True, + "isSessionActivelyViewed": False, + }) + + assert outcome["status"] == "restored" + roles_and_contents = [(item["role"], item["content"]) for item in outcome["messages"]] + assert ("assistant", "**Connection interrupted:** The browser lost the live SSE connection before the response finished.") in roles_and_contents, ( + f"terminal error marker missing: {roles_and_contents}" + ) + assert outcome["terminalMarkerCount"] == 1, f"expected exactly one terminal marker, got {outcome['terminalMarkerCount']}" + assert any("Visible assistant answer segment two" in content for _, content in roles_and_contents), ( + "visible transcript tail was dropped after stale shorter restore" + ) + + +def test_terminal_error_restore_replaces_when_snapshots_are_fuller(driver_path): + """A fuller settled snapshot must replace the visible transcript to keep the pane current.""" + outcome = _run_scenario(driver_path, { + "action": "restore_fuller_terminal", + "state": { + "session": {"session_id": "session-5224", "message_count": 1}, + "messages": [ + {"role": "assistant", "content": "Old interrupted shell", "_ts": "old"}, + ], + "activeStreamId": "stream-5224", + }, + "apiPayload": { + "session": { + "session_id": "session-5224", + "active_stream_id": None, + "pending_user_message": None, + "messages": [ + {"role": "user", "content": "Question about data?", "_ts": "u1"}, + {"role": "assistant", "content": "Settled first segment", "_ts": "a1"}, + {"role": "assistant", "content": "Settled second segment", "_ts": "a2"}, + {"role": "assistant", "content": "Settled third segment", "_ts": "a3"}, + ], + }, + }, + "activeSid": "session-5224", + "streamId": "stream-5224", + "isActiveSession": True, + "isSessionCurrentPane": True, + "isSessionActivelyViewed": False, + }) + + assert outcome["status"] == "restored" + expected = [ + ("user", "Question about data?"), + ("assistant", "Settled first segment"), + ("assistant", "Settled second segment"), + ("assistant", "Settled third segment"), + ] + observed = [(item["role"], item["content"]) for item in outcome["messages"]] + assert observed == expected, f"fuller settled snapshot should replace visible messages: {observed}" + + +def test_terminal_error_marker_is_single_instance_and_not_duplicated(driver_path): + """Appending terminal error marker repeatedly should keep one marker and remove duplicates.""" + outcome = _run_scenario(driver_path, { + "action": "terminal_marker_idempotent", + "messages": [ + {"role": "user", "content": "Question", "_ts": "u1"}, + {"role": "assistant", "content": "Answer", "_ts": "a1"}, + ], + "activeSid": "session-5224", + "streamId": "stream-5224", + "isActiveSession": True, + "isSessionCurrentPane": True, + }) + + assert outcome["terminalMarkerCount"] == 1, f"terminal marker should be deduped to one, got {outcome['terminalMarkerCount']}" + assert outcome["messages"][-1]["content"].startswith("**Connection interrupted:**")