loopx-agent · Oct 26, 2026 · Oct 26, 2026 · Oct 26, 2026
@@ -540,13 +540,13 @@ async function checkSessionHistoryRecoveryContract() {
   globalThis.fetch = (async (input) => {
     const url = String(input);
     requests.push(url);
-    const body = url.includes("?")
+    const body = new URL(url, "http://localhost").pathname === "/api/chat/sessions"
       ? {
           ok: true,
           schema_version: "loopx_chat_session_list_v1",
           sessions: [failed.session, ready.session],
         }
-      : url.endsWith("session-failed")
+      : new URL(url, "http://localhost").pathname.endsWith("session-failed")
         ? failed
         : ready;
     return new Response(JSON.stringify(body), {
@@ -676,8 +676,66 @@ async function checkActiveTurnStreamRecoveryContract() {
   }
 }
 
+async function checkPagedHistoryContract() {
+  const originalFetch = globalThis.fetch;
+  const rows = Array.from({length: 451}, (_, index) => ({
+    message_id: `original-${index}`, turn_id: null, role: index % 2 ? "agent" : "user",
+    text: index === 201 ? "完整正文".repeat(3000) : "Deliberately repeated text",
+    created_at: "2026-08-10T02:00:00Z",
+    ...(index === 201 ? {attachments: [{id: "original-image", name: "original.png", mime_type: "image/png", size: 1, data_url: "data:image/png;base64,AA=="}]} : {}),
+  }));
+  const original = historySnapshot("paged-original", "ready", rows);
+  let failure = "", reads: string[] = [], sourceReads = 0;
+  const bindingId = "a".repeat(24);
+  globalThis.fetch = (async (input, init) => {
+    check(!init?.method || init.method === "GET", "History paging must not execute work");
+    const url = new URL(String(input), "http://localhost");
+    if (url.pathname === "/api/chat/sessions") {
+      sourceReads += 1;
+      return Response.json({ok: true, schema_version: "loopx_chat_session_list_v1",
+        conversation_binding_id: failure === "revoked" && sourceReads > 1 ? "b".repeat(24) : bindingId, sessions: [original.session]});
+    }
+    const cursor = url.searchParams.get("message_cursor");
+    reads.push(cursor ?? "first");
+    const offset = cursor === "page-200" ? 200 : cursor === "page-400" ? 400 : 0;
+    if (offset && failure === "transport") return Response.json({error: "interrupted page"}, {status: 503});
+    const messages = rows.slice(offset, offset + 200);
+    const next = offset === 0 ? "page-200" : offset === 200 ? "page-400" : null;
+    return Response.json({...original, messages,
+      session: {...original.session, ...(offset && failure === "identity" ? {channel_id: "another-channel"} : {})},
+      message_page: {schema_version: "loopx_chat_message_page_v1", offset: offset && failure === "offset" ? 0 : offset,
+        total: offset && failure === "total" ? 452 : rows.length,
+        snapshot_sha256: offset && failure === "digest" ? "b".repeat(64) : "a".repeat(64),
+        next_cursor: offset && failure === "cursor" ? cursor : next}});
+  }) as typeof fetch;
+  try {
+    const history = await fetchChatHistory({channelId: "manager"});
+    check(history.messages.length === rows.length && history.unavailableSessionIds.length === 0,
+      "The default history owner must drain all original pages before claiming a complete snapshot");
+    check(reads.join(",") === "first,page-200,page-400", "Read the original advancing cursors exactly once");
+    check(JSON.stringify(history.snapshots[0].messages) === JSON.stringify(rows),
+      "Page boundaries must retain append order, repeated bodies, full long text and attachments");
+    for (failure of ["transport", "identity", "offset", "total", "digest", "cursor"]) {
+      reads = [];
+      const incomplete = await fetchChatHistory({channelId: "manager"});
+      check(incomplete.messages.length === 0 && incomplete.unavailableSessionIds[0] === original.session.session_id,
+        `An interrupted or inconsistent ${failure} page must not become a complete history`);
+      check(reads.length === 2, "A failed cursor must stop the read, without a loop or replay");
+    }
+    failure = "";
+    check((await fetchChatHistory({channelId: "manager"})).messages.length === rows.length,
+      "Retry must recover the full original snapshot after a partial-page failure");
+    failure = "revoked"; sourceReads = 0;
+    let refused = false;
+    try { await fetchChatHistory({channelId: "manager", bound: {bindingId, sessionId: original.session.session_id}}); }
+    catch { refused = true; }
+    check(refused && sourceReads === 2, "Recheck the original source after all pages; a revoked binding cannot publish the old read");
+  } finally { globalThis.fetch = originalFetch; }
+}
+
 void checkSessionCloseContract()
   .then(checkSessionHistoryRecoveryContract)
+  .then(checkPagedHistoryContract)
   .then(checkActiveTurnStreamRecoveryContract)
   .then(() => console.log("chat-route-smoke: ok"))
   .catch((error) => {

@@ -6,6 +6,7 @@
 import sys
 import tempfile
 import threading
+from urllib.parse import parse_qs, urlparse
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[4]))
 
@@ -16,7 +17,9 @@
 
 class HistoryReadFault(ChatRequestHandler):
     def _session_snapshot(self, session_id: str) -> None:
-        if session_id == self.server.unavailable_session_id:
+        query = parse_qs(urlparse(self.path).query)
+        if (session_id == self.server.unavailable_session_id
+                or (session_id == "old" and self.server.unavailable_message_pages and "message_cursor" in query)):
             self._send_error("Synthetic history read unavailable", status=503)
         else:
             super()._session_snapshot(session_id)
@@ -50,6 +53,7 @@ def main() -> None:
         server.chat_store = store
         server.runtime_controller = runtime
         server.unavailable_session_id = "old"
+        server.unavailable_message_pages = False
         thread = threading.Thread(target=server.serve_forever, daemon=True)
         thread.start()
         print(json.dumps({"origin": f"http://127.0.0.1:{server.server_port}"}), flush=True)
@@ -58,7 +62,19 @@ def main() -> None:
                 command = command.strip()
                 if command == "recover":
                     server.unavailable_session_id = ""
+                    server.unavailable_message_pages = False
                     print(json.dumps({"recovered": True}), flush=True)
+                elif command == "history":
+                    for index in range(450):
+                        store.append_message("old", role="agent", message_id=f"history-{index}",
+                            text="Original full body. " * 600 if index == 201 else f"Paged original row {index}")
+                    before = {str(path.relative_to(store.sessions_root)): path.read_bytes()
+                              for path in store.sessions_root.rglob("*") if path.is_file()}
+                    print(json.dumps({"history_rows": len(store.messages("old"))}), flush=True)
+                elif command == "page-fault":
+                    server.unavailable_session_id = ""
+                    server.unavailable_message_pages = True
+                    print(json.dumps({"unavailable_message_pages": True}), flush=True)
                 elif command == "fault":
                     server.unavailable_session_id = "old"
                     print(json.dumps({"unavailable": "old"}), flush=True)

@@ -26,14 +26,16 @@ async function next() {
 }
 async function command(value) { fixture.stdin.write(`${value}\n`); return next(); }
 let server, browser, workspace;
+const reads = [];
 try {
   const { origin } = await next();
   await command("recover");
+  assert.deepEqual(await command("history"), {history_rows: 451});
+  await command("page-fault");
   server = await startServer();
   const url = `http://127.0.0.1:${port}/chat/?statusUrl=/status.json`;
   await waitForHttp(url);
   browser = await launchBrowser(loadPlaywright().chromium);
-  const reads = [];
   workspace = await openWorkspacePage({ newPage: options => browser.newPage({ locale: "zh-CN", ...options }) }, url, {
     beforeGoto(_api, page) {
       return page.route("**/api/chat/sessions**", async route => {
@@ -53,15 +55,43 @@ try {
         }
         assert.equal(route.request().method(), "GET", "Observation cannot start work");
         const response = await route.fetch({ url: new URL(path + new URL(route.request().url()).search, origin).href });
-        reads.push({ path, status: response.status() });
+        const snapshot = response.status() === 200 ? await response.json() : null;
+        reads.push({ path, query: new URL(route.request().url()).search, status: response.status(), offset: snapshot?.message_page?.offset,
+          sessions: snapshot?.sessions?.map(row => ({id: row.session_id, revision: row.transcript_revision})) });
         await route.fulfill({ response });
       });
     },
   });
   const { page, api } = workspace;
   const chat = () => page.getByRole("navigation", { name: "管家视图" }).getByRole("button", { name: /^(Chat|对话)$/ }).click();
   await chat();
+  const notice = page.getByRole("status").filter({hasText: "部分历史暂时无法读取"});
+  await notice.waitFor({state: "visible"});
+  assert.ok(reads.some(row => row.path === "/api/chat/sessions/old" && row.status === 503));
+  assert.equal(await page.getByText("Earlier public report", {exact: true}).count(), 0,
+    "The first real HTTP page must not masquerade as a complete history");
+  const composer = page.getByLabel("向 LoopX 发送消息");
+  await composer.fill("Keep this draft while original history recovers.");
+  await mkdir(outputDir, {recursive: true});
+  await page.screenshot({path: resolve(outputDir, "conversation-pagination-before-desktop.png"), animations: "disabled"});
+  await command("recover");
+  await notice.getByRole("button", {name: "重试读取", exact: true}).click();
   await page.getByText("Earlier public report", { exact: true }).waitFor({ state: "visible" });
+  await page.getByText("Paged original row 449", {exact: true}).waitFor({state: "visible"});
+  assert.equal(await page.getByText("Original full body. ".repeat(600).trim(), {exact: true}).count(), 1);
+  await notice.waitFor({state: "hidden"});
+  assert.deepEqual(reads.filter(row => row.path === "/api/chat/sessions/old" && row.status === 200 && typeof row.offset === "number").slice(-3).map(row => row.offset), [0, 200, 400]);
+  assert.equal(await composer.inputValue(), "Keep this draft while original history recovers.");
+  await page.screenshot({path: resolve(outputDir, "conversation-pagination-after-desktop.png"), animations: "disabled"});
+  await page.setViewportSize({width: 390, height: 844});
+  assert.ok(await page.locator("body").evaluate(body => body.scrollWidth <= innerWidth + 1));
+  await page.screenshot({path: resolve(outputDir, "conversation-pagination-after-narrow.png"), animations: "disabled"});
+  await page.setViewportSize({width: 1512, height: 982});
+  if (process.argv.includes("--pagination-only")) {
+    assert.equal(api.turnRequests.length, 0);
+    assert.deepEqual(await command("inspect"), {store_unchanged: true, turn_count: 0});
+    console.log("conversation-pagination-browser: passed (packaged UI, real HTTP/store, 451 complete rows, second-page failure/retry, full body, draft and zero Turns)");
+  } else {
   await command("result");
   await page.getByText("First checked result", { exact: true }).waitFor({ state: "visible", timeout: 12_000 });
   const firstLink = page.locator('a[href="https://example.org/forms/result"]');
@@ -73,9 +103,9 @@ try {
   await page.waitForTimeout(6500);
   assert.equal(oldReadCount(), settledReads, "Quiet completed transcripts must not be downloaded repeatedly");
   await page.locator(".personal-goal-link").first().click();
+  const failedRead = page.waitForResponse(response => new URL(response.url()).pathname === "/api/chat/sessions/old" && response.status() === 503);
   await command("fault");
   await command("revision");
-  const failedRead = page.waitForResponse(response => new URL(response.url()).pathname === "/api/chat/sessions/old" && response.status() === 503);
   await failedRead;
   const recoveredRead = page.waitForResponse(response => new URL(response.url()).pathname === "/api/chat/sessions/old" && response.status() === 200);
   await command("recover");
@@ -96,6 +126,11 @@ try {
   assert.ok(await page.locator("body").evaluate(body => body.scrollWidth <= innerWidth + 1));
   await page.screenshot({ path: resolve(outputDir, "conversation-return-revisions-mobile.png"), animations: "disabled" });
   console.log("conversation-return-browser: passed (packaged UI, real HTTP/store, quiet reads, background revision, read recovery, no duplicate or model Turn)");
+  }
+} catch (error) {
+  console.error(JSON.stringify({recent_history_reads: reads.filter(row => row.path !== "/api/chat/sessions").slice(-10),
+    recent_indexes: reads.filter(row => row.path === "/api/chat/sessions").slice(-8)}, null, 2));
+  throw error;
 } finally {
   await workspace?.page.unrouteAll({ behavior: "wait" });
   await workspace?.close();

@@ -940,12 +940,69 @@ export type ChatSessionSnapshot = {
   session: ChatSessionSummary;
   messages: ChatVisibleMessage[];
   active_turn: Record<string, unknown> | null;
+  message_page?: {
+    schema_version: "loopx_chat_message_page_v1";
+    offset: number; total: number; snapshot_sha256: string; next_cursor: string | null;
+  };
 };
 
 export async function fetchChatSession(sessionId: string, signal?: AbortSignal) {
   return requestJson<ChatSessionSnapshot>(`/api/chat/sessions/${encodeURIComponent(sessionId)}`, { signal });
 }
 
+const chatMessagePageSchema = z.object({
+  schema_version: z.literal("loopx_chat_message_page_v1"),
+  offset: z.number().int().nonnegative(), total: z.number().int().nonnegative(),
+  snapshot_sha256: z.string().regex(/^[a-f0-9]{64}$/), next_cursor: z.string().min(1).max(1024).nullable(),
+});
+
+/** Hydrate one complete original prefix; partial pages never become a snapshot. */
+async function fetchChatSessionHistory(session: ChatSessionSummary): Promise<ChatSessionSnapshot> {
+  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 {
+    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 &&
+      (["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");
+    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.");
+    }
+    // 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")) {
+      if (cursor !== null) throw new Error("The history page contract changed during read.");
+      return snapshot;
+    }
+    const parsed = chatMessagePageSchema.safeParse(snapshot.message_page);
+    if (!parsed.success) throw new Error("The original history page could not be verified.");
+    const page = parsed.data, end = messages.length + snapshot.messages.length;
+    if (page.offset !== messages.length || snapshot.messages.length > 200 || end > page.total
+      || (total !== undefined && total !== page.total) || (digest !== undefined && digest !== page.snapshot_sha256)
+      || (page.next_cursor !== null) !== (end < page.total)
+      || (page.next_cursor !== null && (!snapshot.messages.length || cursors.has(page.next_cursor)))) {
+      throw new Error("The original history cursor did not advance consistently.");
+    }
+    for (const message of snapshot.messages) {
+      if (typeof message.message_id !== "string" || !message.message_id || ids.has(message.message_id)) {
+        throw new Error("The original history contains conflicting message identities.");
+      }
+      ids.add(message.message_id); messages.push(message);
+    }
+    first ??= snapshot;
+    total = page.total; digest = page.snapshot_sha256; cursor = page.next_cursor;
+    if (cursor !== null) cursors.add(cursor);
+  } while (cursor !== null);
+  // Keep the first read's execution/revision hint. An append between pages is
+  // intentionally outside this prefix and will trigger the normal next refresh.
+  return {...first!, messages, message_page: undefined};
+}
+
 /** Read-only discovery through the existing Core Goal/Agent/Session identity. */
 export async function fetchManagedAgentSessions(goalId: string, agentId: string, signal?: AbortSignal) {
   if (!/^[A-Za-z0-9._-]{1,160}$/.test(goalId) || !/^[A-Za-z0-9][A-Za-z0-9._-]{0,119}$/.test(agentId)) throw new Error("Invalid registered Agent address.");
@@ -1067,7 +1124,7 @@ export async function fetchChatHistory(options: {
   });
   if (exact) for (const session of missing) known.delete(session.session_id);
   const results = await Promise.allSettled(missing.map(async (session) => {
-    const snapshot = await fetchChatSession(session.session_id, AbortSignal.timeout(5000));
+    const snapshot = await fetchChatSessionHistory(session);
     if (snapshot.session.session_id !== session.session_id || snapshot.session.agent_id !== session.agent_id) {
       throw new Error("The original Session identity changed during history read.");
     }
@@ -1076,6 +1133,12 @@ export async function fetchChatHistory(options: {
       || snapshot.session.session_mode !== "managed_runtime")) throw new Error("The original managed Session scope changed during history read.");
     return snapshot;
   }));
+  // A long page sequence must not publish against a connection/Goal removed
+  // while it was in flight. Reuse the original source owner, never a fallback.
+  if (missing.length && results.some(result => result.status === "fulfilled")) {
+    if (options.bound) await resolveBoundChatSession(options.bound, AbortSignal.timeout(5000));
+    if (options.managed) await resolveManagedChatSession(options.managed, AbortSignal.timeout(5000));
+  }
   for (const result of results) {
     if (result.status === "fulfilled") known.set(result.value.session.session_id, result.value);
   }