loopx-agent · Oct 29, 2026
@@ -4,7 +4,7 @@ import { once } from "node:events";
 import { resolve } from "node:path";
 import { createInterface } from "node:readline";
 import { resolveTestPython } from "../../../../scripts/test-python.mjs";
-import { fetchChatHistory } from "../src/data/chat";
+import { fetchChatHistory, fetchChatSessionHistory } from "../src/data/chat";
 import { readConversationReturns } from "../src/data/conversation-return-observation";
 import { conversationReturnSessions, reconcileConversationReturns } from "../src/data/conversation-returns";
 
@@ -93,11 +93,64 @@ try {
   assert.equal(transcript.find(row => row.sourceSessionId === "current"), originalCurrent, "Old updates preserve current text and object identity");
   await observe();
   assert.equal(transcript.filter(row => row.text === revisedResult).length, 1);
+  child.stdin.write("history\n");
+  const expectedRows = (await next()).history_rows;
+  requests.length = 0;
+  const complete = await fetchChatSessionHistory("old");
+  assert.equal(complete.messages.length, expectedRows);
+  assert.equal(new Set(complete.messages.map(row => row.message_id)).size, expectedRows);
+  assert.equal(complete.messages.find(row => row.message_id === "history-201")?.text, "Original full body. ".repeat(600));
+  assert.equal(requests.length, 3, "A complete original requires every real HTTP page");
+  assert.ok(requests.every(path => new URL(path, origin).searchParams.get("message_limit") === "200"));
+  assert.deepEqual((await fetchChatSessionHistory(complete.session)).messages, complete.messages,
+    "Existing discovered-summary history and delegation-ID readers share one complete-prefix owner");
+  child.stdin.write("page-fault\n");
+  assert.equal((await next()).unavailable_message_pages, true);
+  await assert.rejects(() => fetchChatSessionHistory("old"));
+  child.stdin.write("recover\n"); assert.equal((await next()).recovered, true);
+  assert.equal((await fetchChatSessionHistory("old")).messages.length, expectedRows);
+  const routedFetch = globalThis.fetch;
+  // Challenge the original-source and frozen-prefix rules, not only length.
+  for (const mutate of [
+    (value: any) => {value.session.goal_id = "unrelated-goal";},
+    (value: any) => {value.message_page.snapshot_sha256 = "0".repeat(64);},
+    (value: any) => {value.message_page.offset = 0;},
+    (value: any) => {value.messages[0].message_id = complete.messages[0].message_id;},
+    (value: any) => {delete value.message_page;},
+  ]) {
+    globalThis.fetch = async (input, init) => {
+      const response = await routedFetch(input, init);
+      if (!new URL(String(input), origin).searchParams.has("message_cursor")) return response;
+      const value = await response.json(); mutate(value);
+      return new Response(JSON.stringify(value), {status: response.status});
+    };
+    await assert.rejects(() => fetchChatSessionHistory("old"), "Never accept mixed or partial pages");
+  }
+  globalThis.fetch = routedFetch;
+  const abort = new AbortController(); abort.abort(); requests.length = 0;
+  await assert.rejects(() => fetchChatSessionHistory("old", abort.signal));
+  assert.equal(requests.length, 0, "A closed reader must not start another page read");
+  const inFlight = new AbortController();
+  let pageAborted = false;
+  globalThis.fetch = async (input, init) => {
+    if (!new URL(String(input), origin).searchParams.has("message_cursor")) return routedFetch(input, init);
+    assert.ok(init?.signal);
+    inFlight.abort();
+    pageAborted = init.signal.aborted;
+    init.signal.throwIfAborted();
+    throw new Error("The page did not receive the reader's cancellation");
+  };
+  await assert.rejects(() => fetchChatSessionHistory("old", inFlight.signal));
+  assert.equal(pageAborted, true, "The in-flight page shares the reader's cancellation");
+  // The existing full-snapshot service remains compatible on the first read.
+  globalThis.fetch = async () => new Response(JSON.stringify(complete));
+  assert.deepEqual((await fetchChatSessionHistory(complete.session)).messages, complete.messages);
+  globalThis.fetch = routedFetch;
   child.stdin.end("inspect\n");
   assert.deepEqual(await next(), { store_unchanged: true, turn_count: 0 });
   const [exitCode] = await exited;
   assert.equal(exitCode, 0, stderr);
-  console.log("conversation-history: passed (real HTTP/store, partial read, channel isolation, missing-only recovery, scoped identity, quiet index reads, late revisions, recovery, zero read-side writes or Turns)");
+  console.log("conversation-history: passed (real HTTP/store, channel isolation, missing-only recovery, quiet index reads, late revisions, complete bounded pages, source/prefix conflicts, cancellation, legacy parity, zero read-side writes or Turns)");
 } finally {
   globalThis.fetch = originalFetch;
   output.close();

@@ -2,6 +2,7 @@
 // This checks the packaged reader, not multimodal execution or message delivery.
 import assert from 'node:assert/strict';
 import {spawn} from 'node:child_process';
+import {once} from 'node:events';
 import {mkdir, writeFile} from 'node:fs/promises';
 import {resolve} from 'node:path';
 import {createInterface} from 'node:readline';
@@ -16,6 +17,7 @@ const command = testPythonCommand(`import runpy; runpy.run_path(${JSON.stringify
 const child = spawn(command.command, [...command.args, '--status-file', resolve(repo, 'examples/status.example.json'),
   ...(process.env.LOOPX_MEDIA_ASSETS ? ['--assets-dir', process.env.LOOPX_MEDIA_ASSETS] : [])],
   {cwd: repo, env: {...process.env, LOOPX_USAGE_PING: '0'}, stdio: ['pipe', 'pipe', 'pipe']});
+const exited = once(child, 'exit');
 let stderr = '';
 child.stderr.on('data', value => stderr += value);
 const queue = [], waiting = [];
@@ -56,7 +58,7 @@ try {
   });
   let mode = 'images', release, observed;
   const body = '这是原请求的完整正文。Original request evidence. '.repeat(80) + '完整正文结束。';
-  await page.route(`**/api/chat/sessions/${ready.session_id}`, async route => {
+  await page.route(url => url.pathname === `/api/chat/sessions/${ready.session_id}`, async route => {
     assert.equal(route.request().method(), 'GET');
     if (mode === 'unavailable') return route.fulfill({status: 503, json: {ok: false}});
     const response = await route.fetch(), value = await response.json();
@@ -69,6 +71,7 @@ try {
     if (mode === 'broken') source.attachments = [image('broken', '不可读图片.png', 'data:image/png;base64,AA==')];
     value.messages.push({...source, message_id: 'other-turn', turn_id: 'other-turn', text: 'Other turn must stay hidden.',
       attachments: [image('foreign', 'Foreign turn image.png')]});
+    if (value.message_page) value.message_page.total += 1;
     if (mode === 'wrong-session') value.session.session_id = 'foreign-session';
     if (mode === 'delayed') {observed(); await new Promise(resolve => release = resolve);}
     await route.fulfill({response, json: value}).catch(() => {});
@@ -245,4 +248,13 @@ try {
 } catch (error) {
   if (page) {await page.screenshot({path: resolve(output, 'failure.png')}).catch(() => {}); await writeFile(resolve(output, 'failure.txt'), await page.locator('body').innerText());}
   throw error;
-} finally {await page?.unrouteAll({behavior: 'ignoreErrors'}); await browser?.close(); child.stdin.write('stop\n'); child.stdin.end();}
+} finally {
+  await page?.unrouteAll({behavior: 'ignoreErrors'}); await browser?.close(); child.stdin.end('stop\n');
+  let timer;
+  try {
+    const [code] = await Promise.race([exited, new Promise((_, reject) => {
+      timer = setTimeout(() => reject(Error('Owned fixture did not stop')), 15000);
+    })]);
+    assert.equal(code, 0, stderr);
+  } finally {clearTimeout(timer);}
+}
@@ -1,6 +1,7 @@
-// Packaged Chat and real SQLite/Inbox; synthetic data, no model or external effects.
+// Packaged Chat and real File/SQLite Inbox; synthetic data, no model or external effects.
 import assert from 'node:assert/strict';
 import {spawn} from 'node:child_process';
+import {once} from 'node:events';
 import {mkdir, writeFile} from 'node:fs/promises';
 import {resolve} from 'node:path';
 import {createInterface} from 'node:readline';
@@ -12,8 +13,10 @@ await mkdir(output, {recursive: true});
 const fixturePath = resolve(repo, 'examples/personal-workspace-browser/steward_requests_fixture.py');
 const command = testPythonCommand(`import runpy; runpy.run_path(${JSON.stringify(fixturePath)}, run_name="__main__")`, {repoRoot: repo});
 const child = spawn(command.command, [...command.args, '--status-file', resolve(repo, 'examples/status.example.json'),
+  '--long-history', '--canonical-storage', process.env.LOOPX_REQUEST_STORAGE ?? 'sqlite',
   ...(process.env.LOOPX_SEARCH_ASSETS ? ['--assets-dir', process.env.LOOPX_SEARCH_ASSETS] : [])],
   {cwd: repo, env: {...process.env, LOOPX_USAGE_PING: '0'}, stdio: ['pipe', 'pipe', 'pipe']});
+const exited = once(child, 'exit');
 let stderr = '';
 child.stderr.on('data', value => stderr += value);
 const queue = [], waiting = [];
@@ -62,9 +65,53 @@ try {
     await rows(1);
     assert.equal(await trigger.getAttribute('aria-label'), '委派，13 条待跟进');
     await dialog.locator('.conversation-request-row > summary').click();
-    await dialog.getByText('原对话', {exact: true}).click();
-    await dialog.locator('.conversation-request-detail article').first().waitFor();
+    const original = dialog.getByText('原对话', {exact: true});
+    const articles = dialog.locator('.conversation-request-detail article');
+    // Workspace startup may attempt Session admission with this synthetic
+    // executor. It is rejected by the fixture. The original reader must add
+    // no writes of its own; keep that boundary distinct from startup.
+    const beforeOriginal = await fixture('inspect'), originalWrites = writes.length;
+    await fixture('history-corrupt'); await original.click();
+    await dialog.getByRole('alert').waitFor();
+    assert.equal(await articles.count(), 0, 'A damaged original must not look like a complete conversation');
+    await page.screenshot({path: resolve(output, 'original-unavailable.png')});
+    await fixture('history-recover'); await original.click();
+    let releaseHistory, sawHistory;
+    let historyMode = 'hold';
+    const seenHistory = new Promise(resolve => sawHistory = resolve);
+    await page.route(`**/api/chat/sessions/${ready.session_id}?*`, async route => {
+      if (!new URL(route.request().url()).searchParams.has('message_cursor')) return route.continue();
+      if (historyMode === 'fault') return route.fulfill({status: 503, json: {ok: false}});
+      const response = await route.fetch();
+      if (historyMode === 'hold') {sawHistory(); await new Promise(resolve => releaseHistory = resolve);}
+      return route.fulfill({response}).catch(() => {});
+    });
+    await original.click(); await seenHistory;
+    assert.equal(await articles.count(), 0, 'Do not show the first-page question before reading its second-page answer');
+    releaseHistory(); historyMode = 'normal';
+    await articles.nth(1).waitFor();
+    const body = '核对账单导出的字段映射\n\n' + '完整原始要求。Original evidence. '.repeat(600) + '完整正文结束。';
+    assert.equal(await articles.first().locator(':scope > .personal-md').innerText(), body);
+    assert.equal(await articles.count(), 2, 'Both original messages appear once, without unrelated Turns');
     await page.screenshot({path: resolve(output, 'search-original.png')});
+    await page.setViewportSize({width: 390, height: 844});
+    await articles.nth(1).scrollIntoViewIfNeeded();
+    assert.ok(await dialog.evaluate(node => node.scrollWidth <= node.clientWidth + 1));
+    await page.screenshot({path: resolve(output, 'original-narrow-end.png')});
+    await page.setViewportSize({width: 1440, height: 1000});
+    await original.click(); historyMode = 'fault'; await original.click();
+    await dialog.getByRole('alert').waitFor(); assert.equal(await articles.count(), 0);
+    await original.click(); historyMode = 'normal'; await original.click(); await articles.nth(1).waitFor();
+    await original.click(); historyMode = 'hold';
+    const lateHistory = new Promise(resolve => sawHistory = resolve);
+    await original.click(); await lateHistory; await original.click();
+    releaseHistory(); historyMode = 'normal';
+    assert.equal(await articles.count(), 0, 'Closing cancels the full read, including its held later page');
+    await original.click(); await articles.nth(1).waitFor();
+    await page.unroute(`**/api/chat/sessions/${ready.session_id}?*`);
+    const afterOriginal = await fixture('inspect');
+    assert.equal(afterOriginal.posts, beforeOriginal.posts);
+    assert.deepEqual(writes.slice(originalWrites), [], 'Reading, retrying and closing originals cannot submit an operation');
     await search('no matching request'); await settled();
     await dialog.getByText('未找到匹配的委派。', {exact: true}).waitFor();
     await dialog.getByRole('button', {name: '清空', exact: true}).click(); await settled();
@@ -86,7 +133,11 @@ try {
     await page.screenshot({path: resolve(output, 'read-failure.png')});
     await fixture('recover'); await dialog.getByRole('button', {name: '重试', exact: true}).click(); await settled();
     await rows(4);
-    assert.equal((await fixture('inspect')).source_unchanged, true);
+    const inspected = await fixture('inspect');
+    assert.equal(inspected.source_unchanged, true);
+    assert.equal(inspected.chat_unchanged, true);
+    assert.equal(inspected.posts, beforeOriginal.posts);
+    assert.equal(inspected.turn_count, 13, 'Reading original pages cannot execute another Turn');
     await fixture('deliver'); await search('复核迁移说明中的示例'); await settled();
     await dialog.getByText('未找到匹配的委派。', {exact: true}).waitFor();
     await dialog.getByRole('button', {name: '全部记录', exact: true}).click(); await settled();
@@ -128,9 +179,19 @@ try {
     await page.getByRole('searchbox', {name: 'Search delegations'}).waitFor();
     assert.equal(writes.some(path => /request-(?:inventory|control)|team-work/.test(path)), false);
     assert.deepEqual(errors, []);
-    console.log('PASS: packaged search, pre-pagination match, count, original conversation, clear/paging, failure/retry, returned history, stale response, Escape/focus, workspace isolation, narrow layout, English, source refusal/recovery and read-only receipts.');
+    console.log('PASS: packaged search, count, complete original across real pages, long body, corrupt journal and later-page failure/recovery, close/reopen, clear/paging, returned history, stale response, Escape/focus, workspace isolation, narrow layout, English, source refusal/recovery; original-reader POST delta zero, Chat/receipt state unchanged and no new Turns.');
   }
 } catch (error) {
   if (page) {await page.screenshot({path: resolve(output, 'failure.png')}).catch(() => {}); await writeFile(resolve(output, 'failure.txt'), await page.locator('body').innerText());}
   throw error;
-} finally {await browser?.close(); child.stdin.write('stop\n'); child.stdin.end();}
+} finally {
+  await page?.unrouteAll({behavior: 'ignoreErrors'});
+  await browser?.close(); child.stdin.end('stop\n');
+  let timer;
+  try {
+    const [code] = await Promise.race([exited, new Promise((_, reject) => {
+      timer = setTimeout(() => reject(Error('Owned fixture did not stop')), 15000);
+    })]);
+    assert.equal(code, 0, stderr);
+  } finally {clearTimeout(timer);}
+}
@@ -970,22 +970,31 @@ const chatMessagePageSchema = z.object({
 });
 
 /** Hydrate one complete original prefix; partial pages never become a snapshot. */
-async function fetchChatSessionHistory(session: ChatSessionSummary): Promise<ChatSessionSnapshot> {
+export async function fetchChatSessionHistory(session: ChatSessionSummary | string, signal?: AbortSignal): Promise<ChatSessionSnapshot> {
+  const sessionId = typeof session === "string" ? session : session.session_id;
+  // A delegation has the original Session ID; pin its first verified page's
+  // scope. History discovery callers keep validating against their own source.
+  let expectedSession = typeof session === "string" ? undefined : session;
   let first: ChatSessionSnapshot | undefined, cursor: string | null = null;
   let total: number | undefined, digest: string | undefined;
   const messages: ChatVisibleMessage[] = [], cursors = new Set<string>(), ids = new Set<string>();
   do {
+    signal?.throwIfAborted();
     const query = new URLSearchParams({message_limit: "200"});
     if (cursor !== null) query.set("message_cursor", cursor);
     const snapshot = await requestJson<ChatSessionSnapshot>(
-      `/api/chat/sessions/${encodeURIComponent(session.session_id)}?${query}`, {signal: AbortSignal.timeout(5000)});
-    const sameSource = snapshot.session?.session_id === session.session_id &&
+      `/api/chat/sessions/${encodeURIComponent(sessionId)}?${query}`,
+      {signal: signal ? AbortSignal.any([signal, AbortSignal.timeout(5000)]) : AbortSignal.timeout(5000)});
+    signal?.throwIfAborted();
+    const expected = expectedSession;
+    const sameSource = snapshot.session?.session_id === sessionId && (!expected || (
       (["agent_id", "goal_id", "channel_id", "managed_agent_id", "goal_instance_id", "project_ref"] as const)
-        .every(key => (snapshot.session[key] ?? null) === (session[key] ?? null)) &&
-      (snapshot.session.session_mode ?? "managed_runtime") === (session.session_mode ?? "managed_runtime");
+        .every(key => (snapshot.session[key] ?? null) === (expected[key] ?? null)) &&
+      (snapshot.session.session_mode ?? "managed_runtime") === (expected.session_mode ?? "managed_runtime")));
     if (snapshot.ok !== true || snapshot.schema_version !== "loopx_chat_store_v1" || !sameSource || !Array.isArray(snapshot.messages)) {
       throw new Error("The original Session scope changed during history read.");
     }
+    expectedSession ??= snapshot.session;
     // Older services return their complete v1 snapshot for this read. A later
     // page cannot silently lose the cursor contract and masquerade as complete.
     if (!Object.hasOwn(snapshot, "message_page")) {

@@ -1,6 +1,6 @@
 import {useEffect, useRef, useState} from "react";
 import {ChevronRight, CircleAlert, ListChecks, RefreshCw, X} from "lucide-react";
-import {fetchChatSession, fetchChatSessions, requestJson, type ChatImageAttachment, type ChatVisibleMessage, type ChatReplyContext} from "../../data/chat";
+import {fetchChatSessionHistory, fetchChatSessions, requestJson, type ChatImageAttachment, type ChatVisibleMessage, type ChatReplyContext} from "../../data/chat";
 import {z} from "zod";
 import {fetchConversationRequests, type ConversationRequest, type ConversationRequestInventory} from "../../data/conversation-requests";
 import {ContextHelp} from "./context-help";
@@ -170,9 +170,8 @@ function OriginalRequest({row, zh, revision, active}: {row: ConversationRequest;
     setError(false);
     if (!open || !active) {setMessages(null); return;}
     const abort = new AbortController();
-    void fetchChatSession(row.source_session_id, abort.signal).then(value => {
+    void fetchChatSessionHistory(row.source_session_id, abort.signal).then(value => {
       if (abort.signal.aborted) return;
-      if (value.session.session_id !== row.source_session_id) throw new Error("Original conversation scope changed.");
       const source = value.messages.filter(message => message.turn_id === row.source_turn_id);
       if (!source.length) throw new Error("Original turn unavailable.");
       setMessages(source);