diff options
Diffstat (limited to 'src/features/chat')
22 files changed, 2816 insertions, 81 deletions
diff --git a/src/features/chat/adapter/pending-inputs.ts b/src/features/chat/adapter/pending-inputs.ts new file mode 100644 index 0000000..5798fdb --- /dev/null +++ b/src/features/chat/adapter/pending-inputs.ts @@ -0,0 +1,49 @@ +import type { PendingInput, PendingInputStorage } from "../pending-inputs"; + +export function createPendingInputStorage( + storage: Storage, + conversationId: string, +): PendingInputStorage { + const prefix = `dispatch.pending-inputs:${conversationId}:`; + let known = new Map<string, string>(); + return { + load() { + const inputs: PendingInput[] = []; + const loaded = new Map<string, string>(); + for (let i = 0; i < storage.length; i++) { + const key = storage.key(i); + if (key === null || !key.startsWith(prefix)) continue; + const raw = storage.getItem(key); + if (raw === null) continue; + const input = JSON.parse(raw); + if ( + (input?.request?.type !== "chat.send" && input?.request?.type !== "chat.queue") || + input.request.conversationId !== conversationId || + typeof input.request.requestId !== "string" || + key !== prefix + input.request.requestId || + (input.request.type === "chat.send" + ? typeof input.request.message !== "string" + : typeof input.request.text !== "string") || + (input.rejection !== undefined && typeof input.rejection !== "string") + ) + throw new Error("Invalid saved input"); + inputs.push(input); + loaded.set(input.request.requestId, raw); + } + known = loaded; + return inputs; + }, + save(inputs) { + // Per-request keys prevent another browser tab's send/receipt from overwriting ours. + // Quota/security failures reach the composer BEFORE it clears its staged input. + const next = new Map(inputs.map((input) => [input.request.requestId, JSON.stringify(input)])); + for (const [id, raw] of next) { + if (known.get(id) !== raw) storage.setItem(prefix + id, raw); + } + for (const id of known.keys()) { + if (!next.has(id)) storage.removeItem(prefix + id); + } + known = next; + }, + }; +} diff --git a/src/features/chat/delivery.test.ts b/src/features/chat/delivery.test.ts new file mode 100644 index 0000000..99ad2d4 --- /dev/null +++ b/src/features/chat/delivery.test.ts @@ -0,0 +1,146 @@ +import { describe, expect, it, vi } from "vitest"; +import { createPendingInputStorage } from "./adapter/pending-inputs"; +import { createChatStore } from "./store.svelte"; +import { + createFakeCache, + createFakeHistorySync, + createFakeMetricsSync, + createFakeTransport, +} from "./test-helpers"; + +function storage(): Storage { + const values = new Map<string, string>(); + return { + get length() { + return values.size; + }, + getItem: (key) => values.get(key) ?? null, + setItem: (key, value) => { + values.set(key, value); + }, + removeItem: (key) => { + values.delete(key); + }, + clear: () => values.clear(), + key: (index) => [...values.keys()][index] ?? null, + }; +} + +function fixture(saved = storage(), metricsSync = createFakeMetricsSync().impl) { + const transport = createFakeTransport(); + let id = 0; + const store = createChatStore({ + conversationId: "c", + createRequestId: () => `request-${++id}`, + pendingInputStorage: createPendingInputStorage(saved, "c"), + transport: transport.impl, + historySync: createFakeHistorySync().impl, + metricsSync, + cache: createFakeCache().impl, + }); + return { store, transport, saved }; +} + +const images = [{ url: "data:image/png;base64,AAAA", mimeType: "image/png" }]; + +describe("input delivery", () => { + it.each([ + "send", + "queueMessage", + ] as const)("saves %s input through refresh without automatically retransmitting", (method) => { + const f = fixture(); + f.store[method]("keep me", images); + expect(f.store.pendingInputs).toHaveLength(1); + f.store.dispose(); + const reloaded = fixture(f.saved); + expect(reloaded.store.pendingInputs).toMatchObject([{ request: { images } }]); + expect(JSON.stringify(reloaded.store.pendingInputs)).toContain("keep me"); + expect(reloaded.transport.sent).toEqual([]); + expect(reloaded.transport.sentQueue).toEqual([]); + reloaded.store.dispose(); + }); + + it("streamed user echoes and wrong receipts cannot discard an unconfirmed send", () => { + const f = fixture(); + f.store.send("mine", images); + f.store.handleDelta({ + type: "chat.delta", + event: { type: "user-message", conversationId: "c", turnId: "t", text: "mine" }, + }); + f.store.handleDelta({ type: "chat.accepted", conversationId: "other", requestId: "request-1" }); + f.store.handleDelta({ type: "chat.accepted", conversationId: "c", requestId: "other" }); + expect(f.store.pendingInputs).toHaveLength(1); + f.store.handleDelta({ type: "chat.accepted", conversationId: "c", requestId: "request-1" }); + expect(f.store.pendingInputs).toEqual([]); + expect(createPendingInputStorage(f.saved, "c").load()).toEqual([]); + f.store.dispose(); + }); + + it("retains explicitly rejected input and images for recovery until discarded", () => { + const f = fixture(); + f.store.send("mine", images); + f.store.handleDelta({ + type: "chat.error", + conversationId: "c", + requestId: "request-1", + message: "disk full", + }); + const reloaded = fixture(f.saved); + expect(reloaded.store.pendingInputs).toMatchObject([ + { rejection: "disk full", request: { message: "mine", images } }, + ]); + reloaded.store.discardPendingInput("request-1"); + expect(createPendingInputStorage(f.saved, "c").load()).toEqual([]); + f.store.dispose(); + reloaded.store.dispose(); + }); + + it.each([ + "send", + "queueMessage", + ] as const)("does not clear or transmit %s input if saving fails", (method) => { + const saved = storage(); + saved.setItem = () => { + throw new Error("quota exceeded"); + }; + const f = fixture(saved); + expect(() => f.store[method]("mine", images)).toThrow("quota exceeded"); + expect(f.store.pendingInputs).toEqual([]); + expect(f.store.chunks).toEqual([]); + expect(f.transport.sent).toEqual([]); + expect(f.transport.sentQueue).toEqual([]); + f.store.dispose(); + }); + + it("coalesces a reconnect burst and permits one metrics catch-up after completion", async () => { + let resolve!: (value: { turns: [] }) => void; + const first = { + promise: new Promise<{ turns: [] }>((done) => { + resolve = done; + }), + resolve: (value: { turns: [] }) => resolve(value), + }; + const metricsSync = vi.fn().mockReturnValueOnce(first.promise).mockResolvedValue({ turns: [] }); + const f = fixture(storage(), metricsSync); + for (let i = 0; i < 100; i++) f.store.resync(); + expect(metricsSync).toHaveBeenCalledTimes(1); + first.resolve({ turns: [] }); + await vi.waitFor(() => expect(metricsSync).toHaveBeenCalledTimes(2)); + f.store.dispose(); + }); +}); + +it("concurrent browser tabs cannot overwrite each other's saved messages or receipts", () => { + const saved = storage(); + const first = createPendingInputStorage(saved, "c"); + const second = createPendingInputStorage(saved, "c"); + first.load(); + second.load(); + const input = (requestId: string) => ({ + request: { type: "chat.send" as const, conversationId: "c", requestId, message: requestId }, + }); + first.save([input("first")]); + second.save([input("second")]); + first.save([]); + expect(createPendingInputStorage(saved, "c").load()).toEqual([input("second")]); +}); diff --git a/src/features/chat/index.ts b/src/features/chat/index.ts index 59da39b..41cb06f 100644 --- a/src/features/chat/index.ts +++ b/src/features/chat/index.ts @@ -6,8 +6,11 @@ export type { } from "../../core/chunks"; export { groupRenderedChunks, resolveImageUrl, viewProviderRetry } from "../../core/chunks"; export type { TurnMetricsEntry } from "../../core/metrics"; +export { createPendingInputStorage } from "./adapter/pending-inputs"; export type { NamedAgentLike, NamedAgentSnapshotLike } from "./model-select"; export { + CLAUDE_SUBSCRIPTION_KEY, + isClaudeSubscriptionModel, isNamedAgentSelection, isNumberedOpenCode, isVisionModel, @@ -20,6 +23,7 @@ export { openCodeNumber, providerKeysWithNamedAgents, } from "./model-select"; +export type { PendingInput, PendingInputStorage } from "./pending-inputs"; export type { ChatTransport, HistorySync, HistoryWindow, MetricsSync } from "./ports"; export type { EffortOption, @@ -51,6 +55,7 @@ export { default as CompactionView } from "./ui/CompactionView.svelte"; export type { ComposerStatus } from "./ui/Composer.svelte"; export { default as Composer } from "./ui/Composer.svelte"; export { default as ModelSelector } from "./ui/ModelSelector.svelte"; +export { default as PendingInputs } from "./ui/PendingInputs.svelte"; export { default as ReasoningEffortSelector } from "./ui/ReasoningEffortSelector.svelte"; /** Public module manifest — aggregated by the shell's "Loaded Modules" view. */ diff --git a/src/features/chat/logic/composer.test.ts b/src/features/chat/logic/composer.test.ts new file mode 100644 index 0000000..86e9dc4 --- /dev/null +++ b/src/features/chat/logic/composer.test.ts @@ -0,0 +1,20 @@ +import { describe, expect, it } from "vitest"; +import { stopButtonLabels } from "./composer"; + +describe("stopButtonLabels", () => { + it.each([undefined, 0])("preserves plain Stop with queued count %s", (count) => { + expect(stopButtonLabels(count)).toEqual({ + text: "Stop", + ariaLabel: "Stop generation", + title: undefined, + }); + }); + + it.each([1, 2])("labels the queued Stop action Flush with queued count %s", (count) => { + expect(stopButtonLabels(count)).toEqual({ + text: "Flush", + ariaLabel: "Flush", + title: "Stop the current response and start the queued message now", + }); + }); +}); diff --git a/src/features/chat/logic/composer.ts b/src/features/chat/logic/composer.ts new file mode 100644 index 0000000..e56a007 --- /dev/null +++ b/src/features/chat/logic/composer.ts @@ -0,0 +1,11 @@ +/** Stop still uses the same action; the backend carries queued input into a new turn. */ +export function stopButtonLabels(queuedMessageCount: number | undefined) { + if (queuedMessageCount !== undefined && queuedMessageCount >= 1) { + return { + text: "Flush", + ariaLabel: "Flush", + title: "Stop the current response and start the queued message now", + }; + } + return { text: "Stop", ariaLabel: "Stop generation", title: undefined }; +} 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/model-select.test.ts b/src/features/chat/model-select.test.ts index 57de11d..523f1d8 100644 --- a/src/features/chat/model-select.test.ts +++ b/src/features/chat/model-select.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from "vitest"; import { + isClaudeSubscriptionModel, isNamedAgentSelection, isVisionModel, joinModelName, @@ -192,3 +193,23 @@ describe("named-agent provider helpers", () => { expect(namedAgentSelectionLabel("Coding")).toBe("Named Agents -> Coding"); }); }); + +describe("isClaudeSubscriptionModel", () => { + it("is true for claude/<model> (the Claude Code subscription credential)", () => { + expect(isClaudeSubscriptionModel("claude/claude-opus-4-6")).toBe(true); + expect(isClaudeSubscriptionModel("claude/claude-sonnet-4-6")).toBe(true); + }); + + it("is false for Claude models served through other credentials", () => { + expect(isClaudeSubscriptionModel("opencode-1/claude-sonnet-4.5")).toBe(false); + expect(isClaudeSubscriptionModel("openrouter/anthropic/claude-opus-4")).toBe(false); + expect(isClaudeSubscriptionModel("openai/gpt-4o")).toBe(false); + }); + + it("is false for a bare key, a lookalike key, or undefined", () => { + expect(isClaudeSubscriptionModel("claude")).toBe(false); + expect(isClaudeSubscriptionModel("claude/")).toBe(false); + expect(isClaudeSubscriptionModel("claude-2/claude-opus-4-6")).toBe(false); + expect(isClaudeSubscriptionModel(undefined)).toBe(false); + }); +}); diff --git a/src/features/chat/model-select.ts b/src/features/chat/model-select.ts index 6f198f3..bebad2c 100644 --- a/src/features/chat/model-select.ts +++ b/src/features/chat/model-select.ts @@ -63,6 +63,25 @@ export function isNumberedOpenCode(id: string): boolean { } /** + * The credential key of the Claude Code subscription (the external Claude + * extension — `auth-claude` + `provider-anthropic`). The backend registers it + * under `DISPATCH_CLAUDE_CREDENTIAL`, default `"claude"`, so its models arrive + * as `claude/<model>`. + */ +export const CLAUDE_SUBSCRIPTION_KEY = "claude"; + +/** + * True iff `full` is a model served through the Claude Code subscription + * (`claude/<model>`). Gates the cache timer (that subscription's prompt cache + * expires after 5 minutes). A bare key with no model is not a model. + */ +export function isClaudeSubscriptionModel(full: string | undefined): boolean { + if (full === undefined) return false; + const { key, model } = splitModelName(full); + return key === CLAUDE_SUBSCRIPTION_KEY && model !== ""; +} + +/** * A friendly human label for a model key/provider id: `opencode-<n>` → * "OpenCode #<n>"; any other id is returned unchanged. The RAW id is still what * gets selected/persisted — this only affects display. diff --git a/src/features/chat/pending-inputs.ts b/src/features/chat/pending-inputs.ts new file mode 100644 index 0000000..d7fb0aa --- /dev/null +++ b/src/features/chat/pending-inputs.ts @@ -0,0 +1,12 @@ +import type { ChatQueueMessage, ChatSendMessage } from "@dispatch/transport-contract"; + +/** A saved copy, never automatically retransmitted: requestId is not idempotent. */ +export interface PendingInput { + readonly request: (ChatSendMessage | ChatQueueMessage) & { readonly requestId: string }; + readonly rejection?: string; +} + +export interface PendingInputStorage { + load(): readonly PendingInput[]; + save(inputs: readonly PendingInput[]): void; +} diff --git a/src/features/chat/queued.test.ts b/src/features/chat/queued.test.ts new file mode 100644 index 0000000..3b579c6 --- /dev/null +++ b/src/features/chat/queued.test.ts @@ -0,0 +1,207 @@ +import type { ChatQueuedMessage } from "@dispatch/transport-contract"; +import type { AgentEvent } from "@dispatch/wire"; +import { describe, expect, it, vi } from "vitest"; +import { createChatStore } from "./store.svelte"; +import { + createFakeCache, + createFakeHistorySync, + createFakeMetricsSync, + createFakeTransport, +} from "./test-helpers"; + +const images = [ + { url: "data:image/png;base64,AAAA", mimeType: "image/png" }, + { url: "data:image/png;base64,BBBB" }, +]; + +function fixture(batched = false) { + const transport = createFakeTransport(); + const history = createFakeHistorySync(); + let nextId = 0; + const scheduled: (() => void)[] = []; + const store = createChatStore({ + conversationId: "c1", + createRequestId: () => `r${++nextId}`, + transport: transport.impl, + historySync: history.impl, + metricsSync: createFakeMetricsSync().impl, + cache: createFakeCache().impl, + ...(batched ? { scheduleFlush: (flush: () => void) => scheduled.push(flush) } : {}), + }); + function ack(requestId: string | undefined = "r1", conversationId = "c1") { + const msg: ChatQueuedMessage = { + type: "chat.queued", + conversationId, + reason: "compaction", + messageId: "q1", + ...(requestId !== undefined ? { requestId } : {}), + }; + store.handleDelta(msg); + } + function event(event: AgentEvent) { + store.handleDelta({ type: "chat.delta", event }); + } + function start(text?: string, turnId = "t1") { + event({ type: "turn-start", conversationId: "c1", turnId }); + if (text !== undefined) event({ type: "user-message", conversationId: "c1", turnId, text }); + } + return { store, transport, history, scheduled, ack, event, start }; +} + +describe("compaction-deferred chat.send", () => { + it.each([ + "hello", + "", + ])("removes just the accepted local text/images for %j without error, retry or queue sends", (text) => { + const f = fixture(); + f.store.send(text, images); + expect(f.store.chunks).toHaveLength(text ? 3 : 2); + f.ack(); + f.ack(); + expect(f.store.chunks).toEqual([]); + expect(f.store.generating).toBe(false); + expect(f.store.error).toBeNull(); + expect(f.transport.sent).toHaveLength(1); + expect(f.transport.sent[0]).toMatchObject({ requestId: "r1", images, message: text }); + expect(f.transport.sentQueue).toEqual([]); + f.store.dispose(); + }); + + it("uses distinct request identities for identical sends and preserves unrelated content", () => { + const f = fixture(); + f.start("another client's prompt"); + f.event({ type: "text-delta", conversationId: "c1", turnId: "t1", delta: "working" }); + f.store.send("identical", images); + f.store.send("identical", images); + expect(f.transport.sent.map((m) => m.requestId)).toEqual(["r1", "r2"]); + f.ack("r1"); + f.ack("r1"); + expect(f.store.chunks.map((c) => c.chunk)).toEqual([ + { type: "text", text: "another client's prompt" }, + { type: "text", text: "working" }, + { type: "text", text: "identical" }, + ...images.map((i) => ({ type: "image", ...i })), + ]); + f.ack("r2"); + expect(f.store.chunks.map((c) => c.chunk)).toEqual([ + { type: "text", text: "another client's prompt" }, + { type: "text", text: "working" }, + ]); + expect(f.store.generating).toBe(true); + f.store.dispose(); + }); + + it.each([ + "missing", + "unknown", + "wrong-conversation", + ])("tolerates %s correlation without deleting any echo", (kind) => { + const f = fixture(); + f.store.send("hello", images); + f.store.handleDelta({ + type: "chat.queued", + conversationId: kind === "wrong-conversation" ? "c2" : "c1", + reason: "compaction", + messageId: "q1", + ...(kind === "missing" ? {} : { requestId: kind === "unknown" ? "r999" : "r1" }), + }); + expect(f.store.chunks).toHaveLength(3); + expect(f.store.error).toBeNull(); + f.store.dispose(); + }); + + it.each([ + false, + true, + ])("preserves a live confirmed prompt before delayed ack (batched=%s)", (batched) => { + const f = fixture(batched); + f.store.send("hello", images); + f.start("hello"); + f.event({ type: "text-delta", conversationId: "c1", turnId: "t1", delta: "answer" }); + f.ack(); + f.ack(); + for (const flush of f.scheduled) flush(); + expect(f.store.chunks.map((c) => c.chunk)).toEqual([ + { type: "text", text: "hello" }, + { type: "text", text: "answer" }, + ]); + expect(f.store.generating).toBe(true); + expect(f.store.error).toBeNull(); + f.store.dispose(); + }); + + it("turn-start tagging alone does not protect a still-local echo", () => { + const f = fixture(true); + f.store.send("hello", images); + f.start(); + f.ack(); + expect(f.store.chunks).toEqual([]); + expect(f.store.generating).toBe(true); + f.event({ type: "user-message", conversationId: "c1", turnId: "t1", text: "hello" }); + for (const flush of f.scheduled) flush(); + expect(f.store.chunks.map((c) => c.chunk)).toEqual([{ type: "text", text: "hello" }]); + f.store.dispose(); + }); + + it("keeps a combined queue opening and steering; removes only the original individual echoes", () => { + const f = fixture(true); + f.store.send("one", images); + f.store.send("two", images); + f.start("one\n\ntwo"); + f.event({ type: "steering", conversationId: "c1", turnId: "t1", text: "steer" }); + f.ack("r2"); + f.ack("r1"); + expect(f.store.chunks.map((c) => c.chunk)).toEqual([ + { type: "text", text: "one\n\ntwo" }, + { type: "text", text: "steer" }, + ]); + expect(f.store.generating).toBe(true); + f.store.dispose(); + }); + + it.each([ + "ack-first", + "reset-first", + "history-first", + ])("handles epoch replacement idempotently: %s", async (order) => { + const f = fixture(); + await f.store.load(); + f.store.send("hello", images); + if (order === "ack-first") f.ack(); + f.history.historyEpoch = 1; + f.history.returnChunks = [ + { seq: 1, role: "user", turnId: "t1", chunk: { type: "text", text: "hello" } }, + { seq: 2, role: "user", turnId: "t1", chunk: { type: "image", url: "/images/c1/saved.png" } }, + ]; + f.store.resync(1); + if (order === "reset-first") f.ack(); + await vi.waitFor(() => expect(f.store.pendingSync).toBe(false)); + f.start("hello"); + const before = f.store.chunks; + f.ack(); + f.ack(); + expect(f.store.chunks).toEqual(before); + expect(f.store.chunks.map((c) => c.chunk)).toEqual(f.history.returnChunks.map((c) => c.chunk)); + expect(f.store.generating).toBe(true); + expect(f.store.error).toBeNull(); + expect(f.transport.sent).toHaveLength(1); + f.store.dispose(); + }); + + it("keeps normal immediate-send text/image deduplication and authoritative history", async () => { + const f = fixture(); + f.store.send("hello", images); + f.start("hello"); + expect(f.store.chunks).toHaveLength(3); + f.history.returnChunks = [ + { seq: 1, role: "user", turnId: "t1", chunk: { type: "text", text: "hello" } }, + { seq: 2, role: "user", turnId: "t1", chunk: { type: "image", url: "/images/c1/saved.png" } }, + ]; + f.event({ type: "turn-sealed", conversationId: "c1", turnId: "t1" }); + await vi.waitFor(() => expect(f.store.pendingSync).toBe(false)); + expect(f.store.chunks.map((c) => c.chunk)).toEqual(f.history.returnChunks.map((c) => c.chunk)); + expect(f.store.chunks.every((c) => !c.provisional)).toBe(true); + expect(f.transport.sent).toHaveLength(1); + f.store.dispose(); + }); +}); 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 7f8e113..dacbe06 100644 --- a/src/features/chat/store.svelte.ts +++ b/src/features/chat/store.svelte.ts @@ -1,9 +1,13 @@ import type { + ChatAcceptedMessage, ChatDeltaMessage, ChatErrorMessage, ChatQueueCancelMessage, + ChatQueuedMessage, ChatQueueMessage, + ChatResyncRequiredMessage, ChatSendMessage, + ChatSubscriptionRejectedMessage, } from "@dispatch/transport-contract"; import type { ChatMessage, ImageInput, StoredChunk, TurnProviderRetryEvent } from "@dispatch/wire"; import type { RenderedChunk, TranscriptState } from "../../core/chunks"; @@ -14,7 +18,9 @@ import { foldEvent, initialState, initialWindowSize, + markHistoryRecovering, normalizeChatLimit, + removeLocalEcho, restoreEarlier, selectChunks, selectGenerating, @@ -25,20 +31,28 @@ import { unloadCount, windowTranscript, } from "../../core/chunks"; -import type { MetricsState, TurnMetricsEntry } from "../../core/metrics"; +import type { CacheTimerState, MetricsState, TurnMetricsEntry } from "../../core/metrics"; import { applyDurableMetrics, + foldCacheTimerEvent, foldMetricsEvent, + initialCacheTimerState, initialMetricsState, + seedTurnGap, selectCurrentContextSize, selectOrderedTurnMetrics, + stopTurnGap, } from "../../core/metrics"; import type { ConversationCache } from "../conversation-cache"; import type { NamedAgentSelectionSnapshot } from "../named-agents"; +import type { PendingInput, PendingInputStorage } from "./pending-inputs"; import type { ChatTransport, HistorySync, MetricsSync } from "./ports"; export interface ChatStoreDependencies { + readonly pendingInputStorage?: PendingInputStorage; readonly conversationId: string; + /** Injected unique client-local correlation token for each new send. */ + readonly createRequestId: () => string; readonly model?: string; readonly transport: ChatTransport; readonly historySync: HistorySync; @@ -78,9 +92,18 @@ export interface ChatStoreDependencies { * folded synchronously (the original behavior; used by tests). */ 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 { + readonly pendingInputs: readonly PendingInput[]; + discardPendingInput(requestId: string): void; readonly messages: readonly ChatMessage[]; readonly chunks: readonly RenderedChunk[]; readonly turnMetrics: readonly TurnMetricsEntry[]; @@ -107,6 +130,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; @@ -122,7 +147,29 @@ export interface ChatStore { * a trim (see `TranscriptState.hiddenThinkingCount`). */ readonly thinkingKeyBase: number; - handleDelta(msg: ChatDeltaMessage | ChatErrorMessage): void; + /** + * The cache timer (see `core/metrics/cache-timer.ts`): the running gap + * between model requests (a step's tools executing, or the wait after a turn + * ended) plus the stamped durations of finished gaps, keyed by `stepId` / + * `turnId`. Timed with the local clock from live events; in-memory only (a + * reload keeps just a turn gap seeded via `noteLastActivity`). + */ + readonly cacheTimer: CacheTimerState; + /** + * Seed the turn gap from an authoritative activity timestamp (the + * conversation list's `lastActivityAt`) — call only for an IDLE conversation. + * Ignored while a turn is generating; a running turn gap keeps the later start. + */ + noteLastActivity(at: number): void; + handleDelta( + msg: + | ChatAcceptedMessage + | ChatSubscriptionRejectedMessage + | 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 @@ -139,9 +186,10 @@ export interface ChatStore { * auto-starts a turn with the message as its opening prompt (equivalent to * `chat.send`). No optimistic transcript echo — the queue SURFACE carries the * pending message until drain; the `steering` event places it in the - * transcript. `text` must be non-empty (the server 400/errors otherwise). + * transcript. Empty `text` is allowed when images are supplied. WS delivery + * is fire-and-forget; transport send is not a persistence acknowledgement. */ - queueMessage(text: string): void; + queueMessage(text: string, images?: readonly ImageInput[]): void; /** * Cancel (remove) a single queued steering message by id so it never runs * (`chat.queue.cancel` WS op). Fire-and-forget + idempotent: success is @@ -189,18 +237,79 @@ export interface ChatStore { * running, the server's post-subscribe replay re-asserts `generating`. The * app store pairs this with a `chat.subscribe` for the conversation. */ - resync(): void; + resync(historyEpoch?: number): void; + /** Discard pre-replacement transcript state without loading an unseen conversation. */ + invalidateHistory(historyEpoch: number): void; dispose(): void; } export function createChatStore(deps: ChatStoreDependencies): ChatStore { + let pendingInputs = $state<readonly PendingInput[]>(deps.pendingInputStorage?.load() ?? []); + function savePendingInputs(next: readonly PendingInput[]): void { + deps.pendingInputStorage?.save(next); + pendingInputs = next; + } + function confirmInput(requestId: string): void { + savePendingInputs(pendingInputs.filter((input) => input.request.requestId !== requestId)); + } let transcript = $state<TranscriptState>(initialState()); let metrics = $state<MetricsState>(initialMetricsState()); let _pendingSync = $state(false); let _error = $state<string | null>(null); let _model = $state<string | undefined>(deps.model); let _namedAgent = $state<NamedAgentSelectionSnapshot | null>(null); + let cacheTimer = $state<CacheTimerState>(initialCacheTimerState()); + const now = deps.now ?? Date.now; let disposed = false; + let rerunRequested = false; + let historyEpoch: number | undefined; + 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 = []; + } + function hasEarlier(): boolean { + return selectHasEarlier(transcript, earliestSeq); + } + + /** The most recent turn's id (resolves a seeded turn gap when it is stamped). */ + function lastTurnIdOf(m: MetricsState): string | undefined { + const entries = selectOrderedTurnMetrics(m); + return entries[entries.length - 1]?.turnId; + } let chatLimit = normalizeChatLimit(deps.chatLimit); @@ -242,16 +351,42 @@ 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) { t = foldEvent(t, event); m = foldMetricsEvent(m, event); - if (t.sealedTurnId !== null) sealed = true; + const mNow = m; + c = foldCacheTimerEvent(c, event, at, () => lastTurnIdOf(mNow)); + 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) { @@ -280,27 +415,142 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { * between the cache and the fetched window. */ async function syncTail(coldLimit?: number): Promise<void> { - if (disposed || _pendingSync) return; + if (disposed) return; + if (_pendingSync) { + rerunRequested = true; + return; + } _pendingSync = true; - try { - const since = await deps.cache.sinceSeq(deps.conversationId); - const window = since === 0 && coldLimit !== undefined ? { limit: coldLimit } : undefined; - const res = await deps.historySync(deps.conversationId, since, window); - const merged = await deps.cache.commit(deps.conversationId, res.chunks); - transcript = applyHistory(transcript, merged); - maybeTrim(); - _error = null; - } catch (err) { - _error = err instanceof Error ? err.message : String(err); - } finally { - _pendingSync = false; + // One request plus at most two sequential reruns. Continued churn waits for + // the next reconnect/seal/explicit resync, never a microtask refetch loop. + let attempts = 0; + do { + attempts++; + rerunRequested = false; + try { + const currentRevision = revision; + let cachedEpoch = await deps.cache.historyEpoch(deps.conversationId); + if (historyEpoch !== undefined && cachedEpoch !== historyEpoch) { + await deps.cache.reset(deps.conversationId, historyEpoch); + cachedEpoch = await deps.cache.historyEpoch(deps.conversationId); + } + const since = + reloadWindow || cachedEpoch === undefined + ? 0 + : await deps.cache.sinceSeq(deps.conversationId); + if (disposed) break; + if (revision !== currentRevision) { + rerunRequested = true; + continue; + } + const limit = + since === 0 + ? reloadWindow || recovering || finishedRecoveryTurns.size > 0 + ? initialWindowSize(chatLimit) + : coldLimit + : undefined; + const finishedAtRequest = [...finishedRecoveryTurns]; + finalSyncRequested = false; + const res = await deps.historySync( + deps.conversationId, + since, + limit === undefined ? undefined : { limit }, + ); + if (disposed) break; + if (revision !== currentRevision) { + rerunRequested = true; + continue; + } + const mismatch = + (cachedEpoch !== undefined && cachedEpoch !== res.historyEpoch) || + (historyEpoch !== undefined && historyEpoch !== res.historyEpoch); + if (mismatch && since !== 0) { + // A cursor response cannot replace a different history. Confirm with + // a new full-window request; don't mistake a lower epoch for staleness. + reloadWindow = true; + rerunRequested = true; + continue; + } + let merged: readonly StoredChunk[]; + if (mismatch || cachedEpoch === undefined) { + const replaced = await deps.cache.replace( + deps.conversationId, + res.chunks, + res.historyEpoch, + cachedEpoch, + ); + const replacedEpoch = await deps.cache.historyEpoch(deps.conversationId); + if (disposed) break; + if (revision !== currentRevision || !replaced || replacedEpoch !== res.historyEpoch) { + rerunRequested = true; + continue; + } + if (mismatch) resetTranscript(res.historyEpoch); + merged = res.chunks; + } else { + merged = await deps.cache.commit(deps.conversationId, res.chunks, res.historyEpoch); + const committedEpoch = await deps.cache.historyEpoch(deps.conversationId); + if (disposed) break; + if (revision !== currentRevision || committedEpoch !== res.historyEpoch) { + rerunRequested = true; + continue; + } + } + reloadWindow = false; + historyEpoch = res.historyEpoch; + if (since === 0 && (limit === undefined || res.chunks.length < limit)) + earliestSeq = merged[0]?.seq ?? 0; + flushEvents(); + transcript = applyHistory(transcript, merged, finishedAtRequest); + for (const turnId of finishedAtRequest) finishedRecoveryTurns.delete(turnId); + maybeTrim(); + _error = null; + } catch (err) { + _error = err instanceof Error ? err.message : String(err); + } + } 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)); + } + } + + let pendingMetrics: Promise<void> | undefined; + let rerunMetrics = false; + function syncMetrics(): Promise<void> { + if (pendingMetrics !== undefined) { + rerunMetrics = true; + return pendingMetrics; } + pendingMetrics = (async () => { + for (let attempt = 0; attempt < 2; attempt++) { + rerunMetrics = false; + await fetchMetrics(); + if (!rerunMetrics || disposed) break; + } + })().finally(() => { + pendingMetrics = undefined; + }); + return pendingMetrics; } - async function syncMetrics(): Promise<void> { + async function fetchMetrics(): Promise<void> { if (disposed) return; try { const res = await deps.metricsSync(deps.conversationId); + if (disposed) return; metrics = applyDurableMetrics(metrics, res.turns); } catch (err) { // Metrics fetch failure must not block history sync or throw; @@ -318,14 +568,27 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { * newest `count` of them). Shared by `showEarlier` and the raise-refill. */ async function backfillOlder(oldest: number, want: number): Promise<readonly StoredChunk[]> { + const currentRevision = revision; let earlier = (await deps.cache.load(deps.conversationId)).filter((c) => c.seq < oldest); + if (disposed || revision !== currentRevision) return []; const oldestKnown = earlier[0]?.seq ?? oldest; - if (earlier.length < want && oldestKnown > 1) { + if (earlier.length < want && (earliestSeq === null || oldestKnown > earliestSeq)) { const res = await deps.historySync(deps.conversationId, 0, { beforeSeq: oldestKnown, limit: want - earlier.length, }); - const merged = await deps.cache.commit(deps.conversationId, res.chunks); + if (disposed || revision !== currentRevision) return []; + if (res.historyEpoch !== historyEpoch) { + // Confirm a backfill mismatch with a fresh full window as well. Bump + // the local revision so an overlapping old tail cannot be applied. + revision++; + reloadWindow = true; + await syncTail(initialWindowSize(chatLimit)); + return []; + } + const merged = await deps.cache.commit(deps.conversationId, res.chunks, res.historyEpoch); + if (disposed || revision !== currentRevision) return []; + if (res.chunks.length < want - earlier.length) earliestSeq = merged[0]?.seq ?? 0; earlier = merged.filter((c) => c.seq < oldest); } return earlier; @@ -335,15 +598,15 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { * Refill toward the fresh-load window after a limit RAISE: pull older * history (cache first, then server) so the loaded set grows to match what a * fresh `load()` would show at the new limit. No-op when already at the - * origin (seq 1) or already within the window. `restoreEarlier` re-derives + * known history boundary or already within the window. `restoreEarlier` re-derives * the window start at apply time, so a delta landing during the await can't * corrupt the merge. NOT gated (refilling prepends above the viewport; the * caller preserves scroll position). */ async function refill(): Promise<void> { if (disposed) return; + if (!hasEarlier()) return; const oldest = transcript.committed[0]?.seq ?? transcript.hiddenBeforeSeq; - if (oldest <= 1) return; const want = initialWindowSize(chatLimit) - transcript.committed.length; if (want <= 0) return; try { @@ -357,6 +620,10 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { } return { + get pendingInputs() { + return pendingInputs; + }, + discardPendingInput: confirmInput, get messages(): readonly ChatMessage[] { return selectMessages(transcript); }, @@ -375,6 +642,9 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { get providerRetry(): TurnProviderRetryEvent | null { return selectProviderRetry(transcript); }, + get recovering(): boolean { + return recovering; + }, get pendingSync(): boolean { return _pendingSync; }, @@ -385,18 +655,90 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { return _model; }, get hasEarlier(): boolean { - return selectHasEarlier(transcript); + return hasEarlier(); }, get thinkingKeyBase(): number { return transcript.hiddenThinkingCount; }, + get cacheTimer(): CacheTimerState { + return cacheTimer; + }, - handleDelta(msg: ChatDeltaMessage | ChatErrorMessage): void { + noteLastActivity(at: number): void { + if (selectGenerating(transcript)) return; + cacheTimer = seedTurnGap(cacheTimer, at); + }, + + handleDelta( + msg: + | ChatAcceptedMessage + | ChatSubscriptionRejectedMessage + | ChatDeltaMessage + | ChatErrorMessage + | ChatQueuedMessage + | ChatResyncRequiredMessage, + ): void { + if (disposed) return; + if (msg.type === "chat.accepted") { + if (msg.conversationId === deps.conversationId) confirmInput(msg.requestId); + return; + } + if (msg.type === "chat.subscription-rejected") { + if (msg.conversationId === deps.conversationId) { + _error = + "Live updates could not be subscribed. Message delivery is independent; saved inputs remain available below."; + void syncTail(); + } + 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; } _error = msg.message; + if (msg.requestId !== undefined) { + savePendingInputs( + pendingInputs.map((input) => + input.request.requestId === msg.requestId + ? { ...input, rejection: msg.message } + : input, + ), + ); + transcript = removeLocalEcho(transcript, msg.requestId); + } + return; + } + if (msg.type === "chat.queued") { + if (disposed || msg.conversationId !== deps.conversationId) return; + // A queued live opening may already have confirmed the echoed text. + // Fold it first; turn-start alone must not protect still-local artifacts. + flushEvents(); + if (msg.requestId !== undefined) { + confirmInput(msg.requestId); + transcript = removeLocalEcho(transcript, msg.requestId); + } return; } if (msg.event.conversationId !== deps.conversationId) { @@ -415,10 +757,12 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { send(text: string, images?: readonly ImageInput[]): void { flushEvents(); - transcript = appendUserMessage(transcript, text, images); - maybeTrim(); - const msg: ChatSendMessage = { + // Sending ends the wait since the last turn: stamp it on that turn. + cacheTimer = stopTurnGap(cacheTimer, now(), () => lastTurnIdOf(metrics)); + const requestId = deps.createRequestId(); + const msg: ChatSendMessage & { requestId: string } = { type: "chat.send", + requestId, conversationId: deps.conversationId, message: text, ...(_model !== undefined ? { model: _model } : {}), @@ -426,18 +770,34 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { ...(deps.workspaceId !== undefined ? { workspaceId: deps.workspaceId } : {}), ...(images !== undefined && images.length > 0 ? { images: [...images] } : {}), }; + try { + savePendingInputs([...pendingInputs, { request: msg }]); + } catch (error) { + deps.onError?.("Unable to save input; the message has not been sent", error); + throw error; + } + transcript = appendUserMessage(transcript, text, images, requestId); + maybeTrim(); deps.transport.send(msg); }, - queueMessage(text: string): void { + queueMessage(text: string, images?: readonly ImageInput[]): void { const trimmed = text.trim(); - if (trimmed.length === 0) return; - const msg: ChatQueueMessage = { + if (trimmed.length === 0 && !images?.length) return; + const msg: ChatQueueMessage & { requestId: string } = { type: "chat.queue", + requestId: deps.createRequestId(), conversationId: deps.conversationId, text: trimmed, + ...(images !== undefined && images.length > 0 ? { images: [...images] } : {}), ...(deps.workspaceId !== undefined ? { workspaceId: deps.workspaceId } : {}), }; + try { + savePendingInputs([...pendingInputs, { request: msg }]); + } catch (error) { + deps.onError?.("Unable to save input; the message has not been queued", error); + throw error; + } deps.transport.send(msg); }, @@ -481,8 +841,11 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { // server instead (CR-5 `?limit=`), so a huge conversation never ships // whole. The post-sync window re-asserts the cap either way. const windowSize = initialWindowSize(chatLimit); + const currentRevision = revision; + const cachedEpoch = await deps.cache.historyEpoch(deps.conversationId); const cached = await deps.cache.load(deps.conversationId); - if (cached.length > 0) { + if (!disposed && revision === currentRevision && cachedEpoch !== undefined) { + historyEpoch = cachedEpoch; transcript = windowTranscript(applyHistory(transcript, cached), windowSize); } await syncTail(windowSize); @@ -493,7 +856,7 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { async showEarlier(): Promise<void> { if (disposed) return; const oldest = transcript.committed[0]?.seq ?? transcript.hiddenBeforeSeq; - if (oldest <= 1) return; + if (!hasEarlier()) return; const want = unloadCount(chatLimit); try { const earlier = await backfillOlder(oldest, want); @@ -504,19 +867,33 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { } }, - resync(): void { + invalidateHistory(epoch): void { + if (!disposed && (historyEpoch === undefined || epoch > historyEpoch)) resetTranscript(epoch); + }, + + resync(epoch?: number): void { if (disposed) return; + if (epoch !== undefined && (historyEpoch === undefined || epoch > historyEpoch)) + resetTranscript(epoch); // A turn may have sealed while we were disconnected (missed `turn-sealed`): // 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/store.test.ts b/src/features/chat/store.test.ts index f9b4b38..5040e67 100644 --- a/src/features/chat/store.test.ts +++ b/src/features/chat/store.test.ts @@ -1,5 +1,8 @@ import type { AgentEvent, ImageInput, StepId, StoredChunk } from "@dispatch/wire"; +import { IDBFactory, IDBKeyRange } from "fake-indexeddb"; import { describe, expect, it, vi } from "vitest"; +import { createIdbChunkStore } from "../../adapters/idb"; +import { createConversationCache } from "../conversation-cache"; import { createChatStore } from "./store.svelte"; import { createFakeCache, @@ -29,6 +32,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -61,6 +65,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -104,6 +109,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -128,6 +134,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, model: "openai/gpt-4", transport: transport.impl, @@ -150,6 +157,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -182,6 +190,7 @@ describe("createChatStore", () => { it("send omits images on the wire when none are staged (backward compatible)", () => { const transport = createFakeTransport(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: createFakeHistorySync().impl, @@ -198,6 +207,7 @@ describe("createChatStore", () => { it("send omits images on the wire for an empty array", () => { const transport = createFakeTransport(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: createFakeHistorySync().impl, @@ -214,6 +224,7 @@ describe("createChatStore", () => { it("send allows an images-only message (empty text)", () => { const transport = createFakeTransport(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: createFakeHistorySync().impl, @@ -238,6 +249,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -252,16 +264,52 @@ describe("createChatStore", () => { expect(transport.sentQueue[0]?.type).toBe("chat.queue"); expect(transport.sentQueue[0]?.conversationId).toBe(CONV_ID); expect(transport.sentQueue[0]?.text).toBe("steer left"); + expect(transport.sentQueue[0]).not.toHaveProperty("images"); store.dispose(); }); + it.each([" describe this ", "", " "])("queues images with text %j", (text) => { + const transport = createFakeTransport(); + const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), + conversationId: CONV_ID, + workspaceId: "images-workspace", + transport: transport.impl, + historySync: createFakeHistorySync().impl, + metricsSync: createFakeMetricsSync().impl, + cache: createFakeCache().impl, + }); + const images: readonly ImageInput[] = [ + { url: "data:image/png;base64,AAAA", mimeType: "image/png" }, + { url: "https://example.com/photo.jpg" }, + ]; + + store.queueMessage(text, images); + + expect(transport.sentQueue).toEqual([ + { + type: "chat.queue", + requestId: expect.any(String), + conversationId: CONV_ID, + workspaceId: "images-workspace", + text: text.trim(), + images, + }, + ]); + expect(transport.sentQueue[0]?.images).not.toBe(images); + expect(transport.sent).toHaveLength(0); + expect(store.chunks).toHaveLength(0); + store.dispose(); + }); + it("trims whitespace before sending", () => { const transport = createFakeTransport(); const historySync = createFakeHistorySync(); const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -282,6 +330,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -291,6 +340,7 @@ describe("createChatStore", () => { store.queueMessage(" "); store.queueMessage(""); + store.queueMessage("", []); expect(transport.sentQueue).toHaveLength(0); @@ -303,6 +353,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -325,6 +376,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -350,6 +402,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -371,6 +424,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -396,6 +450,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -425,6 +480,7 @@ describe("createChatStore", () => { historySync.returnChunks = [makeStoredChunk(3, "assistant")]; const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -453,6 +509,7 @@ describe("createChatStore", () => { historySync.returnChunks = [makeStoredChunk(1, "assistant")]; const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -474,6 +531,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -506,6 +564,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -528,7 +587,7 @@ describe("createChatStore", () => { store.dispose(); }); - it("overlapping syncs are guarded", async () => { + it("overlapping syncs coalesce into a sequential rerun", async () => { const transport = createFakeTransport(); const _historySync = createFakeHistorySync(); const metricsSync = createFakeMetricsSync(); @@ -546,10 +605,11 @@ describe("createChatStore", () => { if (callCount === 1) { await firstSyncPromise; } - return { chunks: [], latestSeq: sinceSeq }; + return { chunks: [], latestSeq: sinceSeq, historyEpoch: 0 }; }; const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: slowHistorySync, @@ -573,7 +633,7 @@ describe("createChatStore", () => { // Release the first sync resolveFirstSync?.(); - await new Promise((r) => setTimeout(r, 10)); + await vi.waitFor(() => expect(callCount).toBe(2)); store.dispose(); }); @@ -584,6 +644,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -629,6 +690,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, model: "openai/gpt-4", transport: transport.impl, @@ -653,6 +715,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -676,6 +739,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -706,6 +770,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -726,6 +791,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -750,6 +816,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -788,6 +855,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -849,6 +917,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -893,6 +962,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -965,6 +1035,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1031,6 +1102,7 @@ describe("createChatStore", () => { ]; const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1055,6 +1127,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1086,6 +1159,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1110,6 +1184,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1144,6 +1219,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1190,6 +1266,7 @@ describe("createChatStore", () => { const cache = createFakeCache(); let atBottom = false; // reader scrolled up const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1244,6 +1321,7 @@ describe("createChatStore", () => { const cache = createFakeCache(); let atBottom = false; const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1282,6 +1360,7 @@ describe("createChatStore", () => { ); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1311,6 +1390,7 @@ describe("createChatStore", () => { historySync.returnChunks = Array.from({ length: 500 }, (_, i) => makeStoredChunk(i + 1)); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1346,6 +1426,7 @@ describe("createChatStore", () => { historySync.returnChunks = [makeStoredChunk(3)]; const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1370,6 +1451,7 @@ describe("createChatStore", () => { historySync.returnChunks = Array.from({ length: 500 }, (_, i) => makeStoredChunk(i + 1)); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1407,6 +1489,7 @@ describe("createChatStore", () => { ); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1439,6 +1522,7 @@ describe("createChatStore", () => { ); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1470,6 +1554,7 @@ describe("createChatStore", () => { ); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1502,6 +1587,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1539,6 +1625,7 @@ describe("createChatStore", () => { Array.from({ length: 200 }, (_, i) => makeStoredChunk(i + 1)), ); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1570,6 +1657,7 @@ describe("createChatStore", () => { // Server holds 200; cold-cache load at limit 100 → window 75 → seqs 126..200. historySync.returnChunks = Array.from({ length: 200 }, (_, i) => makeStoredChunk(i + 1)); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1597,6 +1685,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1636,6 +1725,7 @@ describe("createChatStore", () => { Array.from({ length: 50 }, (_, i) => makeStoredChunk(i + 1)), ); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1645,6 +1735,7 @@ describe("createChatStore", () => { }); await store.load(); // only 50 chunks → all loaded, window starts at seq 1 expect(store.chunks).toHaveLength(50); + await store.showEarlier(); // Probe server exhaustion; seq 1 alone is not a boundary. expect(store.hasEarlier).toBe(false); const callsAfterLoad = historySync.calls.length; @@ -1662,6 +1753,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1691,6 +1783,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1712,6 +1805,7 @@ describe("createChatStore — batched delta application (scheduleFlush)", () => const scheduled: (() => void)[] = []; const historySync = createFakeHistorySync(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: createFakeTransport().impl, historySync: historySync.impl, @@ -1789,3 +1883,383 @@ describe("createChatStore — batched delta application (scheduleFlush)", () => expect(store.chunks).toHaveLength(0); }); }); + +describe("createChatStore — cacheTimer", () => { + function makeStore(now: () => number) { + return createChatStore({ + createRequestId: () => crypto.randomUUID(), + conversationId: CONV_ID, + transport: createFakeTransport().impl, + historySync: createFakeHistorySync().impl, + metricsSync: createFakeMetricsSync().impl, + cache: createFakeCache().impl, + now, + }); + } + const call = (stepId: string, toolCallId: string) => + deltaEvent({ + type: "tool-call", + conversationId: CONV_ID, + turnId: "t1", + stepId: stepId as StepId, + toolCallId, + toolName: "read_file", + input: {}, + }); + const result = (stepId: string, toolCallId: string) => + deltaEvent({ + type: "tool-result", + conversationId: CONV_ID, + turnId: "t1", + stepId: stepId as StepId, + toolCallId, + toolName: "read_file", + content: "ok", + isError: false, + }); + const stepComplete = (stepId: string) => + deltaEvent({ + type: "step-complete", + conversationId: CONV_ID, + turnId: "t1", + stepId: stepId as StepId, + }); + const done = (turnId: string) => + deltaEvent({ type: "done", conversationId: CONV_ID, turnId, reason: "end-turn" }); + + it("is idle until something happens", () => { + const store = makeStore(() => 1000); + expect(store.cacheTimer.running).toBeNull(); + store.handleDelta(deltaEvent({ type: "turn-start", conversationId: CONV_ID, turnId: "t1" })); + expect(store.cacheTimer.running).toBeNull(); + store.dispose(); + }); + + it("times a step's tool execution with the injected clock and stamps it", () => { + let clock = 0; + const store = makeStore(() => clock); + store.handleDelta(call("t1#0", "a")); + clock = 1_000; + store.handleDelta(stepComplete("t1#0")); + expect(store.cacheTimer.running).toMatchObject({ kind: "step", startedAt: 1_000 }); + clock = 31_000; + store.handleDelta(result("t1#0", "a")); + expect(store.cacheTimer.running).toBeNull(); + expect(store.cacheTimer.stepStamps.get("t1#0")).toBe(30_000); + store.dispose(); + }); + + it("starts a turn gap on done and stamps it on send", () => { + let clock = 5_000; + const store = makeStore(() => clock); + store.handleDelta(done("t1")); + expect(store.cacheTimer.running).toEqual({ kind: "turn", turnId: "t1", startedAt: 5_000 }); + clock = 125_000; + store.send("next"); + expect(store.cacheTimer.running).toBeNull(); + expect(store.cacheTimer.turnStamps.get("t1")).toBe(120_000); + store.dispose(); + }); + + it("ignores another conversation's events", () => { + const store = makeStore(() => 5_000); + store.handleDelta( + deltaEvent({ type: "done", conversationId: "other", turnId: "t1", reason: "end-turn" }), + ); + expect(store.cacheTimer.running).toBeNull(); + store.dispose(); + }); + + it("noteLastActivity seeds a turn gap, keeps the later start, and is ignored while generating", () => { + const store = makeStore(() => 5_000); + store.noteLastActivity(3_000); + expect(store.cacheTimer.running).toEqual({ kind: "turn", turnId: null, startedAt: 3_000 }); + store.handleDelta(done("t1")); + expect(store.cacheTimer.running).toEqual({ kind: "turn", turnId: "t1", startedAt: 5_000 }); + store.noteLastActivity(4_000); // an older seed never rewinds it + expect(store.cacheTimer.running).toMatchObject({ startedAt: 5_000 }); + store.dispose(); + + const busy = makeStore(() => 5_000); + busy.handleDelta(deltaEvent({ type: "turn-start", conversationId: CONV_ID, turnId: "t2" })); + busy.noteLastActivity(4_000); + expect(busy.cacheTimer.running).toBeNull(); + busy.dispose(); + }); +}); + +describe("history epochs", () => { + it.each([ + "tail", + "backfill", + ])("recovers missed compaction on %s and exhausts non-1 history", async (path) => { + const history = createFakeHistorySync(); + const cache = createFakeCache(); + history.returnChunks = Array.from({ length: 20 }, (_, i) => makeStoredChunk(i + 1)); + const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), + conversationId: CONV_ID, + transport: createFakeTransport().impl, + historySync: history.impl, + metricsSync: createFakeMetricsSync().impl, + cache: cache.impl, + chatLimit: 10, + }); + await store.load(); + history.historyEpoch = 1; + history.returnChunks = [makeStoredChunk(101), makeStoredChunk(102)]; + if (path === "tail") { + store.resync(); + await vi.waitFor(() => expect(store.chunks[0]?.seq).toBe(101)); + } else await store.showEarlier(); + expect(store.chunks.map((c) => c.seq)).toEqual([101, 102]); + expect((await cache.impl.load(CONV_ID)).map((c) => c.seq)).toEqual([101, 102]); + expect(history.calls.at(-1)).toMatchObject({ sinceSeq: 0, window: { limit: 7 } }); + expect(store.hasEarlier).toBe(false); + const calls = history.calls.length; + await store.showEarlier(); + expect(history.calls).toHaveLength(calls); + store.dispose(); + }); +}); + +it("a compaction broadcast during an in-flight fetch discards the old response", async () => { + const history = createFakeHistorySync(); + history.returnChunks = [makeStoredChunk(101)]; + let resolve!: (res: import("@dispatch/transport-contract").ConversationHistoryResponse) => void; + const first = new Promise<import("@dispatch/transport-contract").ConversationHistoryResponse>( + (r) => { + resolve = r; + }, + ); + const sync = vi + .fn<import("./ports").HistorySync>() + .mockImplementationOnce(() => first) + .mockImplementation(history.impl); + const cache = createFakeCache(); + const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), + conversationId: CONV_ID, + transport: createFakeTransport().impl, + historySync: sync, + metricsSync: createFakeMetricsSync().impl, + cache: cache.impl, + }); + const loading = store.load(); + await vi.waitFor(() => expect(sync).toHaveBeenCalledTimes(1)); + history.historyEpoch = 1; + store.resync(1); + resolve({ chunks: [makeStoredChunk(1)], latestSeq: 1, historyEpoch: 0 }); + await loading; + expect(sync).toHaveBeenCalledTimes(2); + expect(store.chunks.map((c) => c.seq)).toEqual([101]); + expect((await cache.impl.load(CONV_ID)).map((c) => c.seq)).toEqual([101]); + store.dispose(); +}); + +it("reloads unknown legacy cache epochs rather than merging pre-upgrade rows", async () => { + vi.stubGlobal("IDBKeyRange", IDBKeyRange); + const storage = createIdbChunkStore({ indexedDB: new IDBFactory() }); + const cache = createConversationCache(storage); + await storage.append(CONV_ID, [makeStoredChunk(1)]); + const history = createFakeHistorySync(); + history.returnChunks = [makeStoredChunk(101)]; + const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), + conversationId: CONV_ID, + transport: createFakeTransport().impl, + historySync: history.impl, + metricsSync: createFakeMetricsSync().impl, + cache, + }); + try { + await store.load(); + expect(history.calls[0]).toMatchObject({ sinceSeq: 0 }); + expect(store.chunks.map((c) => c.seq)).toEqual([101]); + expect(await cache.historyEpoch(CONV_ID)).toBe(0); + expect((await cache.load(CONV_ID)).map((c) => c.seq)).toEqual([101]); + } finally { + store.dispose(); + vi.unstubAllGlobals(); + } +}); + +describe("server epoch rollback", () => { + it.each([ + "tail", + "backfill", + ])("converges on %s with a real cache and a fresh bounded fetch", async (path) => { + vi.stubGlobal("IDBKeyRange", IDBKeyRange); + const storage = createIdbChunkStore({ indexedDB: new IDBFactory() }); + const cache = createConversationCache(storage); + const old = Array.from({ length: 20 }, (_, i) => makeStoredChunk(i + 1)); + await cache.commit(CONV_ID, path === "backfill" ? old.slice(-7) : old, 5); + const history = createFakeHistorySync(); + history.historyEpoch = 5; + const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), + conversationId: CONV_ID, + transport: createFakeTransport().impl, + historySync: history.impl, + metricsSync: createFakeMetricsSync().impl, + cache, + chatLimit: 10, + }); + try { + await store.load(); + history.historyEpoch = 0; + // Recreated ids can reuse seqs with completely different content. + const restored: StoredChunk = { + seq: 1, + role: "user", + chunk: { type: "text", text: "restored" }, + }; + history.returnChunks = [restored]; + const before = history.calls.length; + if (path === "tail") { + store.resync(); + await vi.waitFor(() => expect(store.pendingSync).toBe(false)); + } else await store.showEarlier(); + expect(history.calls.length - before).toBeLessThanOrEqual(2); + expect(history.calls.at(-1)).toMatchObject({ sinceSeq: 0, window: { limit: 7 } }); + expect(await cache.historyEpoch(CONV_ID)).toBe(0); + expect(await cache.load(CONV_ID)).toEqual([restored]); + expect(store.chunks.map((c) => c.chunk)).toEqual([restored.chunk]); + expect(store.hasEarlier).toBe(false); + expect(store.error).toBeNull(); + } finally { + store.dispose(); + vi.unstubAllGlobals(); + } + }); + + it("bounds reruns even when every response is invalidated by another compaction", async () => { + let calls = 0; + let churn = true; + let store!: ReturnType<typeof createChatStore>; + const sync: import("./ports").HistorySync = async () => { + calls++; + // Stop the old broken loop safely rather than hanging the test worker. + if (churn && calls <= 6) store.resync(calls); + return { chunks: [], latestSeq: 0, historyEpoch: calls - 1 }; + }; + store = createChatStore({ + createRequestId: () => crypto.randomUUID(), + conversationId: CONV_ID, + transport: createFakeTransport().impl, + historySync: sync, + metricsSync: createFakeMetricsSync().impl, + cache: createFakeCache().impl, + }); + try { + await store.load(); + expect(calls).toBe(3); + expect(store.pendingSync).toBe(false); + expect(store.error).toMatch(/history changed repeatedly/i); + churn = false; + store.resync(); + await vi.waitFor(() => expect(store.pendingSync).toBe(false)); + expect(calls).toBe(4); + expect(store.error).toBeNull(); + } finally { + store.dispose(); + } + }); +}); + +it.each([ + "tail", + "backfill", +])("rejects a delayed old %s after a lower-epoch history is observed", async (delayed) => { + vi.stubGlobal("IDBKeyRange", IDBKeyRange); + const cache = createConversationCache(createIdbChunkStore({ indexedDB: new IDBFactory() })); + await cache.commit( + CONV_ID, + Array.from({ length: 7 }, (_, i) => makeStoredChunk(14 + i)), + 5, + ); + const restored: StoredChunk = { + seq: 1, + role: "user", + chunk: { type: "text", text: "new history" }, + }; + type Response = import("@dispatch/transport-contract").ConversationHistoryResponse; + let resolve!: (res: Response) => void; + const slow = new Promise<Response>((r) => { + resolve = r; + }); + let delayNext = false; + let epoch = 5; + const sync = vi.fn<import("./ports").HistorySync>().mockImplementation(async () => { + if (delayNext) { + delayNext = false; + return slow; + } + return { + chunks: epoch === 5 ? [] : [restored], + latestSeq: epoch === 5 ? 20 : 1, + historyEpoch: epoch, + }; + }); + const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), + conversationId: CONV_ID, + transport: createFakeTransport().impl, + historySync: sync, + metricsSync: createFakeMetricsSync().impl, + cache, + chatLimit: 10, + }); + try { + await store.load(); + delayNext = true; + let earlier: Promise<void>; + if (delayed === "tail") { + store.resync(); + await vi.waitFor(() => expect(sync).toHaveBeenCalledTimes(2)); + epoch = 0; + earlier = store.showEarlier(); + await earlier; // Invalidates the old tail and requests a sequential rerun. + } else { + earlier = store.showEarlier(); + await vi.waitFor(() => expect(sync).toHaveBeenCalledTimes(2)); + epoch = 0; + store.resync(); + await vi.waitFor(() => expect(store.chunks.map((c) => c.chunk)).toEqual([restored.chunk])); + } + resolve({ chunks: [makeStoredChunk(21)], latestSeq: 21, historyEpoch: 5 }); + await earlier; + await vi.waitFor(() => expect(store.pendingSync).toBe(false)); + expect(await cache.historyEpoch(CONV_ID)).toBe(0); + expect(await cache.load(CONV_ID)).toEqual([restored]); + expect(store.chunks.map((c) => c.chunk)).toEqual([restored.chunk]); + expect(sync).toHaveBeenCalledTimes(4); + } finally { + store.dispose(); + vi.unstubAllGlobals(); + } +}); + +it("replaces an empty high-epoch cache with the server's empty lower epoch in one request", async () => { + const cache = createConversationCache(createIdbChunkStore({ indexedDB: new IDBFactory() })); + await cache.reset(CONV_ID, 8); + const history = createFakeHistorySync(); + const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), + conversationId: CONV_ID, + transport: createFakeTransport().impl, + historySync: history.impl, + metricsSync: createFakeMetricsSync().impl, + cache, + }); + try { + await store.load(); + expect(history.calls).toHaveLength(1); + expect(await cache.historyEpoch(CONV_ID)).toBe(0); + expect(await cache.load(CONV_ID)).toEqual([]); + expect(store.chunks).toEqual([]); + expect(store.hasEarlier).toBe(false); + } finally { + store.dispose(); + } +}); diff --git a/src/features/chat/test-helpers.ts b/src/features/chat/test-helpers.ts index c99d1f4..41cfb3b 100644 --- a/src/features/chat/test-helpers.ts +++ b/src/features/chat/test-helpers.ts @@ -43,14 +43,22 @@ export interface FakeHistorySync { readonly calls: Array<{ conversationId: string; sinceSeq: number; window?: HistoryWindow }>; /** Set the chunks to return on the next call. */ returnChunks: readonly StoredChunk[]; + historyEpoch: number; readonly impl: HistorySync; } export function createFakeHistorySync(): FakeHistorySync { const calls: Array<{ conversationId: string; sinceSeq: number; window?: HistoryWindow }> = []; let returnChunks: readonly StoredChunk[] = []; + let historyEpoch = 0; return { calls, + get historyEpoch() { + return historyEpoch; + }, + set historyEpoch(v: number) { + historyEpoch = v; + }, get returnChunks() { return returnChunks; }, @@ -72,7 +80,7 @@ export function createFakeHistorySync(): FakeHistorySync { chunks = chunks.slice(-window.limit); } const latestSeq = chunks.length > 0 ? Math.max(...chunks.map((c) => c.seq)) : sinceSeq; - return { chunks, latestSeq }; + return { chunks, latestSeq, historyEpoch }; }, }; } @@ -122,13 +130,28 @@ export interface FakeCache { export function createFakeCache(): FakeCache { const store = new Map<string, StoredChunk[]>(); + const epochs = new Map<string, number>(); return { store, impl: { + async historyEpoch(id) { + return epochs.get(id) ?? 0; + }, + async reset(id, epoch) { + store.delete(id); + epochs.set(id, epoch); + }, + async replace(id, incoming, epoch, expectedEpoch) { + if ((epochs.get(id) ?? 0) !== expectedEpoch) return false; + store.set(id, [...incoming]); + epochs.set(id, epoch); + return true; + }, async load(conversationId) { return store.get(conversationId) ?? []; }, - async commit(conversationId, incoming) { + async commit(conversationId, incoming, epoch) { + if (epoch !== undefined) epochs.set(conversationId, epoch); const existing = store.get(conversationId) ?? []; const seen = new Set(existing.map((c) => c.seq)); const toAppend = incoming.filter((c) => !seen.has(c.seq)); diff --git a/src/features/chat/ui.test.ts b/src/features/chat/ui.test.ts index 91585e2..74c0619 100644 --- a/src/features/chat/ui.test.ts +++ b/src/features/chat/ui.test.ts @@ -3,8 +3,13 @@ import { render, screen, within } from "@testing-library/svelte"; import userEvent from "@testing-library/user-event"; import { describe, expect, it, vi } from "vitest"; import type { RenderedChunk } from "../../core/chunks"; -import type { TurnMetricsEntry } from "../../core/metrics"; +import { + type CacheTimerState, + initialCacheTimerState, + type TurnMetricsEntry, +} from "../../core/metrics"; import type { ConvertedJpeg } from "./adapter/image-encode"; +import CacheTimerBadge from "./ui/CacheTimerBadge.svelte"; import ChatView from "./ui/ChatView.svelte"; import Composer from "./ui/Composer.svelte"; import ModelSelector from "./ui/ModelSelector.svelte"; @@ -51,6 +56,25 @@ describe("ChatView", () => { expect(screen.getByText("Hello!")).toBeInTheDocument(); }); + it("renders local-path Markdown images through the backend image route", () => { + const chunks: RenderedChunk[] = [ + { + seq: 1, + role: "assistant", + chunk: { type: "text", text: "" }, + provisional: false, + }, + ]; + + const { container } = render(ChatView, { + props: { chunks, apiBaseUrl: "http://localhost:24203", cwd: "/w/foo" }, + }); + + expect(container.querySelector("img")?.getAttribute("src")).toBe( + "http://localhost:24203/local-image?path=%2Fhome%2Fu%2Fpic.png", + ); + }); + it("shows the show-earlier button only when earlier history is unloaded, and pages it in", async () => { const chunks: RenderedChunk[] = [ { seq: 26, role: "user", chunk: { type: "text", text: "later" }, provisional: false }, @@ -825,6 +849,57 @@ describe("ChatView", () => { }); describe("Composer", () => { + it.each([ + "running", + "queued", + ] as const)("relabels Stop only with queued input while %s and preserves its action", async (status) => { + const onStop = vi.fn(); + const onSend = vi.fn(); + const onQueue = vi.fn(); + const user = userEvent.setup(); + const { rerender } = render(Composer, { props: { onSend, onQueue, onStop, status } }); + const expectPlainStop = () => { + const button = screen.getByRole("button", { name: "Stop generation" }); + expect(button).toHaveTextContent(/^Stop$/); + expect(button).not.toHaveAttribute("title"); + }; + expectPlainStop(); // No queue snapshot available. + await rerender({ queuedMessageCount: 0 }); + expectPlainStop(); + await user.click(screen.getByRole("button", { name: "Stop generation" })); + expect(onStop).toHaveBeenCalledTimes(1); + + for (const queuedMessageCount of [1, 2]) { + await rerender({ queuedMessageCount }); + const button = screen.getByRole("button", { name: "Flush" }); + expect(button).toHaveTextContent(/^Flush$/); + expect(button).toHaveAttribute( + "title", + "Stop the current response and start the queued message now", + ); + } + await user.click(screen.getByRole("button", { name: "Flush" })); + expect(onStop).toHaveBeenCalledTimes(2); + expect(onSend).not.toHaveBeenCalled(); + expect(onQueue).not.toHaveBeenCalled(); + + const input = screen.getByRole("textbox", { name: "Message input" }); + await user.type(input, "next"); + expect(screen.getByRole("button", { name: "Queue" })).toBeInTheDocument(); + expect(screen.queryByRole("button", { name: "Flush" })).toBeNull(); + await user.click(screen.getByRole("button", { name: "Queue" })); + expect(onQueue).toHaveBeenCalledExactlyOnceWith("next", undefined); + expect(screen.getByRole("button", { name: "Flush" })).toBeInTheDocument(); + await rerender({ queuedMessageCount: 0 }); + expectPlainStop(); // Queue drained or cancelled. + await rerender({ queuedMessageCount: 1, status: "idle" }); + expect(screen.getByRole("button", { name: "Send" })).toBeDisabled(); + await rerender({ status: "error" }); + expect(screen.getByRole("button", { name: "Send" })).toBeDisabled(); + await rerender({ status, queuedMessageCount: undefined }); + expectPlainStop(); + }); + it("calls onSend with the typed text and clears", async () => { const onSend = vi.fn(); const user = userEvent.setup(); @@ -1005,17 +1080,28 @@ describe("Composer", () => { expect(screen.queryByRole("button", { name: "Remove image" })).not.toBeInTheDocument(); }); - it("queues (steers) text-only and never forwards images", async () => { - // While running, the Send button becomes "Queue"; steering is text-only. + it("still queues text-only with images omitted", async () => { + const onQueue = vi.fn(); + const onSend = vi.fn(); + const user = userEvent.setup(); + render(Composer, { props: { onSend, onQueue, status: "running" } }); + await user.type(screen.getByRole("textbox", { name: "Message input" }), "steer{Enter}"); + expect(onQueue).toHaveBeenCalledExactlyOnceWith("steer", undefined); + expect(onSend).not.toHaveBeenCalled(); + }); + + it("queues text and images while running and clears the composer", async () => { const onQueue = vi.fn(); const onSend = vi.fn(); const user = userEvent.setup(); - const { container } = render(Composer, { props: { onSend, onQueue, status: "running" } }); + const { container } = render(Composer, { + props: { onSend, onQueue, status: "running", encodeJpeg: async () => null }, + }); const textarea = screen.getByRole("textbox", { name: "Message input" }); await user.type(textarea, "steer here"); - // Also stage an image — it must NOT be forwarded on a queue. + // Stage an image along with the steering text. const file = new File(["PNG"], "shot.png", { type: "image/png" }); const input = container.querySelector('input[type="file"]') as HTMLInputElement; Object.defineProperty(input, "files", { value: [file], writable: false }); @@ -1025,8 +1111,66 @@ describe("Composer", () => { }); await user.click(screen.getByRole("button", { name: "Queue" })); - expect(onQueue).toHaveBeenCalledWith("steer here"); + expect(onQueue).toHaveBeenCalledExactlyOnceWith("steer here", [ + { url: "data:image/png;base64,UE5H", mimeType: "image/png" }, + ]); + expect(onSend).not.toHaveBeenCalled(); + expect(textarea).toHaveValue(""); + expect(screen.queryByRole("button", { name: "Remove image" })).not.toBeInTheDocument(); + }); + + it.each(["running", "queued"] as const)("queues images-only while %s", async (status) => { + const onQueue = vi.fn(); + const onSend = vi.fn(); + const user = userEvent.setup(); + const { container } = render(Composer, { props: { onSend, onQueue, status } }); + const input = container.querySelector('input[type="file"]') as HTMLInputElement; + Object.defineProperty(input, "files", { + value: [new File(["GIF"], "shot.gif", { type: "image/gif" })], + }); + input.dispatchEvent(new Event("change", { bubbles: true })); + await screen.findByRole("button", { name: "Remove image" }); + + await user.click(screen.getByRole("button", { name: "Queue" })); + + expect(onQueue).toHaveBeenCalledExactlyOnceWith("", [ + { url: "data:image/gif;base64,R0lG", mimeType: "image/gif" }, + ]); + expect(onSend).not.toHaveBeenCalled(); + expect(screen.queryByRole("button", { name: "Remove image" })).not.toBeInTheDocument(); + }); + + it.each([ + "running", + "queued", + ] as const)("keeps images and text staged while %s without onQueue", async (status) => { + const onSend = vi.fn(); + const user = userEvent.setup(); + const { container, rerender } = render(Composer, { props: { onSend, status } }); + const input = container.querySelector('input[type="file"]') as HTMLInputElement; + Object.defineProperty(input, "files", { + value: [new File(["GIF"], "shot.gif", { type: "image/gif" })], + }); + input.dispatchEvent(new Event("change", { bubbles: true })); + await screen.findByRole("button", { name: "Remove image" }); + const textarea = screen.getByRole("textbox", { name: "Message input" }); + await user.type(textarea, "{Enter}"); // image-only: no queue callback + await user.type(textarea, "keep this{Enter}"); + container + .querySelector("form") + ?.dispatchEvent(new Event("submit", { bubbles: true, cancelable: true })); + expect(onSend).not.toHaveBeenCalled(); + expect(textarea).toHaveValue("keep this"); + expect(screen.getByRole("button", { name: "Remove image" })).toBeInTheDocument(); + expect(screen.getByRole("button", { name: "Queue" })).toBeDisabled(); + // Once idle, the same staged input can actually be sent. + await rerender({ status: "idle" }); + await user.click(screen.getByRole("button", { name: "Send" })); + expect(onSend).toHaveBeenCalledExactlyOnceWith("keep this", [ + { url: "data:image/gif;base64,R0lG", mimeType: "image/gif" }, + ]); + expect(screen.queryByRole("button", { name: "Remove image" })).not.toBeInTheDocument(); }); }); @@ -1542,3 +1686,192 @@ describe("ReasoningEffortSelector", () => { }); }); }); + +describe("ChatView — cache timer", () => { + const usage = { inputTokens: 1000, outputTokens: 10, cacheReadTokens: 900 }; + function step(stepId: string) { + return { stepId: stepId as StepId, usage, decodeMs: 100 }; + } + function turn(turnId: string, stepIds: string[] = [`${turnId}#0`]): TurnMetricsEntry { + const steps = stepIds.map(step); + return { turnId, steps, total: { turnId, usage, steps } }; + } + function exchange(turnId: string, base: number): RenderedChunk[] { + return [ + { + seq: base, + role: "user", + chunk: { type: "text", text: `q-${turnId}` }, + provisional: false, + turnId, + }, + { + seq: base + 1, + role: "assistant", + chunk: { type: "text", text: `a-${turnId}` }, + provisional: false, + turnId, + }, + ]; + } + function toolChunk(turnId: string, stepId: string, seq: number): RenderedChunk { + return { + seq, + role: "assistant", + chunk: { + type: "tool-call", + toolCallId: `call-${stepId}`, + toolName: "read_file", + input: {}, + stepId: stepId as StepId, + }, + provisional: false, + turnId, + }; + } + function state(partial: Partial<CacheTimerState>): CacheTimerState { + return { ...initialCacheTimerState(), ...partial }; + } + + it("renders nothing when the feature is off (cacheTimer null)", () => { + render(ChatView, { props: { chunks: exchange("t1", 1), turnMetrics: [turn("t1")] } }); + expect(screen.queryByTestId("cache-timer-live")).toBeNull(); + expect(screen.queryByTestId("cache-timer-stamp")).toBeNull(); + }); + + it("shows a live timer under executing tools, counting up to 10+ min", async () => { + vi.useFakeTimers(); + try { + const chunks = [exchange("t1", 1)[0] as RenderedChunk, toolChunk("t1", "t1#0", 2)]; + render(ChatView, { + props: { + chunks, + turnMetrics: [{ turnId: "t1", steps: [step("t1#0")], total: null }], + cacheTimer: state({ + running: { + kind: "step", + turnId: "t1", + stepId: "t1#0", + startedAt: Date.now(), + toolCount: 1, + pending: new Set(["call-t1#0"]), + maxToolMs: 0, + }, + }), + }, + }); + const row = screen.getByTestId("cache-timer-tools"); + expect(row).toHaveTextContent(/Running tools/); + expect(screen.getByTestId("cache-timer-live")).toHaveTextContent("0:00"); + await vi.advanceTimersByTimeAsync(5 * 60_000 + 30_000); + expect(screen.getByTestId("cache-timer-live")).toHaveTextContent("5:30"); + await vi.advanceTimersByTimeAsync(5 * 60_000); + expect(screen.getByTestId("cache-timer-live")).toHaveTextContent("10+ min"); + } finally { + vi.useRealTimers(); + } + }); + + it("hides the live tools row for a step with no tool calls", () => { + render(ChatView, { + props: { + chunks: exchange("t1", 1), + turnMetrics: [{ turnId: "t1", steps: [step("t1#0")], total: null }], + cacheTimer: state({ + running: { + kind: "step", + turnId: "t1", + stepId: "t1#0", + startedAt: Date.now(), + toolCount: 0, + pending: new Set(), + maxToolMs: 0, + }, + }), + }, + }); + expect(screen.queryByTestId("cache-timer-tools")).toBeNull(); + }); + + it("stamps a finished step gap on that step's line, next to its cache badge", () => { + const chunks = [exchange("t1", 1)[0] as RenderedChunk, toolChunk("t1", "t1#0", 2)]; + render(ChatView, { + props: { + chunks, + turnMetrics: [{ turnId: "t1", steps: [step("t1#0")], total: null }], + cacheTimer: state({ stepStamps: new Map([["t1#0", 65_000]]) }), + }, + }); + const stamp = screen.getByTestId("cache-timer-stamp"); + expect(stamp).toHaveTextContent("1:05"); + // Sized to match the step's (xs) cache badge. + expect(stamp).toHaveClass("badge-xs"); + expect(stamp).not.toHaveClass("badge-sm"); + // Same line as the step's cache badge. + expect(stamp.parentElement).toContainElement(screen.getByTestId("step-cache-badge")); + expect(screen.queryByTestId("cache-timer-tools")).toBeNull(); + }); + + it("runs a turn gap live on that turn's row, and shows stamps on earlier turns", () => { + render(ChatView, { + props: { + chunks: [...exchange("t1", 1), ...exchange("t2", 3)], + turnMetrics: [turn("t1"), turn("t2")], + cacheTimer: state({ + running: { kind: "turn", turnId: "t2", startedAt: Date.now() }, + turnStamps: new Map([["t1", 11 * 60_000]]), + }), + }, + }); + const live = screen.getByTestId("cache-timer-live"); + expect(live.closest(".chat-bubble")).toHaveTextContent(/turn 2/); + expect(live).toHaveClass("badge-sm"); + const stamp = screen.getByTestId("cache-timer-stamp"); + expect(stamp).toHaveTextContent("10+ min"); + expect(stamp.closest(".chat-bubble")).toHaveTextContent(/turn 1/); + }); + + it("a live timer mounts without re-running its effect on a moving clock (no render stall)", async () => { + // Regression: the badge's effect once READ the `current` state it wrote, so + // with a real (always-advancing) clock it re-ran itself until Svelte threw + // `effect_update_depth_exceeded` — aborting the flush, so turn data showed late. + let calls = 0; + const now = () => { + calls++; + return 1_000_000 + calls; + }; + render(CacheTimerBadge, { props: { since: 1_000_000, now } }); + await new Promise((r) => setTimeout(r, 20)); + expect(calls).toBeLessThan(5); + expect(screen.getByTestId("cache-timer-live")).toHaveTextContent("0:00"); + }); + + it("renders the turn's metrics alongside a live turn timer", () => { + render(ChatView, { + props: { + chunks: exchange("t1", 1), + turnMetrics: [turn("t1")], + cacheTimer: state({ running: { kind: "turn", turnId: "t1", startedAt: Date.now() } }), + }, + }); + expect(screen.getByText(/turn 1 · 10 tok out/)).toBeInTheDocument(); + expect(screen.getByText("a-t1")).toBeInTheDocument(); + expect(screen.getByTestId("cache-timer-live")).toHaveTextContent(/^0:00$/); + }); + + it("puts a SEEDED turn gap (no turnId) on the most recent turn's row", () => { + render(ChatView, { + props: { + chunks: [...exchange("t1", 1), ...exchange("t2", 3)], + turnMetrics: [turn("t1"), turn("t2")], + cacheTimer: state({ + running: { kind: "turn", turnId: null, startedAt: Date.now() - 2 * 60_000 }, + }), + }, + }); + const badges = screen.getAllByTestId("cache-timer-live"); + expect(badges).toHaveLength(1); + expect(badges[0]).toHaveTextContent("2:00"); + expect(badges[0]?.closest(".chat-bubble")).toHaveTextContent(/turn 2/); + }); +}); diff --git a/src/features/chat/ui/CacheTimerBadge.svelte b/src/features/chat/ui/CacheTimerBadge.svelte new file mode 100644 index 0000000..9e6f96d --- /dev/null +++ b/src/features/chat/ui/CacheTimerBadge.svelte @@ -0,0 +1,70 @@ +<script lang="ts"> + /** + * One cache-timer badge. LIVE when given `since` (counts up from that epoch + * ms, ticking once a second, and stops ticking at the 10-minute cap); STAMPED + * when given `elapsedMs` (a finished gap — static). Formatting lives in + * `core/metrics` (`viewCacheTimer`). Owning the tick here means only this + * badge re-renders each second, not the transcript. + */ + import { untrack } from "svelte"; + import { CACHE_TIMER_MAX_MS, viewCacheTimer } from "../../../core/metrics"; + + let { + since, + elapsedMs, + size = "sm", + now = Date.now, + }: { + /** Live: epoch ms the gap started. */ + since?: number; + /** Stamped: the gap's final duration (wins over `since`). */ + elapsedMs?: number; + /** Badge size — match the neighbouring cache badge (`xs` on step lines). */ + size?: "xs" | "sm"; + /** Wall clock (injected for tests). */ + now?: () => number; + } = $props(); + + // Initial read only (the effect below keeps it current while live). + let current = $state(untrack(() => now())); + + // Depends ONLY on `since`/`elapsedMs`/`now`. It must never READ `current` + // (it writes it): reading its own write makes the effect re-run itself on + // every clock change → `effect_update_depth_exceeded`, which aborts the whole + // reactive flush and stalls the transcript (turn data rendering late). + $effect(() => { + if (elapsedMs !== undefined || since === undefined) return; + const start = since; + const clock = now; + const t0 = clock(); + current = t0; + if (t0 - start >= CACHE_TIMER_MAX_MS) return; + const id = setInterval(() => { + const t = clock(); + current = t; + if (t - start >= CACHE_TIMER_MAX_MS) clearInterval(id); + }, 1000); + return () => clearInterval(id); + }); + + const badgeClass = { + success: "badge-success", + warning: "badge-warning", + error: "badge-error", + } as const; + + const live = $derived(elapsedMs === undefined); + const view = $derived(viewCacheTimer(elapsedMs ?? current - (since ?? current))); +</script> + +<span + class="badge font-mono tabular-nums {badgeClass[view.level]}" + class:badge-xs={size === "xs"} + class:badge-sm={size === "sm"} + class:badge-outline={!live} + title={live + ? "Cache timer (running): time since the last model request. The Claude prompt cache expires after 5 minutes idle." + : "Cache timer (stopped): how long the cache sat idle before the next request. The Claude prompt cache expires after 5 minutes idle."} + data-testid={live ? "cache-timer-live" : "cache-timer-stamp"} + >{view.label}</span +> diff --git a/src/features/chat/ui/ChatView.svelte b/src/features/chat/ui/ChatView.svelte index 18fea7c..c7f566a 100644 --- a/src/features/chat/ui/ChatView.svelte +++ b/src/features/chat/ui/ChatView.svelte @@ -2,7 +2,9 @@ import type { TurnProviderRetryEvent } from "@dispatch/wire"; import { groupRenderedChunks, resolveImageUrl, type RenderedChunk, viewProviderRetry } from "../index"; import { + type CacheTimerState, interleaveTurnMetrics, + lastTurnRowIndex, viewCacheRate, viewExpectedCache, viewStepMetrics, @@ -10,6 +12,7 @@ type TurnMetricsEntry, } from "../../../core/metrics"; import { Markdown } from "../../markdown"; + import CacheTimerBadge from "./CacheTimerBadge.svelte"; import LazyCollapse from "./LazyCollapse.svelte"; import ToolCard from "./ToolCard.svelte"; @@ -21,14 +24,18 @@ let { chunks, + recovering = false, turnMetrics = [], hasEarlier = false, onShowEarlier, thinkingKeyBase = 0, providerRetry = null, apiBaseUrl = "", + cwd = null, + cacheTimer = null, }: { chunks: readonly RenderedChunk[]; + recovering?: boolean; turnMetrics?: readonly TurnMetricsEntry[]; /** Earlier history is unloaded (chat limit) and can be paged back in. */ hasEarlier?: boolean; @@ -55,10 +62,25 @@ * "" (root-relative — a browser resolves `/images/…` against its origin). */ apiBaseUrl?: string; + /** + * The active conversation's working directory (null when unset). Relative + * Markdown image srcs (``) resolve against it; local absolute + * and `file://` paths do not need it. Passed to `Markdown` with `apiBaseUrl`. + */ + cwd?: string | null; + /** + * The cache timer (gaps between model requests): a running step gap shows + * live at the bottom (under the executing tools), a running turn gap live + * on that turn's metrics row; finished gaps are stamped on their step / + * turn metrics line. `null` (the default) hides it entirely — the caller + * gates on the model. + */ + cacheTimer?: CacheTimerState | null; } = $props(); // True while a show-earlier page-in is awaited (disables the button). let loadingEarlier = $state(false); + let failedImages = $state<string[]>([]); async function showEarlier() { if (!onShowEarlier || loadingEarlier) return; @@ -70,10 +92,35 @@ } } - const groups = $derived(groupRenderedChunks(chunks)); + const groups = $derived(groupRenderedChunks(chunks, recovering)); const rows = $derived(interleaveTurnMetrics(groups, turnMetrics)); + // A SEEDED turn gap (no turnId — e.g. after a reload) renders on the most + // recent turn's row (when no user message follows it). + const seededTurnRow = $derived( + cacheTimer?.running?.kind === "turn" && cacheTimer.running.turnId === null + ? lastTurnRowIndex(rows) + : null, + ); + + /** Start of the running turn gap if it belongs on this turn row, else undefined. */ + function liveTurnSince(turnId: string, rowIndex: number): number | undefined { + const r = cacheTimer?.running; + if (r?.kind !== "turn") return undefined; + if (r.turnId === turnId || (r.turnId === null && rowIndex === seededTurnRow)) { + return r.startedAt; + } + return undefined; + } + + // A running step gap with tool calls → the live timer under the executing tools. + const liveStepGap = $derived( + cacheTimer?.running?.kind === "step" && cacheTimer.running.toolCount > 0 + ? cacheTimer.running + : null, + ); + // Tool cards use toolCallId keys across pending/result/seal and orphan page-in. // Stable per-row keys. Thinking blocks get an ordinal key (`think<n>`) that // survives the provisional→committed (seq null → seq N) transition, so the @@ -105,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 @@ -117,13 +170,22 @@ {#if rendered.chunk.type === "text"} <p>{rendered.chunk.text}</p> {:else if rendered.chunk.type === "image"} - <img - src={resolveImageUrl(rendered.chunk.url, apiBaseUrl)} - alt={rendered.chunk.mimeType ?? "pasted image"} - loading="lazy" - decoding="async" - class="max-h-80 max-w-full rounded" - /> + {#if failedImages.includes(rendered.chunk.url)} + <span class="text-sm opacity-70">image expired</span> + {:else} + <img + src={resolveImageUrl(rendered.chunk.url, apiBaseUrl)} + alt={rendered.chunk.mimeType ?? "pasted image"} + loading="lazy" + decoding="async" + onerror={() => { + if (rendered.chunk.type === "image" && rendered.chunk.url.startsWith("/images/")) { + failedImages = [...failedImages, rendered.chunk.url]; + } + }} + class="max-h-80 max-w-full rounded" + /> + {/if} {/if} </div> </div> @@ -150,7 +212,11 @@ <div class="chat chat-start [&>.chat-bubble]:max-w-5xl"> <div class="chat-bubble w-full bg-transparent"> {#if rendered.chunk.type === "text"} - <Markdown text={rendered.chunk.text} streaming={rendered.streaming ?? false} /> + <Markdown + text={rendered.chunk.text} + streaming={rendered.streaming ?? false} + image={{ apiBase: apiBaseUrl, cwd }} + /> {:else if rendered.chunk.type === "error"} <div class="text-error" role="alert"> {rendered.chunk.message} @@ -180,13 +246,14 @@ </button> </div> {/if} - {#each keyedRows as { row, key } (key)} + {#each keyedRows as { row, key }, i (key)} <!-- `.chat-row`: content-visibility:auto — the browser skips layout/paint for rows outside the viewport (see app.css). --> <div class="chat-row"> {#if row.kind === "step-metrics"} {@const sv = viewStepMetrics(row.step, row.index)} {@const stepCache = viewCacheRate(row.step.usage)} + {@const stepStamp = cacheTimer?.stepStamps.get(row.step.stepId)} <div class="chat chat-start"> <div class="chat-bubble w-full max-w-5xl bg-transparent p-0"> <div class="flex flex-wrap items-center gap-x-1 text-xs"> @@ -203,6 +270,9 @@ class="badge badge-xs {badgeClass[stepCache.level]}" data-testid="step-cache-badge">{stepCache.pct}%</span > + {#if stepStamp !== undefined} + <CacheTimerBadge elapsedMs={stepStamp} size="xs" /> + {/if} </div> </div> </div> @@ -211,6 +281,8 @@ {@const lastCache = viewCacheRate(row.turn.usage)} {@const chatCache = viewCacheRate(row.cumulativeUsage)} {@const retention = viewExpectedCache(row.turn.steps[0]?.usage, row.prevLastStepUsage)} + {@const turnLive = liveTurnSince(row.turn.turnId, i)} + {@const turnStamp = cacheTimer?.turnStamps.get(row.turn.turnId)} <div class="chat chat-start"> <div class="chat-bubble w-full max-w-5xl bg-transparent p-0"> <div class="flex flex-col gap-1 text-xs"> @@ -236,6 +308,11 @@ <span class="badge badge-sm {badgeClass[retention.level]}">{retention.pct}%</span> </span> {/if} + {#if turnLive !== undefined} + <CacheTimerBadge since={turnLive} /> + {:else if turnStamp !== undefined} + <CacheTimerBadge elapsedMs={turnStamp} /> + {/if} </div> </div> </div> @@ -252,6 +329,18 @@ {/if} </div> {/each} + {#if liveStepGap} + <!-- Live cache timer while the last step's tools execute; stamped onto that + step's metrics line once they return (or the next step starts). --> + <div class="chat chat-start"> + <div class="chat-bubble w-full max-w-5xl bg-transparent p-0"> + <div class="flex items-center gap-1 text-xs" data-testid="cache-timer-tools"> + <span class="opacity-70">Running tools</span> + <CacheTimerBadge since={liveStepGap.startedAt} size="xs" /> + </div> + </div> + </div> + {/if} {#if providerRetry} {@const rv = viewProviderRetry(providerRetry)} <!-- Transient yellow warning: a provider error is being retried with backoff. diff --git a/src/features/chat/ui/ChatView.test.ts b/src/features/chat/ui/ChatView.test.ts index 77f832d..a6bd8ac 100644 --- a/src/features/chat/ui/ChatView.test.ts +++ b/src/features/chat/ui/ChatView.test.ts @@ -1,5 +1,5 @@ import type { StepId } from "@dispatch/wire"; -import { render, screen } from "@testing-library/svelte"; +import { fireEvent, render, screen } from "@testing-library/svelte"; import userEvent from "@testing-library/user-event"; import { describe, expect, it } from "vitest"; import type { RenderedChunk } from "../../../core/chunks"; @@ -497,3 +497,62 @@ it("renders a per-tool Phosphor icon in the header", () => { // Different tools → different glyphs. expect(icons[0]?.innerHTML).not.toBe(icons[1]?.innerHTML); }); + +it("replaces failed persisted images with an expired placeholder, without marking inline images expired", async () => { + const chunks: RenderedChunk[] = ["/images/conv/gone.png", "data:image/png;base64,AAAA"].map( + (url, i) => ({ seq: i + 1, role: "user", chunk: { type: "image", url }, provisional: false }), + ); + render(ChatView, { chunks, apiBaseUrl: "https://backend.test" }); + const [stored, inline] = screen.getAllByRole("img"); + if (!stored || !inline) throw new Error("expected both images"); + expect(stored).toHaveAttribute("src", "https://backend.test/images/conv/gone.png"); + await fireEvent.error(inline); + expect(screen.queryByText("image expired")).toBeNull(); + await fireEvent.error(stored); + 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(); +}); diff --git a/src/features/chat/ui/Composer.svelte b/src/features/chat/ui/Composer.svelte index c7d9d02..a43ed53 100644 --- a/src/features/chat/ui/Composer.svelte +++ b/src/features/chat/ui/Composer.svelte @@ -1,6 +1,7 @@ <script lang="ts"> import type { ImageInput } from "@dispatch/wire"; import { convertToJpeg, type ConvertedJpeg, type ImageJpegEncoder } from "../adapter/image-encode"; + import { stopButtonLabels } from "../logic/composer"; import { MAX_IMAGE_EDGE_PX, shouldConvertImage } from "../logic/image-compress"; import { computeContextUsage, formatCompactTokens } from "../../../core/metrics"; @@ -59,6 +60,7 @@ contextSize = undefined, contextWindow = undefined, status = "idle", + queuedMessageCount = undefined, }: { /** * Send a message (start a turn via `chat.send`). Carries any staged images @@ -70,10 +72,10 @@ /** * Enqueue a steering message (`chat.queue`). When provided AND the status * is `running`, the send button becomes a "Queue" button that steers the - * in-flight turn instead of starting a new one. Steering is text-only — - * it never carries images (a mid-turn injection has no image surface). + * in-flight turn instead of starting a new one. Carries staged images just + * like onSend; without this callback, in-flight input stays staged. */ - onQueue?: (text: string) => void; + onQueue?: (text: string, images?: ImageInput[]) => void; /** Stop the in-flight generation (`POST /conversations/:id/stop`). */ onStop?: () => void; /** @@ -95,6 +97,8 @@ * `running` for the send button (steer/stop). */ status?: ComposerStatus; + /** Current conversation's queued snapshot size; undefined when unavailable. */ + queuedMessageCount?: number | undefined; } = $props(); export type ComposerStatus = "idle" | "running" | "queued" | "error"; @@ -107,25 +111,23 @@ const hasText = $derived(text.trim().length > 0); const hasImages = $derived(images.length > 0); - const canSend = $derived(hasText || hasImages); + const inFlight = $derived(status === "running" || status === "queued"); + const canSend = $derived((hasText || hasImages) && (!inFlight || onQueue !== undefined)); const effectiveMax = $derived(contextWindow ?? FALLBACK_CONTEXT_WINDOW); const usage = $derived(computeContextUsage(contextSize, effectiveMax)); // One button, three modes: // - idle → "Send" (starts a turn via chat.send) - // - running/queued + text → "Queue" (steers via chat.queue — text only) + // - running/queued + text/images → "Queue" (steers via chat.queue) // - running/queued + empty → "Stop" (aborts via POST /stop) // (`queued` behaves like `running` — the turn is in flight, just waiting for a // concurrency slot; the user can still steer or stop it.) - // Steering never carries images: when running with images staged but no text, - // the images stay staged (queue is text-only). Images-without-text while running - // is an unusual case that still sends (the server auto-starts/resolves). - const inFlight = $derived(status === "running" || status === "queued"); const buttonMode = $derived.by<"send" | "queue" | "stop">(() => { if (inFlight && !hasText && !hasImages && onStop !== undefined) return "stop"; - if (inFlight && hasText && onQueue !== undefined) return "queue"; + if (inFlight) return "queue"; return "send"; }); + const stopButton = $derived(stopButtonLabels(queuedMessageCount)); const placeholder = $derived( status === "queued" ? "Queued for a slot…" @@ -283,15 +285,15 @@ const trimmed = text.trim(); // Allow a send with images even when text is empty (an image-only turn). if (trimmed.length === 0 && !hasImages) return; - if (buttonMode === "queue") { - // Steering is text-only — never forward images. - onQueue?.(trimmed); + const toSend = hasImages ? images.map((img) => imageInput(img)) : undefined; + if (inFlight) { + if (onQueue === undefined) return; // Keep staged input if no queue port is wired. + onQueue(trimmed, toSend); } else { - const toSend: ImageInput[] | undefined = hasImages - ? images.map((img) => imageInput(img)) - : undefined; onSend(trimmed, toSend); } + // Clear only after handing the complete input to a callback. WS is still + // fire-and-forget: this does not imply backend storage has completed. text = ""; images = []; } @@ -322,7 +324,7 @@ <!-- Top bar: expanding textarea + image-attach button + single context-aware button --> <div class="flex items-end gap-2 px-4 pt-3 pb-2"> <div - class="flex-1" + class="min-w-0 flex-1" onpaste={handlePaste} class:border-2={dragOver} class:border-primary={dragOver} @@ -374,12 +376,17 @@ {#if buttonMode === "stop"} <button - class="btn btn-error w-20 shrink-0" + class="btn btn-error shrink-0" + class:w-20={stopButton.text === "Stop"} + class:w-36={stopButton.text !== "Stop"} + class:whitespace-normal={stopButton.text !== "Stop"} + class:leading-tight={stopButton.text !== "Stop"} type="button" - aria-label="Stop generation" + aria-label={stopButton.ariaLabel} + title={stopButton.title} onclick={() => onStop?.()} > - Stop + {stopButton.text} </button> {:else} <button class="btn btn-primary w-20 shrink-0" type="submit" disabled={!canSend}> diff --git a/src/features/chat/ui/PendingInputWarning.svelte b/src/features/chat/ui/PendingInputWarning.svelte new file mode 100644 index 0000000..c9e675e --- /dev/null +++ b/src/features/chat/ui/PendingInputWarning.svelte @@ -0,0 +1,47 @@ +<script lang="ts"> + import { onMount } from "svelte"; + import type { PendingInput } from "../pending-inputs"; + + let { + input, + onDiscard, + }: { + input: PendingInput; + onDiscard: (requestId: string) => void; + } = $props(); + + let visible = $state(false); + + // Presentation only: each keyed row gets one grace period, including restored + // or rejected inputs. Prop updates never restart it. Confirmation/discard + // removes the row, cancelling the timer along with any other unmount. + onMount(() => { + const timer = setTimeout(() => { + visible = true; + }, 1000); + return () => clearTimeout(timer); + }); +</script> + +{#if visible} + <div class="m-3 rounded border border-warning p-3 text-sm" role="status"> + <p> + {input.rejection + ? `Message rejected: ${input.rejection}` + : "Delivery unconfirmed. Your input is saved on this device."} + </p> + <p>Check the conversation history before sending again to avoid a duplicate.</p> + <textarea + class="textarea textarea-bordered mt-2 w-full" + aria-label="Saved message" + readonly + value={input.request.type === "chat.send" ? input.request.message : input.request.text} + ></textarea> + {#each input.request.images ?? [] as image} + <img class="my-2 max-h-32" src={image.url} alt="Saved attachment" /> + {/each} + <button class="btn btn-sm mt-2" onclick={() => onDiscard(input.request.requestId)} + >Discard saved copy</button + > + </div> +{/if} diff --git a/src/features/chat/ui/PendingInputs.svelte b/src/features/chat/ui/PendingInputs.svelte new file mode 100644 index 0000000..dcb7aee --- /dev/null +++ b/src/features/chat/ui/PendingInputs.svelte @@ -0,0 +1,16 @@ +<script lang="ts"> + import type { PendingInput } from "../pending-inputs"; + import PendingInputWarning from "./PendingInputWarning.svelte"; + + let { + inputs, + onDiscard, + }: { + inputs: readonly PendingInput[]; + onDiscard: (requestId: string) => void; + } = $props(); +</script> + +{#each inputs as input (input.request.requestId)} + <PendingInputWarning {input} {onDiscard} /> +{/each} diff --git a/src/features/chat/ui/PendingInputs.test.ts b/src/features/chat/ui/PendingInputs.test.ts new file mode 100644 index 0000000..cea770d --- /dev/null +++ b/src/features/chat/ui/PendingInputs.test.ts @@ -0,0 +1,217 @@ +import { cleanup, fireEvent, render, screen } from "@testing-library/svelte"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { PendingInput } from "../pending-inputs"; +import { createChatStore } from "../store.svelte"; +import { + createFakeCache, + createFakeHistorySync, + createFakeMetricsSync, + createFakeTransport, +} from "../test-helpers"; +import PendingInputs from "./PendingInputs.svelte"; + +function input(requestId: string, type: "chat.send" | "chat.queue" = "chat.send"): PendingInput { + return { + request: { + ...(type === "chat.send" ? { type, message: requestId } : { type, text: requestId }), + conversationId: "c", + requestId, + images: [{ url: "data:image/png;base64,AAAA", mimeType: "image/png" }], + }, + }; +} + +beforeEach(() => vi.useFakeTimers()); +afterEach(() => { + cleanup(); + const remaining = vi.getTimerCount(); + vi.useRealTimers(); + expect(remaining).toBe(0); +}); + +describe("pending input warning grace period", () => { + it.each([ + "chat.send", + "chat.queue", + ] as const)("renders no %s recovery UI before 1000 ms, then keeps its content stable", async (type) => { + const { container } = render(PendingInputs, { + inputs: [input("saved message", type)], + onDiscard: vi.fn(), + }); + expect(container.children).toHaveLength(0); + expect(container.textContent).toBe(""); + await vi.advanceTimersByTimeAsync(999); + expect(container.children).toHaveLength(0); + expect(container.textContent).toBe(""); + expect(screen.queryByRole("status")).toBeNull(); + await vi.advanceTimersByTimeAsync(1); + const warning = screen.getByRole("status"); + expect(warning).toHaveTextContent("Delivery unconfirmed. Your input is saved on this device."); + expect(warning).toHaveTextContent( + "Check the conversation history before sending again to avoid a duplicate.", + ); + expect(screen.getByRole("textbox", { name: "Saved message" })).toHaveValue("saved message"); + expect(screen.getByRole("img", { name: "Saved attachment" })).toHaveAttribute( + "src", + "data:image/png;base64,AAAA", + ); + const content = warning.innerHTML; + await vi.advanceTimersByTimeAsync(10_000); + expect(screen.getByRole("status")).toBe(warning); + expect(warning.innerHTML).toBe(content); + }); + + it.each([ + "confirmation", + "discard", + ])("%s before the threshold cancels the timer and never flashes later", async (resolution) => { + const transport = createFakeTransport(); + const save = vi.fn(); + const store = createChatStore({ + conversationId: "c", + createRequestId: () => "first", + transport: transport.impl, + historySync: createFakeHistorySync().impl, + metricsSync: createFakeMetricsSync().impl, + cache: createFakeCache().impl, + pendingInputStorage: { load: () => [], save }, + }); + try { + store.send("first"); + // Saving and transmission stay immediate, independent of warning display. + expect(save).toHaveBeenCalledExactlyOnceWith(store.pendingInputs); + expect(transport.sent).toHaveLength(1); + const { container, rerender } = render(PendingInputs, { + inputs: store.pendingInputs, + onDiscard: store.discardPendingInput, + }); + await vi.advanceTimersByTimeAsync(999); + if (resolution === "confirmation") { + store.handleDelta({ type: "chat.accepted", conversationId: "c", requestId: "first" }); + } else { + store.discardPendingInput("first"); + } + expect(store.pendingInputs).toEqual([]); + expect(save).toHaveBeenLastCalledWith([]); + await rerender({ inputs: store.pendingInputs }); + expect(vi.getTimerCount()).toBe(0); + expect(container.children).toHaveLength(0); + expect(container.textContent).toBe(""); + await vi.advanceTimersByTimeAsync(10_000); + expect(container.children).toHaveLength(0); + expect(container.textContent).toBe(""); + } finally { + store.dispose(); + } + }); + + it("hides an already visible warning immediately on confirmation", async () => { + const { container, rerender } = render(PendingInputs, { + inputs: [input("first")], + onDiscard: vi.fn(), + }); + await vi.advanceTimersByTimeAsync(1000); + expect(screen.getByRole("status")).toBeInTheDocument(); + await rerender({ inputs: [] }); + expect(container.children).toHaveLength(0); + expect(container.textContent).toBe(""); + }); + + it("preserves the discard action and hides the saved copy without another delay", async () => { + const onDiscard = vi.fn(); + const { container, rerender } = render(PendingInputs, { + inputs: [input("first")], + onDiscard, + }); + await vi.advanceTimersByTimeAsync(1000); + await fireEvent.click(screen.getByRole("button", { name: "Discard saved copy" })); + expect(onDiscard).toHaveBeenCalledExactlyOnceWith("first"); + await rerender({ inputs: [] }); + expect(container.children).toHaveLength(0); + expect(container.textContent).toBe(""); + }); + + it("cancels all waiting timers on unmount", async () => { + const { unmount } = render(PendingInputs, { + inputs: [input("first"), input("second")], + onDiscard: vi.fn(), + }); + expect(vi.getTimerCount()).toBe(2); + await vi.advanceTimersByTimeAsync(500); + unmount(); + expect(vi.getTimerCount()).toBe(0); + await vi.advanceTimersByTimeAsync(10_000); + expect(screen.queryByRole("status")).toBeNull(); + }); + + it("gives successive inputs independent delays without restarting existing rows", async () => { + const first = input("first"); + const second = input("second"); + const { rerender } = render(PendingInputs, { inputs: [first], onDiscard: vi.fn() }); + await vi.advanceTimersByTimeAsync(600); + await rerender({ inputs: [second, { ...first }], onDiscard: vi.fn() }); + await vi.advanceTimersByTimeAsync(399); + expect(screen.queryByRole("status")).toBeNull(); + await vi.advanceTimersByTimeAsync(1); + const warning = screen.getByRole("status"); + expect(screen.getByRole("textbox")).toHaveValue("first"); + await rerender({ inputs: [{ ...second }, { ...first }], onDiscard: vi.fn() }); + await vi.advanceTimersByTimeAsync(599); + expect(screen.getAllByRole("status")).toEqual([warning]); + await vi.advanceTimersByTimeAsync(1); + expect( + screen.getAllByRole("textbox").map((node) => (node as HTMLTextAreaElement).value), + ).toEqual(["second", "first"]); + // Removing the older warning must not re-hide/restart the newer warning. + await rerender({ inputs: [second] }); + expect(screen.getByRole("textbox")).toHaveValue("second"); + expect(vi.getTimerCount()).toBe(0); + }); + + it("starts a fresh delay after an earlier input was confirmed during its grace period", async () => { + const { rerender } = render(PendingInputs, { inputs: [input("first")], onDiscard: vi.fn() }); + await vi.advanceTimersByTimeAsync(600); + await rerender({ inputs: [] }); + await rerender({ inputs: [input("second")] }); + await vi.advanceTimersByTimeAsync(999); + expect(screen.queryByRole("status")).toBeNull(); + await vi.advanceTimersByTimeAsync(1); + expect(screen.getByRole("textbox")).toHaveValue("second"); + }); + + it("keeps the original deadline when rejection arrives, and updates visible content immediately", async () => { + const pending = input("first"); + const { rerender } = render(PendingInputs, { inputs: [pending], onDiscard: vi.fn() }); + await vi.advanceTimersByTimeAsync(600); + await rerender({ inputs: [{ ...pending, rejection: "disk full" }] }); + await vi.advanceTimersByTimeAsync(399); + expect(screen.queryByRole("status")).toBeNull(); + await vi.advanceTimersByTimeAsync(1); + const warning = screen.getByRole("status"); + expect(warning).toHaveTextContent("Message rejected: disk full"); + expect(warning).not.toHaveTextContent("Delivery unconfirmed"); + await rerender({ inputs: [{ ...pending, rejection: "still rejected" }] }); + expect(screen.getByRole("status")).toBe(warning); + expect(warning).toHaveTextContent("Message rejected: still rejected"); + expect(vi.getTimerCount()).toBe(0); + }); + + it.each([ + undefined, + "disk full", + ])("restored entries (rejection: %s) use the same grace period and retain recovery content", async (rejection) => { + const restored = input("restored saved message"); + render(PendingInputs, { + inputs: [rejection === undefined ? restored : { ...restored, rejection }], + onDiscard: vi.fn(), + }); + await vi.advanceTimersByTimeAsync(999); + expect(screen.queryByRole("status")).toBeNull(); + await vi.advanceTimersByTimeAsync(1); + expect(screen.getByRole("status")).toHaveTextContent( + rejection === undefined ? "Delivery unconfirmed" : "Message rejected: disk full", + ); + expect(screen.getByRole("textbox")).toHaveValue("restored saved message"); + expect(screen.getByRole("button", { name: "Discard saved copy" })).toBeEnabled(); + }); +}); |