summaryrefslogtreecommitdiffhomepage
path: root/src/features/chat
diff options
context:
space:
mode:
Diffstat (limited to 'src/features/chat')
-rw-r--r--src/features/chat/logic/tool-summary.ts8
-rw-r--r--src/features/chat/recovery.test.ts529
-rw-r--r--src/features/chat/store.svelte.ts128
-rw-r--r--src/features/chat/ui/ChatView.svelte10
-rw-r--r--src/features/chat/ui/ChatView.test.ts44
5 files changed, 709 insertions, 10 deletions
diff --git a/src/features/chat/logic/tool-summary.ts b/src/features/chat/logic/tool-summary.ts
index 566e40f..f6d1569 100644
--- a/src/features/chat/logic/tool-summary.ts
+++ b/src/features/chat/logic/tool-summary.ts
@@ -381,7 +381,9 @@ export function viewToolEntry(entry: ToolEntry): ToolCardView {
const builtInSummary = summarize?.(args).trim();
const callSummary =
entry.call === null
- ? "Call in earlier history"
+ ? entry.recovering
+ ? "Recovering tool input…"
+ : "Call in earlier history"
: isEmptyInput(input)
? ""
: oneLine(
@@ -416,7 +418,9 @@ export function viewToolEntry(entry: ToolEntry): ToolCardView {
entry.call === null
? {
kind: "unavailable",
- text: "This call is earlier in the conversation. Show earlier messages to see it.",
+ text: entry.recovering
+ ? "Tool input is unavailable until it is persisted. Recovering from conversation history…"
+ : "This call is earlier in the conversation. Show earlier messages to see it.",
}
: detail(entry.toolName, input, args),
resultSummary,
diff --git a/src/features/chat/recovery.test.ts b/src/features/chat/recovery.test.ts
new file mode 100644
index 0000000..4b46288
--- /dev/null
+++ b/src/features/chat/recovery.test.ts
@@ -0,0 +1,529 @@
+import type {
+ ChatResyncRequiredMessage,
+ ConversationHistoryResponse,
+} from "@dispatch/transport-contract";
+import type { AgentEvent, StepId, StoredChunk, TurnEndedEvent } from "@dispatch/wire";
+import { IDBFactory, IDBKeyRange } from "fake-indexeddb";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { createIdbChunkStore } from "../../adapters/idb";
+import { createConversationCache } from "../conversation-cache";
+import type { HistorySync } from "./ports";
+import { createChatStore } from "./store.svelte";
+import {
+ createFakeCache,
+ createFakeHistorySync,
+ createFakeMetricsSync,
+ createFakeTransport,
+} from "./test-helpers";
+
+const step = "t1#0" as StepId;
+const opening: StoredChunk = {
+ seq: 1,
+ role: "user",
+ turnId: "t1",
+ chunk: { type: "text", text: "prompt" },
+};
+const call: StoredChunk = {
+ seq: 3,
+ role: "assistant",
+ turnId: "t1",
+ chunk: {
+ type: "tool-call",
+ toolCallId: "a",
+ toolName: "run_shell",
+ stepId: step,
+ input: { command: "full input" },
+ },
+};
+const result: StoredChunk = {
+ seq: 4,
+ role: "tool",
+ turnId: "t1",
+ chunk: {
+ type: "tool-result",
+ toolCallId: "a",
+ toolName: "run_shell",
+ stepId: step,
+ content: "full output",
+ isError: false,
+ },
+};
+const committed: StoredChunk[] = [
+ opening,
+ { seq: 2, role: "assistant", turnId: "t1", chunk: { type: "text", text: "complete text" } },
+ call,
+ result,
+];
+
+function fixture(historySync?: HistorySync, batched = false) {
+ const history = createFakeHistorySync();
+ const transport = createFakeTransport();
+ const cache = createFakeCache();
+ const scheduled: (() => void)[] = [];
+ const store = createChatStore({
+ conversationId: "c1",
+ createRequestId: () => "r1",
+ historySync: historySync ?? history.impl,
+ metricsSync: createFakeMetricsSync().impl,
+ cache: cache.impl,
+ transport: transport.impl,
+ ...(batched ? { scheduleFlush: (fn: () => void) => scheduled.push(fn) } : {}),
+ });
+ const event = (event: AgentEvent) => store.handleDelta({ type: "chat.delta", event });
+ const notice = (
+ turnId: string | undefined = "t1",
+ reason: ChatResyncRequiredMessage["reason"] = "replay-limit",
+ ) =>
+ store.handleDelta({
+ type: "chat.resync-required",
+ conversationId: "c1",
+ reason,
+ ...(turnId !== undefined ? { turnId } : {}),
+ });
+ const end = (reason: TurnEndedEvent["reason"] = "sealed", turnId = "t1") =>
+ event({ type: "turn-ended", conversationId: "c1", turnId, reason });
+ const text = (delta: string, turnId = "t1") =>
+ event({ type: "text-delta", conversationId: "c1", turnId, delta });
+ return { store, history, transport, cache, scheduled, event, notice, end, text };
+}
+
+async function settled(store: ReturnType<typeof createChatStore>) {
+ for (let i = 0; i < 80 && store.pendingSync; i++) await Promise.resolve();
+ expect(store.pendingSync).toBe(false);
+}
+function response(chunks: readonly StoredChunk[], historyEpoch = 0): ConversationHistoryResponse {
+ return { chunks, historyEpoch, latestSeq: chunks.at(-1)?.seq ?? 0 };
+}
+function deferred() {
+ let resolve!: (value: ConversationHistoryResponse) => void;
+ const promise = new Promise<ConversationHistoryResponse>((r) => {
+ resolve = r;
+ });
+ return { promise, resolve };
+}
+
+afterEach(() => {
+ vi.useRealTimers();
+ vi.unstubAllGlobals();
+});
+
+describe("omitted chat recovery", () => {
+ it.each([
+ false,
+ true,
+ ])("marks omitted replay active without content and recovers full HTTP payload (batched=%s)", async (batched) => {
+ vi.useFakeTimers();
+ const f = fixture(undefined, batched);
+ f.text("partial");
+ f.notice();
+ expect(f.store.generating).toBe(true);
+ expect(f.store.recovering).toBe(true);
+ expect(f.store.chunks.map((r) => r.chunk)).toEqual([{ type: "text", text: "partial" }]);
+ await settled(f.store);
+ // No initial commit yet. A tool result follows an OMITTED call input.
+ f.event({
+ type: "tool-result",
+ conversationId: "c1",
+ turnId: "t1",
+ stepId: step,
+ toolCallId: "a",
+ toolName: "run_shell",
+ content: "short replay",
+ isError: false,
+ });
+ f.history.returnChunks = committed;
+ await vi.advanceTimersByTimeAsync(1000);
+ expect(f.store.chunks.filter((r) => r.chunk.type === "tool-call")).toEqual([
+ expect.objectContaining({ chunk: call.chunk, provisional: false }),
+ ]);
+ expect(f.store.chunks.filter((r) => r.chunk.type === "tool-result")).toHaveLength(1);
+ expect(f.store.chunks.find((r) => r.chunk.type === "tool-result")?.chunk).toEqual(result.chunk);
+ expect(f.store.generating).toBe(true);
+ expect(f.store.error).toBeNull();
+ expect(f.transport.sent).toEqual([]);
+ f.store.dispose();
+ expect(vi.getTimerCount()).toBe(0);
+ });
+
+ it("coalesces repeat notices and polls after step notification races persistence, without parallel requests", async () => {
+ vi.useFakeTimers();
+ const pending = deferred();
+ let calls = 0;
+ const f = fixture(async () => {
+ calls++;
+ return calls === 1 ? pending.promise : response(committed);
+ });
+ f.notice();
+ for (let i = 0; i < 20; i++) f.notice();
+ for (let i = 0; i < 15; i++) await Promise.resolve();
+ expect(calls).toBe(1);
+ f.event({ type: "step-complete", conversationId: "c1", turnId: "t1", stepId: step });
+ await vi.advanceTimersByTimeAsync(10000);
+ expect(calls).toBe(1);
+ pending.resolve(response([]));
+ await settled(f.store);
+ expect(calls).toBe(1);
+ await vi.advanceTimersByTimeAsync(999);
+ expect(calls).toBe(1);
+ await vi.advanceTimersByTimeAsync(1);
+ expect(calls).toBe(2);
+ expect(f.store.chunks.map((r) => r.chunk)).toEqual(committed.map((r) => r.chunk));
+ f.store.dispose();
+ });
+
+ it.each([
+ "sealed",
+ "aborted",
+ "error",
+ "rejected",
+ ] as const)("final-syncs %s behind an in-flight fetch and preserves immediate queued carry", async (reason) => {
+ vi.useFakeTimers();
+ const pending = deferred();
+ let calls = 0;
+ const f = fixture(async () => (++calls === 1 ? pending.promise : response(committed)));
+ f.notice();
+ for (let i = 0; i < 10; i++) await Promise.resolve();
+ f.text("old tail");
+ f.end(reason);
+ f.event({ type: "user-message", conversationId: "c1", turnId: "t2", text: "queued input" });
+ f.event({ type: "turn-start", conversationId: "c1", turnId: "t2" });
+ f.text("carried tail", "t2");
+ pending.resolve(response([opening]));
+ await settled(f.store);
+ expect(calls).toBe(2);
+ expect(f.store.recovering).toBe(false);
+ expect(f.store.generating).toBe(true);
+ expect(f.store.chunks.filter((r) => r.provisional).map((r) => r.chunk)).toEqual([
+ { type: "text", text: "queued input" },
+ { type: "text", text: "carried tail" },
+ ]);
+ expect(f.store.chunks.filter((r) => r.turnId === "t1").every((r) => !r.provisional)).toBe(true);
+ await vi.advanceTimersByTimeAsync(5000);
+ expect(calls).toBe(2);
+ f.end("sealed", "t2");
+ expect(f.store.generating).toBe(false);
+ f.store.dispose();
+ });
+
+ it.each([
+ false,
+ true,
+ ])("notice during %s-cache load requests one rerun, preserving current window", async (warm) => {
+ vi.useFakeTimers();
+ const pending = deferred();
+ let calls = 0;
+ const f = fixture(async (_id, _since, window) => {
+ calls++;
+ if (!warm) expect(window?.limit).toBe(192);
+ return calls === 1 ? pending.promise : response([opening]);
+ });
+ if (warm) f.cache.store.set("c1", [opening]);
+ const load = f.store.load();
+ for (let i = 0; i < 15; i++) await Promise.resolve();
+ f.notice();
+ pending.resolve(response([opening]));
+ await load;
+ await settled(f.store);
+ expect(calls).toBe(2);
+ expect(f.store.chunks).toHaveLength(1);
+ expect(f.store.recovering).toBe(true);
+ expect(f.store.generating).toBe(true);
+ f.store.dispose();
+ });
+
+ it("payload notice after a completed HTTP load preserves live tail, unrelated turn and tool progress", async () => {
+ vi.useFakeTimers();
+ const f = fixture();
+ await f.store.load();
+ f.event({ type: "turn-start", conversationId: "c1", turnId: "other" });
+ f.text("unrelated", "other");
+ f.event({ type: "turn-start", conversationId: "c1", turnId: "t1" });
+ f.text("partial text");
+ f.event({
+ type: "tool-call",
+ conversationId: "c1",
+ turnId: "t1",
+ stepId: step,
+ toolCallId: "a",
+ toolName: "run_shell",
+ input: { command: "short" },
+ });
+ f.event({
+ type: "tool-result",
+ conversationId: "c1",
+ turnId: "t1",
+ stepId: step,
+ toolCallId: "a",
+ toolName: "run_shell",
+ content: "short",
+ isError: false,
+ });
+ f.text("complete text"); // same payload as earlier step; NOT the same identity
+ f.event({
+ type: "tool-output",
+ conversationId: "c1",
+ turnId: "t1",
+ toolCallId: "next",
+ data: "progress",
+ stream: "stdout",
+ });
+ f.history.returnChunks = committed;
+ f.notice("t1", "payload-limit");
+ await settled(f.store);
+ expect(f.store.chunks.filter((r) => r.provisional).map((r) => r.chunk)).toEqual([
+ { type: "text", text: "unrelated" },
+ { type: "text", text: "complete text" },
+ ]);
+ expect(f.store.chunks.filter((r) => r.chunk.type === "tool-call")).toHaveLength(1);
+ expect(f.store.chunks.filter((r) => r.chunk.type === "tool-result")).toHaveLength(1);
+ f.store.dispose();
+ });
+
+ it("bounds epoch churn during recovery and cancels its retry timer on disposal", async () => {
+ vi.useFakeTimers();
+ let calls = 0;
+ const f = fixture(async () => {
+ f.store.invalidateHistory(++calls);
+ return response([], calls - 1);
+ });
+ f.notice();
+ await settled(f.store);
+ expect(calls).toBe(3);
+ expect(f.store.error).toContain("sync paused");
+ await vi.advanceTimersByTimeAsync(10000);
+ expect(calls).toBe(3);
+ f.store.dispose();
+ await vi.advanceTimersByTimeAsync(10000);
+ expect(calls).toBe(3);
+ expect(vi.getTimerCount()).toBe(0);
+ });
+
+ it("ignores other conversations and disposed notices; accepts a notice without turnId", async () => {
+ vi.useFakeTimers();
+ const f = fixture();
+ f.store.handleDelta({
+ type: "chat.resync-required",
+ conversationId: "other",
+ reason: "payload-limit",
+ });
+ expect(f.store.recovering).toBe(false);
+ f.store.handleDelta({
+ type: "chat.resync-required",
+ conversationId: "c1",
+ reason: "payload-limit",
+ });
+ await settled(f.store);
+ expect(f.store.generating).toBe(false);
+ expect(f.store.recovering).toBe(false);
+ expect(vi.getTimerCount()).toBe(0);
+ f.event({ type: "turn-start", conversationId: "c1", turnId: "t1" });
+ f.store.handleDelta({
+ type: "chat.resync-required",
+ conversationId: "c1",
+ reason: "payload-limit",
+ });
+ expect(f.store.generating).toBe(true);
+ expect(f.store.recovering).toBe(true);
+ f.end("rejected");
+ await settled(f.store);
+ expect(f.store.recovering).toBe(false);
+ expect(f.store.generating).toBe(false);
+ f.store.dispose();
+ f.notice();
+ expect(vi.getTimerCount()).toBe(0);
+ });
+});
+
+it("recovers a windowed rollback with real IndexedDB cache while preserving lifecycle evidence", async () => {
+ vi.stubGlobal("IDBKeyRange", IDBKeyRange);
+ const cache = createConversationCache(createIdbChunkStore({ indexedDB: new IDBFactory() }));
+ await cache.commit(
+ "c1",
+ [{ seq: 1, role: "assistant", chunk: { type: "text", text: "old epoch" } }],
+ 5,
+ );
+ const history = createFakeHistorySync();
+ history.historyEpoch = 5;
+ const timers = new Map<object, () => void>();
+ const store = createChatStore({
+ conversationId: "c1",
+ createRequestId: () => "r",
+ cache,
+ historySync: history.impl,
+ metricsSync: createFakeMetricsSync().impl,
+ transport: createFakeTransport().impl,
+ chatLimit: 10,
+ recoveryTimer: {
+ schedule(fn) {
+ const key = {};
+ timers.set(key, fn);
+ return key;
+ },
+ cancel(key) {
+ timers.delete(key as object);
+ },
+ },
+ });
+ try {
+ await store.load();
+ history.historyEpoch = 0;
+ history.returnChunks = Array.from({ length: 20 }, (_, i) => ({
+ seq: i + 1,
+ role: "assistant" as const,
+ turnId: "t1",
+ chunk: { type: "text" as const, text: `restored-${i}` },
+ }));
+ const before = history.calls.length;
+ store.handleDelta({
+ type: "chat.resync-required",
+ conversationId: "c1",
+ turnId: "t1",
+ reason: "replay-limit",
+ });
+ await vi.waitFor(() => expect(store.pendingSync).toBe(false));
+ expect(history.calls.length - before).toBe(2);
+ expect(history.calls.at(-1)).toMatchObject({ sinceSeq: 0, window: { limit: 7 } });
+ expect(await cache.historyEpoch("c1")).toBe(0);
+ expect(store.chunks.map((c) => c.seq)).toEqual([14, 15, 16, 17, 18, 19, 20]);
+ expect(store.hasEarlier).toBe(true);
+ expect(store.generating).toBe(true);
+ expect(timers.size).toBe(1);
+ store.handleDelta({
+ type: "chat.delta",
+ event: { type: "turn-ended", conversationId: "c1", turnId: "t1", reason: "aborted" },
+ });
+ await vi.waitFor(() => expect(store.pendingSync).toBe(false));
+ expect(store.generating).toBe(false);
+ expect(timers.size).toBe(0);
+ expect(store.hasEarlier).toBe(true);
+ } finally {
+ store.dispose();
+ }
+});
+
+it("ignores an unrelated terminal and stays recovering through done/seal until its actual terminal", async () => {
+ vi.useFakeTimers();
+ const f = fixture();
+ f.notice();
+ await settled(f.store);
+ f.end("aborted", "unrelated");
+ f.event({ type: "done", conversationId: "c1", turnId: "t1", reason: "aborted" });
+ expect(f.store.recovering).toBe(true);
+ expect(f.store.generating).toBe(true);
+ f.event({ type: "turn-sealed", conversationId: "c1", turnId: "t1" });
+ await settled(f.store);
+ expect(f.store.recovering).toBe(true);
+ expect(f.store.generating).toBe(true);
+ f.end("aborted");
+ await settled(f.store);
+ expect(f.store.recovering).toBe(false);
+ expect(f.store.generating).toBe(false);
+ expect(vi.getTimerCount()).toBe(0);
+ f.store.dispose();
+});
+
+it("makes a post-terminal request even when the terminal races the third bounded epoch attempt", async () => {
+ vi.useFakeTimers();
+ const third = deferred();
+ let calls = 0;
+ const f = fixture(async () => {
+ calls++;
+ if (calls < 3) {
+ f.store.invalidateHistory(calls);
+ return response([], calls - 1);
+ }
+ return calls === 3 ? third.promise : response(committed, 2);
+ });
+ f.notice();
+ for (let i = 0; i < 60 && calls < 3; i++) await Promise.resolve();
+ expect(calls).toBe(3);
+ f.end("error");
+ third.resolve(response([], 2));
+ await settled(f.store);
+ expect(calls).toBe(4);
+ expect(f.store.chunks.map((r) => r.chunk)).toEqual(committed.map((r) => r.chunk));
+ expect(f.store.recovering).toBe(false);
+ expect(vi.getTimerCount()).toBe(0);
+ f.store.dispose();
+});
+
+it("retains active recovery across a compaction epoch refresh and cancels it on reconnect/dispose", async () => {
+ vi.useFakeTimers();
+ const f = fixture();
+ f.notice();
+ await settled(f.store);
+ f.history.historyEpoch = 1;
+ f.store.resync(1);
+ await settled(f.store);
+ expect(f.store.recovering).toBe(true);
+ expect(f.store.generating).toBe(true);
+ expect(vi.getTimerCount()).toBe(1);
+ f.store.resync();
+ await settled(f.store);
+ expect(f.store.recovering).toBe(false);
+ expect(f.store.generating).toBe(false);
+ expect(vi.getTimerCount()).toBe(0);
+ f.store.dispose();
+});
+
+it("flushes a batched aborted terminal plus carried turn before applying an older HTTP response", async () => {
+ vi.useFakeTimers();
+ const pending = deferred();
+ let calls = 0;
+ const f = fixture(async () => (++calls === 1 ? pending.promise : response(committed)), true);
+ f.notice();
+ for (let i = 0; i < 15; i++) await Promise.resolve();
+ f.text("old partial");
+ f.end("aborted");
+ f.event({ type: "user-message", conversationId: "c1", turnId: "t2", text: "queued" });
+ f.event({ type: "turn-start", conversationId: "c1", turnId: "t2" });
+ f.text("live carried output", "t2");
+ pending.resolve(response([opening]));
+ await settled(f.store);
+ expect(calls).toBe(2);
+ expect(f.store.chunks.filter((r) => r.provisional).map((r) => r.chunk)).toEqual([
+ { type: "text", text: "queued" },
+ { type: "text", text: "live carried output" },
+ ]);
+ expect(f.store.generating).toBe(true);
+ expect(f.store.recovering).toBe(false);
+ expect(vi.getTimerCount()).toBe(0);
+ f.store.dispose();
+});
+
+it("does not truncate a payload larger than the WS budget recovered over HTTP", async () => {
+ vi.useFakeTimers();
+ const large = "x".repeat(4 * 1024 * 1024 + 1);
+ const full: StoredChunk[] = [
+ {
+ ...call,
+ chunk: {
+ ...call.chunk,
+ type: "tool-call",
+ toolCallId: "a",
+ toolName: "run_shell",
+ input: { command: large },
+ stepId: step,
+ },
+ },
+ {
+ ...result,
+ chunk: {
+ type: "tool-result",
+ toolCallId: "a",
+ toolName: "run_shell",
+ content: large,
+ isError: false,
+ stepId: step,
+ },
+ },
+ ];
+ const f = fixture();
+ f.history.returnChunks = full;
+ f.notice();
+ await settled(f.store);
+ expect(f.store.chunks).toHaveLength(2);
+ expect(f.store.chunks.map((r) => r.chunk)).toEqual(full.map((r) => r.chunk));
+ expect(f.transport.sent).toEqual([]);
+ f.store.dispose();
+});
diff --git a/src/features/chat/store.svelte.ts b/src/features/chat/store.svelte.ts
index 66ee5ee..bb0b218 100644
--- a/src/features/chat/store.svelte.ts
+++ b/src/features/chat/store.svelte.ts
@@ -4,6 +4,7 @@ import type {
ChatQueueCancelMessage,
ChatQueuedMessage,
ChatQueueMessage,
+ ChatResyncRequiredMessage,
ChatSendMessage,
} from "@dispatch/transport-contract";
import type { ChatMessage, ImageInput, StoredChunk, TurnProviderRetryEvent } from "@dispatch/wire";
@@ -15,6 +16,7 @@ import {
foldEvent,
initialState,
initialWindowSize,
+ markHistoryRecovering,
normalizeChatLimit,
removeLocalEcho,
restoreEarlier,
@@ -88,6 +90,11 @@ export interface ChatStoreDependencies {
readonly scheduleFlush?: (flush: () => void) => void;
/** Wall clock (epoch ms) — times the cache-timer gaps. Absent → `Date.now`. */
readonly now?: () => number;
+ /** Recovery polling clock; browser effects are replaceable by deterministic tests. */
+ readonly recoveryTimer?: {
+ schedule(callback: () => void, delayMs: number): unknown;
+ cancel(handle: unknown): void;
+ };
}
export interface ChatStore {
@@ -117,6 +124,8 @@ export interface ChatStore {
* to the newest attempt + delay; cleared when content resumes or the turn ends.
*/
readonly providerRetry: TurnProviderRetryEvent | null;
+ /** Omitted live payloads may not exist in HTTP history until step persistence. */
+ readonly recovering: boolean;
readonly pendingSync: boolean;
readonly error: string | null;
readonly model: string | undefined;
@@ -146,7 +155,9 @@ export interface ChatStore {
* Ignored while a turn is generating; a running turn gap keeps the later start.
*/
noteLastActivity(at: number): void;
- handleDelta(msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage): void;
+ handleDelta(
+ msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage | ChatResyncRequiredMessage,
+ ): void;
/**
* Send a user message (start a turn via `chat.send`). Optimistically echoes
* the text + any `images` as provisional user chunks (`[text, image, …]` in
@@ -235,12 +246,38 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
let revision = 0;
let reloadWindow = false;
let earliestSeq = $state<number | null>(null);
+ let recoveringTurnId: string | undefined;
+ let recovering = $state(false);
+ const finishedRecoveryTurns = new Set<string>();
+ let finalSyncRequested = false;
+ let recoveryPaused = false;
+ const timer = deps.recoveryTimer ?? {
+ schedule: (callback: () => void, delay: number) => setTimeout(callback, delay),
+ cancel: (handle: unknown) => clearTimeout(handle as ReturnType<typeof setTimeout>),
+ };
+ let recoveryTimer: unknown;
+
+ function cancelRecoveryTimer(): void {
+ if (recoveryTimer !== undefined) timer.cancel(recoveryTimer);
+ recoveryTimer = undefined;
+ }
+
+ function scheduleRecovery(): void {
+ if (disposed || !recovering || recoveryPaused || recoveryTimer !== undefined) return;
+ recoveryTimer = timer.schedule(() => {
+ recoveryTimer = undefined;
+ flushEvents();
+ void syncTail(initialWindowSize(chatLimit));
+ }, 1000);
+ }
function resetTranscript(epoch: number): void {
historyEpoch = epoch;
revision++;
reloadWindow = true;
transcript = initialState();
+ if (recoveringTurnId !== undefined)
+ transcript = markHistoryRecovering(transcript, recoveringTurnId);
earliestSeq = null;
pendingEvents = [];
}
@@ -294,6 +331,7 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
let m = metrics;
let sealed = false;
let failed = false;
+ let recoveryEnded = false;
let c = cacheTimer;
const at = now();
for (const event of events) {
@@ -301,14 +339,34 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
m = foldMetricsEvent(m, event);
const mNow = m;
c = foldCacheTimerEvent(c, event, at, () => lastTurnIdOf(mNow));
- if (t.sealedTurnId !== null) sealed = true;
+ if (
+ event.type === "turn-ended" &&
+ recovering &&
+ (recoveringTurnId === undefined || event.turnId === recoveringTurnId)
+ ) {
+ finishedRecoveryTurns.add(event.turnId);
+ finalSyncRequested = true;
+ recoveringTurnId = undefined;
+ recovering = false;
+ recoveryEnded = true;
+ cancelRecoveryTimer();
+ }
+ if (event.type === "turn-sealed") sealed = true;
else if (event.type === "error") failed = true;
}
transcript = t;
metrics = m;
cacheTimer = c;
maybeTrim();
- if (sealed) {
+ if (recovering && recoveringTurnId !== undefined) {
+ // done/error/seal can precede persistence and the real terminal. Keep
+ // lifecycle evidence until turn-ended, not merely until a content event.
+ transcript = markHistoryRecovering(transcript, recoveringTurnId);
+ }
+ if (recoveryEnded) {
+ void syncTail(initialWindowSize(chatLimit));
+ void syncMetrics();
+ } else if (sealed) {
void syncTail();
void syncMetrics();
} else if (failed) {
@@ -366,7 +424,13 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
continue;
}
const limit =
- since === 0 ? (reloadWindow ? initialWindowSize(chatLimit) : coldLimit) : undefined;
+ since === 0
+ ? reloadWindow || recovering || finishedRecoveryTurns.size > 0
+ ? initialWindowSize(chatLimit)
+ : coldLimit
+ : undefined;
+ const finishedAtRequest = [...finishedRecoveryTurns];
+ finalSyncRequested = false;
const res = await deps.historySync(
deps.conversationId,
since,
@@ -416,7 +480,9 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
historyEpoch = res.historyEpoch;
if (since === 0 && (limit === undefined || res.chunks.length < limit))
earliestSeq = merged[0]?.seq ?? 0;
- transcript = applyHistory(transcript, merged);
+ flushEvents();
+ transcript = applyHistory(transcript, merged, finishedAtRequest);
+ for (const turnId of finishedAtRequest) finishedRecoveryTurns.delete(turnId);
maybeTrim();
_error = null;
} catch (err) {
@@ -424,9 +490,21 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
}
} while (rerunRequested && !disposed && attempts < 3);
if (rerunRequested && !disposed) {
+ recoveryPaused = true;
+ cancelRecoveryTimer();
_error = "Conversation history changed repeatedly; sync paused until the next refresh.";
}
_pendingSync = false;
+ // Schedule AFTER the bounded sync burst. No interval can queue unbounded
+ // reruns behind a slow request; repeated notices coalesce on the same clock.
+ scheduleRecovery();
+ // A terminal landing during the third (bounded epoch) attempt still owns
+ // one post-terminal request. It is a new external trigger, not an epoch
+ // retry loop; consume the flag before starting its bounded burst.
+ if (finalSyncRequested && !disposed) {
+ finalSyncRequested = false;
+ void syncTail(initialWindowSize(chatLimit));
+ }
}
async function syncMetrics(): Promise<void> {
@@ -520,6 +598,9 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
get providerRetry(): TurnProviderRetryEvent | null {
return selectProviderRetry(transcript);
},
+ get recovering(): boolean {
+ return recovering;
+ },
get pendingSync(): boolean {
return _pendingSync;
},
@@ -544,7 +625,32 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
cacheTimer = seedTurnGap(cacheTimer, at);
},
- handleDelta(msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage): void {
+ handleDelta(
+ msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage | ChatResyncRequiredMessage,
+ ): void {
+ if (disposed) return;
+ if (msg.type === "chat.resync-required") {
+ if (msg.conversationId !== deps.conversationId) return;
+ flushEvents();
+ const target =
+ msg.turnId ?? (transcript.generating ? transcript.currentTurnId : null) ?? undefined;
+ // Oversized status packets need not identify an active turn. Refresh
+ // once, but do not invent generation or poll an idle conversation.
+ if (target === undefined) {
+ void syncTail(initialWindowSize(chatLimit));
+ return;
+ }
+ const first = !recovering || recoveryPaused;
+ recoveryPaused = false;
+ recovering = true;
+ recoveringTurnId = target;
+ if (recoveringTurnId !== undefined)
+ transcript = markHistoryRecovering(transcript, recoveringTurnId);
+ // Immediate first handoff, including a coalesced rerun behind a cold
+ // load. Repeats wait for the existing fetch/one-second timer.
+ if (first) void syncTail(initialWindowSize(chatLimit));
+ return;
+ }
if (msg.type === "chat.error") {
if (msg.conversationId !== undefined && msg.conversationId !== deps.conversationId) {
return;
@@ -685,13 +791,21 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore {
// clear the now-stale spinner BEFORE re-subscribing, so a finished turn
// doesn't spin forever. A still-running turn's replay re-asserts it.
flushEvents();
- transcript = clearGenerating(transcript);
+ recoveryPaused = false;
+ if (epoch === undefined) {
+ recovering = false;
+ recoveringTurnId = undefined;
+ cancelRecoveryTimer();
+ transcript = clearGenerating(transcript);
+ }
void syncTail();
void syncMetrics();
},
dispose(): void {
disposed = true;
+ cancelRecoveryTimer();
+ finishedRecoveryTurns.clear();
pendingEvents = [];
},
};
diff --git a/src/features/chat/ui/ChatView.svelte b/src/features/chat/ui/ChatView.svelte
index b473766..c7f566a 100644
--- a/src/features/chat/ui/ChatView.svelte
+++ b/src/features/chat/ui/ChatView.svelte
@@ -24,6 +24,7 @@
let {
chunks,
+ recovering = false,
turnMetrics = [],
hasEarlier = false,
onShowEarlier,
@@ -34,6 +35,7 @@
cacheTimer = null,
}: {
chunks: readonly RenderedChunk[];
+ recovering?: boolean;
turnMetrics?: readonly TurnMetricsEntry[];
/** Earlier history is unloaded (chat limit) and can be paged back in. */
hasEarlier?: boolean;
@@ -90,7 +92,7 @@
}
}
- const groups = $derived(groupRenderedChunks(chunks));
+ const groups = $derived(groupRenderedChunks(chunks, recovering));
const rows = $derived(interleaveTurnMetrics(groups, turnMetrics));
@@ -150,6 +152,12 @@
});
</script>
+{#if recovering}
+ <p role="status" class="px-4 py-2 text-sm text-base-content/70">
+ Recovering conversation history… Omitted tool input/output is unavailable until persisted.
+ </p>
+{/if}
+
{#snippet chunkRow(rendered: RenderedChunk)}
{#if rendered.role === "user"}
<!-- User: a speech bubble, left-aligned. A user message may be multi-chunk
diff --git a/src/features/chat/ui/ChatView.test.ts b/src/features/chat/ui/ChatView.test.ts
index ad50784..a6bd8ac 100644
--- a/src/features/chat/ui/ChatView.test.ts
+++ b/src/features/chat/ui/ChatView.test.ts
@@ -512,3 +512,47 @@ it("replaces failed persisted images with an expired placeholder, without markin
expect(screen.getByText("image expired")).toBeInTheDocument();
expect(screen.getAllByRole("img")).toEqual([inline]);
});
+
+it("shows omitted in-progress tool input as recovering/unavailable, never No arguments", async () => {
+ const { container, rerender } = render(ChatView, {
+ recovering: true,
+ chunks: [
+ {
+ seq: null,
+ role: "tool",
+ provisional: true,
+ chunk: {
+ type: "tool-result",
+ toolCallId: "missing",
+ toolName: "run_shell",
+ content: "real output",
+ isError: false,
+ },
+ },
+ ],
+ });
+ expect(screen.getByRole("status").textContent).toContain("unavailable until persisted");
+ expect(screen.getByText("Recovering tool input…")).toBeTruthy();
+ await openAllCollapses(container);
+ expect(screen.getByText(/Tool input is unavailable until it is persisted/)).toBeTruthy();
+ expect(screen.queryByText("No arguments")).toBeNull();
+ expect(screen.getByText("real output")).toBeTruthy();
+ await rerender({
+ recovering: false,
+ chunks: [
+ {
+ seq: 1,
+ role: "assistant",
+ provisional: false,
+ chunk: {
+ type: "tool-call",
+ toolCallId: "missing",
+ toolName: "run_shell",
+ input: { command: "recovered command" },
+ },
+ },
+ ],
+ });
+ expect(screen.queryByRole("status")).toBeNull();
+ expect(screen.queryByText("Recovering tool input…")).toBeNull();
+});
generated by cgit v1.2.3 (git 2.39.1) at 2026-10-25 23:11:15 +0000