fix(chat): preserve visible transcript on terminal stream failure (#5224)

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);

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:**")