fix(session): restore paging after compress; suffix-exclusive settle strategy (review round 3)

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.

tests/test_7628_recovery_paging.py

@@ -0,0 +1,363 @@
+"""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