diff --git a/static/sessions.js b/static/sessions.js index d2de922e..f2f6ed31 100644 --- a/static/sessions.js +++ b/static/sessions.js @@ -2033,17 +2033,8 @@ async function loadSession(sid){ if(typeof startSessionStream==='function') startSessionStream(S.session.session_id); - function _mergePendingSessionMessage(session,messages){ - if(!Array.isArray(messages)) return false; - const pendingMsg=typeof getPendingSessionMessage==='function'?getPendingSessionMessage(session,messages):null; - if(!pendingMsg) return false; - const liveAssistantIdx=messages.findIndex(m=>m&&m.role==='assistant'&&m._live); - const currentTurnMessages=liveAssistantIdx>=0?messages.slice(0,liveAssistantIdx):messages; - if(_hasCurrentTailUserDuplicate(currentTurnMessages,pendingMsg)) return false; - if(liveAssistantIdx>=0) messages.splice(liveAssistantIdx,0,pendingMsg); - else messages.push(pendingMsg); - return true; - } + // _mergePendingSessionMessage is the global identity-aware helper shared by + // loadSession and refreshSession; see its definition below. // Phase 2a: If session is streaming, restore the persisted transcript first, // then merge the local INFLIGHT live tail. INFLIGHT is a recovery tail, not a @@ -3298,6 +3289,32 @@ function _hasCurrentTailUserDuplicate(messages,candidate){ return !!(existing&&_sameTranscriptMessage(existing,candidate)); } +// Keep pending-user recovery ordering identical across load, reconnect, and +// explicit refresh paths. The pending prompt owns the live assistant tail and +// must be projected before it, regardless of which recovery response arrived. +function _mergePendingSessionMessage(session,messages){ + if(!Array.isArray(messages)) return false; + const liveAssistantIdx=messages.findIndex(m=>m&&m.role==='assistant'&&m._live); + const currentTurnMessages=liveAssistantIdx>=0?messages.slice(0,liveAssistantIdx):messages; + const pendingMsg=typeof getPendingSessionMessage==='function'?getPendingSessionMessage(session,currentTurnMessages):null; + if(!pendingMsg) return false; + if(_hasCurrentTailUserDuplicate(currentTurnMessages,pendingMsg)) return false; + if(liveAssistantIdx>=0){ + const misplacedIdx=messages.findIndex((m,idx)=> + idx>liveAssistantIdx&&m&&m.role==='user'&&_sameTranscriptMessage(m,pendingMsg) + ); + if(misplacedIdx>=0){ + const [misplacedUser]=messages.splice(misplacedIdx,1); + messages.splice(liveAssistantIdx,0,misplacedUser); + }else{ + messages.splice(liveAssistantIdx,0,pendingMsg); + } + }else{ + messages.push(pendingMsg); + } + return true; +} + function _currentTurnAssistantText(messages){ const list=Array.isArray(messages)?messages:[]; let start=-1; diff --git a/static/ui.js b/static/ui.js index 71de6a9a..d2c0a85f 100644 --- a/static/ui.js +++ b/static/ui.js @@ -9569,9 +9569,11 @@ async function refreshSession() { S.messages = data.session.messages || []; _messagesTruncated = !!data.session._messages_truncated; _oldestIdx = data.session._messages_offset || 0; - const pendingMsg=getPendingSessionMessage(data.session,S.messages); - if(pendingMsg) S.messages.push(pendingMsg); - S.activeStreamId=data.session.active_stream_id||null; + if (typeof _mergePendingSessionMessage !== 'function') { + throw new Error('Pending-session merge helper unavailable'); + } + _mergePendingSessionMessage(data.session, S.messages); + S.activeStreamId = data.session.active_stream_id || null; syncTopbar(); _renderMessagesWithScrollSnapshot(); showToast('Conversation refreshed'); diff --git a/tests/test_cross_session_message_load_isolation.py b/tests/test_cross_session_message_load_isolation.py index 141f25ff..b7dd5460 100644 --- a/tests/test_cross_session_message_load_isolation.py +++ b/tests/test_cross_session_message_load_isolation.py @@ -91,6 +91,7 @@ LOAD_SESSION_SRC = _extract_function(SESSIONS_SRC, "loadSession") ENSURE_MESSAGES_LOADED_SRC = _extract_function(SESSIONS_SRC, "_ensureMessagesLoaded") INFLIGHT_HAS_VISIBLE_STATE_SRC = _extract_function(SESSIONS_SRC, "_inflightHasVisibleLiveState") SELECT_LIVE_RECOVERY_INFLIGHT_SRC = _extract_function(SESSIONS_SRC, "_selectLiveRecoveryInflight") +MERGE_PENDING_SESSION_MESSAGE_SRC = _extract_function(SESSIONS_SRC, "_mergePendingSessionMessage") def _normalise_ws(s: str) -> str: @@ -348,6 +349,7 @@ let toastCalls = []; // Source under test __INFLIGHT_HAS_VISIBLE_STATE_SRC__ __SELECT_LIVE_RECOVERY_INFLIGHT_SRC__ +__MERGE_PENDING_SESSION_MESSAGE_SRC__ __LOAD_SESSION_SRC__ __ENSURE_MESSAGES_LOADED_SRC__ @@ -576,10 +578,12 @@ runAll() ''' -def _run_node(script: str) -> dict: +def _run_node(script: str, tmp_path: Path) -> dict: assert NODE is not None, "node is required" + script_path = tmp_path / "cross-session-message-load-isolation.mjs" + script_path.write_text(script, encoding="utf-8") completed = subprocess.run( - [NODE, "--input-type=module", "-e", script], + [NODE, str(script_path)], cwd=str(REPO), text=True, capture_output=True, @@ -593,7 +597,7 @@ def _run_node(script: str) -> dict: @pytest.mark.skipif(NODE is None, reason="node not on PATH") -def test_loadsession_cross_session_ordering_and_stale_reject_behavior(): +def test_loadsession_cross_session_ordering_and_stale_reject_behavior(tmp_path): script = ( _NODE_SCRIPT_TEMPLATE.replace( "__INFLIGHT_HAS_VISIBLE_STATE_SRC__", INFLIGHT_HAS_VISIBLE_STATE_SRC @@ -601,10 +605,13 @@ def test_loadsession_cross_session_ordering_and_stale_reject_behavior(): .replace( "__SELECT_LIVE_RECOVERY_INFLIGHT_SRC__", SELECT_LIVE_RECOVERY_INFLIGHT_SRC ) + .replace( + "__MERGE_PENDING_SESSION_MESSAGE_SRC__", MERGE_PENDING_SESSION_MESSAGE_SRC + ) .replace("__LOAD_SESSION_SRC__", LOAD_SESSION_SRC) .replace("__ENSURE_MESSAGES_LOADED_SRC__", ENSURE_MESSAGES_LOADED_SRC) ) - body = _run_node(script) + body = _run_node(script, tmp_path) cross = body["crossSessionOrdering"] stale = body["staleIdleCatch"] diff --git a/tests/test_gateway_sse_reconnect_dedupe.py b/tests/test_gateway_sse_reconnect_dedupe.py index 4001a965..05db8e06 100644 --- a/tests/test_gateway_sse_reconnect_dedupe.py +++ b/tests/test_gateway_sse_reconnect_dedupe.py @@ -129,7 +129,9 @@ if(!_isDuplicateGatewaySessionSnapshot([null, {session_id:'web-2', session_sourc def test_load_session_persists_only_after_metadata_loads(): """Do not overwrite the last good localStorage sid before /api/session succeeds.""" src = _read(SESSIONS_JS) - load = _block(src, "async function loadSession(sid)", "function _mergePendingSessionMessage") + # _mergePendingSessionMessage was lifted to a top-level helper for #6419, + # so use a stable boundary inside loadSession instead. + load = _block(src, "async function loadSession(sid)", "// Phase 2a:") api_pos = load.index("data = await api(`/api/session") persist_pos = load.index("localStorage.setItem('hermes-webui-session',S.session.session_id)") diff --git a/tests/test_issue6419_pending_user_reconnect_order.py b/tests/test_issue6419_pending_user_reconnect_order.py new file mode 100644 index 00000000..d8424833 --- /dev/null +++ b/tests/test_issue6419_pending_user_reconnect_order.py @@ -0,0 +1,263 @@ +# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd. +# SPDX-License-Identifier: AGPL-3.0 + +"""#6419: mid-stream reconnect must keep pending user message before live response.""" + +import json +import shutil +import subprocess +from pathlib import Path + +import pytest + + +ROOT = Path(__file__).resolve().parents[1] +NODE = shutil.which("node") +UI_JS = (ROOT / "static" / "ui.js").read_text(encoding="utf-8") +SESSIONS_JS = (ROOT / "static" / "sessions.js").read_text(encoding="utf-8") + + +def _function_source(src: str, name: str) -> str: + start = src.find(f"function {name}(") + assert start != -1, f"{name} not found" + brace = src.find("{", start) + depth = 0 + for i, ch in enumerate(src[brace:]): + if ch == "{": + depth += 1 + elif ch == "}": + depth -= 1 + if depth == 0: + return src[start : brace + i + 1] + raise AssertionError(f"{name} body unterminated") + + +def _function_body(src: str, name: str) -> str: + source = _function_source(src, name) + return source[source.find("{") :] + + +# ─── Shared helper contract ──────────────────────────────────────────────── + +def test_merge_pending_session_message_is_a_global_helper(): + """The fix must expose _mergePendingSessionMessage as a single, top-level + identity-aware helper rather than duplicated insertion logic nested inside + individual recovery paths.""" + import re + + # Exactly one definition in sessions.js. + defs = [m.start() for m in re.finditer(r"\bfunction _mergePendingSessionMessage\b", SESSIONS_JS)] + assert len(defs) == 1, ( + f"_mergePendingSessionMessage should be defined exactly once; found {len(defs)}. " + "The fix must consolidate helper logic at one chokepoint, not duplicate it." + ) + # The sole definition must be at a top-level scope (zero indentation), not nested. + line_start = SESSIONS_JS.rfind("\n", 0, defs[0]) + 1 + indent = len(SESSIONS_JS[line_start:defs[0]]) - len(SESSIONS_JS[line_start:defs[0]].lstrip()) + assert indent == 0, ( + f"_mergePendingSessionMessage must be at module top level (indent=0); found indent={indent}." + ) + + +def test_merge_helper_inserts_pending_user_before_live_assistant(): + """Given a live assistant row, the pending user row must appear before it.""" + body = _function_source(SESSIONS_JS, "_mergePendingSessionMessage") + assert "getPendingSessionMessage" in body, "helper derives the candidate row" + assert "findIndex(m=>m&&m.role==='assistant'&&m._live)" in body, ( + "helper looks for the live assistant boundary" + ) + assert "messages.splice(liveAssistantIdx,0,pendingMsg)" in body, ( + "helper inserts pending user before the live assistant" + ) + assert "messages.push(pendingMsg)" in body, ( + "helper falls back to append only when there is no live assistant" + ) + assert "_hasCurrentTailUserDuplicate" in body, ( + "helper deduplicates against the current tail user row" + ) + + +def test_refreshSession_uses_shared_helper(): + """refreshSession() must no longer append pending_user_message at the end + unconditionally; it must route through _mergePendingSessionMessage so that a + live assistant tail during recovery keeps the user prompt before it.""" + body = _function_body(UI_JS, "refreshSession") + assert "_mergePendingSessionMessage" in body, ( + "refreshSession uses the shared identity-aware merge helper" + ) + assert body.find("_mergePendingSessionMessage") < body.find("renderMessages"), ( + "merge must happen before the transcript is re-rendered" + ) + # The old unconditional push must be gone. + assert "if(pendingMsg) S.messages.push(pendingMsg)" not in body, ( + "refreshSession must not unconditionally push the pending message" + ) + + +# ─── loadSession reattach path (was already correct before #6419) ───────────────── + +def test_loadSession_inflight_reattach_merges_pending_user_before_render(): + """Regression for the #2341 contract: loadSession INFLIGHT branch must call + the shared helper and render afterwards.""" + start = SESSIONS_JS.find("if(INFLIGHT[sid]){") + assert start != -1, "loadSession INFLIGHT branch not found" + end = SESSIONS_JS.find("}else{", start) + assert end != -1, "loadSession INFLIGHT branch end not found" + block = SESSIONS_JS[start:end] + + merge_pos = block.find("_mergePendingSessionMessage") + render_pos = block.find("renderMessages(") + assert merge_pos != -1, "INFLIGHT branch must merge pending user message" + assert render_pos != -1, "INFLIGHT branch must render messages" + assert merge_pos < render_pos, ( + "pending user row must be merged before renderMessages() rebuilds the transcript" + ) + + +def _run_node(script: str) -> dict: + completed = subprocess.run( + [NODE, "-e", script], + cwd=ROOT, + text=True, + capture_output=True, + check=False, + ) + assert completed.returncode == 0, completed.stderr or completed.stdout + return json.loads(completed.stdout) + + +@pytest.mark.skipif(NODE is None, reason="node not on PATH") +def test_merge_helper_repairs_malformed_order_and_is_idempotent(): + """A reconnect can already contain [live assistant, pending user]. The + canonical helper must move that exact user row before the live boundary, + preserve its metadata/attachments, and remain ordered on repeated probes.""" + helpers = "\n".join( + [ + _function_source(SESSIONS_JS, "_messageComparableText"), + _function_source(SESSIONS_JS, "_stripAttachedFilesMarker"), + _function_source(SESSIONS_JS, "_stripForcedSkillEnvelope"), + _function_source(SESSIONS_JS, "_normalizeUserTranscriptText"), + _function_source(SESSIONS_JS, "_sameTranscriptMessage"), + _function_source(SESSIONS_JS, "_currentTailUserMessage"), + _function_source(SESSIONS_JS, "_hasCurrentTailUserDuplicate"), + _function_source(SESSIONS_JS, "_mergePendingSessionMessage"), + ] + ) + script = f""" +{helpers} +function getPendingSessionMessage(session, messages){{ + const text=String(session.pending_user_message||'').trim(); + if(!text) return null; + return {{role:'user',content:text,_pending:true,_ts:session.pending_started_at}}; +}} +const historical={{role:'user',content:'same prompt',_ts:1}}; +const settled={{role:'assistant',content:'old answer',_ts:2}}; +const live={{role:'assistant',content:'working',_live:true,_ts:4}}; +const misplaced={{ + role:'user',content:'same prompt',_pending:true,_ts:3, + attachments:[{{path:'proof.txt'}}],_source:'resume' +}}; +const messages=[historical,settled,live,misplaced]; +const session={{pending_user_message:'same prompt',pending_started_at:3}}; +const first=_mergePendingSessionMessage(session,messages); +const afterFirst=messages.map(m=>({{role:m.role,content:m.content,live:!!m._live,ts:m._ts,attachments:m.attachments,source:m._source}})); +const second=_mergePendingSessionMessage(session,messages); +process.stdout.write(JSON.stringify({{first,second,afterFirst,final:messages}})); +""" + result = _run_node(script) + + assert result["first"] is True + assert result["second"] is False + assert [row["role"] for row in result["afterFirst"]] == [ + "user", + "assistant", + "user", + "assistant", + ] + repaired = result["afterFirst"][2] + assert repaired["ts"] == 3 + assert repaired["attachments"] == [{"path": "proof.txt"}] + assert repaired["source"] == "resume" + assert sum(1 for row in result["final"] if row.get("_ts") == 3) == 1 + + +@pytest.mark.skipif(NODE is None, reason="node not on PATH") +def test_refresh_session_uses_canonical_helper_before_render(): + """The actual refresh path must project [user, live assistant] and render + once when the canonical helper is available.""" + helpers = "\n".join( + [ + _function_source(SESSIONS_JS, "_messageComparableText"), + _function_source(SESSIONS_JS, "_stripAttachedFilesMarker"), + _function_source(SESSIONS_JS, "_stripForcedSkillEnvelope"), + _function_source(SESSIONS_JS, "_normalizeUserTranscriptText"), + _function_source(SESSIONS_JS, "_sameTranscriptMessage"), + _function_source(SESSIONS_JS, "_currentTailUserMessage"), + _function_source(SESSIONS_JS, "_hasCurrentTailUserDuplicate"), + _function_source(SESSIONS_JS, "_mergePendingSessionMessage"), + ] + ) + refresh = "async " + _function_source(UI_JS, "refreshSession") + script = f""" +{helpers} +let rendered=0; +let status=''; +const S={{session:{{session_id:'sid-1'}},messages:[]}}; +const window={{_restartingForUpdate:false}}; +function dismissReconnect(){{}} +function getPendingSessionMessage(session, messages){{ + const text=String(session.pending_user_message||'').trim(); + if(!text) return null; + return {{role:'user',content:text,_pending:true,_ts:session.pending_started_at}}; +}} +async function api(){{return {{session:{{ + session_id:'sid-1', + active_stream_id:'stream-1', + pending_user_message:'prompt', + pending_started_at:3, + messages:[{{role:'assistant',content:'working',_live:true,_ts:4}}] +}}}};}} +function syncTopbar(){{}} +function _renderMessagesWithScrollSnapshot(){{rendered+=1;}} +function showToast(){{}} +function setStatus(value){{status=value;}} +{refresh} +refreshSession().then(()=>process.stdout.write(JSON.stringify({{rendered,status,roles:S.messages.map(m=>m.role)}}))); +""" + result = _run_node(script) + + assert result["rendered"] == 1 + assert result["status"] == "" + assert result["roles"] == ["user", "assistant"] + + +@pytest.mark.skipif(NODE is None, reason="node not on PATH") +def test_refresh_session_fails_closed_when_canonical_helper_is_unavailable(): + """Partial bundles must not reintroduce the old append path. If the canonical + helper is unavailable, refreshSession reports failure and never renders an + incorrectly ordered transcript.""" + refresh = "async " + _function_source(UI_JS, "refreshSession") + script = f""" +let rendered=0; +let status=''; +const S={{session:{{session_id:'sid-1'}},messages:[]}}; +const window={{_restartingForUpdate:false}}; +function dismissReconnect(){{}} +async function api(){{return {{session:{{ + session_id:'sid-1', + active_stream_id:'stream-1', + pending_user_message:'prompt', + messages:[{{role:'assistant',content:'working',_live:true}}] +}}}};}} +function syncTopbar(){{}} +function _renderMessagesWithScrollSnapshot(){{rendered+=1;}} +function showToast(){{}} +function setStatus(value){{status=value;}} +{refresh} +refreshSession().then(()=>process.stdout.write(JSON.stringify({{rendered,status,roles:S.messages.map(m=>m.role)}}))); +""" + result = _run_node(script) + + assert result["rendered"] == 0 + assert result["roles"] == ["assistant"] + assert "Pending-session merge helper unavailable" in result["status"]