From c4397f633415a46852f0e3b49c78ab7b2d87d4c1 Mon Sep 17 00:00:00 2001 From: Adam Malczewski Date: Mon, 19 Oct 2026 13:38:57 +0900 Subject: feat(workspaces): block creating a workspace whose slug is archived --- src/features/workspaces/adapter/http.test.ts | 30 +++++++++-- src/features/workspaces/adapter/http.ts | 24 ++++++--- src/features/workspaces/index.ts | 6 +++ src/features/workspaces/logic/create.test.ts | 50 ++++++++++++++++++ src/features/workspaces/logic/create.ts | 64 +++++++++++++++++++++++ src/features/workspaces/store.svelte.ts | 15 ++++++ src/features/workspaces/store.test.ts | 61 ++++++++++++++++++++- src/features/workspaces/ui/WorkspacesHome.svelte | 7 +-- src/features/workspaces/ui/WorkspacesHome.test.ts | 38 +++++++++++++- 9 files changed, 277 insertions(+), 18 deletions(-) create mode 100644 src/features/workspaces/logic/create.test.ts create mode 100644 src/features/workspaces/logic/create.ts (limited to 'src') diff --git a/src/features/workspaces/adapter/http.test.ts b/src/features/workspaces/adapter/http.test.ts index 2e34822..d6409ca 100644 --- a/src/features/workspaces/adapter/http.test.ts +++ b/src/features/workspaces/adapter/http.test.ts @@ -79,15 +79,37 @@ describe("createWorkspaceHttp", () => { expect(result).toEqual({ ok: false, error: "invalid slug" }); }); - it("get returns null on 404", async () => { + it("get reports a missing lookup on 404", async () => { const http = createWorkspaceHttp(BASE, fakeFetch([{ status: 404 }])); - expect(await http.get("nope")).toBeNull(); + expect(await http.get("nope")).toEqual({ kind: "missing" }); }); it("get returns the workspace on 200", async () => { - const ws = { id: "x", title: "X", defaultCwd: "/home", createdAt: 1, lastActivityAt: 2 }; + const ws = { + id: "x", + title: "X", + defaultCwd: "/home", + defaultComputerId: null, + starred: false, + archived: false, + createdAt: 1, + lastActivityAt: 2, + }; const http = createWorkspaceHttp(BASE, fakeFetch([{ body: ws }])); - expect(await http.get("x")).toEqual(ws); + expect(await http.get("x")).toEqual({ kind: "found", workspace: ws }); + }); + + it("get distinguishes a failed response from 404 (not 'missing')", async () => { + const http = createWorkspaceHttp( + BASE, + fakeFetch([{ status: 500, body: { error: "db down" } }]), + ); + expect(await http.get("x")).toEqual({ kind: "failed", error: "db down" }); + }); + + it("get reports a network error as failed (not 'missing')", async () => { + const http = createWorkspaceHttp(BASE, fakeFetch([new Error("network")])); + expect(await http.get("x")).toEqual({ kind: "failed", error: "network" }); }); it("setTitle PUTs the title", async () => { diff --git a/src/features/workspaces/adapter/http.ts b/src/features/workspaces/adapter/http.ts index de95a61..3f7f058 100644 --- a/src/features/workspaces/adapter/http.ts +++ b/src/features/workspaces/adapter/http.ts @@ -9,6 +9,7 @@ import type { WorkspaceListResponse, WorkspaceResponse, } from "@dispatch/transport-contract"; +import type { WorkspaceLookup } from "../logic/create"; /** * Workspace HTTP effects — the injected edge that talks to the backend's @@ -21,7 +22,7 @@ import type { * Endpoints (transport-contract@0.16.0): * - `GET /workspaces` (optional `?archived=true|false|all`, default `false` = active only) → list * - `PUT /workspaces/:id` (create-on-miss, idempotent) → ensure - * - `GET /workspaces/:id` (404 → null) → get + * - `GET /workspaces/:id` (found | 404-missing | failed) → get * - `PUT /workspaces/:id/title` → rename * - `PUT /workspaces/:id/default-cwd` → set/clear default cwd * - `PUT /workspaces/:id/default-computer` → set/clear default computer (SSH handoff #2) @@ -41,7 +42,12 @@ export type WorkspaceArchivedFilter = "true" | "false" | "all"; export interface WorkspaceHttp { list(archived?: WorkspaceArchivedFilter): Promise; ensure(id: string, body?: EnsureWorkspaceRequest): Promise>; - get(id: string): Promise; + /** + * Authoritative existence probe. 404 → `missing`, a network/server failure → + * `failed` (distinct on purpose: the create gate must not treat a broken + * check as "doesn't exist"), 200 → the workspace (`found`). + */ + get(id: string): Promise; setTitle(id: string, title: string): Promise>; setDefaultCwd(id: string, defaultCwd: string | null): Promise>; setDefaultComputer(id: string, computerId: string | null): Promise>; @@ -96,13 +102,17 @@ export function createWorkspaceHttp(httpBase: string, fetchImpl: typeof fetch): } }, - async get(id): Promise { + async get(id): Promise { try { const res = await fetchImpl(`${httpBase}/workspaces/${encodeURIComponent(id)}`); - if (res.status === 404 || !res.ok) return null; - return (await res.json()) as WorkspaceResponse; - } catch { - return null; + if (res.status === 404) return { kind: "missing" }; + if (!res.ok) return { kind: "failed", error: await errText(res) }; + return { kind: "found", workspace: (await res.json()) as WorkspaceResponse }; + } catch (err) { + return { + kind: "failed", + error: err instanceof Error ? err.message : "Workspace lookup failed", + }; } }, diff --git a/src/features/workspaces/index.ts b/src/features/workspaces/index.ts index 1a60bb7..c7f57d5 100644 --- a/src/features/workspaces/index.ts +++ b/src/features/workspaces/index.ts @@ -1,6 +1,12 @@ export type { WorkspaceHttp, WorkspaceResult } from "./adapter/http"; export { createWorkspaceHttp } from "./adapter/http"; export { createTileLayoutHttp, type TileLayoutHttp } from "./adapter/tile-layout-http"; +export type { CreateGate, WorkspaceLookup } from "./logic/create"; +export { + archivedCreateMessage, + checkFailedCreateMessage, + gateCreate, +} from "./logic/create"; export type { DragEmit, DragEvent as DragControllerEvent, diff --git a/src/features/workspaces/logic/create.test.ts b/src/features/workspaces/logic/create.test.ts new file mode 100644 index 0000000..278bcd6 --- /dev/null +++ b/src/features/workspaces/logic/create.test.ts @@ -0,0 +1,50 @@ +import type { Workspace } from "@dispatch/wire"; +import { describe, expect, it } from "vitest"; +import { + archivedCreateMessage, + checkFailedCreateMessage, + gateCreate, + type WorkspaceLookup, +} from "./create"; + +function workspace(overrides: Partial = {}): Workspace { + return { + id: "old", + title: "Old", + defaultCwd: null, + defaultComputerId: null, + starred: false, + archived: false, + createdAt: 1, + lastActivityAt: 2, + ...overrides, + }; +} + +describe("gateCreate", () => { + it("blocks a slug whose workspace is archived, with the can't-create message", () => { + const lookup: WorkspaceLookup = { kind: "found", workspace: workspace({ archived: true }) }; + const gate = gateCreate("old", lookup); + expect(gate).toEqual({ ok: false, error: archivedCreateMessage("old") }); + expect(gate.ok === false && gate.error).toContain( + 'A workspace named "old" already exists and is archived — you can\'t create it.', + ); + }); + + it("proceeds for an existing ACTIVE workspace (ensure is idempotent)", () => { + const lookup: WorkspaceLookup = { kind: "found", workspace: workspace({ archived: false }) }; + expect(gateCreate("old", lookup)).toEqual({ ok: true }); + }); + + it("proceeds for a missing slug", () => { + expect(gateCreate("fresh", { kind: "missing" })).toEqual({ ok: true }); + }); + + it("blocks when the existence probe failed (never guesses)", () => { + const gate = gateCreate("maybe", { kind: "failed", error: "db down" }); + expect(gate).toEqual({ ok: false, error: checkFailedCreateMessage("maybe", "db down") }); + expect(gate.ok === false && gate.error).toContain( + 'Couldn\'t check whether a workspace named "maybe"', + ); + }); +}); diff --git a/src/features/workspaces/logic/create.ts b/src/features/workspaces/logic/create.ts new file mode 100644 index 0000000..c9385e1 --- /dev/null +++ b/src/features/workspaces/logic/create.ts @@ -0,0 +1,64 @@ +/** + * Pure create-gate logic for the New-workspace form — zero DOM, zero effects. + * + * `PUT /workspaces/:id` (ensure) is idempotent, so creating a slug that belongs + * to an ARCHIVED workspace would silently "succeed": the backend returns the + * archived workspace, the active list (which excludes archived) never shows it, + * and the home grid would still place a tile for something invisible. The + * frontend therefore probes `GET /workspaces/:id` first and refuses the create + * when the slug exists and is archived. + */ + +import type { Workspace } from "@dispatch/wire"; + +/** + * The authoritative existence probe's outcome (`GET /workspaces/:id`). The + * adapter maps HTTP status to these kinds; `failed` is deliberately distinct + * from `missing` (404) so a broken check is never mistaken for "no such + * workspace". + */ +export type WorkspaceLookup = + | { readonly kind: "found"; readonly workspace: Workspace } + | { readonly kind: "missing" } + | { readonly kind: "failed"; readonly error: string }; + +/** Whether a create may proceed: `ok` or a user-facing error message. */ +export type CreateGate = { readonly ok: true } | { readonly ok: false; readonly error: string }; + +/** + * The user-facing message when a slug belongs to an archived workspace. It + * states plainly that the create can't happen and points at the existing + * recovery affordance (the Archived workspaces list) without adding UI. + */ +export function archivedCreateMessage(id: string): string { + return `A workspace named "${id}" already exists and is archived — you can't create it. Unarchive it from Archived workspaces instead.`; +} + +/** + * The user-facing message when the existence probe itself failed. Proceeding + * blindly could recreate the silent archived-tile bug, so the create is + * refused with a retryable message instead. + */ +export function checkFailedCreateMessage(id: string, error: string): string { + return `Couldn't check whether a workspace named "${id}" already exists (${error}). Try again.`; +} + +/** + * Decide whether a create may proceed, given the probe outcome. Pure: + * `(id, lookup) → gate`. + * + * - archived slug → blocked with the "can't create" message; + * - missing slug or an ACTIVE existing slug → proceed (`ensure` is idempotent, + * so an active slug keeps today's behavior); + * - a failed probe → blocked too (plan §3.4): never guess, because proceeding + * could ensure an archived workspace's tile. + */ +export function gateCreate(id: string, lookup: WorkspaceLookup): CreateGate { + if (lookup.kind === "failed") { + return { ok: false, error: checkFailedCreateMessage(id, lookup.error) }; + } + if (lookup.kind === "found" && lookup.workspace.archived) { + return { ok: false, error: archivedCreateMessage(id) }; + } + return { ok: true }; +} diff --git a/src/features/workspaces/store.svelte.ts b/src/features/workspaces/store.svelte.ts index 57e34b2..b9b646b 100644 --- a/src/features/workspaces/store.svelte.ts +++ b/src/features/workspaces/store.svelte.ts @@ -1,6 +1,7 @@ import type { EnsureWorkspaceRequest } from "@dispatch/transport-contract"; import type { Workspace, WorkspaceEntry } from "@dispatch/wire"; import type { WorkspaceHttp, WorkspaceResult } from "./adapter/http"; +import { gateCreate } from "./logic/create"; import { applyArchived, applyStarred, sortWorkspaces } from "./logic/view-model"; /** @@ -27,6 +28,14 @@ export interface WorkspaceStore { refreshArchived(): Promise; /** `PUT /workspaces/:id` (create-on-miss). Returns the workspace or an error. */ ensure(id: string, body?: EnsureWorkspaceRequest): Promise>; + /** + * Create a workspace via the New-workspace flow. Probes `GET /workspaces/:id` + * first and refuses a slug that belongs to an ARCHIVED workspace (the + * idempotent `ensure` would otherwise silently resurrect an invisible tile). + * A failed probe is refused too (see `gateCreate`). On success it falls + * through to `ensure`, unchanged. + */ + create(id: string, body?: EnsureWorkspaceRequest): Promise>; /** Rename a workspace (display only; id unchanged). */ rename(id: string, title: string): Promise>; /** Set/clear a workspace's default cwd. */ @@ -102,6 +111,12 @@ export function createWorkspaceStore(http: WorkspaceHttp): WorkspaceStore { } }, + async create(id, body): Promise> { + const gate = gateCreate(id, await http.get(id)); + if (!gate.ok) return { ok: false, error: gate.error }; + return this.ensure(id, body); + }, + async ensure(id, body): Promise> { const result = await http.ensure(id, body); if (result.ok) void this.refresh(); diff --git a/src/features/workspaces/store.test.ts b/src/features/workspaces/store.test.ts index a44d8a6..faec5bf 100644 --- a/src/features/workspaces/store.test.ts +++ b/src/features/workspaces/store.test.ts @@ -1,6 +1,7 @@ import type { Workspace, WorkspaceEntry } from "@dispatch/wire"; import { describe, expect, it, vi } from "vitest"; import type { WorkspaceResult } from "./adapter/http"; +import { archivedCreateMessage, type WorkspaceLookup } from "./logic/create"; import { createWorkspaceStore } from "./store.svelte"; function entry(overrides: Partial = {}): WorkspaceEntry { @@ -25,14 +26,23 @@ function fakeHttp(opts: { unstar?: (id: string) => Promise>; archive?: (id: string) => Promise>; unarchive?: (id: string) => Promise>; + get?: (id: string) => Promise; + ensure?: (id: string) => Promise>; }) { return { list: vi.fn( async (_archived?: "true" | "false" | "all"): Promise => opts.list ?? [], ), - ensure: vi.fn(), - get: vi.fn(), + ensure: + opts.ensure ?? + vi.fn( + async (id: string): Promise> => ({ + ok: true, + value: entry({ id }), + }), + ), + get: vi.fn(opts.get ?? (async (): Promise => ({ kind: "missing" }))), setTitle: vi.fn(), setDefaultCwd: vi.fn(), setDefaultComputer: vi.fn(), @@ -224,6 +234,53 @@ describe("createWorkspaceStore — setArchived", () => { }); }); +describe("createWorkspaceStore — create (archived-slug gate)", () => { + it("refuses an archived slug with the can't-create message and never calls ensure", async () => { + const http = fakeHttp({ + get: async () => ({ kind: "found", workspace: entry({ id: "old", archived: true }) }), + }); + const store = createWorkspaceStore(http); + + const result = await store.create("old"); + + expect(result).toEqual({ ok: false, error: archivedCreateMessage("old") }); + expect(http.ensure).not.toHaveBeenCalled(); + }); + + it("proceeds for an existing non-archived slug (ensure is called as before)", async () => { + const http = fakeHttp({ + get: async () => ({ kind: "found", workspace: entry({ id: "taken", archived: false }) }), + }); + const store = createWorkspaceStore(http); + + const result = await store.create("taken"); + + expect(result.ok).toBe(true); + expect(http.ensure).toHaveBeenCalledTimes(1); + }); + + it("proceeds for a brand-new slug (404 → missing)", async () => { + const http = fakeHttp({ get: async () => ({ kind: "missing" }) }); + const store = createWorkspaceStore(http); + + const result = await store.create("fresh"); + + expect(result.ok).toBe(true); + expect(http.ensure).toHaveBeenCalledTimes(1); + }); + + it("refuses when the existence check itself failed, without calling ensure (§3.4)", async () => { + const http = fakeHttp({ get: async () => ({ kind: "failed", error: "db down" }) }); + const store = createWorkspaceStore(http); + + const result = await store.create("maybe"); + + expect(result.ok).toBe(false); + if (!result.ok) expect(result.error).toContain("Couldn't check whether"); + expect(http.ensure).not.toHaveBeenCalled(); + }); +}); + describe("createWorkspaceStore — refreshArchived", () => { it("loads the archived list via list('true') without touching the main list", async () => { const http = fakeHttp({ list: [entry({ id: "active", archived: false })] }); diff --git a/src/features/workspaces/ui/WorkspacesHome.svelte b/src/features/workspaces/ui/WorkspacesHome.svelte index 630c67f..65e1593 100644 --- a/src/features/workspaces/ui/WorkspacesHome.svelte +++ b/src/features/workspaces/ui/WorkspacesHome.svelte @@ -215,12 +215,13 @@ if (result.ok) deselect(); } - // ── Create (header popover): create on the backend — do NOT navigate — - // place the new tile in the first empty slot, and go straight into rename. + // ── Create (header popover): check the slug resolves to no archived + // workspace, create on the backend — do NOT navigate — place the new tile in + // the first empty slot, and go straight into rename. async function createWorkspace( slug: string, ): Promise<{ ok: true } | { ok: false; error: string }> { - const result = await store.ensure(slug); + const result = await store.create(slug); if (!result.ok) return { ok: false, error: result.error }; await store.refresh(); // Idempotent: if the list refresh's sync already appended the id, this diff --git a/src/features/workspaces/ui/WorkspacesHome.test.ts b/src/features/workspaces/ui/WorkspacesHome.test.ts index 6316930..d4316b1 100644 --- a/src/features/workspaces/ui/WorkspacesHome.test.ts +++ b/src/features/workspaces/ui/WorkspacesHome.test.ts @@ -5,6 +5,7 @@ import userEvent from "@testing-library/user-event"; import { describe, expect, it, vi } from "vitest"; import type { WorkspaceHttp, WorkspaceResult } from "../adapter/http"; import type { TileLayoutHttp } from "../adapter/tile-layout-http"; +import type { WorkspaceLookup } from "../logic/create"; import { createTileLayoutStore } from "../tile-layout-store.svelte"; import WorkspacesHome from "./WorkspacesHome.svelte"; @@ -24,7 +25,10 @@ function ws(overrides: Partial = {}): WorkspaceEntry { } /** A fake WorkspaceHttp edge with a controllable list (no mocks of our modules). */ -function fakeHttp(list: WorkspaceEntry[] = [ws()]): WorkspaceHttp { +function fakeHttp( + list: WorkspaceEntry[] = [ws()], + lookup: (id: string) => Promise = async () => ({ kind: "missing" }), +): WorkspaceHttp { return { list: vi.fn(async (): Promise => list), // Create-on-miss: the backend persists the new workspace, so the next @@ -33,7 +37,7 @@ function fakeHttp(list: WorkspaceEntry[] = [ws()]): WorkspaceHttp { if (!list.some((w) => w.id === id)) list.push(ws({ id, title: id })); return { ok: true, value: ws({ id, title: id }) }; }), - get: vi.fn(), + get: vi.fn(lookup), setTitle: vi.fn(), setDefaultCwd: vi.fn(), setDefaultComputer: vi.fn(), @@ -140,6 +144,36 @@ describe("WorkspacesHome (tile canvas)", () => { expect(store.list.map((w) => w.id)).toContain("fresh-ws"); }); + it("refuses to create an archived slug: message in the popover, form stays open, no tile", async () => { + const user = userEvent.setup(); + const list = [ws({ id: "a", title: "A" })]; + const http = fakeHttp(list, async (id) => ({ + kind: "found", + workspace: ws({ id, title: id, archived: true }), + })); + const store = (await import("../store.svelte")).createWorkspaceStore(http); + render(WorkspacesHome, { + props: { + store, + onNavigate: vi.fn(), + computers: [], + tiles: createTileLayoutStore({ http: memoryLayout() }), + }, + }); + + await screen.findByRole("button", { name: "Open workspace A" }); + await user.click(screen.getByRole("button", { name: "New workspace" })); + await user.type(screen.getByRole("textbox", { name: "New workspace" }), "old-ws{Enter}"); + + expect(await screen.findByText(/already exists and is archived/)).toBeInTheDocument(); + // The popover stays open (the form/input is still mounted). + expect(screen.getByRole("textbox", { name: "New workspace" })).toBeInTheDocument(); + // Nothing was ensured, no tile appeared, and no rename mode was entered. + expect(http.ensure).not.toHaveBeenCalled(); + expect(store.list.map((w) => w.id)).toEqual(["a"]); + expect(screen.queryByLabelText("Workspace title")).toBeNull(); + }); + it("right-clicking a tile opens the action bar; Delete keeps the confirm + store call", async () => { const user = userEvent.setup(); const confirmSpy = vi.spyOn(window, "confirm").mockReturnValue(true); -- cgit v1.2.3 From 01f9fb2a4ad6110def531ead16b370c818e8ee81 Mon Sep 17 00:00:00 2001 From: Adam Malczewski Date: Mon, 19 Oct 2026 17:02:40 +0900 Subject: feat(chat): cache timer for Claude Code subscription models Per-step and per-turn timers for the gaps between model requests (the Claude prompt cache expires after 5 min idle). A step gap runs live under the executing tools from step-complete until its last tool result (or the next step's output), then is stamped on that step's metrics line beside its cache badge. A turn gap runs live on the turn's metrics line from done/error until the next message is sent, then is stamped there. Counts m:ss up to 10 min, then '10+ min'; colour tracks the 5-min TTL. Gated on claude/. Pure reducer in core/metrics/cache-timer.ts; the turn gap is seeded from an idle conversation's lastActivityAt so it survives a reload. Stamps are in-memory only. --- src/app/store.svelte.ts | 7 + src/app/store.test.ts | 48 ++++ src/app/views/ChatPane.svelte | 8 +- src/core/metrics/cache-timer.test.ts | 344 ++++++++++++++++++++++++++++ src/core/metrics/cache-timer.ts | 285 +++++++++++++++++++++++ src/core/metrics/index.ts | 15 ++ src/features/chat/index.ts | 2 + src/features/chat/model-select.test.ts | 21 ++ src/features/chat/model-select.ts | 19 ++ src/features/chat/store.svelte.ts | 45 +++- src/features/chat/store.test.ts | 103 +++++++++ src/features/chat/ui.test.ts | 163 ++++++++++++- src/features/chat/ui/CacheTimerBadge.svelte | 58 +++++ src/features/chat/ui/ChatView.svelte | 62 ++++- 14 files changed, 1176 insertions(+), 4 deletions(-) create mode 100644 src/core/metrics/cache-timer.test.ts create mode 100644 src/core/metrics/cache-timer.ts create mode 100644 src/features/chat/ui/CacheTimerBadge.svelte (limited to 'src') diff --git a/src/app/store.svelte.ts b/src/app/store.svelte.ts index a3bc226..fafd4d7 100644 --- a/src/app/store.svelte.ts +++ b/src/app/store.svelte.ts @@ -1474,6 +1474,8 @@ export function createAppStore(opts?: CreateAppStoreOptions): AppStore { const model = conv.model ?? activeModel; const store = createChatFor(conv.id, model, conv.workspaceId); chatStores.set(conv.id, store); + // Seed the cache timer's turn gap (only an idle conversation is waiting on the user). + if (conv.status === "idle") store.noteLastActivity(conv.lastActivityAt); void store.load(); subscribeChat(conv.id); tabsStore.openTab({ @@ -1492,6 +1494,11 @@ export function createAppStore(opts?: CreateAppStoreOptions): AppStore { // authoritative conversation record is the source of truth, so a // differing tab workspace is corrected here. tabsStore.setTitle(conv.id, conv.title); + // Seed the cache timer's turn gap from the backend (a tab restored + // from localStorage at boot has no live `done` to go by). Only an + // idle conversation is waiting on the user. + if (conv.status === "idle") + chatStores.get(conv.id)?.noteLastActivity(conv.lastActivityAt); const tab = tabsStore.tabs.find((t) => t.conversationId === conv.id); if (tab?.workspaceId !== conv.workspaceId) { tabsStore.setWorkspaceId(conv.id, conv.workspaceId); diff --git a/src/app/store.test.ts b/src/app/store.test.ts index 89e418e..675fa64 100644 --- a/src/app/store.test.ts +++ b/src/app/store.test.ts @@ -855,6 +855,54 @@ describe("createAppStore", () => { store.dispose(); }); + it("fetchOpenConversations seeds the chat store's cache-timer turn gap from an idle conversation's lastActivityAt", async () => { + const ws = fakeSocket(); + const fetchImpl: typeof fetch = async (input: string | URL | Request): Promise => { + const url = typeof input === "string" ? input : input instanceof URL ? input.href : input.url; + if (url.includes("/conversations?status=")) { + return new Response( + JSON.stringify({ + conversations: [ + { + id: "conv-seed", + createdAt: 1000, + lastActivityAt: 7777, + title: "Seed conv", + status: "idle", + workspaceId: "default", + model: "claude/claude-opus-4-6", + }, + ], + }), + { status: 200 }, + ); + } + if (url.endsWith("/metrics")) { + return new Response(JSON.stringify({ turns: [] }), { status: 200 }); + } + return new Response(JSON.stringify({ chunks: [], latestSeq: 0 }), { status: 200 }); + }; + + const store = createAppStore({ + socketFactory: () => ws, + fetchImpl, + localStorage: createFakeStorage(), + }); + ws.resolveOpen(); + + await vi.waitFor(() => { + expect(store.tabs.some((t) => t.conversationId === "conv-seed")).toBe(true); + }); + store.selectTab("conv-seed"); + expect(store.activeChat.cacheTimer.running).toEqual({ + kind: "turn", + turnId: null, + startedAt: 7777, + }); + + store.dispose(); + }); + it("conversation.open WS message carrying a model stamps the tab with it", async () => { const ws = fakeSocket(); const store = createAppStore({ diff --git a/src/app/views/ChatPane.svelte b/src/app/views/ChatPane.svelte index 636cd7c..31e719e 100644 --- a/src/app/views/ChatPane.svelte +++ b/src/app/views/ChatPane.svelte @@ -7,13 +7,18 @@ * preservation, the ScrollToBottom affordance, and the empty-state * watermark. Keyed on the active conversation exactly like today's ChatView. */ - import { ChatView } from "../../features/chat"; + import { ChatView, isClaudeSubscriptionModel } from "../../features/chat"; import { createSmartScrollController, ScrollToBottom } from "../../features/smart-scroll"; import { preserveViewportAcross, type AppViewPorts } from "../view-ports"; let { ports }: { ports: AppViewPorts } = $props(); const store = $derived(ports.store); + // Cache timer: Claude Code subscription models only (5-minute prompt-cache expiry). + const cacheTimer = $derived( + isClaudeSubscriptionModel(store.activeModel) ? store.activeChat.cacheTimer : null, + ); + // One controller owns the chat scroll region; the effects below feed it the // edges (identical to the App.svelte wiring this moved out of). const smartScroll = createSmartScrollController(); @@ -81,6 +86,7 @@ thinkingKeyBase={store.activeChat.thinkingKeyBase} providerRetry={store.activeChat.providerRetry} apiBaseUrl={store.httpBase} + {cacheTimer} /> {/key} diff --git a/src/core/metrics/cache-timer.test.ts b/src/core/metrics/cache-timer.test.ts new file mode 100644 index 0000000..e55eff4 --- /dev/null +++ b/src/core/metrics/cache-timer.test.ts @@ -0,0 +1,344 @@ +import type { AgentEvent, StepId, TurnMetrics } from "@dispatch/wire"; +import { describe, expect, it } from "vitest"; +import type { RenderGroup } from "../chunks"; +import { + CACHE_TIMER_MAX_MS, + CACHE_TTL_MS, + type CacheTimerState, + foldCacheTimerEvent, + initialCacheTimerState, + lastTurnRowIndex, + seedTurnGap, + stopTurnGap, + viewCacheTimer, +} from "./cache-timer"; +import type { MetricsRow } from "./types"; + +const SEC = 1000; +const MIN = 60 * SEC; +const C = "conv"; + +const ev = { + turnStart: (turnId: string): AgentEvent => ({ type: "turn-start", conversationId: C, turnId }), + userMessage: (turnId: string): AgentEvent => ({ + type: "user-message", + conversationId: C, + turnId, + text: "hi", + }), + text: (turnId: string): AgentEvent => ({ + type: "text-delta", + conversationId: C, + turnId, + delta: "x", + }), + call: (turnId: string, stepId: string, toolCallId: string): AgentEvent => ({ + type: "tool-call", + conversationId: C, + turnId, + stepId: stepId as StepId, + toolCallId, + toolName: "t", + input: {}, + }), + result: ( + turnId: string, + stepId: string, + toolCallId: string, + durationMs?: number, + ): AgentEvent => ({ + type: "tool-result", + conversationId: C, + turnId, + stepId: stepId as StepId, + toolCallId, + toolName: "t", + content: "ok", + isError: false, + ...(durationMs !== undefined ? { durationMs } : {}), + }), + stepComplete: (turnId: string, stepId: string): AgentEvent => ({ + type: "step-complete", + conversationId: C, + turnId, + stepId: stepId as StepId, + }), + usage: (turnId: string, stepId: string): AgentEvent => ({ + type: "usage", + conversationId: C, + turnId, + stepId: stepId as StepId, + usage: { inputTokens: 1, outputTokens: 1 }, + }), + done: (turnId: string): AgentEvent => ({ + type: "done", + conversationId: C, + turnId, + reason: "end-turn", + }), + error: (turnId: string): AgentEvent => ({ + type: "error", + conversationId: C, + turnId, + message: "boom", + }), +}; + +const noTurn = () => undefined; + +/** Fold `[time, event]` pairs from the initial state. */ +function run( + steps: readonly (readonly [number, AgentEvent])[], + start: CacheTimerState = initialCacheTimerState(), + lastTurnId: () => string | undefined = noTurn, +): CacheTimerState { + return steps.reduce((s, [at, e]) => foldCacheTimerEvent(s, e, at, lastTurnId), start); +} + +describe("foldCacheTimerEvent — step gaps", () => { + it("runs from step-complete while tools execute, and stamps when the last result returns", () => { + const s1 = run([ + [0, ev.turnStart("t1")], + [100, ev.call("t1", "t1#0", "a")], + [110, ev.call("t1", "t1#0", "b")], + [1_000, ev.stepComplete("t1", "t1#0")], + ]); + expect(s1.running).toMatchObject({ + kind: "step", + stepId: "t1#0", + startedAt: 1_000, + toolCount: 2, + }); + + const s2 = run([[4_000, ev.result("t1", "t1#0", "a")]], s1); + expect(s2.running?.kind).toBe("step"); // still waiting on "b" + expect(s2.stepStamps.size).toBe(0); + + const s3 = run([[13_000, ev.result("t1", "t1#0", "b")]], s2); + expect(s3.running).toBeNull(); + expect(s3.stepStamps.get("t1#0")).toBe(12_000); + }); + + it("stamps each step of a multi-step turn separately", () => { + const s = run([ + [0, ev.call("t1", "t1#0", "a")], + [1_000, ev.stepComplete("t1", "t1#0")], + [3_000, ev.result("t1", "t1#0", "a")], + [5_000, ev.call("t1", "t1#1", "b")], + [6_000, ev.stepComplete("t1", "t1#1")], + [66_000, ev.result("t1", "t1#1", "b")], + ]); + expect(s.stepStamps.get("t1#0")).toBe(2_000); + expect(s.stepStamps.get("t1#1")).toBe(60_000); + }); + + it("uses the backend tool duration when it exceeds the wall clock (late-join replay burst)", () => { + const s = run([ + [5_000, ev.call("t1", "t1#0", "a")], + [5_000, ev.stepComplete("t1", "t1#0")], + [5_000, ev.result("t1", "t1#0", "a", 42_000)], + ]); + expect(s.stepStamps.get("t1#0")).toBe(42_000); + }); + + it("stops at the next step's output when a result never matched (fallback)", () => { + const s = run([ + [0, ev.call("t1", "t1#0", "a")], + [1_000, ev.stepComplete("t1", "t1#0")], + [9_000, ev.text("t1")], + ]); + expect(s.running).toBeNull(); + expect(s.stepStamps.get("t1#0")).toBe(8_000); + }); + + it("stops on a tool-call or usage from a DIFFERENT step, not its own", () => { + const base = run([ + [0, ev.call("t1", "t1#0", "a")], + [1_000, ev.stepComplete("t1", "t1#0")], + ]); + expect(run([[2_000, ev.usage("t1", "t1#0")]], base).running?.kind).toBe("step"); + expect(run([[3_000, ev.usage("t1", "t1#1")]], base).stepStamps.get("t1#0")).toBe(2_000); + expect(run([[4_000, ev.call("t1", "t1#1", "z")]], base).stepStamps.get("t1#0")).toBe(3_000); + }); + + it("discards a step with no tool calls (the final step of a turn)", () => { + const s = run([ + [1_000, ev.stepComplete("t1", "t1#0")], + [1_050, ev.done("t1")], + ]); + expect(s.stepStamps.size).toBe(0); + expect(s.running).toEqual({ kind: "turn", turnId: "t1", startedAt: 1_050 }); + }); + + it("stamps a step whose tools were still running when the turn ended (stopped)", () => { + const s = run([ + [0, ev.call("t1", "t1#0", "a")], + [1_000, ev.stepComplete("t1", "t1#0")], + [21_000, ev.done("t1")], + ]); + expect(s.stepStamps.get("t1#0")).toBe(20_000); + expect(s.running?.kind).toBe("turn"); + }); +}); + +describe("foldCacheTimerEvent — turn gaps", () => { + it("starts at done and is stamped on that turn when the next turn starts", () => { + const s = run([ + [1_000, ev.done("t1")], + [91_000, ev.userMessage("t2")], + [91_500, ev.turnStart("t2")], + ]); + expect(s.running).toBeNull(); + expect(s.turnStamps.get("t1")).toBe(90_000); + }); + + it("an error also ends the turn", () => { + expect(run([[7, ev.error("t1")]]).running).toEqual({ + kind: "turn", + turnId: "t1", + startedAt: 7, + }); + }); + + it("stopTurnGap stamps on a local send", () => { + const s = stopTurnGap(run([[1_000, ev.done("t1")]]), 4_000, noTurn); + expect(s.running).toBeNull(); + expect(s.turnStamps.get("t1")).toBe(3_000); + }); + + it("stopTurnGap is a no-op while a step gap runs", () => { + const base = run([ + [0, ev.call("t1", "t1#0", "a")], + [1_000, ev.stepComplete("t1", "t1#0")], + ]); + expect(stopTurnGap(base, 2_000, noTurn)).toBe(base); + }); +}); + +describe("seedTurnGap", () => { + it("seeds a turn gap with no turnId, stamped on the most recent turn when stopped", () => { + const seeded = seedTurnGap(initialCacheTimerState(), 1_000); + expect(seeded.running).toEqual({ kind: "turn", turnId: null, startedAt: 1_000 }); + const s = stopTurnGap(seeded, 61_000, () => "t9"); + expect(s.turnStamps.get("t9")).toBe(60_000); + }); + + it("drops a seeded gap when the most recent turn is unknown", () => { + const s = stopTurnGap(seedTurnGap(initialCacheTimerState(), 1_000), 2_000, noTurn); + expect(s.running).toBeNull(); + expect(s.turnStamps.size).toBe(0); + }); + + it("keeps the later start, never overrides a step gap, ignores non-finite", () => { + const live = run([[5_000, ev.done("t1")]]); + expect(seedTurnGap(live, 4_000).running).toEqual({ + kind: "turn", + turnId: "t1", + startedAt: 5_000, + }); + expect(seedTurnGap(live, 6_000).running).toEqual({ + kind: "turn", + turnId: "t1", + startedAt: 6_000, + }); + const step = run([ + [0, ev.call("t1", "t1#0", "a")], + [1_000, ev.stepComplete("t1", "t1#0")], + ]); + expect(seedTurnGap(step, 9_000)).toBe(step); + expect(seedTurnGap(initialCacheTimerState(), Number.NaN).running).toBeNull(); + }); +}); + +describe("viewCacheTimer", () => { + it("formats m:ss up to 10 minutes", () => { + expect(viewCacheTimer(0).label).toBe("0:00"); + expect(viewCacheTimer(9 * SEC + 999).label).toBe("0:09"); + expect(viewCacheTimer(MIN + 5 * SEC).label).toBe("1:05"); + expect(viewCacheTimer(5 * MIN + 30 * SEC).label).toBe("5:30"); + expect(viewCacheTimer(9 * MIN + 59 * SEC).label).toBe("9:59"); + expect(viewCacheTimer(9 * MIN + 59 * SEC).capped).toBe(false); + }); + + it("reads '10+ min' and is capped at and past 10 minutes", () => { + expect(viewCacheTimer(CACHE_TIMER_MAX_MS)).toEqual({ + label: "10+ min", + level: "error", + capped: true, + }); + expect(viewCacheTimer(3 * 60 * MIN).label).toBe("10+ min"); + }); + + it("colours by the 5-minute cache TTL", () => { + expect(viewCacheTimer(3 * MIN + 59 * SEC).level).toBe("success"); + expect(viewCacheTimer(4 * MIN).level).toBe("warning"); + expect(viewCacheTimer(CACHE_TTL_MS).level).toBe("error"); + expect(viewCacheTimer(7 * MIN).level).toBe("error"); + }); + + it("clamps negative / non-finite elapsed (clock skew) to 0:00", () => { + expect(viewCacheTimer(-5 * SEC).label).toBe("0:00"); + expect(viewCacheTimer(Number.NaN).label).toBe("0:00"); + }); +}); + +function userRow(text = "hi"): MetricsRow { + const group: RenderGroup = { + kind: "single", + chunk: { seq: null, role: "user", chunk: { type: "text", text }, provisional: true }, + }; + return { kind: "group", group }; +} + +function assistantRow(turnId: string): MetricsRow { + const group: RenderGroup = { + kind: "single", + chunk: { + seq: 1, + role: "assistant", + chunk: { type: "text", text: "ok" }, + provisional: false, + turnId, + }, + }; + return { kind: "group", group }; +} + +function turnRow(turnId: string, turnNumber: number): MetricsRow { + const usage = { inputTokens: 10, outputTokens: 5 }; + const turn: TurnMetrics = { turnId, usage, steps: [{ stepId: `${turnId}#0` as StepId, usage }] }; + return { + kind: "turn-metrics", + turn, + turnNumber, + cumulativeUsage: usage, + prevLastStepUsage: null, + }; +} + +describe("lastTurnRowIndex", () => { + it("picks the LAST turn-metrics row", () => { + const rows = [ + userRow(), + assistantRow("t1"), + turnRow("t1", 1), + userRow(), + assistantRow("t2"), + turnRow("t2", 2), + ]; + expect(lastTurnRowIndex(rows)).toBe(5); + }); + + it("is null once a user message follows the last turn", () => { + expect(lastTurnRowIndex([userRow(), turnRow("t1", 1), userRow("next")])).toBeNull(); + }); + + it("skips trailing non-user rows after the turn row", () => { + expect(lastTurnRowIndex([userRow(), turnRow("t1", 1), assistantRow("t1")])).toBe(1); + }); + + it("is null with no turn-metrics rows", () => { + expect(lastTurnRowIndex([])).toBeNull(); + expect(lastTurnRowIndex([userRow(), assistantRow("t1")])).toBeNull(); + }); +}); diff --git a/src/core/metrics/cache-timer.ts b/src/core/metrics/cache-timer.ts new file mode 100644 index 0000000..66d5fac --- /dev/null +++ b/src/core/metrics/cache-timer.ts @@ -0,0 +1,285 @@ +import type { AgentEvent } from "@dispatch/wire"; +import type { MetricsRow } from "./types"; + +/** + * Pure core for the **cache timer** — how long the prompt cache has sat idle + * between model requests. Anthropic's prompt cache (as used by the Claude Code + * subscription) expires 5 minutes after its last hit, so the gaps BETWEEN + * requests decide whether the next request reads from cache: + * + * - **step gap**: a step (one LLM round-trip) finished and its tools are + * executing. Starts at `step-complete`; stops when the step's last tool + * result arrives (the next request fires right after) or, failing that, when + * anything from the next step shows up. Its final value is **stamped** on + * that step (keyed by `stepId`). + * - **turn gap**: the turn ended (`done` / `error`) and the conversation is + * waiting on the user. Stops when a new message is sent (locally, or a + * `user-message` / `turn-start` from another client) and is stamped on that + * turn (keyed by `turnId`). + * + * At most ONE gap runs at a time (`running`). Zero DOM, zero clock: every fold + * takes the current time as an argument. + */ + +/** Prompt-cache time-to-live: 5 minutes (drives the badge colour). */ +export const CACHE_TTL_MS = 5 * 60 * 1000; + +/** The timer counts up to this, then reads "10+ min". */ +export const CACHE_TIMER_MAX_MS = 10 * 60 * 1000; + +/** Within this much of the TTL the timer warns (colour only). */ +const WARN_AT_MS = CACHE_TTL_MS - 60 * 1000; + +/** A step gap in progress: the step's tools are executing. */ +export interface RunningStepGap { + readonly kind: "step"; + readonly turnId: string; + readonly stepId: string; + /** Epoch ms the step completed. */ + readonly startedAt: number; + /** How many tool calls the step made (`0` → nothing to show; never stamped). */ + readonly toolCount: number; + /** Tool calls still awaiting their result. */ + readonly pending: ReadonlySet; + /** Longest backend-reported tool `durationMs` seen so far for this step. */ + readonly maxToolMs: number; +} + +/** A turn gap in progress: the turn ended and nothing has been sent since. */ +export interface RunningTurnGap { + readonly kind: "turn"; + /** + * The turn that ended, or `null` when seeded from the backend's + * `lastActivityAt` after a reload (no live `done` seen) — then it belongs to + * the most recent turn, resolved positionally / via `lastTurnId` on stop. + */ + readonly turnId: string | null; + /** Epoch ms the turn ended. */ + readonly startedAt: number; +} + +export type RunningGap = RunningStepGap | RunningTurnGap; + +export interface CacheTimerState { + readonly running: RunningGap | null; + /** Stamped step gaps: `stepId` → elapsed ms. */ + readonly stepStamps: ReadonlyMap; + /** Stamped turn gaps: `turnId` → elapsed ms. */ + readonly turnStamps: ReadonlyMap; + /** Tool calls seen per step in the CURRENT turn (`stepId` → call ids). */ + readonly calls: ReadonlyMap>; + /** Tool results seen per step in the CURRENT turn (`stepId` → call ids). */ + readonly results: ReadonlyMap>; + /** Longest tool `durationMs` per step in the CURRENT turn. */ + readonly toolMs: ReadonlyMap; +} + +export function initialCacheTimerState(): CacheTimerState { + return { + running: null, + stepStamps: new Map(), + turnStamps: new Map(), + calls: new Map(), + results: new Map(), + toolMs: new Map(), + }; +} + +function addTo( + map: ReadonlyMap>, + key: string, + value: string, +): ReadonlyMap> { + const next = new Map(map); + next.set(key, new Set([...(map.get(key) ?? []), value])); + return next; +} + +function withEntry(map: ReadonlyMap, key: string, value: number) { + const next = new Map(map); + next.set(key, value); + return next; +} + +/** + * Stop the running STEP gap (if any). A step that made tool calls is stamped + * with `max(wall-clock elapsed, longest backend tool duration)` — the backend + * figure keeps a late-join replay (whole step delivered in one burst, ~0 + * elapsed) honest. A step with no tool calls is discarded. + */ +function stopStep(state: CacheTimerState, now: number): CacheTimerState { + const r = state.running; + if (r === null || r.kind !== "step") return state; + if (r.toolCount === 0) return { ...state, running: null }; + const elapsed = Math.max(0, now - r.startedAt, r.maxToolMs); + return { ...state, running: null, stepStamps: withEntry(state.stepStamps, r.stepId, elapsed) }; +} + +/** + * Stop the running TURN gap (if any) and stamp it on its turn. A seeded gap + * (`turnId === null`) is stamped on `lastTurnId()` — the most recent turn's + * id — or dropped when that is unknown. + */ +export function stopTurnGap( + state: CacheTimerState, + now: number, + lastTurnId: () => string | undefined, +): CacheTimerState { + const r = state.running; + if (r === null || r.kind !== "turn") return state; + const turnId = r.turnId ?? lastTurnId(); + if (turnId === undefined) return { ...state, running: null }; + const elapsed = Math.max(0, now - r.startedAt); + return { ...state, running: null, turnStamps: withEntry(state.turnStamps, turnId, elapsed) }; +} + +/** A turn ended: close any step gap, forget the turn's tool bookkeeping, start the turn gap. */ +function endTurn(state: CacheTimerState, turnId: string, now: number): CacheTimerState { + const s = stopStep(state, now); + return { + ...s, + running: { kind: "turn", turnId, startedAt: now }, + calls: new Map(), + results: new Map(), + toolMs: new Map(), + }; +} + +/** + * Fold one live event into the cache-timer state at time `now`. `lastTurnId` + * resolves a seeded turn gap's turn when a new turn starts (see + * `stopTurnGap`); it is only called in that case. + */ +export function foldCacheTimerEvent( + state: CacheTimerState, + event: AgentEvent, + now: number, + lastTurnId: () => string | undefined, +): CacheTimerState { + const r = state.running; + switch (event.type) { + case "user-message": + case "turn-start": + // A new turn is starting — the user (here or elsewhere) sent a message. + return stopTurnGap(state, now, lastTurnId); + + case "tool-call": { + // A call from a DIFFERENT step means the next step is already streaming. + const s = r?.kind === "step" && r.stepId !== event.stepId ? stopStep(state, now) : state; + return { ...s, calls: addTo(s.calls, event.stepId, event.toolCallId) }; + } + + case "tool-result": { + const results = addTo(state.results, event.stepId, event.toolCallId); + const dur = event.durationMs ?? 0; + const toolMs = + dur > (state.toolMs.get(event.stepId) ?? 0) + ? withEntry(state.toolMs, event.stepId, dur) + : state.toolMs; + let s: CacheTimerState = { ...state, results, toolMs }; + if (r?.kind === "step" && r.stepId === event.stepId) { + const pending = new Set(r.pending); + pending.delete(event.toolCallId); + s = { ...s, running: { ...r, pending, maxToolMs: Math.max(r.maxToolMs, dur) } }; + // Every tool of the step has returned → the next request fires now. + if (pending.size === 0) s = stopStep(s, now); + } + return s; + } + + case "step-complete": { + const s = stopStep(stopTurnGap(state, now, lastTurnId), now); + const calls = s.calls.get(event.stepId) ?? new Set(); + const done = s.results.get(event.stepId) ?? new Set(); + const pending = new Set([...calls].filter((id) => !done.has(id))); + return { + ...s, + running: { + kind: "step", + turnId: event.turnId, + stepId: event.stepId, + startedAt: now, + toolCount: calls.size, + pending, + maxToolMs: s.toolMs.get(event.stepId) ?? 0, + }, + }; + } + + case "text-delta": + case "reasoning-delta": + case "provider-retry": + // Output (or a retry) from the next request: the gap is over. + return r?.kind === "step" ? stopStep(state, now) : state; + + case "usage": + return r?.kind === "step" && event.stepId !== undefined && event.stepId !== r.stepId + ? stopStep(state, now) + : state; + + case "done": + case "error": + return endTurn(state, event.turnId, now); + + default: + return state; + } +} + +/** + * Seed a turn gap from an authoritative "last activity" timestamp (the + * backend's `lastActivityAt`, for a conversation that is idle — e.g. after a + * reload, when no live `done` was seen). Never overrides a running step gap; + * a running turn gap keeps the LATER start. Non-finite input is ignored. + */ +export function seedTurnGap(state: CacheTimerState, at: number): CacheTimerState { + if (!Number.isFinite(at)) return state; + const r = state.running; + if (r === null) return { ...state, running: { kind: "turn", turnId: null, startedAt: at } }; + if (r.kind === "turn" && at > r.startedAt) return { ...state, running: { ...r, startedAt: at } }; + return state; +} + +// ── View ────────────────────────────────────────────────────────────────────── + +/** Formatted cache-timer view: label + colour severity + capped flag. */ +export interface CacheTimerView { + /** `"m:ss"` up to 10 minutes, then `"10+ min"`. */ + readonly label: string; + /** Colour by the 5-minute cache TTL (maps to DaisyUI `badge-{level}`). */ + readonly level: "success" | "warning" | "error"; + /** Whether the 10-minute display cap was reached (a live timer may stop ticking). */ + readonly capped: boolean; +} + +/** + * Build the view for `elapsedMs`. Colour tracks the cache TTL: success under + * 4 min, warning in the last minute, error once the cache has likely expired. + * Negative / non-finite elapsed (clock skew) clamps to `0:00`. + */ +export function viewCacheTimer(elapsedMs: number): CacheTimerView { + const ms = Number.isFinite(elapsedMs) && elapsedMs > 0 ? elapsedMs : 0; + const level = ms >= CACHE_TTL_MS ? "error" : ms >= WARN_AT_MS ? "warning" : "success"; + if (ms >= CACHE_TIMER_MAX_MS) return { label: "10+ min", level, capped: true }; + const totalSeconds = Math.floor(ms / 1000); + const minutes = Math.floor(totalSeconds / 60); + const seconds = totalSeconds % 60; + return { label: `${minutes}:${seconds.toString().padStart(2, "0")}`, level, capped: false }; +} + +/** + * Index of the most recent turn's metrics row — the LAST `turn-metrics` row, + * provided no user message follows it. Where a SEEDED turn gap (no `turnId`) + * renders. `null` when there is no such row. + */ +export function lastTurnRowIndex(rows: readonly MetricsRow[]): number | null { + for (let i = rows.length - 1; i >= 0; i--) { + const row = rows[i]; + if (row === undefined) continue; + if (row.kind === "turn-metrics") return i; + if (row.kind === "group" && row.group.kind === "single" && row.group.chunk.role === "user") { + return null; + } + } + return null; +} diff --git a/src/core/metrics/index.ts b/src/core/metrics/index.ts index d3c9669..d9df7f5 100644 --- a/src/core/metrics/index.ts +++ b/src/core/metrics/index.ts @@ -1,3 +1,18 @@ +export { + CACHE_TIMER_MAX_MS, + CACHE_TTL_MS, + type CacheTimerState, + type CacheTimerView, + foldCacheTimerEvent, + initialCacheTimerState, + lastTurnRowIndex, + type RunningGap, + type RunningStepGap, + type RunningTurnGap, + seedTurnGap, + stopTurnGap, + viewCacheTimer, +} from "./cache-timer"; export { type ContextUsage, computeCachePct, diff --git a/src/features/chat/index.ts b/src/features/chat/index.ts index 59da39b..fec2cdb 100644 --- a/src/features/chat/index.ts +++ b/src/features/chat/index.ts @@ -8,6 +8,8 @@ export { groupRenderedChunks, resolveImageUrl, viewProviderRetry } from "../../c export type { TurnMetricsEntry } from "../../core/metrics"; export type { NamedAgentLike, NamedAgentSnapshotLike } from "./model-select"; export { + CLAUDE_SUBSCRIPTION_KEY, + isClaudeSubscriptionModel, isNamedAgentSelection, isNumberedOpenCode, isVisionModel, 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/ (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 @@ -62,6 +62,25 @@ export function isNumberedOpenCode(id: string): boolean { return openCodeNumber(id) !== null; } +/** + * 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/`. + */ +export const CLAUDE_SUBSCRIPTION_KEY = "claude"; + +/** + * True iff `full` is a model served through the Claude Code subscription + * (`claude/`). 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-` → * "OpenCode #"; any other id is returned unchanged. The RAW id is still what diff --git a/src/features/chat/store.svelte.ts b/src/features/chat/store.svelte.ts index 7f8e113..01ad9f9 100644 --- a/src/features/chat/store.svelte.ts +++ b/src/features/chat/store.svelte.ts @@ -25,13 +25,17 @@ 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"; @@ -78,6 +82,8 @@ 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; } export interface ChatStore { @@ -122,6 +128,20 @@ export interface ChatStore { * a trim (see `TranscriptState.hiddenThinkingCount`). */ readonly thinkingKeyBase: number; + /** + * 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: ChatDeltaMessage | ChatErrorMessage): void; /** * Send a user message (start a turn via `chat.send`). Optimistically echoes @@ -200,8 +220,16 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { let _error = $state(null); let _model = $state(deps.model); let _namedAgent = $state(null); + let cacheTimer = $state(initialCacheTimerState()); + const now = deps.now ?? Date.now; let disposed = false; + /** 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); // Memoized projections, keyed by state-object identity (the reducers are @@ -242,14 +270,19 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { let m = metrics; let sealed = false; let failed = false; + let c = cacheTimer; + const at = now(); for (const event of events) { t = foldEvent(t, event); m = foldMetricsEvent(m, event); + const mNow = m; + c = foldCacheTimerEvent(c, event, at, () => lastTurnIdOf(mNow)); if (t.sealedTurnId !== null) sealed = true; else if (event.type === "error") failed = true; } transcript = t; metrics = m; + cacheTimer = c; maybeTrim(); if (sealed) { void syncTail(); @@ -390,6 +423,14 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { get thinkingKeyBase(): number { return transcript.hiddenThinkingCount; }, + get cacheTimer(): CacheTimerState { + return cacheTimer; + }, + + noteLastActivity(at: number): void { + if (selectGenerating(transcript)) return; + cacheTimer = seedTurnGap(cacheTimer, at); + }, handleDelta(msg: ChatDeltaMessage | ChatErrorMessage): void { if (msg.type === "chat.error") { @@ -415,6 +456,8 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { send(text: string, images?: readonly ImageInput[]): void { flushEvents(); + // Sending ends the wait since the last turn: stamp it on that turn. + cacheTimer = stopTurnGap(cacheTimer, now(), () => lastTurnIdOf(metrics)); transcript = appendUserMessage(transcript, text, images); maybeTrim(); const msg: ChatSendMessage = { diff --git a/src/features/chat/store.test.ts b/src/features/chat/store.test.ts index f9b4b38..8604107 100644 --- a/src/features/chat/store.test.ts +++ b/src/features/chat/store.test.ts @@ -1789,3 +1789,106 @@ describe("createChatStore — batched delta application (scheduleFlush)", () => expect(store.chunks).toHaveLength(0); }); }); + +describe("createChatStore — cacheTimer", () => { + function makeStore(now: () => number) { + return createChatStore({ + 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(); + }); +}); diff --git a/src/features/chat/ui.test.ts b/src/features/chat/ui.test.ts index 91585e2..6b7176d 100644 --- a/src/features/chat/ui.test.ts +++ b/src/features/chat/ui.test.ts @@ -3,7 +3,11 @@ 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 ChatView from "./ui/ChatView.svelte"; import Composer from "./ui/Composer.svelte"; @@ -1542,3 +1546,160 @@ 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 { + 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"); + // 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/); + const stamp = screen.getByTestId("cache-timer-stamp"); + expect(stamp).toHaveTextContent("10+ min"); + expect(stamp.closest(".chat-bubble")).toHaveTextContent(/turn 1/); + }); + + 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..e5abed5 --- /dev/null +++ b/src/features/chat/ui/CacheTimerBadge.svelte @@ -0,0 +1,58 @@ + + +{view.label} diff --git a/src/features/chat/ui/ChatView.svelte b/src/features/chat/ui/ChatView.svelte index 18fea7c..e641791 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"; @@ -27,6 +30,7 @@ thinkingKeyBase = 0, providerRetry = null, apiBaseUrl = "", + cacheTimer = null, }: { chunks: readonly RenderedChunk[]; turnMetrics?: readonly TurnMetricsEntry[]; @@ -55,6 +59,14 @@ * "" (root-relative — a browser resolves `/images/…` against its origin). */ apiBaseUrl?: string; + /** + * 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). @@ -74,6 +86,31 @@ 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`) that // survives the provisional→committed (seq null → seq N) transition, so the @@ -180,13 +217,14 @@ {/if} - {#each keyedRows as { row, key } (key)} + {#each keyedRows as { row, key }, i (key)}
{#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)}
@@ -203,6 +241,9 @@ class="badge badge-xs {badgeClass[stepCache.level]}" data-testid="step-cache-badge">{stepCache.pct}% + {#if stepStamp !== undefined} + + {/if}
@@ -211,6 +252,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)}
@@ -236,6 +279,11 @@ {retention.pct}% {/if} + {#if turnLive !== undefined} + + {:else if turnStamp !== undefined} + + {/if}
@@ -252,6 +300,18 @@ {/if}
{/each} + {#if liveStepGap} + +
+
+
+ Running tools + +
+
+
+ {/if} {#if providerRetry} {@const rv = viewProviderRetry(providerRetry)}
onStop?.()} > - Stop + {stopButton.text} {:else} `, "form,[action]"], + ["button formaction", ``, "[formaction]"], + ["input formaction", ``, "input,[formaction]"], + ["stylesheet link", ``, "link"], + ["preload link", ``, "link"], + ["prefetch link", ``, "link"], + ["DNS prefetch", ``, "link"], + ["meta refresh", ``, "meta"], + ["iframe src", ``, "iframe"], + [ + "iframe srcdoc", + ``, + "iframe,[srcdoc]", + ], + ["base URL", ``, "base"], + ["script src", ``, "script"], + ["anchor ping", `link`, "[ping]"], + ["legacy command", ``, "command"], + ["legacy applet", ``, "applet"], + ["non-image src", `
text
`, "[src],[href]"], + ] as const; + + // Inspect the returned DOM, not jsdom's lack of resource fetching. Both + // authoritative renders and streaming tails must remove each active sink. + describe.each([ + ["full", renderMarkdown], + ["tail", renderMarkdownTail], + ] as const)("%s render", (_name, render) => { + it.each(vectors)("neutralizes %s", (_vector, text, forbidden) => { + const doc = new DOMParser().parseFromString(render(text), "text/html"); + expect(doc.querySelector(forbidden)).toBeNull(); + for (const element of doc.body.querySelectorAll("*")) { + for (const attr of element.attributes) { + if (attr.name === "href" && element.tagName === "A") continue; + expect(attr.value).not.toContain("evil.example"); + } + } + }); + }); + + it("also blocks all vectors in settled streaming prefixes", () => { + for (const [_name, text, forbidden] of vectors) { + const state = createStreamingMarkdown(); + advance(state, `${text}\n\ntail`); + const result = advance(state, `${text}\n\ntail grows`); + const doc = new DOMParser().parseFromString( + result.prefixHtml + result.tailHtml, + "text/html", + ); + expect(doc.querySelector(forbidden)).toBeNull(); + } + }); + + it("retains passive Markdown structure, highlighting and disabled task checkboxes", () => { + const html = renderMarkdown( + "- [x] done\n- [ ] pending\n\n| A | B |\n| - | - |\n| x | y |\n\n```js\nconst x = 1;\n```", + ); + const doc = new DOMParser().parseFromString(html, "text/html"); + expect(doc.querySelectorAll('input[type="checkbox"][disabled]')).toHaveLength(2); + expect(doc.querySelector("input[checked]")).not.toBeNull(); + expect(doc.querySelectorAll("td")).toHaveLength(2); + expect(doc.querySelector(".hljs-keyword")).not.toBeNull(); + expect(doc.querySelector('button[data-copy][type="button"]')).not.toBeNull(); + }); + + it("makes raw checkboxes and buttons inert, even inside an outer form", () => { + const doc = new DOMParser().parseFromString( + renderMarkdown(''), + "text/html", + ); + expect(doc.querySelector("input")?.hasAttribute("disabled")).toBe(true); + expect(doc.querySelector("button")?.getAttribute("type")).toBe("button"); + }); + + it("does not permit non-HTTP network schemes on images", () => { + expect(imageSrc(renderMarkdown(''))).toBeNull(); + }); + }); + describe("local-path images", () => { const BASE = "http://localhost:24203"; const image = { apiBase: BASE, cwd: "/w/foo" }; diff --git a/src/features/markdown/logic/markdown.ts b/src/features/markdown/logic/markdown.ts index a65997e..ed287d2 100644 --- a/src/features/markdown/logic/markdown.ts +++ b/src/features/markdown/logic/markdown.ts @@ -185,6 +185,87 @@ function addCopyButtons(html: string): string { .replace(/<\/pre>/g, "
"); } +// Positive policy: chat HTML is passive Markdown structure, not arbitrary web +// content. New browser elements/attributes must not silently add fetch sinks. +// In particular there is no CSS, SVG/MathML, media, form, embed or metadata. +const MARKDOWN_TAGS = [ + "a", + "abbr", + "b", + "blockquote", + "br", + "caption", + "code", + "col", + "colgroup", + "dd", + "del", + "details", + "div", + "dl", + "dt", + "em", + "h1", + "h2", + "h3", + "h4", + "h5", + "h6", + "hr", + "i", + "img", + "kbd", + "li", + "mark", + "ol", + "p", + "pre", + "q", + "rp", + "rt", + "ruby", + "s", + "samp", + "small", + "span", + "strong", + "sub", + "summary", + "sup", + "table", + "tbody", + "td", + "tfoot", + "th", + "thead", + "tr", + "u", + "ul", + "var", + "wbr", + "button", + "input", +]; +const MARKDOWN_ATTRIBUTES = [ + "alt", + "align", + "checked", + "class", + "colspan", + "disabled", + "height", + "href", + "open", + "reversed", + "rowspan", + "scope", + "src", + "start", + "title", + "type", + "width", +]; + /** * A dedicated, render-local DOMPurify instance: hooks never affect another * consumer or retain a conversation's backend allowlist. The document base is @@ -195,7 +276,22 @@ function sanitize(html: string, image?: MarkdownImageOptions): string { const base = document.baseURI; const origin = new URL(base).origin; const backendOrigin = image === undefined ? origin : new URL(image.apiBase || base, base).origin; + purify.addHook("uponSanitizeElement", (node) => { + // Keep GFM task lists, but no image/submit/other form controls. Removing + // only type/src would allow browser defaulting or later attribute changes. + if ( + node.nodeName === "INPUT" && + (node as Element).getAttribute("type")?.toLowerCase() !== "checkbox" + ) { + node.parentNode?.removeChild(node); + } + }); purify.addHook("afterSanitizeAttributes", (node) => { + // URL-bearing attributes are supported ONLY on these two HTML elements. + if (node.tagName !== "A") node.removeAttribute("href"); + if (node.tagName !== "IMG") node.removeAttribute("src"); + if (node.tagName === "INPUT") node.setAttribute("disabled", ""); + if (node.tagName === "BUTTON") node.setAttribute("type", "button"); if (node.tagName === "A" && node.hasAttribute("href")) { node.setAttribute("target", "_blank"); node.setAttribute("rel", "noopener noreferrer"); @@ -210,7 +306,11 @@ function sanitize(html: string, image?: MarkdownImageOptions): string { node.removeAttribute("src"); return; } - if (url.protocol !== "http:" && url.protocol !== "https:") return; + if (url.protocol === "data:") return; // Image context cannot load subresources. + if (url.protocol !== "http:" && url.protocol !== "https:") { + node.removeAttribute("src"); + return; + } if (url.origin === origin || url.origin === backendOrigin) return; // Only safe, inert attributes are carried over. In particular, never copy @@ -225,8 +325,14 @@ function sanitize(html: string, image?: MarkdownImageOptions): string { button.textContent = `Load image from ${url.host}`; node.parentNode?.replaceChild(button, node); }); - // Responsive candidates on / must not bypass the src hook. - return purify.sanitize(html, { FORBID_ATTR: ["srcset"] }); + return purify.sanitize(html, { + ALLOWED_TAGS: MARKDOWN_TAGS, + ALLOWED_ATTR: MARKDOWN_ATTRIBUTES, + // Keep consent/copy data and accessibility labels, never active URL sinks + // such as srcset, style, background, ping, action or namespaced href. + ALLOW_DATA_ATTR: true, + ALLOW_ARIA_ATTR: true, + }); } /** diff --git a/src/features/markdown/ui/markdown.test.ts b/src/features/markdown/ui/markdown.test.ts index 149bff5..b86cc0a 100644 --- a/src/features/markdown/ui/markdown.test.ts +++ b/src/features/markdown/ui/markdown.test.ts @@ -62,6 +62,25 @@ describe("Markdown", () => { ).toBeInTheDocument(); }); + it.each([ + false, + true, + ])("keeps non-image HTML inert before and after image consent (streaming=%s)", async (streaming) => { + const { container } = render(Markdown, { + props: { + text: '
', + streaming, + }, + }); + expect(container.querySelector("video,svg,input,[style],img")).toBeNull(); + await fireEvent.click(screen.getByRole("button", { name: "Load image from evil.example" })); + expect(container.querySelectorAll("img")).toHaveLength(1); + expect(container.querySelector("img")?.getAttribute("src")).toBe( + "https://evil.example/allowed-after-click", + ); + expect(container.querySelector("video,svg,input,[style]")).toBeNull(); + }); + it("rejects an unsafe URL in a forged placeholder", async () => { const { container } = render(Markdown, { props: { text: '' }, diff --git a/vite.config.ts b/vite.config.ts index 4a74b9b..4a45c51 100644 --- a/vite.config.ts +++ b/vite.config.ts @@ -11,27 +11,17 @@ import { svelteTesting } from "@testing-library/svelte/vite"; import { defineConfig, type Plugin, type UserConfig } from "vitest/config"; import { buildAllowedHosts, - checkDevRequest, type DevNetworkPolicy, normalizeHost, - normalizeHostHeader, tailscaleSelf, } from "./scripts/network-policy"; +import { applyViteNetworkPolicy } from "./scripts/vite-network-guard"; async function secureDevConfig(config: UserConfig, serving: boolean): Promise { // Build/test have no listener and need no subprocess or DNS effects. if (!serving) return config; const network = await detectDevNetworkPolicy(); - return { - ...config, - plugins: [devNetworkGuard(network), ...(config.plugins ?? [])], - server: { - ...config.server, - allowedHosts: [...network.allowedHosts], - // Safe reflection: our guard rejects disallowed Origins before Vite's CORS. - cors: { origin: true }, - }, - }; + return applyViteNetworkPolicy(config, network); } async function detectDevNetworkPolicy(): Promise { @@ -75,53 +65,6 @@ async function detectDevNetworkPolicy(): Promise { return { allowedHosts, allowedAddresses: new Set(["127.0.0.1", "::1", ...self.ips]) }; } -function devNetworkGuard(policy: DevNetworkPolicy): Plugin { - return { - name: "dispatch-dev-network-guard", - enforce: "pre", - configureServer(server) { - const check = (req: import("node:http").IncomingMessage) => - checkDevRequest( - { - localAddress: req.socket.localAddress, - host: req.headers.host, - origin: req.headers.origin, - }, - policy, - ); - server.middlewares.use((req, res, next) => { - const decision = check(req); - if (decision !== "ok") { - res.writeHead(decision.status, { "Content-Type": "text/plain" }); - res.end(decision.reason); - return; - } - // Vite's own check is case-sensitive and doesn't strip trailing dots. - req.headers.host = normalizeHostHeader(req.headers.host ?? ""); - next(); - }); - // Vite registers HMR before configureServer. Wrap (rather than prepend to) - // its listeners: EventEmitter otherwise continues into HMR after a denial. - const http = server.httpServer; - if (http) { - const upgrades = http.listeners("upgrade"); - http.removeAllListeners("upgrade"); - http.on("upgrade", (req, socket, head) => { - const decision = check(req); - if (decision !== "ok") { - socket.end( - `HTTP/1.1 ${decision.status} ${decision.status === 421 ? "Misdirected Request" : "Forbidden"}\r\nConnection: close\r\nContent-Length: 0\r\n\r\n`, - ); - return; - } - req.headers.host = normalizeHostHeader(req.headers.host ?? ""); - for (const upgrade of upgrades) upgrade.call(http, req, socket, head); - }); - } - }, - }; -} - /** * SPIKE (plan 03) — the Svelte-runtime import map. * -- cgit v1.2.3 From 0bdb04b0aa0c0500e68e2e985b1a2880730ad7e8 Mon Sep 17 00:00:00 2001 From: Adam Malczewski Date: Wed, 21 Oct 2026 14:46:06 +0900 Subject: fix(review-followup-fec): reconcile epoch rollback and retain reconnect backoff --- src/adapters/ws/index.test.ts | 77 ++++++++++- src/adapters/ws/index.ts | 17 ++- src/features/chat/store.svelte.ts | 68 ++++++++-- src/features/chat/store.test.ts | 176 ++++++++++++++++++++++++++ src/features/chat/test-helpers.ts | 6 + src/features/conversation-cache/cache.test.ts | 35 +++++ src/features/conversation-cache/cache.ts | 24 +++- 7 files changed, 383 insertions(+), 20 deletions(-) (limited to 'src') diff --git a/src/adapters/ws/index.test.ts b/src/adapters/ws/index.test.ts index 9d821e2..6f5e4ef 100644 --- a/src/adapters/ws/index.test.ts +++ b/src/adapters/ws/index.test.ts @@ -6,7 +6,7 @@ interface FakeSocket extends WebSocketLike { sent: string[]; resolveOpen(): void; invokeMessage(data: string): void; - invokeClose(): void; + invokeClose(code?: number): void; } function fakeSocket(): FakeSocket { @@ -44,8 +44,8 @@ function fakeSocket(): FakeSocket { invokeMessage(data: string) { onmessage?.({ data }); }, - invokeClose() { - onclose?.({ code: 1000, reason: "" }); + invokeClose(code = 1000) { + onclose?.({ code, reason: "" }); }, sent, }; @@ -405,3 +405,74 @@ describe("createSurfaceSocket", () => { expect(JSON.parse(ws.sent[0] ?? "")).toEqual({ type: "chat.send", message: "queued" }); }); }); + +describe("reconnect backoff after open", () => { + it.each([1013, 1006, 1011])("grows and caps delays across repeated early %s closes", (code) => { + vi.useFakeTimers(); + const random = vi.spyOn(Math, "random").mockReturnValue(0.5); + const sockets: FakeSocket[] = []; + const handle = createSurfaceSocket({ + url: "ws://test", + onMessage: vi.fn(), + socketFactory: () => { + const ws = fakeSocket(); + sockets.push(ws); + return ws; + }, + }); + try { + for (const delay of [500, 1000, 2000, 4000, 8000, 16000, 30000, 30000]) { + const count = sockets.length; + sockets.at(-1)?.resolveOpen(); + vi.advanceTimersByTime(100); + sockets.at(-1)?.invokeClose(code); + vi.advanceTimersByTime(delay - 1); + expect(sockets).toHaveLength(count); + vi.advanceTimersByTime(1); + expect(sockets).toHaveLength(count + 1); + } + } finally { + handle.close(); + random.mockRestore(); + vi.useRealTimers(); + } + }); + + it.each([ + "healthy", + "normal", + "going away", + ])("resets accumulated backoff after a %s connection", (mode) => { + vi.useFakeTimers(); + const random = vi.spyOn(Math, "random").mockReturnValue(0.5); + const sockets: FakeSocket[] = []; + const handle = createSurfaceSocket({ + url: "ws://test", + onMessage: vi.fn(), + socketFactory: () => { + const ws = fakeSocket(); + sockets.push(ws); + return ws; + }, + }); + try { + sockets[0]?.resolveOpen(); + sockets[0]?.invokeClose(1013); + vi.advanceTimersByTime(500); + sockets[1]?.resolveOpen(); + if (mode === "healthy") vi.advanceTimersByTime(30000); + sockets[1]?.invokeClose(mode === "normal" ? 1000 : mode === "going away" ? 1001 : 1013); + vi.advanceTimersByTime(499); + expect(sockets).toHaveLength(2); + vi.advanceTimersByTime(1); + expect(sockets).toHaveLength(3); + handle.close(); + vi.advanceTimersByTime(60000); + expect(sockets).toHaveLength(3); + } finally { + handle.close(); + random.mockRestore(); + vi.useRealTimers(); + } + }); +}); diff --git a/src/adapters/ws/index.ts b/src/adapters/ws/index.ts index 76d9e75..eeb7901 100644 --- a/src/adapters/ws/index.ts +++ b/src/adapters/ws/index.ts @@ -67,7 +67,12 @@ export function createSurfaceSocket(opts: SurfaceSocketOptions): SurfaceSocketHa let disposed = false; let reconnectAttempt = 0; let reconnectTimer: ReturnType | null = null; + let healthyTimer: ReturnType | null = null; let isOpen = false; + function clearHealthyTimer(): void { + if (healthyTimer !== null) clearTimeout(healthyTimer); + healthyTimer = null; + } const queue: string[] = []; function connect(isReconnect: boolean): void { @@ -77,7 +82,12 @@ export function createSurfaceSocket(opts: SurfaceSocketOptions): SurfaceSocketHa socket.onopen = () => { if (disposed) return; isOpen = true; - reconnectAttempt = 0; + // Opening is not proof of recovery: replay may immediately overflow the + // server's slow-consumer bound. Only a sustained open resets the streak. + healthyTimer = setTimeout(() => { + healthyTimer = null; + reconnectAttempt = 0; + }, 30_000); for (const raw of queue.splice(0)) { socket?.send(raw); } @@ -112,8 +122,10 @@ export function createSurfaceSocket(opts: SurfaceSocketOptions): SurfaceSocketHa } }; - socket.onclose = () => { + socket.onclose = (ev) => { + clearHealthyTimer(); if (disposed) return; + if (isOpen && (ev.code === 1000 || ev.code === 1001)) reconnectAttempt = 0; isOpen = false; scheduleReconnect(); }; @@ -143,6 +155,7 @@ export function createSurfaceSocket(opts: SurfaceSocketOptions): SurfaceSocketHa }, close(): void { disposed = true; + clearHealthyTimer(); if (reconnectTimer !== null) { clearTimeout(reconnectTimer); reconnectTimer = null; diff --git a/src/features/chat/store.svelte.ts b/src/features/chat/store.svelte.ts index 6cfa484..f48d441 100644 --- a/src/features/chat/store.svelte.ts +++ b/src/features/chat/store.svelte.ts @@ -339,15 +339,28 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { return; } _pendingSync = true; + // 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; - const cachedEpoch = await deps.cache.historyEpoch(deps.conversationId); - if (historyEpoch !== undefined && cachedEpoch !== historyEpoch) + 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 = - cachedEpoch === undefined ? 0 : await deps.cache.sinceSeq(deps.conversationId); + reloadWindow || cachedEpoch === undefined + ? 0 + : await deps.cache.sinceSeq(deps.conversationId); + if (disposed) break; + if (revision !== currentRevision) { + rerunRequested = true; + continue; + } const limit = since === 0 ? (reloadWindow ? initialWindowSize(chatLimit) : coldLimit) : undefined; const res = await deps.historySync( @@ -360,21 +373,43 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { rerunRequested = true; continue; } - if ( + const mismatch = (cachedEpoch !== undefined && cachedEpoch !== res.historyEpoch) || - (historyEpoch !== undefined && historyEpoch !== res.historyEpoch) - ) { - resetTranscript(res.historyEpoch); - await deps.cache.reset(deps.conversationId, res.historyEpoch); - if (since !== 0) { + (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; - const merged = await deps.cache.commit(deps.conversationId, res.chunks, res.historyEpoch); - if (disposed || historyEpoch !== res.historyEpoch) continue; if (since === 0 && (limit === undefined || res.chunks.length < limit)) earliestSeq = merged[0]?.seq ?? 0; transcript = applyHistory(transcript, merged); @@ -383,7 +418,10 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { } catch (err) { _error = err instanceof Error ? err.message : String(err); } - } while (rerunRequested && !disposed); + } while (rerunRequested && !disposed && attempts < 3); + if (rerunRequested && !disposed) { + _error = "Conversation history changed repeatedly; sync paused until the next refresh."; + } _pendingSync = false; } @@ -419,8 +457,10 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { }); if (disposed || revision !== currentRevision) return []; if (res.historyEpoch !== historyEpoch) { - resetTranscript(res.historyEpoch); - await deps.cache.reset(deps.conversationId, res.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 []; } diff --git a/src/features/chat/store.test.ts b/src/features/chat/store.test.ts index 8c94316..de84b8a 100644 --- a/src/features/chat/store.test.ts +++ b/src/features/chat/store.test.ts @@ -2023,3 +2023,179 @@ it("reloads unknown legacy cache epochs rather than merging pre-upgrade rows", a 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({ + 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; + 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({ + 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((r) => { + resolve = r; + }); + let delayNext = false; + let epoch = 5; + const sync = vi.fn().mockImplementation(async () => { + if (delayNext) { + delayNext = false; + return slow; + } + return { + chunks: epoch === 5 ? [] : [restored], + latestSeq: epoch === 5 ? 20 : 1, + historyEpoch: epoch, + }; + }); + const store = createChatStore({ + conversationId: CONV_ID, + transport: createFakeTransport().impl, + historySync: sync, + metricsSync: createFakeMetricsSync().impl, + cache, + chatLimit: 10, + }); + try { + await store.load(); + delayNext = true; + let earlier: Promise; + 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({ + 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 c1b020b..41cfb3b 100644 --- a/src/features/chat/test-helpers.ts +++ b/src/features/chat/test-helpers.ts @@ -141,6 +141,12 @@ export function createFakeCache(): FakeCache { 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) ?? []; }, diff --git a/src/features/conversation-cache/cache.test.ts b/src/features/conversation-cache/cache.test.ts index 4306d51..72b4e20 100644 --- a/src/features/conversation-cache/cache.test.ts +++ b/src/features/conversation-cache/cache.test.ts @@ -214,3 +214,38 @@ it("serializes invalidation and rejects stale epoch commits", async () => { expect(await cache.commit("c", [chunk(103)], 1)).toEqual([]); expect(await store.historyEpoch("c")).toBe(2); }); + +it("replaces a confirmed full history at a lower epoch, without trusting stale commits", async () => { + const store = createFakeStore(); + const cache = createConversationCache(store); + await cache.commit("c", [chunk(9)], 5); + expect(await cache.replace("c", [chunk(1)], 1, 5)).toBe(true); + expect(await cache.historyEpoch("c")).toBe(1); + expect(await cache.load("c")).toEqual([chunk(1)]); + expect(await cache.commit("c", [chunk(9)], 5)).toEqual([chunk(1)]); + // A delayed full response started against epoch 5 must not restore it. + expect(await cache.replace("c", [chunk(9)], 5, 5)).toBe(false); + expect(await cache.load("c")).toEqual([chunk(1)]); +}); + +it("serializes a guarded replacement behind newer invalidation", async () => { + const store = createFakeStore(); + const cache = createConversationCache(store); + await cache.commit("c", [chunk(9)], 5); + const resetting = cache.reset("c", 6); + const replacing = cache.replace("c", [chunk(1)], 1, 5); + await resetting; + expect(await replacing).toBe(false); + expect(await cache.historyEpoch("c")).toBe(6); + expect(await cache.load("c")).toEqual([]); +}); + +it("allows an empty replacement and a legacy cache only if its observed epoch still matches", async () => { + const store = createFakeStore(); + const cache = createConversationCache(store); + await store.append("c", [chunk(9)]); + expect(await cache.replace("c", [], 0, undefined)).toBe(true); + expect(await cache.sinceSeq("c")).toBe(0); + expect(await cache.historyEpoch("c")).toBe(0); + expect(await cache.replace("c", [chunk(9)], 2, undefined)).toBe(false); +}); diff --git a/src/features/conversation-cache/cache.ts b/src/features/conversation-cache/cache.ts index 431a2a7..600a6e6 100644 --- a/src/features/conversation-cache/cache.ts +++ b/src/features/conversation-cache/cache.ts @@ -6,6 +6,19 @@ export interface ConversationCache { historyEpoch(conversationId: string): Promise; reset(conversationId: string, historyEpoch: number): Promise; + /** + * Replace from a fresh full-history response, including an epoch decrease. + * Only succeeds if the cache still has the epoch observed BEFORE that request. + * Unlike monotonic reset/commit, this is for confirmed server authority, not + * broadcasts or cursor/backfill responses. False means the response is stale. + */ + replace( + conversationId: string, + incoming: readonly StoredChunk[], + historyEpoch: number, + expectedEpoch: number | undefined, + ): Promise; + /** Load all cached chunks for a conversation. */ load(conversationId: string): Promise; @@ -73,6 +86,13 @@ export function createConversationCache( const current = await store.historyEpoch(id); if (current === undefined || current < epoch) await store.reset(id, epoch); }), + replace: (id, incoming, epoch, expectedEpoch) => + exclusive(id, async () => { + if ((await store.historyEpoch(id)) !== expectedEpoch) return false; + await store.reset(id, epoch); + if (incoming.length > 0) await store.append(id, incoming); + return true; + }), async load(conversationId) { await pending.get(conversationId); return store.load(conversationId); @@ -82,7 +102,9 @@ export function createConversationCache( return exclusive(conversationId, async () => { if (epoch !== undefined) { const current = await store.historyEpoch(conversationId); - if (current !== undefined && current > epoch) return store.load(conversationId); + // Once an epoch is known, only explicit invalidation or guarded + // full-history replacement may change it (in either direction). + if (current !== undefined && current !== epoch) return store.load(conversationId); if (current !== epoch) await store.reset(conversationId, epoch); } const cached = await store.load(conversationId); -- cgit v1.2.3 From e9786bf1dbf258b68f793870d02378c390939d3d Mon Sep 17 00:00:00 2001 From: Adam Malczewski Date: Wed, 21 Oct 2026 14:52:42 +0900 Subject: fix(review-followup-fed): reconcile compaction-queued send acknowledgements --- src/adapters/ws/index.test.ts | 23 +++++ src/adapters/ws/index.ts | 7 +- src/adapters/ws/logic.test.ts | 32 ++++++ src/adapters/ws/logic.ts | 16 +++ src/app/chat-registry.svelte.ts | 1 + src/app/store.test.ts | 85 +++++++++++++++ src/app/ws-handlers.ts | 5 +- src/core/chunks/index.ts | 1 + src/core/chunks/local-echo.test.ts | 45 ++++++++ src/core/chunks/reducer.ts | 31 +++++- src/core/chunks/types.ts | 2 + src/core/wire/conformance.test.ts | 7 ++ src/core/wire/conformance.ts | 2 + src/features/chat/queued.test.ts | 207 +++++++++++++++++++++++++++++++++++++ src/features/chat/store.svelte.ts | 20 +++- src/features/chat/store.test.ts | 63 +++++++++++ 16 files changed, 536 insertions(+), 11 deletions(-) create mode 100644 src/core/chunks/local-echo.test.ts create mode 100644 src/features/chat/queued.test.ts (limited to 'src') diff --git a/src/adapters/ws/index.test.ts b/src/adapters/ws/index.test.ts index 6f5e4ef..6461493 100644 --- a/src/adapters/ws/index.test.ts +++ b/src/adapters/ws/index.test.ts @@ -476,3 +476,26 @@ describe("reconnect backoff after open", () => { } }); }); + +it("routes chat.queued only through the typed chat callback, never surface/error handling", () => { + const ws = fakeSocket(); + const onChat = vi.fn(); + const onMessage = vi.fn(); + const handle = createSurfaceSocket({ + url: "ws://test", + socketFactory: () => ws, + onChat, + onMessage, + }); + const ack = { + type: "chat.queued", + conversationId: "c1", + reason: "compaction", + messageId: "q1", + requestId: "r1", + }; + ws.invokeMessage(JSON.stringify(ack)); + expect(onChat).toHaveBeenCalledExactlyOnceWith(ack); + expect(onMessage).not.toHaveBeenCalled(); + handle.close(); +}); diff --git a/src/adapters/ws/index.ts b/src/adapters/ws/index.ts index eeb7901..25e9911 100644 --- a/src/adapters/ws/index.ts +++ b/src/adapters/ws/index.ts @@ -1,6 +1,7 @@ import type { ChatDeltaMessage, ChatErrorMessage, + ChatQueuedMessage, ConversationCompactedMessage, ConversationOpenMessage, ConversationStatusChangedMessage, @@ -24,7 +25,7 @@ export interface WebSocketLike { export interface SurfaceSocketOptions { url: string; onMessage: (msg: SurfaceServerMessage) => void; - onChat?: (msg: ChatDeltaMessage | ChatErrorMessage) => void; + onChat?: (msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage) => void; /** Broadcast when a conversation is "opened" (e.g. CLI `--open` flag). */ onConversationOpen?: (msg: ConversationOpenMessage) => void; /** Broadcast when a conversation's lifecycle status changes (active/idle/closed). */ @@ -100,8 +101,8 @@ export function createSurfaceSocket(opts: SurfaceSocketOptions): SurfaceSocketHa if (disposed) return; const msg = parseServerMessage(ev.data); if (msg !== null) { - if (msg.type === "chat.delta" || msg.type === "chat.error") { - opts.onChat?.(msg as ChatDeltaMessage | ChatErrorMessage); + if (msg.type === "chat.delta" || msg.type === "chat.error" || msg.type === "chat.queued") { + opts.onChat?.(msg as ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage); } else if (msg.type === "conversation.open") { opts.onConversationOpen?.(msg as ConversationOpenMessage); } else if (msg.type === "conversation.statusChanged") { diff --git a/src/adapters/ws/logic.test.ts b/src/adapters/ws/logic.test.ts index 4be2cd6..db6444a 100644 --- a/src/adapters/ws/logic.test.ts +++ b/src/adapters/ws/logic.test.ts @@ -577,3 +577,35 @@ it("preserves a compaction epoch and rejects absent or invalid epochs", () => { expect(parseServerMessage(JSON.stringify({ ...msg, historyEpoch }))).toBeNull(); } }); + +describe("chat.queued admission acknowledgement", () => { + const ack = { + type: "chat.queued", + conversationId: "c1", + reason: "compaction", + messageId: "q1", + }; + + it.each([undefined, "request-1"])("preserves optional requestId %s", (requestId) => { + const msg = { ...ack, ...(requestId !== undefined ? { requestId } : {}) }; + expect(parseServerMessage(JSON.stringify(msg))).toEqual(msg); + }); + + it.each([ + { conversationId: undefined }, + { conversationId: 12 }, + { messageId: undefined }, + { messageId: null }, + { reason: undefined }, + { reason: "other" }, + { requestId: null }, + { requestId: 12 }, + ])("rejects malformed acknowledgement %j", (bad) => { + expect(parseServerMessage(JSON.stringify({ ...ack, ...bad }))).toBeNull(); + }); + + it("serializes a correlated send", () => { + const msg = { type: "chat.send" as const, message: "hello", requestId: "r1" }; + expect(JSON.parse(serialize(msg))).toEqual(msg); + }); +}); diff --git a/src/adapters/ws/logic.ts b/src/adapters/ws/logic.ts index 858deaa..0efcd90 100644 --- a/src/adapters/ws/logic.ts +++ b/src/adapters/ws/logic.ts @@ -1,6 +1,7 @@ import type { ChatDeltaMessage, ChatErrorMessage, + ChatQueuedMessage, ConversationCompactedMessage, ConversationOpenMessage, ConversationStatusChangedMessage, @@ -25,6 +26,7 @@ const VALID_SERVER_TYPES = new Set([ "error", "chat.delta", "chat.error", + "chat.queued", "conversation.open", "conversation.statusChanged", "conversation.compacted", @@ -111,6 +113,20 @@ export function parseServerMessage(data: string): WsServerMessage | null { if (typeof event.type !== "string") return null; return { type: "chat.delta", event: event as unknown as ChatDeltaMessage["event"] }; } + case "chat.queued": { + if (typeof parsed.conversationId !== "string") return null; + if (typeof parsed.messageId !== "string") return null; + if (parsed.reason !== "compaction") return null; + if (parsed.requestId !== undefined && typeof parsed.requestId !== "string") return null; + const msg: ChatQueuedMessage = { + type: "chat.queued", + conversationId: parsed.conversationId, + messageId: parsed.messageId, + reason: "compaction", + ...(parsed.requestId !== undefined ? { requestId: parsed.requestId } : {}), + }; + return msg; + } case "chat.error": { if (typeof parsed.message !== "string") return null; const conversationId = parsed.conversationId; diff --git a/src/app/chat-registry.svelte.ts b/src/app/chat-registry.svelte.ts index 8e97af1..a9b28ff 100644 --- a/src/app/chat-registry.svelte.ts +++ b/src/app/chat-registry.svelte.ts @@ -78,6 +78,7 @@ export function createChatRegistry(deps: { function createChatFor(conversationId: string, model: string, workspaceId: string): ChatStore { return createChatStore({ conversationId, + createRequestId: randomId, model, workspaceId, transport: { diff --git a/src/app/store.test.ts b/src/app/store.test.ts index f257f3a..40a0fa7 100644 --- a/src/app/store.test.ts +++ b/src/app/store.test.ts @@ -2991,3 +2991,88 @@ it.each([ store.dispose(); } }); + +describe("compaction queued acknowledgement routing", () => { + it.each([ + "surface-first", + "ack-first", + "drained-before-ack", + ])("uses only authoritative queue snapshots: %s", (order) => { + const ws = fakeSocket(); + const store = createAppStore({ + socketFactory: () => ws, + fetchImpl: fakeFetchImpl(), + localStorage: createFakeStorage(), + chatFlushScheduler: null, + }); + ws.resolveOpen(); + store.send("queued text", [{ url: "data:image/png;base64,AAAA" }]); + const cid = activeConversationId(store); + const send = parseSent(ws).find((m) => (m as { type: string }).type === "chat.send") as { + requestId: string; + }; + expect(send.requestId).toEqual(expect.any(String)); + const sourceChat = store.activeChat; + const watch = store.watchConversation("other-conversation"); + watch.send("unrelated local echo"); + ws.feedSurfaceMessage({ + type: "catalog", + catalog: [{ id: "message-queue", region: "inline", title: "Queue", scope: "conversation" }], + }); + const spec = (messages: { id: string; text: string; timestamp: number }[]) => ({ + id: "message-queue", + region: "inline", + title: "Queue", + fields: [{ kind: "custom" as const, rendererId: "message-queue", payload: { messages } }], + }); + const queued = spec([{ id: "q1", text: "queued text", timestamp: 1 }]); + const drained = spec([]); + const ack = () => + ws.feedServerMessage({ + type: "chat.queued", + conversationId: cid, + reason: "compaction", + messageId: "q1", + requestId: send.requestId, + }); + if (order === "ack-first") { + ack(); + expect(store.surface("message-queue", cid)).toBeNull(); + } + ws.feedSurfaceMessage({ type: "surface", conversationId: cid, spec: queued }); + expect(store.surface("message-queue", cid)).toEqual(queued); + if (order === "drained-before-ack") { + ws.feedSurfaceMessage({ + type: "update", + update: { surfaceId: "message-queue", conversationId: cid, spec: drained }, + }); + ws.feedServerMessage({ + type: "chat.delta", + event: { type: "turn-start", conversationId: cid, turnId: "t1" }, + }); + ws.feedServerMessage({ + type: "chat.delta", + event: { type: "user-message", conversationId: cid, turnId: "t1", text: "queued text" }, + }); + } + ack(); + ack(); + expect(store.surface("message-queue", cid)).toEqual( + order === "drained-before-ack" ? drained : queued, + ); + expect(sourceChat.chunks.map((c) => c.chunk)).toEqual( + order === "drained-before-ack" ? [{ type: "text", text: "queued text" }] : [], + ); + expect(sourceChat.error).toBeNull(); + expect(watch.chunks.map((c) => c.chunk)).toEqual([ + { type: "text", text: "unrelated local echo" }, + ]); + expect(parseSent(ws).filter((m) => (m as { type: string }).type === "chat.send")).toHaveLength( + 2, + ); + expect(parseSent(ws).filter((m) => (m as { type: string }).type === "chat.queue")).toHaveLength( + 0, + ); + store.dispose(); + }); +}); diff --git a/src/app/ws-handlers.ts b/src/app/ws-handlers.ts index c76c9c1..aa1e03e 100644 --- a/src/app/ws-handlers.ts +++ b/src/app/ws-handlers.ts @@ -1,6 +1,7 @@ import type { ChatDeltaMessage, ChatErrorMessage, + ChatQueuedMessage, ConversationCompactedMessage, ConversationOpenMessage, ConversationStatusChangedMessage, @@ -52,7 +53,7 @@ export function createWsHandlers(deps: { layoutStore, tileLayoutStore, } = deps; - function handleChatMessage(msg: ChatDeltaMessage | ChatErrorMessage): void { + function handleChatMessage(msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage): void { let targetId: string | undefined; if (msg.type === "chat.delta") { targetId = msg.event.conversationId; @@ -68,6 +69,8 @@ export function createWsHandlers(deps: { } } + if (msg.type === "chat.queued") return; + // fallback: try all stores (chat.error without conversationId) for (const store of registry.chatStores.values()) { store.handleDelta(msg); diff --git a/src/core/chunks/index.ts b/src/core/chunks/index.ts index 3822a79..49381d5 100644 --- a/src/core/chunks/index.ts +++ b/src/core/chunks/index.ts @@ -7,6 +7,7 @@ export { clearGenerating, foldEvent, initialState, + removeLocalEcho, } from "./reducer"; export type { ProviderRetryView } from "./retry-banner"; export { formatRetryDelay, viewProviderRetry } from "./retry-banner"; diff --git a/src/core/chunks/local-echo.test.ts b/src/core/chunks/local-echo.test.ts new file mode 100644 index 0000000..4b5eed4 --- /dev/null +++ b/src/core/chunks/local-echo.test.ts @@ -0,0 +1,45 @@ +import { describe, expect, it } from "vitest"; +import { appendUserMessage, applyHistory, foldEvent, initialState, removeLocalEcho } from "./index"; + +describe("request-local optimistic echo", () => { + it("retains identity through turn-start tags without conflating identical sends", () => { + let state = appendUserMessage(initialState(), "same", [{ url: "data:first" }], "r1"); + state = appendUserMessage(state, "same", [{ url: "data:second" }], "r2"); + state = foldEvent(state, { type: "turn-start", conversationId: "c1", turnId: "t1" }); + state = removeLocalEcho(state, "r1"); + expect(state.provisional.map((p) => p.localRequestId)).toEqual(["r2", "r2"]); + expect(state.provisional.map((p) => p.chunk)).toEqual([ + { type: "text", text: "same" }, + { type: "image", url: "data:second" }, + ]); + expect(state.generating).toBe(true); + expect(removeLocalEcho(state, "r1")).toBe(state); + }); + + it("a live user-message confirms only its deduplicated text, not every identical local send", () => { + let state = appendUserMessage(initialState(), "same", [{ url: "data:first" }], "r1"); + state = appendUserMessage(state, "same", [{ url: "data:second" }], "r2"); + state = foldEvent(state, { type: "turn-start", conversationId: "c1", turnId: "t1" }); + state = foldEvent(state, { + type: "user-message", + conversationId: "c1", + turnId: "t1", + text: "same", + }); + state = removeLocalEcho(removeLocalEcho(state, "r1"), "r2"); + expect(state.provisional).toEqual([ + { role: "user", turnId: "t1", chunk: { type: "text", text: "same" } }, + ]); + }); + + it("does not remove committed rows even if their content is identical", () => { + let state = appendUserMessage(initialState(), "same", undefined, "r1"); + state = applyHistory(state, [{ seq: 1, role: "user", chunk: { type: "text", text: "same" } }]); + const committed = state.committed; + state = removeLocalEcho(state, "r1"); + expect(state.committed).toBe(committed); + expect(state.provisional).toEqual([]); + expect(removeLocalEcho(state, "r1")).toBe(state); + expect(removeLocalEcho(initialState(), "r1")).toEqual(initialState()); + }); +}); diff --git a/src/core/chunks/reducer.ts b/src/core/chunks/reducer.ts index 0af2ddd..53af9c1 100644 --- a/src/core/chunks/reducer.ts +++ b/src/core/chunks/reducer.ts @@ -246,11 +246,17 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState // text) emits no `user-message` and is not de-duped here. if (event.text.length === 0) return state; const run = trailingRun(state.provisional, "user"); - const alreadyEchoed = run.some((p) => p.chunk.type === "text" && p.chunk.text === event.text); - if (alreadyEchoed) { + const echoed = run.find((p) => p.chunk.type === "text" && p.chunk.text === event.text); + if (echoed !== undefined) { return { ...state, - provisional: tagTrailingUserEcho(state.provisional, event.turnId), + // Preserve the live text independently of the local send. A delayed + // queued ack may remove its images, but cannot erase this server event. + provisional: tagTrailingUserEcho(state.provisional, event.turnId).map((p) => { + if (p.chunk !== echoed.chunk || p.localRequestId === undefined) return p; + const { localRequestId: _local, ...confirmed } = p; + return confirmed; + }), generating: true, }; } @@ -463,6 +469,7 @@ export function appendUserMessage( state: TranscriptState, text: string, images?: readonly ImageInput[], + localRequestId?: string, ): TranscriptState { const provisional = flushAccumulating(state.provisional, state.accumulating); const userChunks: Chunk[] = []; @@ -484,7 +491,23 @@ export function appendUserMessage( } return { ...state, - provisional: [...provisional, ...userChunks.map((chunk) => ({ role: "user", chunk }) as const)], + provisional: [ + ...provisional, + ...userChunks.map( + (chunk) => + ({ + role: "user", + chunk, + ...(localRequestId !== undefined ? { localRequestId } : {}), + }) as const, + ), + ], accumulating: null, }; } + +/** Remove only still-local artifacts of one accepted queued send; never history/live text. */ +export function removeLocalEcho(state: TranscriptState, requestId: string): TranscriptState { + const provisional = state.provisional.filter((p) => p.localRequestId !== requestId); + return provisional.length === state.provisional.length ? state : { ...state, provisional }; +} diff --git a/src/core/chunks/types.ts b/src/core/chunks/types.ts index 949857a..f7bfe01 100644 --- a/src/core/chunks/types.ts +++ b/src/core/chunks/types.ts @@ -10,6 +10,8 @@ export interface AccumulatingChunk { /** A provisional chunk that has no authoritative seq yet. */ export interface ProvisionalChunk { + /** Local send identity only; turn-start tagging does NOT confirm this content. */ + readonly localRequestId?: string; readonly role: Role; readonly chunk: Chunk; /** diff --git a/src/core/wire/conformance.test.ts b/src/core/wire/conformance.test.ts index 1c5829b..1179fbc 100644 --- a/src/core/wire/conformance.test.ts +++ b/src/core/wire/conformance.test.ts @@ -174,6 +174,12 @@ describe("classifies every WsServerMessage type", () => { event: { type: "done" as const, conversationId: "c", turnId: "t", reason: "r" }, }, { type: "chat.error" as const, message: "e" }, + { + type: "chat.queued" as const, + conversationId: "conv-1", + reason: "compaction" as const, + messageId: "queued-1", + }, { type: "conversation.open" as const, conversationId: "c1", workspaceId: "w1" }, { type: "conversation.statusChanged" as const, @@ -198,6 +204,7 @@ describe("classifies every WsServerMessage type", () => { "error", "chat.delta", "chat.error", + "chat.queued", "conversation.open", "conversation.statusChanged", "conversation.compacted", diff --git a/src/core/wire/conformance.ts b/src/core/wire/conformance.ts index 9847081..78313f9 100644 --- a/src/core/wire/conformance.ts +++ b/src/core/wire/conformance.ts @@ -85,6 +85,8 @@ export function assertWsServerMessageExhaustive(msg: WsServerMessage): string { return "error"; case "chat.delta": return "chat.delta"; + case "chat.queued": + return "chat.queued"; case "chat.error": return "chat.error"; case "conversation.open": 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/store.svelte.ts b/src/features/chat/store.svelte.ts index f48d441..66ee5ee 100644 --- a/src/features/chat/store.svelte.ts +++ b/src/features/chat/store.svelte.ts @@ -2,6 +2,7 @@ import type { ChatDeltaMessage, ChatErrorMessage, ChatQueueCancelMessage, + ChatQueuedMessage, ChatQueueMessage, ChatSendMessage, } from "@dispatch/transport-contract"; @@ -15,6 +16,7 @@ import { initialState, initialWindowSize, normalizeChatLimit, + removeLocalEcho, restoreEarlier, selectChunks, selectGenerating, @@ -43,6 +45,8 @@ import type { ChatTransport, HistorySync, MetricsSync } from "./ports"; export interface ChatStoreDependencies { 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; @@ -142,7 +146,7 @@ export interface ChatStore { * Ignored while a turn is generating; a running turn gap keeps the later start. */ noteLastActivity(at: number): void; - handleDelta(msg: ChatDeltaMessage | ChatErrorMessage): void; + handleDelta(msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage): void; /** * Send a user message (start a turn via `chat.send`). Optimistically echoes * the text + any `images` as provisional user chunks (`[text, image, …]` in @@ -540,7 +544,7 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { cacheTimer = seedTurnGap(cacheTimer, at); }, - handleDelta(msg: ChatDeltaMessage | ChatErrorMessage): void { + handleDelta(msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage): void { if (msg.type === "chat.error") { if (msg.conversationId !== undefined && msg.conversationId !== deps.conversationId) { return; @@ -548,6 +552,14 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { _error = msg.message; 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) transcript = removeLocalEcho(transcript, msg.requestId); + return; + } if (msg.event.conversationId !== deps.conversationId) { return; } @@ -566,10 +578,12 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { flushEvents(); // Sending ends the wait since the last turn: stamp it on that turn. cacheTimer = stopTurnGap(cacheTimer, now(), () => lastTurnIdOf(metrics)); - transcript = appendUserMessage(transcript, text, images); + const requestId = deps.createRequestId(); + transcript = appendUserMessage(transcript, text, images, requestId); maybeTrim(); const msg: ChatSendMessage = { type: "chat.send", + requestId, conversationId: deps.conversationId, message: text, ...(_model !== undefined ? { model: _model } : {}), diff --git a/src/features/chat/store.test.ts b/src/features/chat/store.test.ts index de84b8a..c24592d 100644 --- a/src/features/chat/store.test.ts +++ b/src/features/chat/store.test.ts @@ -32,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, @@ -64,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, @@ -107,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, @@ -131,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, @@ -153,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, @@ -185,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, @@ -201,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, @@ -217,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, @@ -241,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, @@ -263,6 +272,7 @@ describe("createChatStore", () => { 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, @@ -298,6 +308,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -318,6 +329,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -340,6 +352,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -362,6 +375,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -387,6 +401,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -408,6 +423,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -433,6 +449,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -462,6 +479,7 @@ describe("createChatStore", () => { historySync.returnChunks = [makeStoredChunk(3, "assistant")]; const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -490,6 +508,7 @@ describe("createChatStore", () => { historySync.returnChunks = [makeStoredChunk(1, "assistant")]; const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -511,6 +530,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -543,6 +563,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -587,6 +608,7 @@ describe("createChatStore", () => { }; const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: slowHistorySync, @@ -621,6 +643,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -666,6 +689,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, @@ -690,6 +714,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -713,6 +738,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -743,6 +769,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -763,6 +790,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -787,6 +815,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -825,6 +854,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -886,6 +916,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -930,6 +961,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1002,6 +1034,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1068,6 +1101,7 @@ describe("createChatStore", () => { ]; const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1092,6 +1126,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1123,6 +1158,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1147,6 +1183,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1181,6 +1218,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1227,6 +1265,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, @@ -1281,6 +1320,7 @@ describe("createChatStore", () => { const cache = createFakeCache(); let atBottom = false; const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1319,6 +1359,7 @@ describe("createChatStore", () => { ); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1348,6 +1389,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, @@ -1383,6 +1425,7 @@ describe("createChatStore", () => { historySync.returnChunks = [makeStoredChunk(3)]; const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1407,6 +1450,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, @@ -1444,6 +1488,7 @@ describe("createChatStore", () => { ); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1476,6 +1521,7 @@ describe("createChatStore", () => { ); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1507,6 +1553,7 @@ describe("createChatStore", () => { ); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1539,6 +1586,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1576,6 +1624,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, @@ -1607,6 +1656,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, @@ -1634,6 +1684,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1673,6 +1724,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, @@ -1700,6 +1752,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1729,6 +1782,7 @@ describe("createChatStore", () => { const metricsSync = createFakeMetricsSync(); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: transport.impl, historySync: historySync.impl, @@ -1750,6 +1804,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, @@ -1831,6 +1886,7 @@ describe("createChatStore — batched delta application (scheduleFlush)", () => describe("createChatStore — cacheTimer", () => { function makeStore(now: () => number) { return createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: createFakeTransport().impl, historySync: createFakeHistorySync().impl, @@ -1940,6 +1996,7 @@ describe("history epochs", () => { 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, @@ -1980,6 +2037,7 @@ it("a compaction broadcast during an in-flight fetch discards the old response", .mockImplementation(history.impl); const cache = createFakeCache(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: createFakeTransport().impl, historySync: sync, @@ -2006,6 +2064,7 @@ it("reloads unknown legacy cache epochs rather than merging pre-upgrade rows", a const history = createFakeHistorySync(); history.returnChunks = [makeStoredChunk(101)]; const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: createFakeTransport().impl, historySync: history.impl, @@ -2037,6 +2096,7 @@ describe("server epoch rollback", () => { const history = createFakeHistorySync(); history.historyEpoch = 5; const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: createFakeTransport().impl, historySync: history.impl, @@ -2083,6 +2143,7 @@ describe("server epoch rollback", () => { return { chunks: [], latestSeq: 0, historyEpoch: calls - 1 }; }; store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: createFakeTransport().impl, historySync: sync, @@ -2140,6 +2201,7 @@ it.each([ }; }); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: createFakeTransport().impl, historySync: sync, @@ -2182,6 +2244,7 @@ it("replaces an empty high-epoch cache with the server's empty lower epoch in on await cache.reset(CONV_ID, 8); const history = createFakeHistorySync(); const store = createChatStore({ + createRequestId: () => crypto.randomUUID(), conversationId: CONV_ID, transport: createFakeTransport().impl, historySync: history.impl, -- cgit v1.2.3 From ac43e92d174ed1f081c387944e9044643ee9efc2 Mon Sep 17 00:00:00 2001 From: Adam Malczewski Date: Wed, 21 Oct 2026 15:36:54 +0900 Subject: fix(review-followup-fee): recover omitted chat payloads through history --- src/adapters/ws/index.test.ts | 37 +++ src/adapters/ws/index.ts | 20 +- src/adapters/ws/logic.test.ts | 29 ++ src/adapters/ws/logic.ts | 14 + src/app/store.test.ts | 53 +++ src/app/views/ChatPane.svelte | 1 + src/app/ws-handlers.ts | 7 +- src/core/chunks/groups.test.ts | 22 ++ src/core/chunks/groups.ts | 14 +- src/core/chunks/index.ts | 1 + src/core/chunks/local-echo.test.ts | 2 +- src/core/chunks/reconcile.test.ts | 143 ++++++++ src/core/chunks/reconcile.ts | 54 +++ src/core/chunks/reducer.test.ts | 26 +- src/core/chunks/reducer.ts | 174 +++++++++- src/core/chunks/selectors.test.ts | 30 +- src/core/chunks/selectors.ts | 13 +- src/core/chunks/types.ts | 8 + src/core/wire/conformance.test.ts | 38 ++- src/core/wire/conformance.ts | 2 + src/features/chat/logic/tool-summary.ts | 8 +- src/features/chat/recovery.test.ts | 529 ++++++++++++++++++++++++++++++ src/features/chat/store.svelte.ts | 128 +++++++- src/features/chat/ui/ChatView.svelte | 10 +- src/features/chat/ui/ChatView.test.ts | 44 +++ src/features/heartbeat/ui/RunModal.svelte | 1 + 26 files changed, 1364 insertions(+), 44 deletions(-) create mode 100644 src/core/chunks/reconcile.test.ts create mode 100644 src/core/chunks/reconcile.ts create mode 100644 src/features/chat/recovery.test.ts (limited to 'src') diff --git a/src/adapters/ws/index.test.ts b/src/adapters/ws/index.test.ts index 6461493..e4e4ba4 100644 --- a/src/adapters/ws/index.test.ts +++ b/src/adapters/ws/index.test.ts @@ -499,3 +499,40 @@ it("routes chat.queued only through the typed chat callback, never surface/error expect(onMessage).not.toHaveBeenCalled(); handle.close(); }); + +it("routes resync notices to chat without reconnect/resend or surface errors", () => { + vi.useFakeTimers(); + const ws = fakeSocket(); + const onChat = vi.fn(); + const onMessage = vi.fn(); + const onReopen = vi.fn(); + const factory = vi.fn(() => ws); + const handle = createSurfaceSocket({ + url: "ws://test", + socketFactory: factory, + onChat, + onMessage, + onReopen, + }); + try { + ws.resolveOpen(); + const notice = { + type: "chat.resync-required", + conversationId: "c1", + turnId: "t1", + reason: "replay-limit", + }; + ws.invokeMessage(JSON.stringify(notice)); + ws.invokeMessage(JSON.stringify(notice)); + vi.advanceTimersByTime(1000); + expect(onChat).toHaveBeenCalledTimes(2); + expect(onChat).toHaveBeenCalledWith(notice); + expect(onMessage).not.toHaveBeenCalled(); + expect(onReopen).not.toHaveBeenCalled(); + expect(factory).toHaveBeenCalledTimes(1); + expect(ws.sent).toEqual([]); + } finally { + handle.close(); + vi.useRealTimers(); + } +}); diff --git a/src/adapters/ws/index.ts b/src/adapters/ws/index.ts index 25e9911..b1e17cf 100644 --- a/src/adapters/ws/index.ts +++ b/src/adapters/ws/index.ts @@ -2,6 +2,7 @@ import type { ChatDeltaMessage, ChatErrorMessage, ChatQueuedMessage, + ChatResyncRequiredMessage, ConversationCompactedMessage, ConversationOpenMessage, ConversationStatusChangedMessage, @@ -25,7 +26,9 @@ export interface WebSocketLike { export interface SurfaceSocketOptions { url: string; onMessage: (msg: SurfaceServerMessage) => void; - onChat?: (msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage) => void; + onChat?: ( + msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage | ChatResyncRequiredMessage, + ) => void; /** Broadcast when a conversation is "opened" (e.g. CLI `--open` flag). */ onConversationOpen?: (msg: ConversationOpenMessage) => void; /** Broadcast when a conversation's lifecycle status changes (active/idle/closed). */ @@ -101,8 +104,19 @@ export function createSurfaceSocket(opts: SurfaceSocketOptions): SurfaceSocketHa if (disposed) return; const msg = parseServerMessage(ev.data); if (msg !== null) { - if (msg.type === "chat.delta" || msg.type === "chat.error" || msg.type === "chat.queued") { - opts.onChat?.(msg as ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage); + if ( + msg.type === "chat.delta" || + msg.type === "chat.error" || + msg.type === "chat.queued" || + msg.type === "chat.resync-required" + ) { + opts.onChat?.( + msg as + | ChatDeltaMessage + | ChatErrorMessage + | ChatQueuedMessage + | ChatResyncRequiredMessage, + ); } else if (msg.type === "conversation.open") { opts.onConversationOpen?.(msg as ConversationOpenMessage); } else if (msg.type === "conversation.statusChanged") { diff --git a/src/adapters/ws/logic.test.ts b/src/adapters/ws/logic.test.ts index db6444a..4bc2aad 100644 --- a/src/adapters/ws/logic.test.ts +++ b/src/adapters/ws/logic.test.ts @@ -609,3 +609,32 @@ describe("chat.queued admission acknowledgement", () => { expect(JSON.parse(serialize(msg))).toEqual(msg); }); }); + +describe("chat.resync-required", () => { + it.each(["replay-limit", "payload-limit"])("parses %s with optional turn identity", (reason) => { + const msg = { type: "chat.resync-required", conversationId: "c1", reason }; + expect(parseServerMessage(JSON.stringify(msg))).toEqual(msg); + expect(parseServerMessage(JSON.stringify({ ...msg, turnId: "t1" }))).toEqual({ + ...msg, + turnId: "t1", + }); + }); + it.each([ + {}, + { conversationId: null }, + { conversationId: 1 }, + { reason: "unknown" }, + { reason: null }, + { turnId: null }, + { turnId: 42 }, + ])("rejects malformed notice %j", (bad) => { + const msg = { + type: "chat.resync-required", + conversationId: "c1", + reason: "replay-limit", + ...bad, + }; + if (Object.keys(bad).length === 0) delete (msg as { conversationId?: unknown }).conversationId; + expect(parseServerMessage(JSON.stringify(msg))).toBeNull(); + }); +}); diff --git a/src/adapters/ws/logic.ts b/src/adapters/ws/logic.ts index 0efcd90..abc0464 100644 --- a/src/adapters/ws/logic.ts +++ b/src/adapters/ws/logic.ts @@ -2,6 +2,7 @@ import type { ChatDeltaMessage, ChatErrorMessage, ChatQueuedMessage, + ChatResyncRequiredMessage, ConversationCompactedMessage, ConversationOpenMessage, ConversationStatusChangedMessage, @@ -27,6 +28,7 @@ const VALID_SERVER_TYPES = new Set([ "chat.delta", "chat.error", "chat.queued", + "chat.resync-required", "conversation.open", "conversation.statusChanged", "conversation.compacted", @@ -113,6 +115,18 @@ export function parseServerMessage(data: string): WsServerMessage | null { if (typeof event.type !== "string") return null; return { type: "chat.delta", event: event as unknown as ChatDeltaMessage["event"] }; } + case "chat.resync-required": { + if (typeof parsed.conversationId !== "string") return null; + if (parsed.turnId !== undefined && typeof parsed.turnId !== "string") return null; + if (parsed.reason !== "replay-limit" && parsed.reason !== "payload-limit") return null; + const msg: ChatResyncRequiredMessage = { + type: "chat.resync-required", + conversationId: parsed.conversationId, + reason: parsed.reason, + ...(parsed.turnId !== undefined ? { turnId: parsed.turnId } : {}), + }; + return msg; + } case "chat.queued": { if (typeof parsed.conversationId !== "string") return null; if (typeof parsed.messageId !== "string") return null; diff --git a/src/app/store.test.ts b/src/app/store.test.ts index 40a0fa7..33f93bb 100644 --- a/src/app/store.test.ts +++ b/src/app/store.test.ts @@ -3076,3 +3076,56 @@ describe("compaction queued acknowledgement routing", () => { store.dispose(); }); }); + +it("routes omitted replay only to the matching watch/tab without WS resend or an error banner", async () => { + const ws = fakeSocket(); + const store = createAppStore({ + socketFactory: () => ws, + fetchImpl: fakeFetchImpl(), + indexedDB: new IDBFactory(), + localStorage: createFakeStorage(), + chatFlushScheduler: null, + }); + try { + ws.resolveOpen(); + const tab = store.activeChat; + const watch = store.watchConversation("omitted-watch"); + const sentBefore = ws.sent.length; + ws.feedServerMessage({ + type: "chat.resync-required", + conversationId: "unknown", + turnId: "unknown-turn", + reason: "replay-limit", + }); + expect(watch.recovering).toBe(false); + expect(tab.recovering).toBe(false); + ws.feedServerMessage({ + type: "chat.resync-required", + conversationId: "omitted-watch", + turnId: "t1", + reason: "replay-limit", + }); + expect(watch.recovering).toBe(true); + expect(watch.generating).toBe(true); + expect(tab.recovering).toBe(false); + expect(watch.error).toBeNull(); + expect(tab.error).toBeNull(); + expect(store.fatalError).toBeNull(); + expect(ws.sent).toHaveLength(sentBefore); + ws.feedServerMessage({ + type: "chat.delta", + event: { + type: "turn-ended", + conversationId: "omitted-watch", + turnId: "t1", + reason: "aborted", + }, + }); + await vi.waitFor(() => expect(watch.pendingSync).toBe(false)); + expect(watch.recovering).toBe(false); + expect(watch.generating).toBe(false); + expect(ws.sent).toHaveLength(sentBefore); + } finally { + store.dispose(); + } +}); diff --git a/src/app/views/ChatPane.svelte b/src/app/views/ChatPane.svelte index d1630af..62b9344 100644 --- a/src/app/views/ChatPane.svelte +++ b/src/app/views/ChatPane.svelte @@ -80,6 +80,7 @@ {#key store.activeConversationId} { expect(groups.map((g) => g.kind)).toEqual(["single", "tool"]); }); }); + +it("keeps windowed committed orphan calls in Show earlier even while another payload recovers", () => { + const groups = groupRenderedChunks( + [ + { + seq: 50, + role: "tool", + provisional: false, + chunk: { + type: "tool-result", + toolCallId: "older", + toolName: "run_shell", + content: "output", + isError: false, + }, + }, + ], + true, + ); + expect(groups[0]).toMatchObject({ kind: "tool", entry: { call: null } }); + if (groups[0]?.kind === "tool") expect(groups[0].entry.recovering).toBeUndefined(); +}); diff --git a/src/core/chunks/groups.ts b/src/core/chunks/groups.ts index c8493cc..6e0f8ea 100644 --- a/src/core/chunks/groups.ts +++ b/src/core/chunks/groups.ts @@ -3,6 +3,7 @@ import type { RenderedChunk } from "./types"; /** One tool call and its response. Null call = unloaded history; null result = pending. */ export interface ToolEntry { + readonly recovering?: boolean; readonly toolCallId: string; readonly toolName: string; readonly call: ToolCallChunk | null; @@ -26,7 +27,10 @@ export type RenderGroup = * the stable t render keys are always unique, even on corrupt replay. * Pure: input → output; no DOM, effects, or Svelte. */ -export function groupRenderedChunks(rendered: readonly RenderedChunk[]): readonly RenderGroup[] { +export function groupRenderedChunks( + rendered: readonly RenderedChunk[], + recovering = false, +): readonly RenderGroup[] { const calls = new Map(); const results = new Map(); for (const rc of rendered) { @@ -55,7 +59,13 @@ export function groupRenderedChunks(rendered: readonly RenderedChunk[]): readonl const turnId = callRow?.turnId ?? resultRow?.turnId; groups.push({ kind: "tool", - entry: { toolCallId: c.toolCallId, toolName: call?.toolName ?? c.toolName, call, result }, + entry: { + toolCallId: c.toolCallId, + toolName: call?.toolName ?? c.toolName, + call, + result, + ...(recovering && call === null && resultRow?.provisional ? { recovering: true } : {}), + }, provisional: Boolean(callRow?.provisional || resultRow?.provisional), ...(stepId !== undefined ? { stepId } : {}), ...(turnId !== undefined ? { turnId } : {}), diff --git a/src/core/chunks/index.ts b/src/core/chunks/index.ts index 49381d5..03d687c 100644 --- a/src/core/chunks/index.ts +++ b/src/core/chunks/index.ts @@ -7,6 +7,7 @@ export { clearGenerating, foldEvent, initialState, + markHistoryRecovering, removeLocalEcho, } from "./reducer"; export type { ProviderRetryView } from "./retry-banner"; diff --git a/src/core/chunks/local-echo.test.ts b/src/core/chunks/local-echo.test.ts index 4b5eed4..565017c 100644 --- a/src/core/chunks/local-echo.test.ts +++ b/src/core/chunks/local-echo.test.ts @@ -28,7 +28,7 @@ describe("request-local optimistic echo", () => { }); state = removeLocalEcho(removeLocalEcho(state, "r1"), "r2"); expect(state.provisional).toEqual([ - { role: "user", turnId: "t1", chunk: { type: "text", text: "same" } }, + { role: "user", turnId: "t1", opening: true, chunk: { type: "text", text: "same" } }, ]); }); diff --git a/src/core/chunks/reconcile.test.ts b/src/core/chunks/reconcile.test.ts new file mode 100644 index 0000000..3d90a51 --- /dev/null +++ b/src/core/chunks/reconcile.test.ts @@ -0,0 +1,143 @@ +import type { AgentEvent, StepId, StoredChunk } from "@dispatch/wire"; +import { describe, expect, it } from "vitest"; +import { applyHistory, foldEvent, initialState, markHistoryRecovering } from "./reducer"; +import { selectChunks } from "./selectors"; +import type { TranscriptState } from "./types"; + +const stepId = "t1#1" as StepId; +function fold(s: TranscriptState, e: AgentEvent) { + return foldEvent(s, e); +} +const call = (toolCallId = "a", turnId = "t1", step = stepId): AgentEvent => ({ + type: "tool-call", + conversationId: "c1", + turnId, + stepId: step, + toolCallId, + toolName: "run_shell", + input: "partial", +}); +const result = (toolCallId = "a"): AgentEvent => ({ + type: "tool-result", + conversationId: "c1", + turnId: "t1", + stepId, + toolCallId, + toolName: "run_shell", + content: "partial", + isError: false, +}); +const stored: StoredChunk[] = [ + { seq: 50, role: "assistant", turnId: "t1", chunk: { type: "text", text: "full" } }, + { + seq: 51, + role: "assistant", + turnId: "t1", + chunk: { + type: "tool-call", + stepId, + toolCallId: "a", + toolName: "run_shell", + input: { command: "full" }, + }, + }, + { + seq: 52, + role: "tool", + turnId: "t1", + chunk: { + type: "tool-result", + stepId, + toolCallId: "a", + toolName: "run_shell", + content: "full result", + isError: false, + }, + }, +]; + +describe("history recovery identities", () => { + it.each([ + "before-live", + "after-live", + ])("reconciles committed tool/text identity %s, not payload equality", (order) => { + let s = markHistoryRecovering(initialState(), "t1"); + if (order === "before-live") s = applyHistory(s, stored); + s = fold(s, { type: "text-delta", conversationId: "c1", turnId: "t1", delta: "short" }); + s = fold(s, call()); + s = fold(s, result()); + if (order === "after-live") s = applyHistory(s, stored); + expect(selectChunks(s).map((c) => c.chunk)).toEqual(stored.map((c) => c.chunk)); + // Idempotent refresh; different step's identical text must remain live. + s = fold(s, { type: "text-delta", conversationId: "c1", turnId: "t1", delta: "full" }); + s = applyHistory(s, stored); + expect(selectChunks(s).at(-1)).toMatchObject({ + provisional: true, + streaming: true, + chunk: { type: "text", text: "full" }, + }); + }); + + it("does not conflate call/result, another turn, or another step's tool identity", () => { + let s = markHistoryRecovering(initialState(), "t1"); + s = fold(s, call("a", "other")); + s = fold(s, call("a", "t1", "other-step" as StepId)); + s = fold(s, result("a")); + s = applyHistory(s, stored); + expect(s.provisional).toHaveLength(2); + expect(s.provisional.map((p) => p.turnId)).toEqual(["other", "t1"]); + expect(s.provisional.every((p) => p.chunk.type === "tool-call")).toBe(true); + }); + + it("terminal sync removes only that turn and keeps optimistic/unrelated/carried content", () => { + let s = markHistoryRecovering(initialState(), "t1"); + s = fold(s, call()); + s = fold(s, { type: "turn-sealed", conversationId: "c1", turnId: "t1" }); + s = fold(s, { type: "turn-start", conversationId: "c1", turnId: "t2" }); + s = fold(s, { type: "text-delta", conversationId: "c1", turnId: "t2", delta: "queued tail" }); + s = applyHistory(s, stored, ["t1"]); + expect(s.generating).toBe(true); + expect(s.currentTurnId).toBe("t2"); + expect(s.accumulating?.text).toBe("queued tail"); + expect(s.historyRecoveryTurnIds).toEqual([]); + }); + + it("does not resurrect watermark-hidden history; older content stays available to Show earlier", () => { + let s = markHistoryRecovering({ ...initialState(), hiddenBeforeSeq: 51 }, "t1"); + s = fold(s, call()); + s = applyHistory(s, stored); + expect(s.committed.map((c) => c.seq)).toEqual([51, 52]); + expect(s.provisional).toEqual([]); + expect(s.hiddenBeforeSeq).toBe(51); + }); +}); + +it("keeps identical Stop→carry openings as distinct turn identities", () => { + let s = markHistoryRecovering(initialState(), "t1"); + s = fold(s, { type: "user-message", conversationId: "c1", turnId: "t1", text: "same" }); + s = fold(s, { type: "turn-ended", conversationId: "c1", turnId: "t1", reason: "aborted" }); + s = fold(s, { type: "user-message", conversationId: "c1", turnId: "t2", text: "same" }); + s = fold(s, { type: "turn-start", conversationId: "c1", turnId: "t2" }); + s = applyHistory( + s, + [{ seq: 1, role: "user", turnId: "t1", chunk: { type: "text", text: "same" } }], + ["t1"], + ); + expect(selectChunks(s).map((c) => [c.turnId, c.provisional])).toEqual([ + ["t1", false], + ["t2", true], + ]); + expect(s.generating).toBe(true); +}); + +it("reconciles steering by its completed tool batch, not repeated user text", () => { + let s = markHistoryRecovering(initialState(), "t1"); + s = fold(s, result()); + s = fold(s, { type: "steering", conversationId: "c1", turnId: "t1", text: "short" }); + s = applyHistory(s, [ + ...stored, + { seq: 53, role: "user", turnId: "t1", chunk: { type: "text", text: "full steering" } }, + ]); + expect(s.provisional).toEqual([]); + expect(selectChunks(s).at(-1)?.chunk).toEqual({ type: "text", text: "full steering" }); +}); diff --git a/src/core/chunks/reconcile.ts b/src/core/chunks/reconcile.ts new file mode 100644 index 0000000..277d25b --- /dev/null +++ b/src/core/chunks/reconcile.ts @@ -0,0 +1,54 @@ +import type { StoredChunk } from "@dispatch/wire"; +import type { ProvisionalChunk, TranscriptState } from "./types"; + +/** + * A persisted tool call proves its entire assistant step committed atomically. + * Text/thinking wire chunks have no stepId: their provisional provenance is + * learned from tool-call/step-complete, never from text equality. This also + * handles a window starting midway through the step and an omitted prefix. + * Also usable when HTTP finished before replay/live flush. + */ +export function historyBackedProvisional( + state: TranscriptState, + committed: readonly StoredChunk[] = state.committed, +): ReadonlySet { + const backed = new Set(); + if (!state.historyRecoveryTurnIds?.length) return backed; + const steps = new Set(); + const steeringAnchors = new Set(); + let lastResult: StoredChunk | undefined; + const toolRows = new Map>(); + for (const row of committed) { + const c = row.chunk; + if (row.role === "assistant" || row.turnId !== lastResult?.turnId) lastResult = undefined; + if (c.type === "tool-result") lastResult = row; + if (row.role === "user" && c.type === "text" && lastResult?.chunk.type === "tool-result") { + steeringAnchors.add(`${row.turnId}\0${lastResult.chunk.toolCallId}`); + } + if (c.type === "tool-call" || c.type === "tool-result") { + const key = `${row.turnId}\0${c.type}\0${c.toolCallId}`; + const ids = toolRows.get(key) ?? new Set(); + ids.add(c.stepId); + toolRows.set(key, ids); + if (c.type === "tool-call" && c.stepId !== undefined) steps.add(`${row.turnId}\0${c.stepId}`); + } + } + for (const p of state.provisional) { + if (p.turnId === undefined || !state.historyRecoveryTurnIds?.includes(p.turnId)) continue; + const c = p.chunk; + if (c.type === "tool-call" || c.type === "tool-result") { + const ids = toolRows.get(`${p.turnId}\0${c.type}\0${c.toolCallId}`); + if (ids !== undefined && (c.stepId === undefined || ids.has(undefined) || ids.has(c.stepId))) + backed.add(p); + } else if (p.afterToolCallId !== undefined && p.role === "user") { + if (steeringAnchors.has(`${p.turnId}\0${p.afterToolCallId}`)) backed.add(p); + } else if (p.opening && c.type === "text") { + // The opening is the first user row of this turn; steering is NOT opening. + const first = committed.find((row) => row.turnId === p.turnId); + if (first?.role === "user" && first.chunk.type === "text") backed.add(p); + } else if (p.stepId !== undefined && p.role === "assistant") { + if (steps.has(`${p.turnId}\0${p.stepId}`)) backed.add(p); + } + } + return backed; +} diff --git a/src/core/chunks/reducer.test.ts b/src/core/chunks/reducer.test.ts index ae0aa73..22b94d4 100644 --- a/src/core/chunks/reducer.test.ts +++ b/src/core/chunks/reducer.test.ts @@ -1053,7 +1053,7 @@ describe("turnId provenance on live chunks", () => { }); describe("foldEvent — additive wire compatibility", () => { - it.each(["turn-ended", "future-event"])("preserves transcript identity for %s", (type) => { + it.each(["future-event"])("preserves transcript identity for %s", (type) => { const state = foldEvent(foldEvent(initialState(), turnStart("t1")), textDelta("t1", "partial")); const before = structuredClone(state); const event = { type, conversationId: "c1", turnId: "t1", reason: "sealed" }; @@ -1061,3 +1061,27 @@ describe("foldEvent — additive wire compatibility", () => { expect(state).toEqual(before); }); }); + +describe("foldEvent — authoritative turn terminal", () => { + it.each([ + "sealed", + "aborted", + "error", + "rejected", + ] as const)("ends generation on %s", (reason) => { + const before = foldEvent( + foldEvent(initialState(), turnStart("t1")), + textDelta("t1", "partial"), + ); + const next = foldEvent(before, { + type: "turn-ended", + conversationId: "c1", + turnId: "t1", + reason, + }); + expect(next.generating).toBe(false); + expect(next.accumulating).toBeNull(); + expect(selectChunks(next).map((c) => c.chunk)).toEqual([{ type: "text", text: "partial" }]); + expect(before.accumulating?.text).toBe("partial"); + }); +}); diff --git a/src/core/chunks/reducer.ts b/src/core/chunks/reducer.ts index 53af9c1..ee5858f 100644 --- a/src/core/chunks/reducer.ts +++ b/src/core/chunks/reducer.ts @@ -1,5 +1,6 @@ import type { AgentEvent, Chunk, ImageInput, Role, StoredChunk } from "@dispatch/wire"; import { assertChunkExhaustive } from "../wire/conformance"; +import { historyBackedProvisional } from "./reconcile"; import type { AccumulatingChunk, ProvisionalChunk, TranscriptState } from "./types"; /** The initial empty transcript state. */ @@ -40,7 +41,13 @@ function flushAccumulating( if (acc === null) return provisional; const chunk: Chunk = acc.kind === "text" ? { type: "text", text: acc.text } : { type: "thinking", text: acc.text }; - return [...provisional, withTurn({ role: "assistant", chunk }, acc.turnId)]; + return [ + ...provisional, + withTurn( + { role: "assistant", chunk, ...(acc.stepId !== undefined ? { stepId: acc.stepId } : {}) }, + acc.turnId, + ), + ]; } /** Attach a `turnId` to a provisional chunk (omitted when unknown). */ @@ -129,8 +136,8 @@ function trailingRun( /** * Merge authoritative seq-keyed chunks into the committed history. * Dedupes by seq (new wins), keeps seq-monotonic order, idempotent. - * When sealedTurnId is set, drops all provisional chunks (now superseded) - * and clears sealedTurnId. + * A sealed/finally-synced turn supersedes only its own provisional chunks; + * unrelated turns and an already-running queued carry remain live. * * Chunks below the chat-limit unload watermark (`hiddenBeforeSeq`) are * REJECTED: a full-cache or tail merge must not resurrect what the trim @@ -139,6 +146,7 @@ function trailingRun( export function applyHistory( state: TranscriptState, chunks: readonly StoredChunk[], + finishedTurnIds: readonly string[] = [], ): TranscriptState { const seqMap = new Map(); for (const c of state.committed) seqMap.set(c.seq, c); @@ -149,14 +157,49 @@ export function applyHistory( seqMap.set(c.seq, c); } const committed = Array.from(seqMap.values()).sort((a, b) => a.seq - b.seq); + const identityRows = new Map(seqMap); + for (const c of chunks) identityRows.set(c.seq, c); + const backed = historyBackedProvisional( + state, + [...identityRows.values()].sort((a, b) => a.seq - b.seq), + ); + const finished = new Set(finishedTurnIds); + if (state.sealedTurnId !== null) finished.add(state.sealedTurnId); + state = { + ...state, + provisional: state.provisional.filter((p) => !backed.has(p) && !finished.has(p.turnId ?? "")), + accumulating: + state.accumulating?.turnId !== undefined && finished.has(state.accumulating.turnId) + ? null + : state.accumulating, + }; if (state.sealedTurnId !== null) { + // Legacy (non-recovery) histories may carry no turn provenance at all. + // Retain their seal cleanup, but never erase an already-live queued carry. + const legacySeal = + (state.historyRecoveryTurnIds?.length ?? 0) === 0 && finishedTurnIds.length === 0; return { ...state, committed, - provisional: [], - accumulating: null, + provisional: legacySeal + ? state.provisional.filter( + (p) => + state.generating && + p.turnId === state.currentTurnId && + p.turnId !== state.sealedTurnId, + ) + : state.provisional, + accumulating: + legacySeal && state.accumulating?.turnId === state.sealedTurnId ? null : state.accumulating, sealedTurnId: null, + ...(state.historyRecoveryTurnIds !== undefined + ? { + historyRecoveryTurnIds: state.historyRecoveryTurnIds.filter( + (id) => !finishedTurnIds.includes(id), + ), + } + : {}), }; } @@ -167,7 +210,14 @@ export function applyHistory( // match the trailing provisional user-run against the trailing committed // user-run by content equality and drop the whole echo when it is fully // backed. Leaves the accumulating (streaming) chunk untouched. - if (addedNew && state.generating && state.provisional.length > 0) { + if ( + addedNew && + state.generating && + state.provisional.length > 0 && + !state.provisional.some( + (p) => p.turnId !== undefined && state.historyRecoveryTurnIds?.includes(p.turnId), + ) + ) { const provRun = trailingRun(state.provisional, "user"); if (provRun.length > 0) { const commRun = trailingRun(committed, "user"); @@ -179,7 +229,11 @@ export function applyHistory( commRun.length >= provRun.length && provRun.every((p, i) => { const c = commRun[i]; - return c !== undefined && chunkContentEquals(p.chunk, c.chunk); + return ( + c !== undefined && + (c.turnId === undefined || p.turnId === undefined || c.turnId === p.turnId) && + chunkContentEquals(p.chunk, c.chunk) + ); }); if (fullyBacked) { const dropStart = state.provisional.length - provRun.length; @@ -189,7 +243,17 @@ export function applyHistory( } } - return { ...state, committed }; + return { + ...state, + committed, + ...(state.historyRecoveryTurnIds !== undefined + ? { + historyRecoveryTurnIds: state.historyRecoveryTurnIds.filter( + (id) => !finishedTurnIds.includes(id), + ), + } + : {}), + }; } /** @@ -227,7 +291,11 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState case "turn-start": return { ...state, - provisional: tagTrailingUserEcho(state.provisional, event.turnId), + provisional: tagTrailingUserEcho( + flushAccumulating(state.provisional, state.accumulating), + event.turnId, + ), + accumulating: null, currentTurnId: event.turnId, generating: true, }; @@ -246,16 +314,21 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState // text) emits no `user-message` and is not de-duped here. if (event.text.length === 0) return state; const run = trailingRun(state.provisional, "user"); - const echoed = run.find((p) => p.chunk.type === "text" && p.chunk.text === event.text); + const echoed = run.find( + (p) => + (p.turnId === undefined || p.turnId === event.turnId) && + p.chunk.type === "text" && + p.chunk.text === event.text, + ); if (echoed !== undefined) { return { ...state, // Preserve the live text independently of the local send. A delayed // queued ack may remove its images, but cannot erase this server event. provisional: tagTrailingUserEcho(state.provisional, event.turnId).map((p) => { - if (p.chunk !== echoed.chunk || p.localRequestId === undefined) return p; + if (p.chunk !== echoed.chunk) return p; const { localRequestId: _local, ...confirmed } = p; - return confirmed; + return { ...confirmed, opening: true }; }), generating: true, }; @@ -265,7 +338,12 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState ...state, provisional: [ ...provisional, - { role: "user", chunk: { type: "text", text: event.text }, turnId: event.turnId }, + { + role: "user", + chunk: { type: "text", text: event.text }, + turnId: event.turnId, + opening: true, + }, ], accumulating: null, generating: true, @@ -274,7 +352,7 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState case "text-delta": { const acc = state.accumulating; - if (acc !== null && acc.kind === "text") { + if (acc !== null && acc.kind === "text" && acc.turnId === event.turnId) { return { ...state, accumulating: { ...acc, text: acc.text + event.delta }, @@ -292,7 +370,7 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState case "reasoning-delta": { const acc = state.accumulating; - if (acc !== null && acc.kind === "thinking") { + if (acc !== null && acc.kind === "thinking" && acc.turnId === event.turnId) { return { ...state, accumulating: { ...acc, text: acc.text + event.delta }, @@ -361,8 +439,24 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState return { ...state, latestUsage: event.usage }; case "step-complete": - // Timing metadata — no content chunk; handled by the telemetry reducer. - return state; + // A persistence boundary may lag this notification. Stamp provenance and + // finish the generation chunk, but keep it until history really backs it. + return { + ...state, + provisional: flushAccumulating(state.provisional, state.accumulating), + accumulating: null, + }; + + case "turn-ended": + return { + ...state, + provisional: + state.accumulating?.turnId === event.turnId + ? flushAccumulating(state.provisional, state.accumulating) + : state.provisional, + accumulating: state.accumulating?.turnId === event.turnId ? null : state.accumulating, + generating: state.currentTurnId === event.turnId ? false : state.generating, + }; case "done": { const provisional = flushAccumulating(state.provisional, state.accumulating); @@ -394,11 +488,21 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState // into the transcript by the sender). if (event.text.length === 0) return state; const provisional = flushAccumulating(state.provisional, state.accumulating); + const preceding = [...state.committed, ...provisional] + .reverse() + .find((p) => p.turnId === event.turnId && p.chunk.type === "tool-result"); + const afterToolCallId = + preceding?.chunk.type === "tool-result" ? preceding.chunk.toolCallId : undefined; return { ...state, provisional: [ ...provisional, - { role: "user", chunk: { type: "text", text: event.text }, turnId: event.turnId }, + { + role: "user", + chunk: { type: "text", text: event.text }, + turnId: event.turnId, + ...(afterToolCallId !== undefined ? { afterToolCallId } : {}), + }, ], accumulating: null, generating: true, @@ -440,9 +544,31 @@ const RETRY_CLEARING_EVENTS: ReadonlySet = new Set([ "error", "done", "turn-sealed", + "turn-ended", ]); export function foldEvent(state: TranscriptState, event: AgentEvent): TranscriptState { + if ( + event.type === "tool-call" || + event.type === "tool-result" || + event.type === "step-complete" + ) { + state = { + ...state, + provisional: state.provisional.map((p) => + p.turnId === event.turnId && + p.role === "assistant" && + p.stepId === undefined && + (p.chunk.type === "text" || p.chunk.type === "thinking") + ? { ...p, stepId: event.stepId } + : p, + ), + accumulating: + state.accumulating?.turnId === event.turnId + ? { ...state.accumulating, stepId: event.stepId } + : state.accumulating, + }; + } const next = reduceEvent(state, event); if (event.type === "provider-retry") return next; // set by reduceEvent; not a clearing event if (RETRY_CLEARING_EVENTS.has(event.type) && state.providerRetry !== null) { @@ -511,3 +637,15 @@ export function removeLocalEcho(state: TranscriptState, requestId: string): Tran const provisional = state.provisional.filter((p) => p.localRequestId !== requestId); return provisional.length === state.provisional.length ? state : { ...state, provisional }; } + +/** A resync notice is lifecycle evidence, never fabricated transcript content. */ +export function markHistoryRecovering(state: TranscriptState, turnId: string): TranscriptState { + return { + ...state, + currentTurnId: turnId, + generating: true, + historyRecoveryTurnIds: state.historyRecoveryTurnIds?.includes(turnId) + ? state.historyRecoveryTurnIds + : [...(state.historyRecoveryTurnIds ?? []), turnId], + }; +} diff --git a/src/core/chunks/selectors.test.ts b/src/core/chunks/selectors.test.ts index 57c1415..05dcc4e 100644 --- a/src/core/chunks/selectors.test.ts +++ b/src/core/chunks/selectors.test.ts @@ -1,6 +1,6 @@ import type { AgentEvent, StepId, StoredChunk } from "@dispatch/wire"; import { describe, expect, it } from "vitest"; -import { applyHistory, foldEvent, initialState } from "./reducer"; +import { applyHistory, foldEvent, initialState, markHistoryRecovering } from "./reducer"; import { selectChunks } from "./selectors"; import type { TranscriptState } from "./types"; @@ -131,6 +131,34 @@ describe("selectChunks — de-dup of a late-join replay against committed histor expect(rendered.filter((r) => r.provisional)).toHaveLength(3); }); + it("preserves an uncommitted accumulating tail even when its text repeats history", () => { + const committed: StoredChunk[] = [ + ...committedStep0, + { seq: 5, role: "assistant", chunk: { type: "text", text: "final" }, turnId: T }, + ]; + let s = applyHistory(initialState(), committed); + s = markHistoryRecovering(s, T); + s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("a"), text("final")]); + const rendered = selectChunks(s); + expect(rendered).toHaveLength(6); + expect(rendered.slice(0, 5).every((r) => !r.provisional)).toBe(true); + expect(rendered[5]).toMatchObject({ + provisional: true, + streaming: true, + chunk: { type: "text", text: "final" }, + }); + }); + + it("reconciles the same step identity even when provisional payload differs", () => { + let s = applyHistory(initialState(), committedStep0); + s = markHistoryRecovering(s, T); + s = fold(s, [userMsg, turnStart, think("DIFFERENT"), call("a"), result("a")]); + const rendered = selectChunks(s); + // Authoritative identity wins; payload equality cannot identify omitted/truncated content. + expect(rendered.filter((r) => r.provisional)).toHaveLength(0); + expect(rendered.map((r) => r.chunk)).toEqual(committedStep0.map((r) => r.chunk)); + }); + it("does nothing when the committed tail belongs to another turn", () => { const other = committedStep0.map((c) => ({ ...c, turnId: "older" })); let s = applyHistory(initialState(), other); diff --git a/src/core/chunks/selectors.ts b/src/core/chunks/selectors.ts index d32a628..b507cf7 100644 --- a/src/core/chunks/selectors.ts +++ b/src/core/chunks/selectors.ts @@ -1,4 +1,5 @@ import type { ChatMessage, Chunk, StoredChunk, TurnProviderRetryEvent } from "@dispatch/wire"; +import { historyBackedProvisional } from "./reconcile"; import type { ProvisionalChunk, RenderedChunk, TranscriptState } from "./types"; /** @@ -42,7 +43,8 @@ function sameChunk(p: ProvisionalChunk, c: StoredChunk): boolean { function committedOverlap(state: TranscriptState): { skip: number; hideAccumulating: boolean } { const first = state.provisional[0]; const turnId = first?.turnId ?? state.accumulating?.turnId; - if (turnId === undefined) return { skip: 0, hideAccumulating: false }; + if (turnId === undefined || state.historyRecoveryTurnIds?.includes(turnId)) + return { skip: 0, hideAccumulating: false }; // The committed run for this turn (contiguous at the tail of `committed`). let start = state.committed.length; while (start > 0 && state.committed[start - 1]?.turnId === turnId) start--; @@ -59,7 +61,7 @@ function committedOverlap(state: TranscriptState): { skip: number; hideAccumulat let hideAccumulating = false; const acc = state.accumulating; - if (acc !== null && skip === state.provisional.length && skip < run) { + if (acc !== null && acc.turnId === turnId && skip === state.provisional.length && skip < run) { const c = state.committed[start + skip]; hideAccumulating = c !== undefined && @@ -72,11 +74,12 @@ function committedOverlap(state: TranscriptState): { skip: number; hideAccumulat /** * Select all chunks for rendering: committed first (seq order), - * then provisional (seq: null) — minus provisional chunks already backed by - * committed history (see `committedOverlap`). + * then provisional (seq: null), excluding history-backed identities. Unknown + * accumulating content stays live, even if its text equals an earlier step. */ export function selectChunks(state: TranscriptState): readonly RenderedChunk[] { const result: RenderedChunk[] = []; + const backed = historyBackedProvisional(state); const { skip, hideAccumulating } = committedOverlap(state); for (const c of state.committed) { result.push( @@ -87,7 +90,7 @@ export function selectChunks(state: TranscriptState): readonly RenderedChunk[] { } for (let i = skip; i < state.provisional.length; i++) { const p = state.provisional[i]; - if (p === undefined) continue; + if (p === undefined || backed.has(p)) continue; result.push( p.turnId !== undefined ? { seq: null, role: p.role, chunk: p.chunk, provisional: true, turnId: p.turnId } diff --git a/src/core/chunks/types.ts b/src/core/chunks/types.ts index f7bfe01..2e2ee0c 100644 --- a/src/core/chunks/types.ts +++ b/src/core/chunks/types.ts @@ -2,6 +2,7 @@ import type { Chunk, Role, StoredChunk, TurnProviderRetryEvent, Usage } from "@d /** A chunk being accumulated from streaming deltas (text or thinking). */ export interface AccumulatingChunk { + readonly stepId?: string; readonly kind: "text" | "thinking"; readonly text: string; /** The turn generating this chunk (from the live event's `turnId`). */ @@ -10,6 +11,11 @@ export interface AccumulatingChunk { /** A provisional chunk that has no authoritative seq yet. */ export interface ProvisionalChunk { + /** Local provenance learned from tool-call/step-complete, not content. */ + readonly stepId?: string; + readonly opening?: boolean; + /** Steering is persisted immediately after this completed tool batch. */ + readonly afterToolCallId?: string; /** Local send identity only; turn-start tagging does NOT confirm this content. */ readonly localRequestId?: string; readonly role: Role; @@ -24,6 +30,8 @@ export interface ProvisionalChunk { /** The transcript reducer state. Holds committed history + live in-flight turn. */ export interface TranscriptState { + /** Turns with omitted delivery: never use payload equality as replay identity. */ + readonly historyRecoveryTurnIds?: readonly string[]; readonly committed: readonly StoredChunk[]; readonly provisional: readonly ProvisionalChunk[]; readonly accumulating: AccumulatingChunk | null; diff --git a/src/core/wire/conformance.test.ts b/src/core/wire/conformance.test.ts index 1179fbc..b3c774f 100644 --- a/src/core/wire/conformance.test.ts +++ b/src/core/wire/conformance.test.ts @@ -1,4 +1,8 @@ -import type { ChatSendMessage, ConversationHistoryResponse } from "@dispatch/transport-contract"; +import type { + ChatSendMessage, + ConversationHistoryResponse, + WsServerMessage, +} from "@dispatch/transport-contract"; import type { AgentEvent, StepId, StoredChunk } from "@dispatch/wire"; import { describe, expect, it } from "vitest"; import { @@ -180,6 +184,11 @@ describe("classifies every WsServerMessage type", () => { reason: "compaction" as const, messageId: "queued-1", }, + { + type: "chat.resync-required" as const, + conversationId: "c1", + reason: "replay-limit" as const, + }, { type: "conversation.open" as const, conversationId: "c1", workspaceId: "w1" }, { type: "conversation.statusChanged" as const, @@ -195,7 +204,29 @@ describe("classifies every WsServerMessage type", () => { historyEpoch: 0, messagesKept: 5, }, + { + type: "conversation.tabGroupChanged" as const, + conversationId: "c1", + workspaceId: "w1", + tabGroupParentId: null, + }, + { type: "extensions-changed" as const, action: "scan" as const }, + { + type: "layout" as const, + workspaceId: "w1", + tree: { version: 1 as const, root: { kind: "leaf" as const, id: "leaf1", view: "chat" } }, + }, + { + type: "tile-layout" as const, + mode: "wide" as const, + data: { version: 2 as const, tiles: [] }, + }, ]; + // Runtime samples stay exhaustive too, not just the switch guard. + const covered: Exclude extends never + ? true + : false = true; + expect(covered).toBe(true); const labels = msgs.map(assertWsServerMessageExhaustive); expect(labels).toEqual([ "catalog", @@ -205,9 +236,14 @@ describe("classifies every WsServerMessage type", () => { "chat.delta", "chat.error", "chat.queued", + "chat.resync-required", "conversation.open", "conversation.statusChanged", "conversation.compacted", + "conversation.tabGroupChanged", + "extensions-changed", + "layout", + "tile-layout", ]); }); }); diff --git a/src/core/wire/conformance.ts b/src/core/wire/conformance.ts index 78313f9..3429ed0 100644 --- a/src/core/wire/conformance.ts +++ b/src/core/wire/conformance.ts @@ -85,6 +85,8 @@ export function assertWsServerMessageExhaustive(msg: WsServerMessage): string { return "error"; case "chat.delta": return "chat.delta"; + case "chat.resync-required": + return "chat.resync-required"; case "chat.queued": return "chat.queued"; case "chat.error": diff --git a/src/features/chat/logic/tool-summary.ts b/src/features/chat/logic/tool-summary.ts index 566e40f..f6d1569 100644 --- a/src/features/chat/logic/tool-summary.ts +++ b/src/features/chat/logic/tool-summary.ts @@ -381,7 +381,9 @@ export function viewToolEntry(entry: ToolEntry): ToolCardView { const builtInSummary = summarize?.(args).trim(); const callSummary = entry.call === null - ? "Call in earlier history" + ? entry.recovering + ? "Recovering tool input…" + : "Call in earlier history" : isEmptyInput(input) ? "" : oneLine( @@ -416,7 +418,9 @@ export function viewToolEntry(entry: ToolEntry): ToolCardView { entry.call === null ? { kind: "unavailable", - text: "This call is earlier in the conversation. Show earlier messages to see it.", + text: entry.recovering + ? "Tool input is unavailable until it is persisted. Recovering from conversation history…" + : "This call is earlier in the conversation. Show earlier messages to see it.", } : detail(entry.toolName, input, args), resultSummary, diff --git a/src/features/chat/recovery.test.ts b/src/features/chat/recovery.test.ts new file mode 100644 index 0000000..4b46288 --- /dev/null +++ b/src/features/chat/recovery.test.ts @@ -0,0 +1,529 @@ +import type { + ChatResyncRequiredMessage, + ConversationHistoryResponse, +} from "@dispatch/transport-contract"; +import type { AgentEvent, StepId, StoredChunk, TurnEndedEvent } from "@dispatch/wire"; +import { IDBFactory, IDBKeyRange } from "fake-indexeddb"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { createIdbChunkStore } from "../../adapters/idb"; +import { createConversationCache } from "../conversation-cache"; +import type { HistorySync } from "./ports"; +import { createChatStore } from "./store.svelte"; +import { + createFakeCache, + createFakeHistorySync, + createFakeMetricsSync, + createFakeTransport, +} from "./test-helpers"; + +const step = "t1#0" as StepId; +const opening: StoredChunk = { + seq: 1, + role: "user", + turnId: "t1", + chunk: { type: "text", text: "prompt" }, +}; +const call: StoredChunk = { + seq: 3, + role: "assistant", + turnId: "t1", + chunk: { + type: "tool-call", + toolCallId: "a", + toolName: "run_shell", + stepId: step, + input: { command: "full input" }, + }, +}; +const result: StoredChunk = { + seq: 4, + role: "tool", + turnId: "t1", + chunk: { + type: "tool-result", + toolCallId: "a", + toolName: "run_shell", + stepId: step, + content: "full output", + isError: false, + }, +}; +const committed: StoredChunk[] = [ + opening, + { seq: 2, role: "assistant", turnId: "t1", chunk: { type: "text", text: "complete text" } }, + call, + result, +]; + +function fixture(historySync?: HistorySync, batched = false) { + const history = createFakeHistorySync(); + const transport = createFakeTransport(); + const cache = createFakeCache(); + const scheduled: (() => void)[] = []; + const store = createChatStore({ + conversationId: "c1", + createRequestId: () => "r1", + historySync: historySync ?? history.impl, + metricsSync: createFakeMetricsSync().impl, + cache: cache.impl, + transport: transport.impl, + ...(batched ? { scheduleFlush: (fn: () => void) => scheduled.push(fn) } : {}), + }); + const event = (event: AgentEvent) => store.handleDelta({ type: "chat.delta", event }); + const notice = ( + turnId: string | undefined = "t1", + reason: ChatResyncRequiredMessage["reason"] = "replay-limit", + ) => + store.handleDelta({ + type: "chat.resync-required", + conversationId: "c1", + reason, + ...(turnId !== undefined ? { turnId } : {}), + }); + const end = (reason: TurnEndedEvent["reason"] = "sealed", turnId = "t1") => + event({ type: "turn-ended", conversationId: "c1", turnId, reason }); + const text = (delta: string, turnId = "t1") => + event({ type: "text-delta", conversationId: "c1", turnId, delta }); + return { store, history, transport, cache, scheduled, event, notice, end, text }; +} + +async function settled(store: ReturnType) { + 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((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 void>(); + const store = createChatStore({ + conversationId: "c1", + createRequestId: () => "r", + cache, + historySync: history.impl, + metricsSync: createFakeMetricsSync().impl, + transport: createFakeTransport().impl, + chatLimit: 10, + recoveryTimer: { + schedule(fn) { + const key = {}; + timers.set(key, fn); + return key; + }, + cancel(key) { + timers.delete(key as object); + }, + }, + }); + try { + await store.load(); + history.historyEpoch = 0; + history.returnChunks = Array.from({ length: 20 }, (_, i) => ({ + seq: i + 1, + role: "assistant" as const, + turnId: "t1", + chunk: { type: "text" as const, text: `restored-${i}` }, + })); + const before = history.calls.length; + store.handleDelta({ + type: "chat.resync-required", + conversationId: "c1", + turnId: "t1", + reason: "replay-limit", + }); + await vi.waitFor(() => expect(store.pendingSync).toBe(false)); + expect(history.calls.length - before).toBe(2); + expect(history.calls.at(-1)).toMatchObject({ sinceSeq: 0, window: { limit: 7 } }); + expect(await cache.historyEpoch("c1")).toBe(0); + expect(store.chunks.map((c) => c.seq)).toEqual([14, 15, 16, 17, 18, 19, 20]); + expect(store.hasEarlier).toBe(true); + expect(store.generating).toBe(true); + expect(timers.size).toBe(1); + store.handleDelta({ + type: "chat.delta", + event: { type: "turn-ended", conversationId: "c1", turnId: "t1", reason: "aborted" }, + }); + await vi.waitFor(() => expect(store.pendingSync).toBe(false)); + expect(store.generating).toBe(false); + expect(timers.size).toBe(0); + expect(store.hasEarlier).toBe(true); + } finally { + store.dispose(); + } +}); + +it("ignores an unrelated terminal and stays recovering through done/seal until its actual terminal", async () => { + vi.useFakeTimers(); + const f = fixture(); + f.notice(); + await settled(f.store); + f.end("aborted", "unrelated"); + f.event({ type: "done", conversationId: "c1", turnId: "t1", reason: "aborted" }); + expect(f.store.recovering).toBe(true); + expect(f.store.generating).toBe(true); + f.event({ type: "turn-sealed", conversationId: "c1", turnId: "t1" }); + await settled(f.store); + expect(f.store.recovering).toBe(true); + expect(f.store.generating).toBe(true); + f.end("aborted"); + await settled(f.store); + expect(f.store.recovering).toBe(false); + expect(f.store.generating).toBe(false); + expect(vi.getTimerCount()).toBe(0); + f.store.dispose(); +}); + +it("makes a post-terminal request even when the terminal races the third bounded epoch attempt", async () => { + vi.useFakeTimers(); + const third = deferred(); + let calls = 0; + const f = fixture(async () => { + calls++; + if (calls < 3) { + f.store.invalidateHistory(calls); + return response([], calls - 1); + } + return calls === 3 ? third.promise : response(committed, 2); + }); + f.notice(); + for (let i = 0; i < 60 && calls < 3; i++) await Promise.resolve(); + expect(calls).toBe(3); + f.end("error"); + third.resolve(response([], 2)); + await settled(f.store); + expect(calls).toBe(4); + expect(f.store.chunks.map((r) => r.chunk)).toEqual(committed.map((r) => r.chunk)); + expect(f.store.recovering).toBe(false); + expect(vi.getTimerCount()).toBe(0); + f.store.dispose(); +}); + +it("retains active recovery across a compaction epoch refresh and cancels it on reconnect/dispose", async () => { + vi.useFakeTimers(); + const f = fixture(); + f.notice(); + await settled(f.store); + f.history.historyEpoch = 1; + f.store.resync(1); + await settled(f.store); + expect(f.store.recovering).toBe(true); + expect(f.store.generating).toBe(true); + expect(vi.getTimerCount()).toBe(1); + f.store.resync(); + await settled(f.store); + expect(f.store.recovering).toBe(false); + expect(f.store.generating).toBe(false); + expect(vi.getTimerCount()).toBe(0); + f.store.dispose(); +}); + +it("flushes a batched aborted terminal plus carried turn before applying an older HTTP response", async () => { + vi.useFakeTimers(); + const pending = deferred(); + let calls = 0; + const f = fixture(async () => (++calls === 1 ? pending.promise : response(committed)), true); + f.notice(); + for (let i = 0; i < 15; i++) await Promise.resolve(); + f.text("old partial"); + f.end("aborted"); + f.event({ type: "user-message", conversationId: "c1", turnId: "t2", text: "queued" }); + f.event({ type: "turn-start", conversationId: "c1", turnId: "t2" }); + f.text("live carried output", "t2"); + pending.resolve(response([opening])); + await settled(f.store); + expect(calls).toBe(2); + expect(f.store.chunks.filter((r) => r.provisional).map((r) => r.chunk)).toEqual([ + { type: "text", text: "queued" }, + { type: "text", text: "live carried output" }, + ]); + expect(f.store.generating).toBe(true); + expect(f.store.recovering).toBe(false); + expect(vi.getTimerCount()).toBe(0); + f.store.dispose(); +}); + +it("does not truncate a payload larger than the WS budget recovered over HTTP", async () => { + vi.useFakeTimers(); + const large = "x".repeat(4 * 1024 * 1024 + 1); + const full: StoredChunk[] = [ + { + ...call, + chunk: { + ...call.chunk, + type: "tool-call", + toolCallId: "a", + toolName: "run_shell", + input: { command: large }, + stepId: step, + }, + }, + { + ...result, + chunk: { + type: "tool-result", + toolCallId: "a", + toolName: "run_shell", + content: large, + isError: false, + stepId: step, + }, + }, + ]; + const f = fixture(); + f.history.returnChunks = full; + f.notice(); + await settled(f.store); + expect(f.store.chunks).toHaveLength(2); + expect(f.store.chunks.map((r) => r.chunk)).toEqual(full.map((r) => r.chunk)); + expect(f.transport.sent).toEqual([]); + f.store.dispose(); +}); diff --git a/src/features/chat/store.svelte.ts b/src/features/chat/store.svelte.ts index 66ee5ee..bb0b218 100644 --- a/src/features/chat/store.svelte.ts +++ b/src/features/chat/store.svelte.ts @@ -4,6 +4,7 @@ import type { ChatQueueCancelMessage, ChatQueuedMessage, ChatQueueMessage, + ChatResyncRequiredMessage, ChatSendMessage, } from "@dispatch/transport-contract"; import type { ChatMessage, ImageInput, StoredChunk, TurnProviderRetryEvent } from "@dispatch/wire"; @@ -15,6 +16,7 @@ import { foldEvent, initialState, initialWindowSize, + markHistoryRecovering, normalizeChatLimit, removeLocalEcho, restoreEarlier, @@ -88,6 +90,11 @@ export interface ChatStoreDependencies { readonly scheduleFlush?: (flush: () => void) => void; /** Wall clock (epoch ms) — times the cache-timer gaps. Absent → `Date.now`. */ readonly now?: () => number; + /** Recovery polling clock; browser effects are replaceable by deterministic tests. */ + readonly recoveryTimer?: { + schedule(callback: () => void, delayMs: number): unknown; + cancel(handle: unknown): void; + }; } export interface ChatStore { @@ -117,6 +124,8 @@ export interface ChatStore { * to the newest attempt + delay; cleared when content resumes or the turn ends. */ readonly providerRetry: TurnProviderRetryEvent | null; + /** Omitted live payloads may not exist in HTTP history until step persistence. */ + readonly recovering: boolean; readonly pendingSync: boolean; readonly error: string | null; readonly model: string | undefined; @@ -146,7 +155,9 @@ export interface ChatStore { * Ignored while a turn is generating; a running turn gap keeps the later start. */ noteLastActivity(at: number): void; - handleDelta(msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage): void; + handleDelta( + msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage | ChatResyncRequiredMessage, + ): void; /** * Send a user message (start a turn via `chat.send`). Optimistically echoes * the text + any `images` as provisional user chunks (`[text, image, …]` in @@ -235,12 +246,38 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { let revision = 0; let reloadWindow = false; let earliestSeq = $state(null); + let recoveringTurnId: string | undefined; + let recovering = $state(false); + const finishedRecoveryTurns = new Set(); + 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), + }; + let recoveryTimer: unknown; + + function cancelRecoveryTimer(): void { + if (recoveryTimer !== undefined) timer.cancel(recoveryTimer); + recoveryTimer = undefined; + } + + function scheduleRecovery(): void { + if (disposed || !recovering || recoveryPaused || recoveryTimer !== undefined) return; + recoveryTimer = timer.schedule(() => { + recoveryTimer = undefined; + flushEvents(); + void syncTail(initialWindowSize(chatLimit)); + }, 1000); + } function resetTranscript(epoch: number): void { historyEpoch = epoch; revision++; reloadWindow = true; transcript = initialState(); + if (recoveringTurnId !== undefined) + transcript = markHistoryRecovering(transcript, recoveringTurnId); earliestSeq = null; pendingEvents = []; } @@ -294,6 +331,7 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { let m = metrics; let sealed = false; let failed = false; + let recoveryEnded = false; let c = cacheTimer; const at = now(); for (const event of events) { @@ -301,14 +339,34 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { m = foldMetricsEvent(m, event); const mNow = m; c = foldCacheTimerEvent(c, event, at, () => lastTurnIdOf(mNow)); - if (t.sealedTurnId !== null) sealed = true; + if ( + event.type === "turn-ended" && + recovering && + (recoveringTurnId === undefined || event.turnId === recoveringTurnId) + ) { + finishedRecoveryTurns.add(event.turnId); + finalSyncRequested = true; + recoveringTurnId = undefined; + recovering = false; + recoveryEnded = true; + cancelRecoveryTimer(); + } + if (event.type === "turn-sealed") sealed = true; else if (event.type === "error") failed = true; } transcript = t; metrics = m; cacheTimer = c; maybeTrim(); - if (sealed) { + if (recovering && recoveringTurnId !== undefined) { + // done/error/seal can precede persistence and the real terminal. Keep + // lifecycle evidence until turn-ended, not merely until a content event. + transcript = markHistoryRecovering(transcript, recoveringTurnId); + } + if (recoveryEnded) { + void syncTail(initialWindowSize(chatLimit)); + void syncMetrics(); + } else if (sealed) { void syncTail(); void syncMetrics(); } else if (failed) { @@ -366,7 +424,13 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { continue; } const limit = - since === 0 ? (reloadWindow ? initialWindowSize(chatLimit) : coldLimit) : undefined; + since === 0 + ? reloadWindow || recovering || finishedRecoveryTurns.size > 0 + ? initialWindowSize(chatLimit) + : coldLimit + : undefined; + const finishedAtRequest = [...finishedRecoveryTurns]; + finalSyncRequested = false; const res = await deps.historySync( deps.conversationId, since, @@ -416,7 +480,9 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { historyEpoch = res.historyEpoch; if (since === 0 && (limit === undefined || res.chunks.length < limit)) earliestSeq = merged[0]?.seq ?? 0; - transcript = applyHistory(transcript, merged); + flushEvents(); + transcript = applyHistory(transcript, merged, finishedAtRequest); + for (const turnId of finishedAtRequest) finishedRecoveryTurns.delete(turnId); maybeTrim(); _error = null; } catch (err) { @@ -424,9 +490,21 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { } } while (rerunRequested && !disposed && attempts < 3); if (rerunRequested && !disposed) { + recoveryPaused = true; + cancelRecoveryTimer(); _error = "Conversation history changed repeatedly; sync paused until the next refresh."; } _pendingSync = false; + // Schedule AFTER the bounded sync burst. No interval can queue unbounded + // reruns behind a slow request; repeated notices coalesce on the same clock. + scheduleRecovery(); + // A terminal landing during the third (bounded epoch) attempt still owns + // one post-terminal request. It is a new external trigger, not an epoch + // retry loop; consume the flag before starting its bounded burst. + if (finalSyncRequested && !disposed) { + finalSyncRequested = false; + void syncTail(initialWindowSize(chatLimit)); + } } async function syncMetrics(): Promise { @@ -520,6 +598,9 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { get providerRetry(): TurnProviderRetryEvent | null { return selectProviderRetry(transcript); }, + get recovering(): boolean { + return recovering; + }, get pendingSync(): boolean { return _pendingSync; }, @@ -544,7 +625,32 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { cacheTimer = seedTurnGap(cacheTimer, at); }, - handleDelta(msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage): void { + handleDelta( + msg: ChatDeltaMessage | ChatErrorMessage | ChatQueuedMessage | ChatResyncRequiredMessage, + ): void { + if (disposed) return; + if (msg.type === "chat.resync-required") { + if (msg.conversationId !== deps.conversationId) return; + flushEvents(); + const target = + msg.turnId ?? (transcript.generating ? transcript.currentTurnId : null) ?? undefined; + // Oversized status packets need not identify an active turn. Refresh + // once, but do not invent generation or poll an idle conversation. + if (target === undefined) { + void syncTail(initialWindowSize(chatLimit)); + return; + } + const first = !recovering || recoveryPaused; + recoveryPaused = false; + recovering = true; + recoveringTurnId = target; + if (recoveringTurnId !== undefined) + transcript = markHistoryRecovering(transcript, recoveringTurnId); + // Immediate first handoff, including a coalesced rerun behind a cold + // load. Repeats wait for the existing fetch/one-second timer. + if (first) void syncTail(initialWindowSize(chatLimit)); + return; + } if (msg.type === "chat.error") { if (msg.conversationId !== undefined && msg.conversationId !== deps.conversationId) { return; @@ -685,13 +791,21 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { // clear the now-stale spinner BEFORE re-subscribing, so a finished turn // doesn't spin forever. A still-running turn's replay re-asserts it. flushEvents(); - transcript = clearGenerating(transcript); + recoveryPaused = false; + if (epoch === undefined) { + recovering = false; + recoveringTurnId = undefined; + cancelRecoveryTimer(); + transcript = clearGenerating(transcript); + } void syncTail(); void syncMetrics(); }, dispose(): void { disposed = true; + cancelRecoveryTimer(); + finishedRecoveryTurns.clear(); pendingEvents = []; }, }; diff --git a/src/features/chat/ui/ChatView.svelte b/src/features/chat/ui/ChatView.svelte index b473766..c7f566a 100644 --- a/src/features/chat/ui/ChatView.svelte +++ b/src/features/chat/ui/ChatView.svelte @@ -24,6 +24,7 @@ let { chunks, + recovering = false, turnMetrics = [], hasEarlier = false, onShowEarlier, @@ -34,6 +35,7 @@ cacheTimer = null, }: { chunks: readonly RenderedChunk[]; + recovering?: boolean; turnMetrics?: readonly TurnMetricsEntry[]; /** Earlier history is unloaded (chat limit) and can be paged back in. */ hasEarlier?: boolean; @@ -90,7 +92,7 @@ } } - const groups = $derived(groupRenderedChunks(chunks)); + const groups = $derived(groupRenderedChunks(chunks, recovering)); const rows = $derived(interleaveTurnMetrics(groups, turnMetrics)); @@ -150,6 +152,12 @@ }); +{#if recovering} +

+ Recovering conversation history… Omitted tool input/output is unavailable until persisted. +

+{/if} + {#snippet chunkRow(rendered: RenderedChunk)} {#if rendered.role === "user"}
(menu = null)} > -
+
{#each forest.roots as node (node.tab.conversationId)} {#if node.children.length > 0} diff --git a/src/features/tabs/ui/tab-flip.ts b/src/features/tabs/ui/tab-flip.ts new file mode 100644 index 0000000..9d0f0f8 --- /dev/null +++ b/src/features/tabs/ui/tab-flip.ts @@ -0,0 +1,157 @@ +import { gsap } from "gsap"; +import { Flip } from "gsap/Flip"; +import { isGroupFlipId, leavingFlipIds } from "../tab-flip"; + +gsap.registerPlugin(Flip); + +/** + * GSAP Flip animation for the tab grid — the DOM EFFECT half of the tab + * FLIP (the pure id scheme lives in `../tab-flip.ts`). + * + * Two-phase, driven by the owning component: + * - `capture(nextIds)` — BEFORE the DOM updates: record every `[data-flip-id]` + * element's position (First), and spawn an inert GHOST for each element the + * next render drops, so a closed tab / dissolved group box fades out in + * place instead of vanishing. + * - `play()` — AFTER the DOM updates: Flip every surviving element from its + * recorded position to its new one (Last → Invert → Play), scale new cards + * in, and fade new group boxes' backdrops in. + * + * Data-driven, not click-driven: it animates ANY change to the tab set (a + * summoned subagent, a close from another device, a regroup). Re-entrant — + * `Flip.getState` captures mid-flight positions. Disabled when the user + * prefers reduced motion (or the environment can't tell, e.g. jsdom). + */ +export interface TabFlip { + capture(nextIds: ReadonlySet): void; + play(): void; +} + +const DURATION = 0.4; +const GHOST_ATTR = "data-tab-ghost"; + +export function createTabFlip(getScroller: () => HTMLElement | undefined): TabFlip { + let state: ReturnType | null = null; + + return { + capture(nextIds) { + state = null; + const scroller = getScroller(); + if (scroller === undefined || !motionAllowed()) return; + const els = flipElements(scroller); + if (els.length === 0) return; + state = Flip.getState(els); + const byId = new Map(els.map((el) => [el.dataset.flipId ?? "", el])); + for (const id of leavingFlipIds([...byId.keys()], nextIds)) { + const el = byId.get(id); + if (el !== undefined) spawnGhost(scroller, el, isGroupFlipId(id)); + } + }, + + play() { + const captured = state; + state = null; + const scroller = getScroller(); + if (captured === null || scroller === undefined) return; + Flip.from(captured, { + targets: flipElements(scroller), + duration: DURATION, + ease: "power3.out", + nested: true, + onEnter: (elements) => { + for (const el of elements) { + if (!(el instanceof HTMLElement)) continue; + if (isGroupFlipId(el.dataset.flipId ?? "")) { + // The box's CARDS are flipping into place — fade only its tint. + const backdrop = el.querySelector(":scope > [data-group-backdrop]"); + if (backdrop !== null) { + gsap.fromTo(backdrop, { opacity: 0 }, { opacity: 1, duration: DURATION }); + } + } else { + gsap.fromTo( + el, + { opacity: 0, scale: 0.6 }, + { + opacity: 1, + scale: 1, + duration: DURATION, + delay: DURATION * 0.35, // let the grid open the gap first + ease: "back.out(1.7)", + clearProps: "opacity,transform", + }, + ); + } + } + }, + }); + }, + }; +} + +function motionAllowed(): boolean { + return ( + typeof window.matchMedia === "function" && + !window.matchMedia("(prefers-reduced-motion: reduce)").matches + ); +} + +function flipElements(scroller: HTMLElement): HTMLElement[] { + return [...scroller.querySelectorAll("[data-flip-id]")]; +} + +/** + * An inert visual copy of a leaving element, absolutely positioned over its + * last on-screen rect, that shrinks + fades out and removes itself. + * + * Paint order matters (cards are `relative`, so everything here stacks in DOM + * order): a leaving CARD's ghost is appended to the scroller so it fades out + * ON TOP of the reflow. A leaving GROUP BOX copies only its backdrop (the + * tint — its cards either flip elsewhere or ghost themselves) and is inserted + * as the FIRST child of the box's enclosing grid: above the enclosing box's + * own tint, but BELOW every card, so the dissolving tint never dims the cards + * flying out of it. + */ +function spawnGhost(scroller: HTMLElement, el: HTMLElement, isGroup: boolean): void { + const source = isGroup ? el.querySelector(":scope > [data-group-backdrop]") : el; + if (source === null) return; + const host = isGroup ? el.parentElement : scroller; + if (host === null) return; + const r = el.getBoundingClientRect(); + const h = host.getBoundingClientRect(); + const ghost = source.cloneNode(!isGroup) as HTMLElement; + // A ghost is never a tab/group to queries, tests or assistive tech. + for (const node of [ghost, ...ghost.querySelectorAll("*")]) { + for (const attr of [ + "data-flip-id", + "data-conversation-id", + "data-group-backdrop", + "role", + "id", + ]) { + node.removeAttribute(attr); + } + } + ghost.setAttribute(GHOST_ATTR, ""); + ghost.setAttribute("aria-hidden", "true"); + ghost.inert = true; + Object.assign(ghost.style, { + position: "absolute", + left: `${r.left - h.left - host.clientLeft + host.scrollLeft}px`, + top: `${r.top - h.top - host.clientTop + host.scrollTop}px`, + width: `${r.width}px`, + height: `${r.height}px`, + margin: "0", + transform: "none", + pointerEvents: "none", + zIndex: isGroup ? "auto" : "2", + }); + if (isGroup) host.prepend(ghost); + else host.appendChild(ghost); + gsap.to(ghost, { + opacity: 0, + scale: isGroup ? 1 : 0.6, + duration: DURATION * 0.75, + ease: "power2.in", + onComplete: () => ghost.remove(), + }); +} -- cgit v1.2.3 From 98f0494e46d09aa7799fa3018705d39ae04ea4eb Mon Sep 17 00:00:00 2001 From: Adam Malczewski Date: Wed, 21 Oct 2026 20:53:41 +0900 Subject: feat(tabs): raise the tab grid height to 70vh --- src/features/tabs/ui.test.ts | 4 ++-- src/features/tabs/ui/TabList.svelte | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) (limited to 'src') diff --git a/src/features/tabs/ui.test.ts b/src/features/tabs/ui.test.ts index 743b98d..408ec6e 100644 --- a/src/features/tabs/ui.test.ts +++ b/src/features/tabs/ui.test.ts @@ -186,7 +186,7 @@ describe("TabList", () => { } }); - it("fixes the tab grid region at 40vh so a long set scrolls internally", () => { + it("fixes the tab grid region at 70vh so a long set scrolls internally", () => { render(TabList, { props: { tabs: sampleTabs, @@ -199,7 +199,7 @@ describe("TabList", () => { const tabs = screen.getAllByRole("tab"); const region = tabs[0]?.parentElement?.parentElement; - expect(region).toHaveClass("h-[40vh]"); + expect(region).toHaveClass("h-[70vh]"); expect(region).toHaveClass("overflow-y-auto"); }); diff --git a/src/features/tabs/ui/TabList.svelte b/src/features/tabs/ui/TabList.svelte index 5bfabf6..dfefc9a 100644 --- a/src/features/tabs/ui/TabList.svelte +++ b/src/features/tabs/ui/TabList.svelte @@ -107,13 +107,13 @@ {/snippet}
-
(menu = null)} -- cgit v1.2.3 From bbf9728721bd25fc5b0532c16088e9e025dadb40 Mon Sep 17 00:00:00 2001 From: Adam Malczewski Date: Wed, 21 Oct 2026 21:06:12 +0900 Subject: feat(layout): GSAP height animation when sidebar views are added, closed or changed --- src/features/layout/logic/panel-keys.test.ts | 73 +++++++++++++ src/features/layout/logic/panel-keys.ts | 118 +++++++++++++++++++++ src/features/layout/ui/TilePanels.svelte | 148 +++++++++++++++++++++------ src/features/layout/ui/TileRoot.test.ts | 53 +++++++++- src/features/layout/ui/panel-motion.ts | 83 +++++++++++++++ 5 files changed, 443 insertions(+), 32 deletions(-) create mode 100644 src/features/layout/logic/panel-keys.test.ts create mode 100644 src/features/layout/logic/panel-keys.ts create mode 100644 src/features/layout/ui/panel-motion.ts (limited to 'src') diff --git a/src/features/layout/logic/panel-keys.test.ts b/src/features/layout/logic/panel-keys.test.ts new file mode 100644 index 0000000..b45671a --- /dev/null +++ b/src/features/layout/logic/panel-keys.test.ts @@ -0,0 +1,73 @@ +import { describe, expect, it } from "vitest"; +import { initialPanelKeys, reconcilePanelKeys, swappedPanels } from "./panel-keys"; + +describe("initialPanelKeys", () => { + it("numbers the panels in order and seeds past them", () => { + expect(initialPanelKeys(["tabs", "model"])).toEqual({ + views: ["tabs", "model"], + keys: [0, 1], + seed: 2, + }); + }); +}); + +describe("reconcilePanelKeys", () => { + const start = initialPanelKeys(["tabs", "model", "usage"]); // keys 0 1 2 + + it("closing the MIDDLE panel drops the middle key (not the last)", () => { + const next = reconcilePanelKeys(start, ["tabs", "usage"]); + expect(next.keys).toEqual([0, 2]); + expect(next.seed).toBe(3); + }); + + it("adding a panel gives only the new one a fresh key", () => { + const next = reconcilePanelKeys(start, ["tabs", "model", "usage", "lsp"]); + expect(next.keys).toEqual([0, 1, 2, 3]); + expect(next.seed).toBe(4); + }); + + it("a dropdown swap keeps every key positional", () => { + const next = reconcilePanelKeys(start, ["tabs", "lsp", "usage"]); + expect(next.keys).toEqual([0, 1, 2]); + }); + + it("handles duplicate views and never reuses a key", () => { + const dup = initialPanelKeys(["a", "b", "a"]); // 0 1 2 + const removed = reconcilePanelKeys(dup, ["a", "a"]); + expect(removed.keys).toEqual([0, 2]); + const added = reconcilePanelKeys(removed, ["a", "a", "a"]); + expect(added.keys).toEqual([0, 2, 3]); + expect(new Set(added.keys).size).toBe(3); + }); + + it("a ✕ hint picks the exact panel when identical views make it ambiguous", () => { + const twins = initialPanelKeys(["model", "usage", "usage"]); // 0 1 2 + expect(reconcilePanelKeys(twins, ["model", "usage"]).keys).toEqual([0, 1]); // no hint + expect(reconcilePanelKeys(twins, ["model", "usage"], 1).keys).toEqual([0, 2]); + }); + + it("ignores a stale hint that does not describe the change", () => { + // The tree changed some other way (e.g. a remote edit) — fall back to LCS. + expect(reconcilePanelKeys(start, ["tabs", "model"], 0).keys).toEqual([0, 1]); + expect(reconcilePanelKeys(start, ["tabs", "model", "usage"], 1).keys).toEqual([0, 1, 2]); + }); + + it("an unrelated replacement of different length gets all-fresh keys", () => { + const next = reconcilePanelKeys(start, ["x"]); + expect(next.keys).toEqual([3]); + }); +}); + +describe("swappedPanels", () => { + it("reports a surviving panel whose view changed", () => { + const prev = initialPanelKeys(["tabs", "model"]); + const next = reconcilePanelKeys(prev, ["tabs", "usage"]); + expect([...swappedPanels(prev, next)]).toEqual([[1, { from: "model", to: "usage" }]]); + }); + + it("does not report adds or removes", () => { + const prev = initialPanelKeys(["tabs", "model", "usage"]); + expect(swappedPanels(prev, reconcilePanelKeys(prev, ["tabs", "usage"])).size).toBe(0); + expect(swappedPanels(prev, reconcilePanelKeys(prev, [...prev.views, "lsp"])).size).toBe(0); + }); +}); diff --git a/src/features/layout/logic/panel-keys.ts b/src/features/layout/logic/panel-keys.ts new file mode 100644 index 0000000..76a053f --- /dev/null +++ b/src/features/layout/logic/panel-keys.ts @@ -0,0 +1,118 @@ +/** + * Stable panel keys — the PURE identity scheme behind the panel height + * animation. + * + * Panel leaves carry no required id, so the panels list can't be keyed by the + * tree. Keying by INDEX is wrong for removal: closing the middle of three + * panels would re-use the middle DOM node for the third view and animate the + * LAST panel out. This reconciles a stable numeric key per panel across tree + * updates, from the view ids alone: + * + * - same length → keys stay positional (a dropdown SWAP keeps its panel — + * the panel morphs its height to the new view instead of leaving). + * - length changed → the longest common subsequence of view ids keeps its + * keys; every unmatched new panel gets a fresh key (an ADD), every + * unmatched old key disappears (a REMOVE). + * + * Zero Svelte, zero DOM. + */ + +export interface PanelKeys { + /** View id per panel, in order. */ + readonly views: readonly string[]; + /** Stable key per panel, aligned with `views`. */ + readonly keys: readonly number[]; + /** The next fresh key. */ + readonly seed: number; +} + +/** Keys for a freshly mounted panels list. */ +export function initialPanelKeys(views: readonly string[]): PanelKeys { + return { views: [...views], keys: views.map((_, i) => i), seed: views.length }; +} + +/** + * Reconcile `prev` against the next view list (see the module doc). + * + * `removedIndex` is an optional HINT from a local ✕ click: when the next list + * is exactly `prev` minus that index, that exact panel is the one removed. + * Without it, closing one of two identical adjacent views is ambiguous to the + * subsequence match (it would drop the later one). + */ +export function reconcilePanelKeys( + prev: PanelKeys, + nextViews: readonly string[], + removedIndex?: number, +): PanelKeys { + if (removedIndex !== undefined && isRemovalAt(prev.views, nextViews, removedIndex)) { + return { + views: [...nextViews], + keys: prev.keys.filter((_, i) => i !== removedIndex), + seed: prev.seed, + }; + } + if (nextViews.length === prev.views.length) { + return { views: [...nextViews], keys: [...prev.keys], seed: prev.seed }; + } + const matched = lcsMatch(prev.views, nextViews); + let seed = prev.seed; + const keys = nextViews.map((_, j) => { + const i = matched.get(j); + if (i !== undefined) return prev.keys[i] ?? seed++; + return seed++; + }); + return { views: [...nextViews], keys, seed }; +} + +/** + * The views whose panel SURVIVED but now mounts a different view (a swap): + * key → { from, to }. Adds/removes are not swaps. + */ +export function swappedPanels( + prev: PanelKeys, + next: PanelKeys, +): Map { + const before = new Map(prev.keys.map((k, i) => [k, prev.views[i] ?? ""])); + const out = new Map(); + next.keys.forEach((k, i) => { + const from = before.get(k); + const to = next.views[i] ?? ""; + if (from !== undefined && from !== to) out.set(k, { from, to }); + }); + return out; +} + +function isRemovalAt(prev: readonly string[], next: readonly string[], index: number): boolean { + if (index < 0 || index >= prev.length || next.length !== prev.length - 1) return false; + return next.every((v, j) => v === prev[j < index ? j : j + 1]); +} + +/** LCS alignment: next index → prev index for every matched element. */ +function lcsMatch(a: readonly string[], b: readonly string[]): Map { + const n = a.length; + const m = b.length; + // dp[i][j] = LCS length of a[i..] and b[j..] + const dp: number[][] = Array.from({ length: n + 1 }, () => new Array(m + 1).fill(0)); + for (let i = n - 1; i >= 0; i--) { + for (let j = m - 1; j >= 0; j--) { + const row = dp[i] as number[]; + const below = dp[i + 1] as number[]; + row[j] = a[i] === b[j] ? (below[j + 1] ?? 0) + 1 : Math.max(below[j] ?? 0, row[j + 1] ?? 0); + } + } + const out = new Map(); + let i = 0; + let j = 0; + while (i < n && j < m) { + if (a[i] === b[j]) { + out.set(j, i); + i++; + j++; + } else if ((dp[i + 1]?.[j] ?? 0) >= (dp[i]?.[j + 1] ?? 0)) { + i++; + } else { + j++; + } + } + return out; +} diff --git a/src/features/layout/ui/TilePanels.svelte b/src/features/layout/ui/TilePanels.svelte index 9162b00..cc5fa89 100644 --- a/src/features/layout/ui/TilePanels.svelte +++ b/src/features/layout/ui/TilePanels.svelte @@ -8,9 +8,17 @@ */ import type { LayoutOp, PanelsNode, Path } from "@dispatch/layout-contract"; import PlusIcon from "phosphor-svelte/lib/PlusIcon"; + import { untrack } from "svelte"; import { viewRegistry } from "../logic/registry"; import { addPanelOp, firstPanelView, removePanelOp, swapPanelOp } from "../logic/panels-edit"; + import { + initialPanelKeys, + type PanelKeys, + reconcilePanelKeys, + swappedPanels, + } from "../logic/panel-keys"; import { markerStyle } from "../logic/marker"; + import { morphPanelHeight, panelGrowIn, panelShrinkOut } from "./panel-motion"; import TileLeaf from "./TileLeaf.svelte"; let { @@ -38,45 +46,122 @@ // appear in the dropdowns. Reactive so late registrations re-render options. const panelViews = $derived($viewRegistry.filter((v) => v.panelEligible)); const addView = $derived(firstPanelView(panelViews)); + + // Stable per-panel keys (pure reconcile, memoised across tree updates) so a + // closed panel is the one that animates out, not whichever is last. + // A local ✕ click hints WHICH panel left (consumed by the next tree change; + // ignored if that change isn't exactly this removal, e.g. a remote edit). + let keyMemo: PanelKeys | null = null; + let removeHint: number | undefined; + const keyed = $derived.by(() => { + const views = node.panels.map((p) => p.view); + keyMemo = + keyMemo === null ? initialPanelKeys(views) : reconcilePanelKeys(keyMemo, views, removeHint); + removeHint = undefined; + return keyMemo; + }); + + function removePanel(idx: number): void { + removeHint = idx; + onEdit([removePanelOp(path, idx)]); + } + + // View SWAP: the panel BOX (the well) animates from the old view's height + // to the new one's. Measure before the DOM updates, animate after. (Adds/removes are + // the per-panel in:/out: transitions below.) + let columnEl = $state(); + /** The keys the DOM currently shows (set after each update). */ + let shownKeys: PanelKeys | null = null; + let before: { keys: PanelKeys; heights: Map } | null = null; + $effect.pre(() => { + const next = keyed; + untrack(() => { + const heights = new Map(); + for (const el of panelEls()) { + const well = wellOf(el); + if (well !== null) heights.set(Number(el.dataset.panelKey), well.offsetHeight); + } + before = { keys: shownKeys ?? next, heights }; + }); + }); + $effect(() => { + const now = keyed; + untrack(() => { + const prev = before; + before = null; + shownKeys = now; + if (prev === null) return; + for (const key of swappedPanels(prev.keys, now).keys()) { + const el = panelEls().find((e) => Number(e.dataset.panelKey) === key); + const well = el === undefined ? null : wellOf(el); + const from = prev.heights.get(key); + if (well !== null && from !== undefined) morphPanelHeight(well, from); + } + }); + }); + + function wellOf(panel: HTMLElement): HTMLElement | null { + return panel.querySelector(":scope > [data-panel-well]"); + } + + function panelEls(): HTMLElement[] { + return columnEl === undefined + ? [] + : [...columnEl.querySelectorAll(":scope > [data-panel-key]")]; + } +
- {#each node.panels as panel, idx (idx)} - -
-
- - {#if idx > 0} - + {/if} +
-
- +
+ +
{/each} @@ -85,8 +170,9 @@ + replaced; `shrink-0` so the column can never squash it. The last panel's + `pb-2` is the gap above it; `mb-2` = the same gap below, so it + doesn't sit tight to the end. -->