From affbe50ee52b768db1397c3b83b008fbb3141e40 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 11 Nov 2026 07:29:49 +0800 Subject: [PATCH] Retry transient reads of the original Chat queue Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../smoke/bound-conversation-smoke.ts | 46 ++++++++++++++++++- apps/presentation/dashboard/src/data/chat.ts | 15 +++++- 2 files changed, 58 insertions(+), 3 deletions(-) diff --git a/apps/presentation/dashboard/smoke/bound-conversation-smoke.ts b/apps/presentation/dashboard/smoke/bound-conversation-smoke.ts index 98e60ce114..6b2310568c 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, fetchChatSessionHistory, resolveBoundChatSession, resolveBoundConversation } from "../src/data/chat"; + discoverBoundChatSession, fetchChatHistory, fetchChatQueue, 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 @@ -80,6 +80,50 @@ try { assert.equal(calls[0].path, `/api/chat/sessions?conversation_binding_id=${bindingId}`); const routedRead = globalThis.fetch; if (!fallback) { + // The original idle queue gets one bounded recovery from a temporary + // read failure; no Turn, fallback Session or permission retry is involved. + const queuePath = `/api/chat/sessions/${sessionId}/queue`; + const queueReads: string[] = []; + const unavailable = () => Response.json({ok: false, error: "Queued messages are temporarily unavailable"}, {status: 503}); + const queueInterruption = new Error("Synthetic queue transport interrupted"); + const replies: (Response | Error)[] = [unavailable()]; + globalThis.fetch = async (input, init) => { + if (String(input) !== queuePath) return routedRead(input, init); + queueReads.push(init?.method ?? "GET"); + const reply = replies.shift(); + if (reply instanceof Error) throw reply; + return reply ?? routedRead(input, init); + }; + const observedQueue = await fetchChatQueue(sessionId, AbortSignal.timeout(5000)); + assert.deepEqual([observedQueue.session_id, observedQueue.active_turn_id, observedQueue.turns], [sessionId, null, []]); + assert.deepEqual(queueReads.splice(0), ["GET", "GET"]); + replies.push(queueInterruption); + assert.equal((await fetchChatQueue(sessionId, AbortSignal.timeout(5000))).session_id, sessionId); + assert.deepEqual(queueReads.splice(0), ["GET", "GET"]); + + replies.push(unavailable(), unavailable()); + await assert.rejects(() => fetchChatQueue(sessionId, AbortSignal.timeout(5000)), /temporarily unavailable/); + assert.deepEqual(queueReads.splice(0), ["GET", "GET"], "A continued failure must remain unknown after one retry"); + replies.push(Response.json({ok: false, error: "Original scope revoked"}, {status: 403})); + await assert.rejects(() => fetchChatQueue(sessionId), /scope revoked/); + assert.deepEqual(queueReads.splice(0), ["GET"], "Never retry a denial"); + replies.push(Response.json({ok: true, schema_version: "loopx_chat_store_v1", + session_id: "different-source", active_turn_id: null, turns: []})); + await assert.rejects(() => fetchChatQueue(sessionId), /original Session queue/); + assert.deepEqual(queueReads.splice(0), ["GET"], "Never accept or retry another Session's queue"); + + const controller = new AbortController(); + globalThis.fetch = async (input, init) => { + assert.equal(String(input), queuePath); + queueReads.push(init?.method ?? "GET"); + controller.abort(); + return unavailable(); + }; + await assert.rejects(() => fetchChatQueue(sessionId, controller.signal), /temporarily unavailable/); + assert.deepEqual(queueReads, ["GET"], "Leaving this read prevents another request"); + globalThis.fetch = routedRead; + console.log("bound idle queue: one same-Session retry, denial and cancellation remain failures"); + // Repeated transport loss at later pages must advance one original prefix // through the existing retry caller, without ever publishing partial text. child.stdin.write("history\n"); diff --git a/apps/presentation/dashboard/src/data/chat.ts b/apps/presentation/dashboard/src/data/chat.ts index 92476af64e..c684411979 100644 --- a/apps/presentation/dashboard/src/data/chat.ts +++ b/apps/presentation/dashboard/src/data/chat.ts @@ -1015,8 +1015,19 @@ export type ChatQueuedTurn = z.infer["turns"][number]; /** Bounded metadata from the existing persistent queue, never a local queue. */ export async function fetchChatQueue(sessionId: string, signal?: AbortSignal) { - const result = chatQueueSchema.safeParse(await requestJson( - `/api/chat/sessions/${encodeURIComponent(sessionId)}/queue`, {signal})); + const read = () => requestJson(`/api/chat/sessions/${encodeURIComponent(sessionId)}/queue`, {signal}); + let payload: unknown; + try { + payload = await read(); + } catch (error) { + if (signal?.aborted || !(error instanceof ChatApiError)) throw error; + const {error_code: code, http_status: status} = error.payload; + const connectionInterrupted = code === "chat_api_unavailable" && (status === undefined || status === 200); + if (status !== 503 && !connectionInterrupted) throw error; + // Reuse the original Session and caller signal for one read-only retry. + payload = await read(); + } + const result = chatQueueSchema.safeParse(payload); if (!result.success || result.data.session_id !== sessionId || new Set(result.data.turns.map(row => row.turn_id)).size !== result.data.turns.length) { throw new Error("The original Session queue could not be verified.");