From 94e45ed32b67c143986e808af17a9df683e63488 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 16:48:47 +0800 Subject: [PATCH] fix(chat): allow verified history reads to finish Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../smoke/bound-conversation-http-fixture.py | 25 ++++++++++++++++ .../smoke/bound-conversation-smoke.ts | 30 ++++++++++++++++++- apps/presentation/dashboard/src/data/chat.ts | 15 ++++++---- 3 files changed, 64 insertions(+), 6 deletions(-) diff --git a/apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py b/apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py index ea7bcd8302..58a0014b05 100644 --- a/apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py +++ b/apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py @@ -13,6 +13,7 @@ import sys import tempfile import threading +import time import pytest @@ -135,6 +136,24 @@ def start(self, server): print(json.dumps({"origin": f"http://127.0.0.1:{server.server_port}", "bindingId": binding["binding_id"], "sessionId": sid, "ordinarySessionId": other["session_id"]}), flush=True) settings_handler = server.RequestHandlerClass._private_conversations + history_handler = server.RequestHandlerClass._session_snapshot + sessions_handler = server.RequestHandlerClass._list_sessions + slow_page, slow_source = threading.Event(), threading.Event() + + def delayed_history(handler, selected_session): + if selected_session == sid and slow_page.is_set(): + slow_page.clear() + time.sleep(5.2) + return history_handler(handler, selected_session) + + def delayed_sessions(handler): + if slow_source.is_set() and "conversation_binding_id=" in handler.path: + slow_source.clear() + time.sleep(5.2) + return sessions_handler(handler) + + server.RequestHandlerClass._session_snapshot = delayed_history + server.RequestHandlerClass._list_sessions = delayed_sessions def control(): for line in sys.stdin: @@ -150,6 +169,12 @@ def control(): elif command == "close": server.chat_store.update_session(sid, status="closed") result = {"closed": True} + elif command in {"slow-history", "slow-history-cancel"}: + slow_page.set() + result = {"slow_history": True} + elif command == "slow-history-source": + slow_source.set() + result = {"slow_history_source": True} elif command == "replace-peer" and self.peer_provider is not None: old = next(row for row in bindings.read()["bindings"] if row["transport_ref"] == self.peer_provider.transport_ref) bindings.disconnect(old["binding_id"], expected_revision=bindings.read()["revision"]) diff --git a/apps/presentation/dashboard/smoke/bound-conversation-smoke.ts b/apps/presentation/dashboard/smoke/bound-conversation-smoke.ts index 0b5712e570..b5d3828561 100644 --- a/apps/presentation/dashboard/smoke/bound-conversation-smoke.ts +++ b/apps/presentation/dashboard/smoke/bound-conversation-smoke.ts @@ -5,7 +5,7 @@ import { resolve } from "node:path"; import { createInterface } from "node:readline"; import { resolveTestPython } from "../../../../scripts/test-python.mjs"; import { acceptChatTurn, conversationContextKey, conversationContextOfKey, conversationSupportsStatusProjection, createChatSession, - discoverBoundChatSession, fetchChatHistory, fetchChatProjects, resolveBoundChatSession, resolveBoundConversation } from "../src/data/chat"; + discoverBoundChatSession, fetchChatHistory, fetchChatProjects, fetchChatSessionHistory, resolveBoundChatSession, resolveBoundConversation } from "../src/data/chat"; // Invariant: continuation and refreshed history retain the verified original // Session. An absent/ignored selector, changed source or closed Session cannot @@ -78,6 +78,34 @@ try { await fetchChatHistory(options, history); assert.equal(calls.length, 1, "Quiet refresh verifies the source using only its compact index"); assert.equal(calls[0].path, `/api/chat/sessions?conversation_binding_id=${bindingId}`); + const routedRead = globalThis.fetch; + if (!fallback) { + // Real HTTP responses beyond the old interactive deadline must still hydrate + // the complete original history. Source verification keeps its own deadline. + child.stdin.write("slow-history\n"); + assert.equal((await next()).slow_history, true); + const slow = await fetchChatHistory(options); + assert.deepEqual(slow.messages, history.messages); + assert.deepEqual(slow.unavailableSessionIds, []); + // Delay only the post-history fresh source check, not initial discovery. + const beforeSlowSource = calls.length; + globalThis.fetch = async (input, init) => { + const response = await routedRead(input, init); + if (String(input).startsWith(`/api/chat/sessions/${sessionId}?`)) { + child.stdin.write("slow-history-source\n"); + assert.equal((await next()).slow_history_source, true); + } + return response; + }; + const sourceSlow = await fetchChatHistory(options); + assert.deepEqual(sourceSlow.messages, history.messages); + assert.deepEqual(sourceSlow.unavailableSessionIds, []); + assert.ok(calls.slice(beforeSlowSource).every(row => row.method === "GET")); + globalThis.fetch = routedRead; + child.stdin.write("slow-history-cancel\n"); + assert.equal((await next()).slow_history, true); + await assert.rejects(() => fetchChatSessionHistory(original.session, AbortSignal.timeout(50))); + } let updated = history; if (!fallback) { child.stdin.write("source\n"); diff --git a/apps/presentation/dashboard/src/data/chat.ts b/apps/presentation/dashboard/src/data/chat.ts index 749832fee6..f8f705c65c 100644 --- a/apps/presentation/dashboard/src/data/chat.ts +++ b/apps/presentation/dashboard/src/data/chat.ts @@ -1011,6 +1011,11 @@ const chatMessagePageSchema = z.object({ turn_id: z.string().min(1).optional(), }); +// A long original transcript can outlive the interactive request budget even +// when each verified page succeeds. Apply the same bounded read deadline to +// its pages and fresh source checks; caller cancellation still takes priority. +const CHAT_HISTORY_READ_TIMEOUT_MS = 30_000; + /** Hydrate one complete original prefix; partial pages never become a snapshot. */ export async function fetchChatSessionHistory(session: ChatSessionSummary | string, signal?: AbortSignal, turnId?: string): Promise { const sessionId = typeof session === "string" ? session : session.session_id; @@ -1027,7 +1032,7 @@ export async function fetchChatSessionHistory(session: ChatSessionSummary | stri if (cursor !== null) query.set("message_cursor", cursor); const snapshot = await requestJson( `/api/chat/sessions/${encodeURIComponent(sessionId)}?${query}`, - {signal: signal ? AbortSignal.any([signal, AbortSignal.timeout(5000)]) : AbortSignal.timeout(5000)}); + {signal: signal ? AbortSignal.any([signal, AbortSignal.timeout(CHAT_HISTORY_READ_TIMEOUT_MS)]) : AbortSignal.timeout(CHAT_HISTORY_READ_TIMEOUT_MS)}); signal?.throwIfAborted(); const expected = expectedSession; const sameSource = snapshot.session?.session_id === sessionId && (!expected || ( @@ -1176,8 +1181,8 @@ export async function fetchChatHistory(options: { }, previous?: ChatHistory): Promise { // Source authorization must be re-read even when recovering an older snapshot. const exact = Boolean(options.bound || options.managed); - const listed = options.managed ? {sessions: [await resolveManagedChatSession(options.managed, AbortSignal.timeout(5000))]} - : (!options.bound ? previous : undefined) ?? await fetchChatSessions({ ...options, signal: AbortSignal.timeout(5000) }); + const listed = options.managed ? {sessions: [await resolveManagedChatSession(options.managed, AbortSignal.timeout(CHAT_HISTORY_READ_TIMEOUT_MS))]} + : (!options.bound ? previous : undefined) ?? await fetchChatSessions({ ...options, signal: AbortSignal.timeout(CHAT_HISTORY_READ_TIMEOUT_MS) }); const known = new Map(previous?.snapshots.map((snapshot) => [snapshot.session.session_id, snapshot])); // A failed historical read is not an empty transcript. Retrying only the // missing snapshots keeps this recovery read-only and bounds repeated work. @@ -1202,8 +1207,8 @@ export async function fetchChatHistory(options: { // 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)); + if (options.bound) await resolveBoundChatSession(options.bound, AbortSignal.timeout(CHAT_HISTORY_READ_TIMEOUT_MS)); + if (options.managed) await resolveManagedChatSession(options.managed, AbortSignal.timeout(CHAT_HISTORY_READ_TIMEOUT_MS)); } for (const result of results) { if (result.status === "fulfilled") known.set(result.value.session.session_id, result.value);