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/core/metrics/cache-timer.test.ts | 344 +++++++++++++++++++++++++++++++++++ src/core/metrics/cache-timer.ts | 285 +++++++++++++++++++++++++++++ src/core/metrics/index.ts | 15 ++ 3 files changed, 644 insertions(+) create mode 100644 src/core/metrics/cache-timer.test.ts create mode 100644 src/core/metrics/cache-timer.ts (limited to 'src/core') 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, -- cgit v1.2.3 From 2686cf6bcd247d17528922e36aa380a521c964e0 Mon Sep 17 00:00:00 2001 From: Adam Malczewski Date: Wed, 21 Oct 2026 02:31:38 +0900 Subject: fix(review-3.6): invalidate history caches by epoch and coalesce syncs --- .dispatch/transport-contract.reference.md | 41 ++++++--- .dispatch/wire.reference.md | 31 +++++-- backend-handoff.md | 20 +++++ src/adapters/idb/index.test.ts | 15 ++++ src/adapters/idb/index.ts | 23 +++++ src/adapters/ws/logic.test.ts | 15 ++++ src/adapters/ws/logic.ts | 3 + src/app/App.test.ts | 8 +- src/app/store.svelte.ts | 48 +++++----- src/app/store.test.ts | 36 +++++++- src/core/chunks/trim.test.ts | 21 +++-- src/core/chunks/trim.ts | 22 ++--- src/core/wire/conformance.test.ts | 4 +- src/features/chat/store.svelte.ts | 122 +++++++++++++++++++++----- src/features/chat/store.test.ts | 103 +++++++++++++++++++++- src/features/chat/test-helpers.ts | 21 ++++- src/features/conversation-cache/cache.test.ts | 19 ++++ src/features/conversation-cache/cache.ts | 56 ++++++++++-- src/features/conversation-cache/types.ts | 4 + src/features/layout/adapter/http.test.ts | 2 +- 20 files changed, 509 insertions(+), 105 deletions(-) (limited to 'src/core') diff --git a/.dispatch/transport-contract.reference.md b/.dispatch/transport-contract.reference.md index 95f968e..e798fa7 100644 --- a/.dispatch/transport-contract.reference.md +++ b/.dispatch/transport-contract.reference.md @@ -1,5 +1,7 @@ # `@dispatch/transport-contract` — in-repo reference (read THIS, not node_modules) +> **2026-10-20 review 3.5:** history replacements preserve monotonic seqs (retained history need not start at 1). HTTP history and WS compaction carry required `historyEpoch`; unknown cached epochs must be reloaded. Consumed by review 3.6. +> > MIRRORS the backend's `@dispatch/transport-contract` package source so headless FE agents can read > the transport types WITHOUT following the `file:` dep symlink out of this repo (which hangs on a > permission prompt). Your CODE still imports `@dispatch/transport-contract` normally — this file is for @@ -318,17 +320,24 @@ export interface ModelMetadata { * tail sync = `?sinceSeq=` (no limit); page older history in = * `?beforeSeq=&limit=`. * - * Seq numbering is **1-based and gap-free** (a CONTRACTUAL GUARANTEE — see - * `StoredChunk` in `@dispatch/wire`): a client can derive "older chunks exist" - * purely from `oldestLoaded.seq > 1`; there is deliberately no - * `earliestSeq`/`hasOlder` response field. + * Seq is positive and never reused within a conversation, even when history is + * replaced. The retained log need not start at 1, so `oldestLoaded.seq > 1` + * does not prove older chunks exist. An empty/short backfill page can establish + * that no earlier chunks remain. + * + * `historyEpoch` is the persisted replacement count (missing storage value = 0). + * Every successful replacement, including an empty one, increments it once. + * On a different epoch, clients invalidate their conversation cache and refetch + * with `sinceSeq=0` (optionally windowed), even if this response has no chunks. * * `chunks` is the RAW, append-order, seq-ordered slice of the conversation log * selected + windowed as above. It is NOT reconciled: a dangling tool-call is * returned as-is (rendered as an interrupted call). Reconciliation is a * turn-path concern — the server repairs history only when it feeds a provider, * never on this read path — which is what preserves the per-chunk `seq` cursor - * invariant (a synthesized repair chunk would have no seq). + * invariant (a synthesized repair chunk would have no seq). Provider-opaque + * thinking replay fields (`signature` and `redactedData`) are omitted from + * this client-facing response; chunk presence, text, order, and seq are intact. * * `latestSeq` is the `seq` of the LAST chunk in this response, or — when the * slice is empty (the client is already caught up) — the requested `sinceSeq` @@ -343,8 +352,9 @@ export interface ModelMetadata { * the store contract.) */ export interface ConversationHistoryResponse { - readonly chunks: readonly StoredChunk[]; - readonly latestSeq: number; + readonly chunks: readonly StoredChunk[]; + readonly latestSeq: number; + readonly historyEpoch: number; } /** @@ -945,15 +955,18 @@ export interface ConversationStatusChangedMessage { /** * Broadcast to all connected WS clients when a conversation's history has been - * compacted (summarized). The frontend should reload the conversation history - * via `GET /conversations/:id` to reflect the compacted state. + * compacted (summarized). `historyEpoch` is the persisted replacement count: + * missing storage value = 0; every successful replacement increments it once, + * including empty replacements. On a different epoch, clients invalidate their + * conversation cache and refetch `GET /conversations/:id?sinceSeq=0`. */ export interface ConversationCompactedMessage { - readonly type: "conversation.compacted"; - readonly conversationId: string; - readonly newConversationId: string; - readonly messagesSummarized: number; - readonly messagesKept: number; + readonly type: "conversation.compacted"; + readonly conversationId: string; + readonly newConversationId: string; + readonly messagesSummarized: number; + readonly messagesKept: number; + readonly historyEpoch: number; } /** diff --git a/.dispatch/wire.reference.md b/.dispatch/wire.reference.md index 6e8daff..3743bd5 100644 --- a/.dispatch/wire.reference.md +++ b/.dispatch/wire.reference.md @@ -1,5 +1,7 @@ # `@dispatch/wire` — in-repo reference (read THIS, not node_modules) +> **2026-10-20 review 3.5:** history replacements preserve monotonic seqs (retained history need not start at 1). HTTP history and WS compaction carry required `historyEpoch`; unknown cached epochs must be reloaded. Consumed by review 3.6. +> > MIRRORS the backend's `@dispatch/wire` package source so headless FE agents can read the wire > types WITHOUT following the `file:` dep symlink out of this repo (which hangs on a permission > prompt). Your CODE still imports `@dispatch/wire` normally — this file is for READING only. @@ -258,26 +260,37 @@ export interface ChatMessage { /** * A persisted chunk plus its sync metadata. The append-only conversation log - * stamps every chunk with a **1-based**, monotonic, gap-free, per-conversation + * stamps every chunk with a positive, monotonic, per-conversation * `seq` (the sync cursor, assigned in append order) and records the `role` of * the message it belongs to. This makes a flat seq-ordered stream both * incrementally syncable ("give me chunks after seq N") and regroupable into * messages by the client. * - * The 1-based start is a CONTRACTUAL GUARANTEE (not an implementation detail): - * a conversation's first chunk is always `seq === 1` and numbering never skips, - * so a client holding only a windowed suffix of the log can derive "older - * chunks exist server-side" purely from `oldestLoaded.seq > 1` — no separate - * has-older flag is needed (or provided). `chunk` is the content unit — `Chunk` carries no storage/sync cursor + * Seq is never reused within a conversation, including after compaction replaces + * its history. The retained log need not start at 1 or be gap-free; a client + * cannot infer that older chunks exist solely from `oldestLoaded.seq > 1`. + * `chunk` is the content unit — `Chunk` carries no storage/sync cursor * (`seq` lives here on the envelope, not on the chunk, since it is assigned by * the store and the provider has no use for it). A chunk MAY still carry * generation provenance assigned at production time (e.g. a tool chunk's * `stepId`), which is intrinsic to the content and so travels with it. */ export interface StoredChunk { - readonly seq: number; - readonly role: Role; - readonly chunk: Chunk; + readonly seq: number; + readonly role: Role; + readonly chunk: Chunk; + /** + * The turn that produced this chunk — the same `turnId` carried on every + * live `AgentEvent` of that turn (the turn's opening user message, any + * mid-turn steering message, and every assistant/tool chunk the turn + * generated). The join key between transcript content and the turn's + * `TurnMetrics`, so a client places per-turn metrics EXACTLY (no positional + * guessing) — including while a turn is in flight and when older history has + * been unloaded. Optional: absent on rows persisted before this field + * existed and on chunks written outside any turn (e.g. a compaction summary). + * A consumer must tolerate its absence. + */ + readonly turnId?: string; } // ─── Reasoning effort ─────────────────────────────────────────────────────── diff --git a/backend-handoff.md b/backend-handoff.md index 1986e94..998525b 100644 --- a/backend-handoff.md +++ b/backend-handoff.md @@ -18,6 +18,26 @@ > ?archived=`) was made in the same branch and the FE consumes it via the `file:` deps (dists > rebuilt). Recorded below. +## Milestone: Review 3.6 — history epochs (2026-10-20) + +- Consumed backend Step 3.5: required `historyEpoch` on HTTP history and WS compaction; + unchanged `file:../backend/packages/*` links refreshed with `bun install`. + Versions remain transport-contract **0.26.0**, wire **0.13.0**. Relevant declarations + and documentation re-mirrored in `.dispatch/{transport-contract,wire}.reference.md`. +- IndexedDB meta persists epoch, including empty history; legacy missing epochs are + untrusted and reloaded from cursor 0. Epoch changes clear transcript/watermarks and + cache before a windowed reload. Broadcasts reuse the existing chat store and invalidate + in-flight responses, including modal watches; unseen conversations stay lazy. +- Tail triggers coalesce into sequential reruns. Backfill validates epoch; earlier-history + availability uses a known lower boundary / short-page exhaustion, not `seq > 1`. + With no server `hasOlder` field, a full final page may require one empty-page probe. +- Public FE seams: cache/storage `historyEpoch` + `reset`, optional epoch on cache `commit`, + optional epoch on chat `resync`, lazy transcript `invalidateHistory`, explicit earliest + boundary on `selectHasEarlier`. +- Verification: typecheck/check/build green; full suite **2,186 tests / 115 files** + (run twice on final files; see `reports/review-3.6.md`). Backend health unavailable + on localhost:24203, so no live probe/server started. No new backend asks. + ## Milestone: Local-path images in Markdown (`feature/local-image-markdown`, 2026-10-19) **Slice status:** COMPLETE (frontend half; the backend route is implemented in parallel in diff --git a/src/adapters/idb/index.test.ts b/src/adapters/idb/index.test.ts index 04e2177..aa56f15 100644 --- a/src/adapters/idb/index.test.ts +++ b/src/adapters/idb/index.test.ts @@ -128,3 +128,18 @@ describe("createIdbChunkStore", () => { expect(loaded2[0]?.chunk).toEqual(textChunk("b1")); }); }); + +it("persists epoch across adapter recreation and append, including empty histories", async () => { + const indexedDB = new IDBFactory(); + const store = createIdbChunkStore({ indexedDB }); + await store.append("c", [makeChunk(1, "legacy")]); + expect(await store.historyEpoch("c")).toBeUndefined(); + await store.reset("c", 2); + const reopened = createIdbChunkStore({ indexedDB }); + expect(await reopened.load("c")).toEqual([]); + expect(await reopened.historyEpoch("c")).toBe(2); + await reopened.append("c", [makeChunk(101, "new")]); + expect(await reopened.historyEpoch("c")).toBe(2); + await reopened.delete("c"); + expect(await reopened.historyEpoch("c")).toBeUndefined(); +}); diff --git a/src/adapters/idb/index.ts b/src/adapters/idb/index.ts index 3dffef9..8b1848a 100644 --- a/src/adapters/idb/index.ts +++ b/src/adapters/idb/index.ts @@ -20,6 +20,7 @@ interface ChunkRecord { interface MetaRecord { conversationId: string; lastAccess: number; + historyEpoch?: number; } export interface CreateIdbChunkStoreOptions { @@ -92,6 +93,26 @@ export function createIdbChunkStore(opts?: CreateIdbChunkStoreOptions): Conversa } return { + async historyEpoch(conversationId) { + const db = await getDb(); + const tx = db.transaction(META_STORE, "readonly"); + const meta = await requestToPromise( + tx.objectStore(META_STORE).get(conversationId), + ); + await txComplete(tx); + return meta?.historyEpoch; + }, + async reset(conversationId, historyEpoch) { + const db = await getDb(); + const tx = db.transaction([CHUNKS_STORE, META_STORE], "readwrite"); + tx.objectStore(CHUNKS_STORE).delete(keyRangeFor(conversationId)); + tx.objectStore(META_STORE).put({ + conversationId, + historyEpoch, + lastAccess: Date.now(), + } satisfies MetaRecord); + await txComplete(tx); + }, async load(conversationId: string): Promise { const db = await getDb(); const tx = db.transaction(CHUNKS_STORE, "readonly"); @@ -122,7 +143,9 @@ export function createIdbChunkStore(opts?: CreateIdbChunkStoreOptions): Conversa } satisfies ChunkRecord); } + const meta = await requestToPromise(metaStore.get(conversationId)); metaStore.put({ + ...meta, conversationId, lastAccess: Date.now(), } satisfies MetaRecord); diff --git a/src/adapters/ws/logic.test.ts b/src/adapters/ws/logic.test.ts index 2179ad0..4be2cd6 100644 --- a/src/adapters/ws/logic.test.ts +++ b/src/adapters/ws/logic.test.ts @@ -562,3 +562,18 @@ describe("parseServerMessage — subagent tab groups", () => { ).toBeNull(); }); }); + +it("preserves a compaction epoch and rejects absent or invalid epochs", () => { + const msg = { + type: "conversation.compacted", + conversationId: "c", + newConversationId: "archive", + messagesSummarized: 2, + messagesKept: 1, + historyEpoch: 7, + }; + expect(parseServerMessage(JSON.stringify(msg))).toEqual(msg); + for (const historyEpoch of [undefined, -1, 1.5, "7"]) { + expect(parseServerMessage(JSON.stringify({ ...msg, historyEpoch }))).toBeNull(); + } +}); diff --git a/src/adapters/ws/logic.ts b/src/adapters/ws/logic.ts index cecc66c..858deaa 100644 --- a/src/adapters/ws/logic.ts +++ b/src/adapters/ws/logic.ts @@ -180,7 +180,10 @@ export function parseServerMessage(data: string): WsServerMessage | null { if (typeof parsed.newConversationId !== "string") return null; if (typeof parsed.messagesSummarized !== "number") return null; if (typeof parsed.messagesKept !== "number") return null; + if (!Number.isSafeInteger(parsed.historyEpoch) || (parsed.historyEpoch as number) < 0) + return null; const msg: ConversationCompactedMessage = { + historyEpoch: parsed.historyEpoch as number, type: "conversation.compacted", conversationId: parsed.conversationId, newConversationId: parsed.newConversationId, diff --git a/src/app/App.test.ts b/src/app/App.test.ts index 26f5cdb..183b8b4 100644 --- a/src/app/App.test.ts +++ b/src/app/App.test.ts @@ -104,7 +104,9 @@ function fakeFetchImpl(): typeof fetch { { status: 200 }, ); } - return new Response(JSON.stringify({ chunks: [], latestSeq: 0 }), { status: 200 }); + return new Response(JSON.stringify({ chunks: [], historyEpoch: 0, latestSeq: 0 }), { + status: 200, + }); }; } @@ -564,7 +566,9 @@ describe("App component interaction tests", () => { status: 200, }); } - return new Response(JSON.stringify({ chunks: [], latestSeq: 0 }), { status: 200 }); + return new Response(JSON.stringify({ chunks: [], historyEpoch: 0, latestSeq: 0 }), { + status: 200, + }); }; const ws = fakeSocket(); diff --git a/src/app/store.svelte.ts b/src/app/store.svelte.ts index c354e11..2eb13ef 100644 --- a/src/app/store.svelte.ts +++ b/src/app/store.svelte.ts @@ -704,7 +704,16 @@ function createHistorySync(httpBase: string, fetchImpl: typeof fetch): HistorySy if (!res.ok) { throw new Error(`History sync failed: ${res.status}`); } - return (await res.json()) as ConversationHistoryResponse; + const data = (await res.json()) as ConversationHistoryResponse; + if ( + !data || + !Array.isArray(data.chunks) || + !Number.isSafeInteger(data.historyEpoch) || + data.historyEpoch < 0 + ) { + throw new Error("Invalid history response"); + } + return data; }; } @@ -1483,6 +1492,13 @@ export function createAppStore(opts?: CreateAppStoreOptions): AppStore { if (!res.ok) return; const data = (await res.json()) as ConversationListResponse; + if (!data || !Array.isArray(data.conversations)) { + reportError( + "Failed to load open conversations", + new Error("Invalid conversations response"), + ); + return; + } // Update the status map from the authoritative backend list. const newStatuses = new Map(); for (const conv of data.conversations) { @@ -1628,27 +1644,17 @@ export function createAppStore(opts?: CreateAppStoreOptions): AppStore { } }, onConversationCompacted(msg: ConversationCompactedMessage): void { - // Compaction keeps the conversation ID — the old full history is forked - // to an archive (newConversationId). Just reload the same conversation's - // history (dispose stale store + cache + re-fetch). const cid = msg.conversationId; - const wasActive = tabsStore.activeConversationId === cid; - void cache.delete(cid); - if (!chatStores.has(cid) || !loadedConversationIds.delete(cid)) return; - const store = chatStores.get(cid); - if (store !== undefined) { - store.dispose(); - } - const tab = tabsStore.tabs.find((t) => t.conversationId === cid); - const fresh = createChatFor( - cid, - tab?.model ?? activeModel, - tab?.workspaceId ?? activeWorkspaceId, - ); - chatStores.set(cid, fresh); - ensureLoaded(cid); - if (wasActive) { - refreshActiveChat(); + const store = chatStores.get(cid) ?? watchStores.get(cid); + if (store && loadedConversationIds.has(cid)) { + store.resync(msg.historyEpoch); + } else { + store?.invalidateHistory(msg.historyEpoch); + // Unseen conversations stay lazy. Invalidation is queued before their + // next load; epoch-guarded commits reject any older in-flight response. + void cache + .reset(cid, msg.historyEpoch) + .catch((err) => reportError("Failed to reset conversation cache", err)); } }, onReopen() { diff --git a/src/app/store.test.ts b/src/app/store.test.ts index f5d4723..f257f3a 100644 --- a/src/app/store.test.ts +++ b/src/app/store.test.ts @@ -195,7 +195,8 @@ function fakeFetchImpl(opts?: FakeFetchOptions): typeof fetch { } if (/^\/conversations\/[^/]+$/.test(pathname) && searchParams.has("sinceSeq")) { return Response.json( - history[url] ?? ({ chunks: [], latestSeq: 0 } satisfies ConversationHistoryResponse), + history[url] ?? + ({ chunks: [], historyEpoch: 0, latestSeq: 0 } satisfies ConversationHistoryResponse), ); } } @@ -682,6 +683,7 @@ describe("createAppStore", () => { { seq: 1, role: "user", chunk: { type: "text", text: "hi" } }, { seq: 2, role: "assistant", chunk: { type: "text", text: "hello!" } }, ], + historyEpoch: 0, latestSeq: 2, }; const base = fakeFetchImpl(); @@ -2743,6 +2745,7 @@ describe("conversation load-on-demand", () => { { seq: 1, role: "user", turnId: "t1", chunk: { type: "text", text: "hi" } }, { seq: 2, role: "assistant", turnId: "t1", chunk: { type: "text", text: "hello" } }, ], + historyEpoch: 0, latestSeq: 2, }); } @@ -2907,12 +2910,21 @@ describe("conversation load-on-demand", () => { const { store, requests } = await setup(ws); try { ws.resolveOpen(); + ws.feedServerMessage({ + type: "chat.delta", + event: { type: "turn-start", conversationId: "lazy-1", turnId: "old" }, + }); + ws.feedServerMessage({ + type: "chat.delta", + event: { type: "text-delta", conversationId: "lazy-1", turnId: "old", delta: "stale" }, + }); for (const id of ["lazy-1", "unknown", "lazy-0"]) { ws.feedServerMessage({ type: "conversation.compacted", conversationId: id, newConversationId: `${id}-archive`, messagesSummarized: 10, + historyEpoch: 0, messagesKept: 2, }); } @@ -2957,3 +2969,25 @@ describe("conversation load-on-demand", () => { } }); }); + +it.each([ + null, + {}, + { conversations: {} }, +])("reports malformed conversation lists: %j", async (body) => { + const ws = fakeSocket(); + const base = fakeFetchImpl(); + const store = createAppStore({ + socketFactory: () => ws, + indexedDB: new IDBFactory(), + localStorage: createFakeStorage(), + fetchImpl: async (input, init) => + String(input).includes("/conversations?status=") ? Response.json(body) : base(input, init), + }); + try { + ws.resolveOpen(); + await vi.waitFor(() => expect(store.fatalError).toContain("Invalid conversations response")); + } finally { + store.dispose(); + } +}); diff --git a/src/core/chunks/trim.test.ts b/src/core/chunks/trim.test.ts index 7c4bbef..24c75fd 100644 --- a/src/core/chunks/trim.test.ts +++ b/src/core/chunks/trim.test.ts @@ -146,7 +146,7 @@ describe("windowTranscript", () => { it("is the identity within the window", () => { const state = stateWith(chunks(1, 50)); expect(windowTranscript(state, 75)).toBe(state); - expect(selectHasEarlier(state)).toBe(false); + expect(selectHasEarlier(state, 1)).toBe(false); }); }); @@ -178,21 +178,21 @@ describe("restoreEarlier", () => { expect(selectHasEarlier(restored)).toBe(true); }); - it("restoring down to seq 1 reaches the contractual origin (hasEarlier clears)", () => { + it("restoring down to a known boundary clears hasEarlier", () => { const windowed = windowTranscript(stateWith(chunks(1, 100)), 75); // hidden: 1..25 const restored = restoreEarlier(windowed, chunks(1, 100), 64); expect(restored.committed).toHaveLength(100); expect(restored.committed[0]?.seq).toBe(1); expect(restored.hiddenBeforeSeq).toBe(1); // floor at the origin — inert expect(restored.hiddenThinkingCount).toBe(0); - expect(selectHasEarlier(restored)).toBe(false); + expect(selectHasEarlier(restored, 1)).toBe(false); }); it("is the identity when nothing older is known locally (server may still hold more)", () => { const windowed = windowTranscript(stateWith(chunks(50, 200)), 75); const restored = restoreEarlier(windowed, [], 64); expect(restored).toBe(windowed); - // seqs are 1-based gap-free: window starts at 126 ⇒ older chunks DO exist. + // The server boundary is unknown, so another page may exist. expect(selectHasEarlier(restored)).toBe(true); }); @@ -203,7 +203,7 @@ describe("restoreEarlier", () => { it("works on a server-windowed transcript (no local watermark)", () => { // A cold-cache fresh load with `?limit=` commits a suffix (seq 809..1000) - // with hiddenBeforeSeq still 0 — hasEarlier derives from seq > 1, and a + // with hiddenBeforeSeq still 0 — the server boundary is unknown, and a // backfilled run merges below it. const state = stateWith(chunks(809, 1000)); expect(state.hiddenBeforeSeq).toBe(0); @@ -230,6 +230,15 @@ describe("restoreEarlier", () => { const restored = restoreEarlier(trimmed, original, 1000); expect(restored.committed).toEqual(original); expect(restored.hiddenBeforeSeq).toBe(1); - expect(selectHasEarlier(restored)).toBe(false); + expect(selectHasEarlier(restored, 1)).toBe(false); }); }); + +it("uses a non-1 history boundary rather than the seq origin", () => { + const state = stateWith(chunks(101, 110)); + expect(selectHasEarlier(state, 101)).toBe(false); + expect(selectHasEarlier(state, null)).toBe(true); + const windowed = windowTranscript(state, 5); + expect(selectHasEarlier(windowed, 101)).toBe(true); + expect(selectHasEarlier(restoreEarlier(windowed, state.committed, 5), 101)).toBe(false); +}); diff --git a/src/core/chunks/trim.ts b/src/core/chunks/trim.ts index 9846357..259c101 100644 --- a/src/core/chunks/trim.ts +++ b/src/core/chunks/trim.ts @@ -150,8 +150,7 @@ function oldestLoadedSeq(state: TranscriptState): number { * `?beforeSeq=` backfill; chunks at/inside the window are ignored). The newest * `count` of them are merged back in front of `committed`, and the watermark * follows the new window start so history merges still can't resurrect what - * remains unloaded. Identity when the window already starts at seq 1 (the - * contractual origin) or nothing older is known locally. + * remains unloaded. Identity when nothing older is known locally. */ export function restoreEarlier( state: TranscriptState, @@ -159,7 +158,7 @@ export function restoreEarlier( count: number, ): TranscriptState { const oldest = oldestLoadedSeq(state); - if (oldest <= 1) return state; + if (oldest === 0) return state; const below = earlier.filter((c) => c.seq < oldest).sort((a, b) => a.seq - b.seq); if (below.length === 0) return state; const keep = below.slice(-Math.max(1, count)); @@ -172,14 +171,11 @@ export function restoreEarlier( }; } -/** - * Whether earlier history exists below the loaded window — drives the - * "Show earlier messages" affordance. Derived from the wire@0.6.1 CONTRACT - * that per-conversation seqs are 1-based and gap-free: a loaded window that - * starts above seq 1 means older chunks exist (locally cached or server-side), - * whether the window came from a local trim or a server-windowed (`?limit=`) - * fresh load. - */ -export function selectHasEarlier(state: TranscriptState): boolean { - return oldestLoadedSeq(state) > 1; +/** Earlier availability uses the known server boundary, never a seq origin. */ +export function selectHasEarlier( + state: TranscriptState, + earliestSeq: number | null = null, +): boolean { + const oldest = oldestLoadedSeq(state); + return oldest > 0 && (earliestSeq === null || oldest > earliestSeq); } diff --git a/src/core/wire/conformance.test.ts b/src/core/wire/conformance.test.ts index 2b98ca6..e74a663 100644 --- a/src/core/wire/conformance.test.ts +++ b/src/core/wire/conformance.test.ts @@ -184,6 +184,7 @@ describe("classifies every WsServerMessage type", () => { conversationId: "c1", newConversationId: "c2", messagesSummarized: 10, + historyEpoch: 0, messagesKept: 5, }, ]; @@ -268,6 +269,7 @@ describe("ConversationHistoryResponse shape is constructible", () => { it("constructs a response with chunks", () => { const resp: ConversationHistoryResponse = { chunks: [{ seq: 1, role: "user", chunk: { type: "text", text: "hi" } }], + historyEpoch: 0, latestSeq: 1, }; expect(resp.chunks).toHaveLength(1); @@ -275,7 +277,7 @@ describe("ConversationHistoryResponse shape is constructible", () => { }); it("constructs an empty (caught-up) response", () => { - const resp: ConversationHistoryResponse = { chunks: [], latestSeq: 5 }; + const resp: ConversationHistoryResponse = { chunks: [], historyEpoch: 0, latestSeq: 5 }; expect(resp.chunks).toHaveLength(0); expect(resp.latestSeq).toBe(5); }); diff --git a/src/features/chat/store.svelte.ts b/src/features/chat/store.svelte.ts index 01ad9f9..dab1ba2 100644 --- a/src/features/chat/store.svelte.ts +++ b/src/features/chat/store.svelte.ts @@ -209,7 +209,9 @@ export interface ChatStore { * running, the server's post-subscribe replay re-asserts `generating`. The * app store pairs this with a `chat.subscribe` for the conversation. */ - resync(): void; + resync(historyEpoch?: number): void; + /** Discard pre-replacement transcript state without loading an unseen conversation. */ + invalidateHistory(historyEpoch: number): void; dispose(): void; } @@ -223,6 +225,23 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { let cacheTimer = $state(initialCacheTimerState()); const now = deps.now ?? Date.now; let disposed = false; + let rerunRequested = false; + let historyEpoch: number | undefined; + let revision = 0; + let reloadWindow = false; + let earliestSeq = $state(null); + + function resetTranscript(epoch: number): void { + historyEpoch = epoch; + revision++; + reloadWindow = true; + transcript = initialState(); + earliestSeq = null; + pendingEvents = []; + } + function hasEarlier(): boolean { + return selectHasEarlier(transcript, earliestSeq); + } /** The most recent turn's id (resolves a seeded turn gap when it is stamped). */ function lastTurnIdOf(m: MetricsState): string | undefined { @@ -313,21 +332,58 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { * between the cache and the fetched window. */ async function syncTail(coldLimit?: number): Promise { - if (disposed || _pendingSync) return; - _pendingSync = true; - try { - const since = await deps.cache.sinceSeq(deps.conversationId); - const window = since === 0 && coldLimit !== undefined ? { limit: coldLimit } : undefined; - const res = await deps.historySync(deps.conversationId, since, window); - const merged = await deps.cache.commit(deps.conversationId, res.chunks); - transcript = applyHistory(transcript, merged); - maybeTrim(); - _error = null; - } catch (err) { - _error = err instanceof Error ? err.message : String(err); - } finally { - _pendingSync = false; + if (disposed) return; + if (_pendingSync) { + rerunRequested = true; + return; } + _pendingSync = true; + do { + rerunRequested = false; + try { + const currentRevision = revision; + const cachedEpoch = await deps.cache.historyEpoch(deps.conversationId); + if (historyEpoch !== undefined && cachedEpoch !== historyEpoch) + await deps.cache.reset(deps.conversationId, historyEpoch); + const since = + cachedEpoch === undefined ? 0 : await deps.cache.sinceSeq(deps.conversationId); + const limit = + since === 0 ? (reloadWindow ? initialWindowSize(chatLimit) : coldLimit) : undefined; + const res = await deps.historySync( + deps.conversationId, + since, + limit === undefined ? undefined : { limit }, + ); + if (disposed) break; + if (revision !== currentRevision) { + rerunRequested = true; + continue; + } + if ( + (cachedEpoch !== undefined && cachedEpoch !== res.historyEpoch) || + (historyEpoch !== undefined && historyEpoch !== res.historyEpoch) + ) { + resetTranscript(res.historyEpoch); + await deps.cache.reset(deps.conversationId, res.historyEpoch); + if (since !== 0) { + 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); + maybeTrim(); + _error = null; + } catch (err) { + _error = err instanceof Error ? err.message : String(err); + } + } while (rerunRequested && !disposed); + _pendingSync = false; } async function syncMetrics(): Promise { @@ -351,14 +407,25 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { * newest `count` of them). Shared by `showEarlier` and the raise-refill. */ async function backfillOlder(oldest: number, want: number): Promise { + const currentRevision = revision; let earlier = (await deps.cache.load(deps.conversationId)).filter((c) => c.seq < oldest); + if (disposed || revision !== currentRevision) return []; const oldestKnown = earlier[0]?.seq ?? oldest; - if (earlier.length < want && oldestKnown > 1) { + if (earlier.length < want && (earliestSeq === null || oldestKnown > earliestSeq)) { const res = await deps.historySync(deps.conversationId, 0, { beforeSeq: oldestKnown, limit: want - earlier.length, }); - const merged = await deps.cache.commit(deps.conversationId, res.chunks); + if (disposed || revision !== currentRevision) return []; + if (res.historyEpoch !== historyEpoch) { + resetTranscript(res.historyEpoch); + await deps.cache.reset(deps.conversationId, res.historyEpoch); + await syncTail(initialWindowSize(chatLimit)); + return []; + } + const merged = await deps.cache.commit(deps.conversationId, res.chunks, res.historyEpoch); + if (disposed || revision !== currentRevision) return []; + if (res.chunks.length < want - earlier.length) earliestSeq = merged[0]?.seq ?? 0; earlier = merged.filter((c) => c.seq < oldest); } return earlier; @@ -368,15 +435,15 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { * Refill toward the fresh-load window after a limit RAISE: pull older * history (cache first, then server) so the loaded set grows to match what a * fresh `load()` would show at the new limit. No-op when already at the - * origin (seq 1) or already within the window. `restoreEarlier` re-derives + * known history boundary or already within the window. `restoreEarlier` re-derives * the window start at apply time, so a delta landing during the await can't * corrupt the merge. NOT gated (refilling prepends above the viewport; the * caller preserves scroll position). */ async function refill(): Promise { if (disposed) return; + if (!hasEarlier()) return; const oldest = transcript.committed[0]?.seq ?? transcript.hiddenBeforeSeq; - if (oldest <= 1) return; const want = initialWindowSize(chatLimit) - transcript.committed.length; if (want <= 0) return; try { @@ -418,7 +485,7 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { return _model; }, get hasEarlier(): boolean { - return selectHasEarlier(transcript); + return hasEarlier(); }, get thinkingKeyBase(): number { return transcript.hiddenThinkingCount; @@ -524,8 +591,11 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { // server instead (CR-5 `?limit=`), so a huge conversation never ships // whole. The post-sync window re-asserts the cap either way. const windowSize = initialWindowSize(chatLimit); + const currentRevision = revision; + const cachedEpoch = await deps.cache.historyEpoch(deps.conversationId); const cached = await deps.cache.load(deps.conversationId); - if (cached.length > 0) { + if (!disposed && revision === currentRevision && cachedEpoch !== undefined) { + historyEpoch = cachedEpoch; transcript = windowTranscript(applyHistory(transcript, cached), windowSize); } await syncTail(windowSize); @@ -536,7 +606,7 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { async showEarlier(): Promise { if (disposed) return; const oldest = transcript.committed[0]?.seq ?? transcript.hiddenBeforeSeq; - if (oldest <= 1) return; + if (!hasEarlier()) return; const want = unloadCount(chatLimit); try { const earlier = await backfillOlder(oldest, want); @@ -547,8 +617,14 @@ export function createChatStore(deps: ChatStoreDependencies): ChatStore { } }, - resync(): void { + invalidateHistory(epoch): void { + if (!disposed && (historyEpoch === undefined || epoch > historyEpoch)) resetTranscript(epoch); + }, + + resync(epoch?: number): void { if (disposed) return; + if (epoch !== undefined && (historyEpoch === undefined || epoch > historyEpoch)) + resetTranscript(epoch); // A turn may have sealed while we were disconnected (missed `turn-sealed`): // clear the now-stale spinner BEFORE re-subscribing, so a finished turn // doesn't spin forever. A still-running turn's replay re-asserts it. diff --git a/src/features/chat/store.test.ts b/src/features/chat/store.test.ts index 8604107..bb1e304 100644 --- a/src/features/chat/store.test.ts +++ b/src/features/chat/store.test.ts @@ -1,5 +1,8 @@ import type { AgentEvent, ImageInput, StepId, StoredChunk } from "@dispatch/wire"; +import { IDBFactory, IDBKeyRange } from "fake-indexeddb"; import { describe, expect, it, vi } from "vitest"; +import { createIdbChunkStore } from "../../adapters/idb"; +import { createConversationCache } from "../conversation-cache"; import { createChatStore } from "./store.svelte"; import { createFakeCache, @@ -528,7 +531,7 @@ describe("createChatStore", () => { store.dispose(); }); - it("overlapping syncs are guarded", async () => { + it("overlapping syncs coalesce into a sequential rerun", async () => { const transport = createFakeTransport(); const _historySync = createFakeHistorySync(); const metricsSync = createFakeMetricsSync(); @@ -546,7 +549,7 @@ describe("createChatStore", () => { if (callCount === 1) { await firstSyncPromise; } - return { chunks: [], latestSeq: sinceSeq }; + return { chunks: [], latestSeq: sinceSeq, historyEpoch: 0 }; }; const store = createChatStore({ @@ -573,7 +576,7 @@ describe("createChatStore", () => { // Release the first sync resolveFirstSync?.(); - await new Promise((r) => setTimeout(r, 10)); + await vi.waitFor(() => expect(callCount).toBe(2)); store.dispose(); }); @@ -1645,6 +1648,7 @@ describe("createChatStore", () => { }); await store.load(); // only 50 chunks → all loaded, window starts at seq 1 expect(store.chunks).toHaveLength(50); + await store.showEarlier(); // Probe server exhaustion; seq 1 alone is not a boundary. expect(store.hasEarlier).toBe(false); const callsAfterLoad = historySync.calls.length; @@ -1892,3 +1896,96 @@ describe("createChatStore — cacheTimer", () => { busy.dispose(); }); }); + +describe("history epochs", () => { + it.each([ + "tail", + "backfill", + ])("recovers missed compaction on %s and exhausts non-1 history", async (path) => { + const history = createFakeHistorySync(); + const cache = createFakeCache(); + history.returnChunks = Array.from({ length: 20 }, (_, i) => makeStoredChunk(i + 1)); + const store = createChatStore({ + conversationId: CONV_ID, + transport: createFakeTransport().impl, + historySync: history.impl, + metricsSync: createFakeMetricsSync().impl, + cache: cache.impl, + chatLimit: 10, + }); + await store.load(); + history.historyEpoch = 1; + history.returnChunks = [makeStoredChunk(101), makeStoredChunk(102)]; + if (path === "tail") { + store.resync(); + await vi.waitFor(() => expect(store.chunks[0]?.seq).toBe(101)); + } else await store.showEarlier(); + expect(store.chunks.map((c) => c.seq)).toEqual([101, 102]); + expect((await cache.impl.load(CONV_ID)).map((c) => c.seq)).toEqual([101, 102]); + expect(history.calls.at(-1)).toMatchObject({ sinceSeq: 0, window: { limit: 7 } }); + expect(store.hasEarlier).toBe(false); + const calls = history.calls.length; + await store.showEarlier(); + expect(history.calls).toHaveLength(calls); + store.dispose(); + }); +}); + +it("a compaction broadcast during an in-flight fetch discards the old response", async () => { + const history = createFakeHistorySync(); + history.returnChunks = [makeStoredChunk(101)]; + let resolve!: (res: import("@dispatch/transport-contract").ConversationHistoryResponse) => void; + const first = new Promise( + (r) => { + resolve = r; + }, + ); + const sync = vi + .fn() + .mockImplementationOnce(() => first) + .mockImplementation(history.impl); + const cache = createFakeCache(); + const store = createChatStore({ + conversationId: CONV_ID, + transport: createFakeTransport().impl, + historySync: sync, + metricsSync: createFakeMetricsSync().impl, + cache: cache.impl, + }); + const loading = store.load(); + await vi.waitFor(() => expect(sync).toHaveBeenCalledTimes(1)); + history.historyEpoch = 1; + store.resync(1); + resolve({ chunks: [makeStoredChunk(1)], latestSeq: 1, historyEpoch: 0 }); + await loading; + expect(sync).toHaveBeenCalledTimes(2); + expect(store.chunks.map((c) => c.seq)).toEqual([101]); + expect((await cache.impl.load(CONV_ID)).map((c) => c.seq)).toEqual([101]); + store.dispose(); +}); + +it("reloads unknown legacy cache epochs rather than merging pre-upgrade rows", async () => { + vi.stubGlobal("IDBKeyRange", IDBKeyRange); + const storage = createIdbChunkStore({ indexedDB: new IDBFactory() }); + const cache = createConversationCache(storage); + await storage.append(CONV_ID, [makeStoredChunk(1)]); + const history = createFakeHistorySync(); + history.returnChunks = [makeStoredChunk(101)]; + const store = createChatStore({ + conversationId: CONV_ID, + transport: createFakeTransport().impl, + historySync: history.impl, + metricsSync: createFakeMetricsSync().impl, + cache, + }); + try { + await store.load(); + expect(history.calls[0]).toMatchObject({ sinceSeq: 0 }); + expect(store.chunks.map((c) => c.seq)).toEqual([101]); + expect(await cache.historyEpoch(CONV_ID)).toBe(0); + expect((await cache.load(CONV_ID)).map((c) => c.seq)).toEqual([101]); + } finally { + store.dispose(); + vi.unstubAllGlobals(); + } +}); diff --git a/src/features/chat/test-helpers.ts b/src/features/chat/test-helpers.ts index c99d1f4..c1b020b 100644 --- a/src/features/chat/test-helpers.ts +++ b/src/features/chat/test-helpers.ts @@ -43,14 +43,22 @@ export interface FakeHistorySync { readonly calls: Array<{ conversationId: string; sinceSeq: number; window?: HistoryWindow }>; /** Set the chunks to return on the next call. */ returnChunks: readonly StoredChunk[]; + historyEpoch: number; readonly impl: HistorySync; } export function createFakeHistorySync(): FakeHistorySync { const calls: Array<{ conversationId: string; sinceSeq: number; window?: HistoryWindow }> = []; let returnChunks: readonly StoredChunk[] = []; + let historyEpoch = 0; return { calls, + get historyEpoch() { + return historyEpoch; + }, + set historyEpoch(v: number) { + historyEpoch = v; + }, get returnChunks() { return returnChunks; }, @@ -72,7 +80,7 @@ export function createFakeHistorySync(): FakeHistorySync { chunks = chunks.slice(-window.limit); } const latestSeq = chunks.length > 0 ? Math.max(...chunks.map((c) => c.seq)) : sinceSeq; - return { chunks, latestSeq }; + return { chunks, latestSeq, historyEpoch }; }, }; } @@ -122,13 +130,22 @@ export interface FakeCache { export function createFakeCache(): FakeCache { const store = new Map(); + const epochs = new Map(); return { store, impl: { + async historyEpoch(id) { + return epochs.get(id) ?? 0; + }, + async reset(id, epoch) { + store.delete(id); + epochs.set(id, epoch); + }, async load(conversationId) { return store.get(conversationId) ?? []; }, - async commit(conversationId, incoming) { + async commit(conversationId, incoming, epoch) { + if (epoch !== undefined) epochs.set(conversationId, epoch); const existing = store.get(conversationId) ?? []; const seen = new Set(existing.map((c) => c.seq)); const toAppend = incoming.filter((c) => !seen.has(c.seq)); diff --git a/src/features/conversation-cache/cache.test.ts b/src/features/conversation-cache/cache.test.ts index ce1c60c..4306d51 100644 --- a/src/features/conversation-cache/cache.test.ts +++ b/src/features/conversation-cache/cache.test.ts @@ -15,8 +15,16 @@ const chunk = (seq: number, role: "user" | "assistant" = "user"): StoredChunk => */ function createFakeStore(): ConversationChunkStore { const store = new Map(); + const epochs = new Map(); return { + async historyEpoch(id) { + return epochs.get(id); + }, + async reset(id, epoch) { + store.delete(id); + epochs.set(id, epoch); + }, async load(conversationId) { return store.get(conversationId) ?? []; }, @@ -33,6 +41,7 @@ function createFakeStore(): ConversationChunkStore { async delete(conversationId) { store.delete(conversationId); + epochs.delete(conversationId); }, async index() { @@ -195,3 +204,13 @@ describe("cache.delete", () => { expect(result).toEqual([]); }); }); + +it("serializes invalidation and rejects stale epoch commits", async () => { + const store = createFakeStore(); + const cache = createConversationCache(store); + await store.append("c", [chunk(1)]); // Legacy rows have no trusted epoch. + expect(await cache.commit("c", [chunk(101)], 1)).toEqual([chunk(101)]); + await Promise.all([cache.commit("c", [chunk(102)], 1), cache.reset("c", 2)]); + expect(await cache.commit("c", [chunk(103)], 1)).toEqual([]); + expect(await store.historyEpoch("c")).toBe(2); +}); diff --git a/src/features/conversation-cache/cache.ts b/src/features/conversation-cache/cache.ts index 4953944..431a2a7 100644 --- a/src/features/conversation-cache/cache.ts +++ b/src/features/conversation-cache/cache.ts @@ -3,6 +3,9 @@ import { nextSinceSeq, reconcileCache, selectEvictions } from "./logic"; import type { ConversationChunkStore } from "./types"; export interface ConversationCache { + historyEpoch(conversationId: string): Promise; + reset(conversationId: string, historyEpoch: number): Promise; + /** Load all cached chunks for a conversation. */ load(conversationId: string): Promise; @@ -10,7 +13,11 @@ export interface ConversationCache { * Load + reconcile + append new chunks. * Returns the merged cache (the new authoritative cache for this conversation). */ - commit(conversationId: string, incoming: readonly StoredChunk[]): Promise; + commit( + conversationId: string, + incoming: readonly StoredChunk[], + historyEpoch?: number, + ): Promise; /** Return the `?sinceSeq=` cursor for the next incremental sync. */ sinceSeq(conversationId: string): Promise; @@ -43,21 +50,52 @@ export function createConversationCache( ): ConversationCache { const maxChunks = opts?.maxChunks ?? DEFAULT_MAX_CHUNKS; + // Serialize mutation per conversation, including invalidation, so an old fetch + // cannot resurrect rows after a compaction broadcast has invalidated them. + const pending = new Map>(); + function exclusive(id: string, run: () => Promise): Promise { + const next = (pending.get(id) ?? Promise.resolve()).catch(() => {}).then(run); + pending.set(id, next); + void next + .finally(() => { + if (pending.get(id) === next) pending.delete(id); + }) + .catch(() => {}); + return next; + } return { + historyEpoch: async (id) => { + await pending.get(id); + return store.historyEpoch(id); + }, + reset: (id, epoch) => + exclusive(id, async () => { + const current = await store.historyEpoch(id); + if (current === undefined || current < epoch) await store.reset(id, epoch); + }), async load(conversationId) { + await pending.get(conversationId); return store.load(conversationId); }, - async commit(conversationId, incoming) { - const cached = await store.load(conversationId); - const { merged, toAppend } = reconcileCache(cached, incoming); - if (toAppend.length > 0) { - await store.append(conversationId, toAppend); - } - return merged; + commit(conversationId, incoming, epoch) { + return exclusive(conversationId, async () => { + if (epoch !== undefined) { + const current = await store.historyEpoch(conversationId); + if (current !== undefined && current > epoch) return store.load(conversationId); + if (current !== epoch) await store.reset(conversationId, epoch); + } + const cached = await store.load(conversationId); + const { merged, toAppend } = reconcileCache(cached, incoming); + if (toAppend.length > 0) { + await store.append(conversationId, toAppend); + } + return merged; + }); }, async sinceSeq(conversationId) { + await pending.get(conversationId); const cached = await store.load(conversationId); return nextSinceSeq(cached); }, @@ -72,7 +110,7 @@ export function createConversationCache( }, async delete(conversationId) { - await store.delete(conversationId); + await exclusive(conversationId, () => store.delete(conversationId)); }, }; } diff --git a/src/features/conversation-cache/types.ts b/src/features/conversation-cache/types.ts index 345ab71..a273f34 100644 --- a/src/features/conversation-cache/types.ts +++ b/src/features/conversation-cache/types.ts @@ -17,6 +17,10 @@ export interface ConversationCacheIndexEntry { * All methods MUST be idempotent on `seq`: re-appending an existing seq is a no-op. */ export interface ConversationChunkStore { + historyEpoch(conversationId: string): Promise; + /** Atomically replace cached rows with empty history at this epoch. */ + reset(conversationId: string, historyEpoch: number): Promise; + /** Load all cached chunks for a conversation, seq-ordered. Returns [] if absent. */ load(conversationId: string): Promise; diff --git a/src/features/layout/adapter/http.test.ts b/src/features/layout/adapter/http.test.ts index cc7d7e9..30e194a 100644 --- a/src/features/layout/adapter/http.test.ts +++ b/src/features/layout/adapter/http.test.ts @@ -77,7 +77,7 @@ describe("createLayoutHttp — getTree", () => { it("a malformed 200 (SPA-fallback JSON without a tree) degrades to notFound", async () => { const http = createLayoutHttp( "http://backend.test", - fetchReturning(200, { chunks: [], latestSeq: 0 }), + fetchReturning(200, { chunks: [], historyEpoch: 0, latestSeq: 0 }), ); const result = await http.getTree("ws-a"); expect(result).toMatchObject({ ok: false, notFound: true }); -- cgit v1.2.3 From 0c0f366b3820136c9877f80005939011c30b2de4 Mon Sep 17 00:00:00 2001 From: Adam Malczewski Date: Wed, 21 Oct 2026 02:56:48 +0900 Subject: fix(review-4.4): tolerate additive terminal turn events --- src/core/chunks/reducer.test.ts | 10 ++++++++++ src/core/chunks/reducer.ts | 3 +++ src/core/wire/conformance.test.ts | 6 ++++-- src/core/wire/conformance.ts | 2 ++ 4 files changed, 19 insertions(+), 2 deletions(-) (limited to 'src/core') diff --git a/src/core/chunks/reducer.test.ts b/src/core/chunks/reducer.test.ts index cba6fad..ae0aa73 100644 --- a/src/core/chunks/reducer.test.ts +++ b/src/core/chunks/reducer.test.ts @@ -1051,3 +1051,13 @@ describe("turnId provenance on live chunks", () => { expect(selectChunks(s).map((c) => c.turnId)).toEqual(["t9", undefined]); }); }); + +describe("foldEvent — additive wire compatibility", () => { + it.each(["turn-ended", "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" }; + expect(foldEvent(state, event as Parameters[1])).toBe(state); + expect(state).toEqual(before); + }); +}); diff --git a/src/core/chunks/reducer.ts b/src/core/chunks/reducer.ts index 4eb08c1..0af2ddd 100644 --- a/src/core/chunks/reducer.ts +++ b/src/core/chunks/reducer.ts @@ -408,6 +408,9 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState // (already true from `turn-start`) is left untouched. return { ...state, providerRetry: event }; } + + default: + return state; // Forward-compatible with additive wire events. } } diff --git a/src/core/wire/conformance.test.ts b/src/core/wire/conformance.test.ts index e74a663..1c5829b 100644 --- a/src/core/wire/conformance.test.ts +++ b/src/core/wire/conformance.test.ts @@ -75,6 +75,7 @@ describe("classifies every AgentEvent type", () => { { type: "error", conversationId: "c1", turnId: "t1", message: "oops" }, { type: "done", conversationId: "c1", turnId: "t1", reason: "complete" }, { type: "turn-sealed", conversationId: "c1", turnId: "t1" }, + { type: "turn-ended", conversationId: "c1", turnId: "t1", reason: "sealed" }, { type: "steering", conversationId: "c1", turnId: "t1", text: "steer mid-turn" }, { type: "provider-retry", @@ -103,13 +104,14 @@ describe("classifies every AgentEvent type", () => { "error", "done", "turn-sealed", + "turn-ended", "steering", "provider-retry", ]); }); - it("covers all 15 AgentEvent variants", () => { - expect(samples).toHaveLength(15); + it("covers all 16 AgentEvent variants", () => { + expect(samples).toHaveLength(16); }); }); diff --git a/src/core/wire/conformance.ts b/src/core/wire/conformance.ts index 855f7ea..9847081 100644 --- a/src/core/wire/conformance.ts +++ b/src/core/wire/conformance.ts @@ -32,6 +32,8 @@ export function assertAgentEventExhaustive(event: AgentEvent): string { return "done"; case "turn-sealed": return "turn-sealed"; + case "turn-ended": + return "turn-ended"; case "step-complete": return "step-complete"; case "steering": -- 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/core') 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/core') 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"}