diff --git a/apps/presentation/dashboard/smoke/bound-conversation-smoke.ts b/apps/presentation/dashboard/smoke/bound-conversation-smoke.ts index 6b2310568c..6b840f5af5 100644 --- a/apps/presentation/dashboard/smoke/bound-conversation-smoke.ts +++ b/apps/presentation/dashboard/smoke/bound-conversation-smoke.ts @@ -159,10 +159,31 @@ try { assert.equal(recovered.pendingReads?.size, 0); assert.ok(calls.every(row => row.method === "GET"), "Recovery never starts work"); globalThis.fetch = routedRead; + // If the fresh post-history source read loses its connection, keep the + // verified pages private until a later source check succeeds. + const isBoundDiscovery = (input: RequestInfo | URL) => + new URL(String(input), origin).searchParams.get("conversation_binding_id") === bindingId; + let sourceChecks = 0; + globalThis.fetch = (input, init) => { + if (isBoundDiscovery(input) && ++sourceChecks === 2) throw new Error("Lost post-history source connection"); + return routedRead(input, init); + }; + const awaitingSource = await fetchChatHistory(options); + assert.deepEqual(awaitingSource.unavailableSessionIds, [sessionId]); + assert.deepEqual(awaitingSource.snapshots, []); + assert.deepEqual(awaitingSource.messages, []); + globalThis.fetch = routedRead; + calls.length = 0; + const sourceRecovered = await fetchChatHistory(options, awaitingSource); + assert.deepEqual(sourceRecovered.snapshots[0].messages, complete.messages); + assert.deepEqual(calls, [{path: `/api/chat/sessions?conversation_binding_id=${bindingId}`, method: "GET"}], + "A fresh check of the same unchanged binding reuses the complete verified history"); + // Current source revocation blocks recovery before another history page. child.stdin.write("deny\n"); assert.equal((await next()).source_verified, false); calls.length = 0; await assert.rejects(() => fetchChatHistory(options, interrupted)); + await assert.rejects(() => fetchChatHistory(options, awaitingSource)); assert.ok(calls.every(row => new URL(row.path, origin).pathname === "/api/chat/sessions")); child.stdin.write("recover\n"); assert.equal((await next()).source_verified, true); // Permanent denial, a changed source or malformed prefix cannot be retained @@ -223,6 +244,10 @@ try { assert.equal(resumed.messages.at(-1)?.text, "A full new original result. ".repeat(150)); assert.equal(calls.filter(row => new URL(row.path, origin).pathname === "/api/chat/sessions").length, 2, "The original source is verified again after a suffix read"); + calls.length = 0; + const changedWhileWaiting = await fetchChatHistory(options, awaitingSource); + assert.deepEqual(changedWhileWaiting.messages, resumed.messages); + assert.ok(readRequests().length > 2, "A changed transcript cannot publish the pending history"); child.stdin.write("history-rewrite\n"); assert.equal((await next()).history_rows, added.history_rows); diff --git a/apps/presentation/dashboard/src/data/chat.ts b/apps/presentation/dashboard/src/data/chat.ts index c684411979..d7c740c9f6 100644 --- a/apps/presentation/dashboard/src/data/chat.ts +++ b/apps/presentation/dashboard/src/data/chat.ts @@ -1289,12 +1289,27 @@ export function mergeChatSessionMessages(snapshots: ChatSessionSnapshot[]) { ); } +function sameHistoryRevision(current: ChatSessionSummary, read: ChatSessionSummary) { + return sameHistorySource(current, read) && Boolean(current.transcript_revision) + && current.transcript_revision === read.transcript_revision + && current.updated_at === read.updated_at && current.active_turn_id === read.active_turn_id + && current.status === read.status; +} + +function historyTransportUnavailable(error: unknown) { + return error instanceof ChatApiError && error.payload.error_code === "chat_api_unavailable" + && (error.payload.http_status === undefined || Number(error.payload.http_status) >= 500); +} + export type ChatHistory = { messages: ChatVisibleMessage[]; sessions: ChatSessionSummary[]; snapshots: ChatSessionSnapshot[]; unavailableSessionIds: string[]; pendingReads?: Map; + // Read and verified, but excluded from all displayed history until its + // original binding is checked after the transcript was read. + pendingBoundSnapshot?: {bindingId: string; snapshot: ChatSessionSnapshot}; }; export async function fetchChatHistory(options: { @@ -1312,16 +1327,21 @@ export async function fetchChatHistory(options: { 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])); + const pending = previous?.pendingBoundSnapshot; + const current = listed.sessions[0]; + // This discovery happened after the retained read. An unchanged exact + // source and revision satisfy the pending check without fetching its pages. + if (!options.managed && pending && pending.bindingId === options.bound?.bindingId && current + && sameHistoryRevision(current, pending.snapshot.session)) { + known.set(current.session_id, pending.snapshot); + } const pendingReads = new Map(previous?.pendingReads); for (const id of pendingReads.keys()) if (!listed.sessions.some(session => session.session_id === id)) pendingReads.delete(id); // A failed historical read is not an empty transcript. Retrying only the // missing snapshots keeps this recovery read-only and bounds repeated work. const missing = listed.sessions.filter((session) => { const saved = known.get(session.session_id)?.session; - return !saved || (exact && (!sameHistorySource(saved, session) || !session.transcript_revision - || session.transcript_revision !== saved.transcript_revision - || session.updated_at !== saved.updated_at || session.active_turn_id !== saved.active_turn_id - || session.status !== saved.status)); + return !saved || (exact && !sameHistoryRevision(session, saved)); }); if (exact) for (const session of missing) known.delete(session.session_id); const results = await Promise.allSettled(missing.map(async (session) => { @@ -1337,8 +1357,7 @@ export async function fetchChatHistory(options: { } catch (error) { // Reuse the hook's existing retry schedule only for a lost transport. // Revoked sources, invalid pages and stale cursors must start a fresh read. - if (!(error instanceof ChatApiError && error.payload.error_code === "chat_api_unavailable" - && (error.payload.http_status === undefined || Number(error.payload.http_status) >= 500))) pendingReads.delete(session.session_id); + if (!historyTransportUnavailable(error)) pendingReads.delete(session.session_id); throw error; } pendingReads.delete(session.session_id); @@ -1353,7 +1372,22 @@ 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(CHAT_HISTORY_READ_TIMEOUT_MS)); + if (options.bound) { + try { + await resolveBoundChatSession(options.bound, AbortSignal.timeout(CHAT_HISTORY_READ_TIMEOUT_MS)); + } catch (error) { + const read = results[0]; + if (options.managed || read?.status !== "fulfilled" || !historyTransportUnavailable(error)) throw error; + return { + messages: [], + sessions: listed.sessions, + snapshots: [], + unavailableSessionIds: listed.sessions.map(session => session.session_id), + pendingReads, + pendingBoundSnapshot: {bindingId: options.bound.bindingId, snapshot: read.value}, + }; + } + } if (options.managed) await resolveManagedChatSession(options.managed, AbortSignal.timeout(CHAT_HISTORY_READ_TIMEOUT_MS)); } for (const result of results) { diff --git a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx index c6309840d7..b78a4fe3be 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx @@ -1355,6 +1355,8 @@ export function PersonalWorkspacePage({ const retryingSteering = retainedSteering && retainedSteering.sessionId === conversationSessionId && retainedSteering.text.trim() === composer.trim() ? retainedSteering : undefined; const composerBlocked = Boolean(workspaceConversations?.entryGate) || steering || (!(retryingSteering || queueTurnId) && (sending || conversationTurnRunning)); + const historyBlocksSend = Boolean(conversationHistoryState?.sendBlocked + || (originalConversation && conversationHistoryState?.phase === "unavailable")); const quickPromptBlocked = Boolean(workspaceConversations?.entryGate) || Boolean(originalConversation && !originalConversation.session?.resumable) || steering || sending || conversationTurnRunning; // The Goal overview is for reading; its composer folds to one line until asked for. const [overviewComposerOpen, setOverviewComposerOpen] = useState(false); @@ -2406,7 +2408,7 @@ export function PersonalWorkspacePage({ if (modelSettingsSaving[composerDraftKey]) return; const pendingImages = messageOverride ? [] : imageAttachments; const message = (messageOverride ?? composer).trim() || (pendingImages.length ? t("composer.imageAnalysisPrompt") : ""); - if (!message || composerBlocked || workspaceUnavailable || (originalConversation && !originalConversation.session?.resumable) || (workspaceConversations?.managed && !workspaceConversations.managed.session?.resumable) || conversationHistoryState?.sendBlocked) return; + if (!message || composerBlocked || workspaceUnavailable || (originalConversation && !originalConversation.session?.resumable) || (workspaceConversations?.managed && !workspaceConversations.managed.session?.resumable) || historyBlocksSend) return; // A retained uncertain request must be durably recovered before any new // dispatch. Restoring its text alone cannot turn it into a fresh send. if (conversationSessionId && !readSteeringRequest(composerDraftKey)) { @@ -3066,7 +3068,7 @@ export function PersonalWorkspacePage({ {conversationHistoryState && conversationHistoryState.phase !== "ready" ? (
{t(`history.${conversationHistoryState.phase}`)} - {conversationHistoryState.sendBlocked && conversationHistoryState.phase !== "loading" + {historyBlocksSend && conversationHistoryState.phase !== "loading" ? {t("history.currentSessionRecovering")} : null}
{conversationHistoryState.phase !== "loading" ? : null} - +