happy5318
95c174f038
Two data-integrity blockers from round-3 review (nesquena-hermes 2026-09-19): 1. [CORE] A successful /compress returns the full transcript, but _applyManualCompressionResult kept the bounded preflight's paging signals (_messagesTruncated/_oldestIdx). Render now restores them before renderMessages(), same ordering as settle/cancel paths. 2. [SILENT] _restoreSettledSession preservation used (prefix || suffix), so repeated identical turns (colliding _messageIdentityKey) made BOTH comparisons succeed and the prefix branch spliced at the wrong offset - silently dropping rows and duplicating others. The server truncation signal now decides the strategy: truncated => suffix-exclusive, untruncated => prefix. Tests: node behavioral driver covering /compress paging restore, repeated-turn settle (no drop/dup, authoritative offset), and gap-free msg_before paging after bounded recovery.
363 lines
17 KiB
Python
363 lines
17 KiB
Python
"""Behavioral tests for #7628 review round: /compress paging restore + suffix
|
||
strategy for repeated identical turns.
|
||
|
||
Round-3 review (nesquena-hermes 2026-09-19) found two data-integrity blockers:
|
||
|
||
1. [CORE] A successful /compress returns the FULL transcript, but
|
||
`_applyManualCompressionResult` did not restore `_messagesTruncated` /
|
||
`_oldestIdx` — the bounded preflight's values (e.g. truncated=true,
|
||
oldestIdx=70) survived, producing a false "70 older messages" affordance
|
||
and shifting every absolute transcript index by 70.
|
||
|
||
2. [SILENT] `_restoreSettledSession` preservation used
|
||
`(_stagedMatchesCurrentPrefix || _stagedMatchesCurrentSuffix)`, so when
|
||
identical repeated turns made BOTH comparisons succeed the prefix branch
|
||
spliced at the wrong offset — silently dropping and duplicating rows.
|
||
The server truncation signal now decides: truncated => suffix-exclusive.
|
||
|
||
Node-backed tests extract the real functions from static/ and drive them
|
||
against a stub DOM (same harness pattern as test_issue5224).
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import shutil
|
||
import subprocess
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
_ROOT = Path(__file__).resolve().parents[1]
|
||
_COMMANDS_JS = _ROOT / "static" / "commands.js"
|
||
_MESSAGES_JS = _ROOT / "static" / "messages.js"
|
||
_SESSIONS_JS = _ROOT / "static" / "sessions.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 commandsSrc = fs.readFileSync(process.argv[2], 'utf8');
|
||
const messagesSrc = fs.readFileSync(process.argv[3], 'utf8');
|
||
const sessionsSrc = fs.readFileSync(process.argv[4], 'utf8');
|
||
const scenario = JSON.parse(process.argv[5] || '{}');
|
||
|
||
function extractFunction(source, name) {
|
||
const markers = [`async function ${name}(`, `function ${name}(${name}`, `function ${name}(`, `${name}=async(`, `${name}=async (`];
|
||
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 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 body for ${name}`);
|
||
}
|
||
|
||
function install(name, src) {
|
||
const body = extractFunction(src, name);
|
||
// Drop the `async function name(` or `function name(` prefix, keep body.
|
||
const eq = body.indexOf('{');
|
||
const inner = body.slice(eq);
|
||
globalThis[name] = new Function(`return (${name === 'jumpToSessionStart' ? 'async function' : 'function'} ${inner})`)();
|
||
}
|
||
|
||
// ---- helpers from messages.js (mirror test_issue5224 set) ----
|
||
const MESSAGE_HELPERS = [
|
||
"_isMarkerOnlyAssistantMessage", "_streamRecoveryControlMessageText",
|
||
"_streamRecoveryControlMessage", "_filterRecoveryControlMessages",
|
||
"_replaceMarkerOnlyAssistantWithStreamError", "_messageIdentityKey",
|
||
"_isHistoricalAnchorActivityScene", "_carryForwardEphemeralTurnFields",
|
||
"_isTerminalStreamErrorMarkerMessage", "_restoreSettledSession",
|
||
"_mergeSettledToolCallsWithLiveMetadata",
|
||
];
|
||
for (const h of MESSAGE_HELPERS) {
|
||
const body = extractFunction(messagesSrc, h);
|
||
const factory = new Function(`${body}; return ${h};`);
|
||
globalThis[h] = factory();
|
||
}
|
||
|
||
// ---- compress result applier from commands.js ----
|
||
const COMPRESS_BODY = extractFunction(commandsSrc, '_applyManualCompressionResult');
|
||
// Rebuild as a normal function (it contains `await` – keep async).
|
||
globalThis._applyManualCompressionResult = new Function(
|
||
'return (' + COMPRESS_BODY + ')'
|
||
)();
|
||
|
||
function runtimeStubs() {
|
||
const calls = [];
|
||
globalThis.S = { session: { session_id: 's1' }, messages: [], toolCalls: [] };
|
||
globalThis.activeSid = 's1';
|
||
globalThis.streamId = 'stream-1';
|
||
globalThis.assistantText = false;
|
||
globalThis._messagesTruncated = !!scenario.prestateTruncated;
|
||
globalThis._oldestIdx = scenario.prestateOldestIdx || 0;
|
||
globalThis._messagesGeneration = 1;
|
||
globalThis._loadingOlder = false;
|
||
globalThis._loadingSessionId = null;
|
||
globalThis._messageRenderWindowSize = 50;
|
||
globalThis._msgLimitMax = scenario.msgLimitMax || 200;
|
||
globalThis._INITIAL_MSG_LIMIT = 50;
|
||
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._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._cancelThrottledSnapshotTimer = () => calls.push('cancelThrottledSnapshot');
|
||
globalThis._clearAnchorProseIncrementalNode = () => calls.push('clearAnchorProse');
|
||
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.updateQueueBadge = () => calls.push('updateQueueBadge');
|
||
globalThis._setActivePaneIdleIfOwner = () => calls.push('setActivePaneIdle');
|
||
globalThis.setBusy = () => calls.push('setBusy');
|
||
globalThis.setComposerStatus = () => calls.push('setComposerStatus');
|
||
globalThis.setCompressionUi = () => calls.push('setCompressionUi');
|
||
globalThis.setStatus = () => calls.push('setStatus');
|
||
globalThis.msgContent = (m) => (m && m.content && typeof m.content === 'string' ? m.content : '');
|
||
globalThis._isContextCompactionMessage = () => false;
|
||
globalThis._setCompressionSessionLock = () => calls.push('setCompressionSessionLock');
|
||
globalThis._messageRenderableMessageCount = () => scenario.messageRenderableCount || 50;
|
||
globalThis._currentMessageRenderWindowSize = () => scenario.currentWindowSize || 12;
|
||
globalThis._restoreMessageRenderWindowAfterSettledRender = () => calls.push('restoreRenderWindow');
|
||
globalThis.projectSessionArtifactsForOwner = () => calls.push('projectArtifacts');
|
||
globalThis._streamFinalized = false;
|
||
globalThis._persistTimer = null;
|
||
globalThis._attachProjectedAnchorSceneToLastAssistant = () => calls.push('attachProjected');
|
||
globalThis._isMessagePaneNearBottom = () => true;
|
||
globalThis._isMessageReaderUnpinned = () => false;
|
||
globalThis._queueDrainSid = null;
|
||
// reducer used by _restoreSettledSession (mirrors live code)
|
||
return calls;
|
||
}
|
||
|
||
(async () => {
|
||
const calls = runtimeStubs();
|
||
|
||
if (scenario.action === 'compress_restores_paging') {
|
||
// Pre-state: bounded preflight left truncated=true, oldestIdx=70.
|
||
globalThis._messagesTruncated = true;
|
||
globalThis._oldestIdx = 70;
|
||
const fullRows = [];
|
||
for (let i = 0; i < 100; i++) {
|
||
fullRows.push({ role: i % 2 === 0 ? 'user' : 'assistant', content: `row-${i}`, _ts: `ts-${i}` });
|
||
}
|
||
const payload = {
|
||
session: {
|
||
session_id: 's1',
|
||
message_count: 100,
|
||
messages: fullRows,
|
||
tool_calls: [],
|
||
_messages_truncated: false,
|
||
_messages_offset: 0,
|
||
},
|
||
};
|
||
await globalThis._applyManualCompressionResult(payload, '', 100, '/compress');
|
||
console.log(JSON.stringify({
|
||
action: 'compress_restores_paging',
|
||
messagesTruncated: !!globalThis._messagesTruncated,
|
||
oldestIdx: globalThis._oldestIdx,
|
||
messageCountAfter: Array.isArray(S.messages) ? S.messages.length : -1,
|
||
calls,
|
||
}));
|
||
return;
|
||
}
|
||
|
||
if (scenario.action === 'settle_repeated_turns') {
|
||
// Bounded settle against a transcript containing repeated identical turns
|
||
// (same role/content but distinct _ids). Round-2 bug: BOTH prefix and
|
||
// suffix matched, prefix won, and rows were dropped+duplicated.
|
||
const state = scenario.state || {};
|
||
globalThis.S.session = { session_id: 's1', message_count: state.messageCount };
|
||
globalThis.S.messages = state.messages || [];
|
||
globalThis.streamId = 'stream-7628';
|
||
globalThis.S.activeStreamId = 'stream-7628';
|
||
globalThis.api = async () => scenario.apiPayload || { session: null };
|
||
const status = await globalThis._restoreSettledSession({}, { status: true, preserveVisibleOnShorterTerminalSnapshot: true });
|
||
const msgs = Array.isArray(globalThis.S.messages) ? globalThis.S.messages : [];
|
||
console.log(JSON.stringify({
|
||
action: 'settle_repeated_turns',
|
||
status,
|
||
messages: msgs.map((m) => ({ role: m.role, content: m.content, _ts: m._ts, _id: m._id || null })),
|
||
oldestIdx: globalThis._oldestIdx,
|
||
messagesTruncated: !!globalThis._messagesTruncated,
|
||
terminalMarkerCount: msgs.filter(globalThis._isTerminalStreamErrorMarkerMessage).length,
|
||
calls,
|
||
}));
|
||
return;
|
||
}
|
||
|
||
if (scenario.action === 'load_older_paging') {
|
||
// After a bounded recovery (offset=70), "Load earlier" must request a page
|
||
// keyed off _oldestIdx (no gap and no overlap), and merge must close the
|
||
// window: final oldestIdx === requested offset.
|
||
globalThis._messagesTruncated = true;
|
||
globalThis._oldestIdx = scenario.oldestIdx || 70;
|
||
const existing = [];
|
||
for (let i = 70; i < 100; i++) existing.push({ role: i % 2 === 0 ? 'user' : 'assistant', content: `row-${i}`, _ts: `ts-${i}` });
|
||
globalThis.S.messages = existing;
|
||
globalThis.S.session = { session_id: 's1', message_count: 100 };
|
||
const requested = [];
|
||
globalThis.api = async (url, opts) => {
|
||
requested.push(String(url));
|
||
// Return one older page of size 50 (rows 20..69).
|
||
const older = [];
|
||
for (let i = 20; i < 70; i++) older.push({ role: i % 2 === 0 ? 'user' : 'assistant', content: `row-${i}`, _ts: `ts-${i}` });
|
||
return {
|
||
session: { session_id: 's1', message_count: 100, messages: older, _messages_truncated: true, _messages_offset: 20 },
|
||
};
|
||
};
|
||
// Extract the live loader from sessions.js.
|
||
const loaderBody = extractFunction(sessionsSrc, '_loadOlderMessages');
|
||
const loaderFactory = new Function(`${loaderBody}; return _loadOlderMessages;`);
|
||
const loader = loaderFactory();
|
||
await loader.call(globalThis);
|
||
console.log(JSON.stringify({
|
||
action: 'load_older_paging',
|
||
requested,
|
||
messagesTruncated: !!globalThis._messagesTruncated,
|
||
oldestIdx: globalThis._oldestIdx,
|
||
messagesLen: Array.isArray(S.messages) ? S.messages.length : -1,
|
||
loadingOlder: !!globalThis._loadingOlder,
|
||
}));
|
||
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("7628_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(_COMMANDS_JS), str(_MESSAGES_JS), str(_SESSIONS_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_compress_success_restores_paging_state(driver_path):
|
||
"""A successful /compress returning the full transcript must reset the
|
||
bounded preflight's paging signals before render (#7628 blocker 1)."""
|
||
outcome = _run_scenario(driver_path, {"action": "compress_restores_paging"})
|
||
assert outcome["messagesTruncated"] is False, "compress success must clear _messagesTruncated"
|
||
assert outcome["oldestIdx"] == 0, f"compress success must reset _oldestIdx to 0, got {outcome['oldestIdx']}"
|
||
assert outcome["messageCountAfter"] == 100
|
||
|
||
|
||
def test_settle_bounded_preserves_no_drop_no_dup(driver_path):
|
||
"""A bounded settle against repeated identical turns must not drop or
|
||
duplicate rows (#7628 blocker 2, exact reviewer reproduction).
|
||
|
||
All user rows share one identity key (same role/content/_ts), all assistant
|
||
rows share one too — _messageIdentityKey collides across every pair. A
|
||
bounded tail [u3,a3,u2,a2] therefore matches BOTH the visible prefix
|
||
(u1,a1,u2,a2…) and the visible suffix (…u2,a2) at key level. The round-2
|
||
`||` code preferred the prefix branch: splice offset = staged.length
|
||
dropped u1/a1 and re-emitted u3/a3 (duplicate). The server truncation
|
||
signal must force suffix-exclusive splicing.
|
||
"""
|
||
_u = {"role": "user", "content": "What's the status?", "_ts": "t-u"}
|
||
_a = {"role": "assistant", "content": "All good.", "_ts": "t-a"}
|
||
# Visible before settle: u1,a1,u2,a2,u3,a3 + Connection-interrupted marker.
|
||
state_messages = [
|
||
{**_u, "_id": 1}, {**_a, "_id": 2},
|
||
{**_u, "_id": 3}, {**_a, "_id": 4},
|
||
{**_u, "_id": 5}, {**_a, "_id": 6},
|
||
{"role": "assistant", "content": "**Connection interrupted:** The browser lost the live SSE connection before the response finished.", "_ts": "t-err", "_id": 7},
|
||
]
|
||
outcome = _run_scenario(driver_path, {
|
||
"action": "settle_repeated_turns",
|
||
"isSessionCurrentPane": True,
|
||
"state": {
|
||
"messageCount": 12,
|
||
"messages": state_messages,
|
||
},
|
||
"apiPayload": {
|
||
"session": {
|
||
"session_id": "s1",
|
||
"active_stream_id": None,
|
||
"pending_user_message": None,
|
||
"message_count": 12,
|
||
"messages": [
|
||
# Reviewer's tail: [u3,a3,u2,a2] — identity keys collide
|
||
# with u1/a1/u2/a2/u3/a3 above, so prefix AND suffix both
|
||
# compare equal.
|
||
{**_u, "_id": 8}, {**_a, "_id": 9},
|
||
{**_u, "_id": 10}, {**_a, "_id": 11},
|
||
],
|
||
"_messages_truncated": True,
|
||
"_messages_offset": 4,
|
||
},
|
||
},
|
||
})
|
||
assert outcome["status"] == "restored", outcome
|
||
ids = [m["_id"] for m in outcome["messages"]]
|
||
# No dropped id, no dup id (the prefix-wins bug duplicated u3/a3 and
|
||
# dropped u1/a1 — every retained row must be unique).
|
||
assert len(ids) == len(set(ids)), f"duplicate rows survived settle: {ids}"
|
||
# Staged (authoritative) rows must all be present, plus the marker once.
|
||
assert set(ids) == {8, 9, 10, 11, 7}, f"unexpected transcript after fix: {ids}"
|
||
# Identity (role|_ts|content) counts: staged rows legitimately repeat
|
||
# (two identical user pairs in the tail), but the bug added a THIRD copy by
|
||
# splicing the visible tail back in. user<=2, assistant<=2, marker exactly 1.
|
||
keys = [(m["role"], m["_ts"], m["content"]) for m in outcome["messages"]]
|
||
kc = {}
|
||
for k in keys:
|
||
kc[k] = kc.get(k, 0) + 1
|
||
user_key = ("user", "t-u", "What's the status?")
|
||
asst_key = ("assistant", "t-a", "All good.")
|
||
assert kc.get(user_key, 0) == 2, f"user rows must appear exactly twice (staged), got {kc.get(user_key, 0)}: {keys}"
|
||
assert kc.get(asst_key, 0) == 2, f"assistant rows must appear exactly twice (staged), got {kc.get(asst_key, 0)}: {keys}"
|
||
assert outcome["terminalMarkerCount"] == 1, outcome
|
||
assert outcome["oldestIdx"] == 4, f"offset must follow server, got {outcome['oldestIdx']}"
|
||
assert outcome["messagesTruncated"] is True
|
||
|
||
|
||
def test_bounded_recovery_prepares_gap_free_load_earlier(driver_path):
|
||
"""After a bounded recovery (oldestIdx=70), Load Earlier must page from the
|
||
authoritative offset so the fetched range has no gap and no overlap
|
||
(#7628)."""
|
||
outcome = _run_scenario(driver_path, {"action": "load_older_paging", "oldestIdx": 70, "msgLimitMax": 80})
|
||
assert outcome["requested"], "loader must have issued an api call"
|
||
url = outcome["requested"][0]
|
||
assert "msg_before=70" in url, f"load-earlier must page from authoritative offset: {url}"
|
||
assert "msg_limit=" in url, url |