diff options
Diffstat (limited to 'src/core')
28 files changed, 5683 insertions, 1009 deletions
diff --git a/src/core/chunks/groups.test.ts b/src/core/chunks/groups.test.ts new file mode 100644 index 0000000..c8b0fa2 --- /dev/null +++ b/src/core/chunks/groups.test.ts @@ -0,0 +1,125 @@ +import type { Role, StepId } from "@dispatch/wire"; +import { describe, expect, it } from "vitest"; +import { groupRenderedChunks } from "./groups"; +import type { RenderedChunk } from "./types"; + +const text = (seq: number, role: Role, t: string, provisional = false): RenderedChunk => ({ + seq, + role, + chunk: { type: "text", text: t }, + provisional, +}); + +const call = (seq: number, id: string, stepId?: string, provisional = false): RenderedChunk => ({ + seq, + role: "assistant", + chunk: { + type: "tool-call", + toolCallId: id, + toolName: `tool-${id}`, + input: { id }, + ...(stepId !== undefined ? { stepId: stepId as StepId } : {}), + }, + provisional, +}); + +const result = (seq: number, id: string, stepId?: string, provisional = false): RenderedChunk => ({ + seq, + role: "tool", + chunk: { + type: "tool-result", + toolCallId: id, + toolName: `tool-${id}`, + content: `result-${id}`, + isError: false, + ...(stepId !== undefined ? { stepId: stepId as StepId } : {}), + }, + provisional, +}); + +describe("groupRenderedChunks", () => { + it("returns no groups for an empty stream", () => { + expect(groupRenderedChunks([])).toEqual([]); + }); + + it("passes non-tool chunks through as single groups, in order", () => { + const groups = groupRenderedChunks([text(1, "user", "hi"), text(2, "assistant", "hello")]); + expect(groups).toHaveLength(2); + expect(groups.every((g) => g.kind === "single")).toBe(true); + }); + + it("does NOT batch a single tool call (one per step) — call+result stay separate singles", () => { + const groups = groupRenderedChunks([call(1, "a", "s1"), result(2, "a", "s1")]); + expect(groups).toHaveLength(2); + expect(groups.map((g) => g.kind)).toEqual(["single", "single"]); + }); + + it("does NOT batch tool calls that have no stepId (pre-0.2.0 replay)", () => { + const groups = groupRenderedChunks([ + call(1, "a"), + call(2, "b"), + result(3, "a"), + result(4, "b"), + ]); + expect(groups).toHaveLength(4); + expect(groups.every((g) => g.kind === "single")).toBe(true); + }); + + it("batches 2+ calls sharing a stepId into one group, pairing each with its result", () => { + const groups = groupRenderedChunks([ + call(1, "a", "s1"), + call(2, "b", "s1"), + result(3, "a", "s1"), + result(4, "b", "s1"), + ]); + expect(groups).toHaveLength(1); + const g = groups[0]; + if (g?.kind !== "tool-batch") throw new Error("expected a tool-batch group"); + expect(g.stepId).toBe("s1"); + expect(g.entries).toHaveLength(2); + expect(g.entries[0]?.call.toolCallId).toBe("a"); + expect(g.entries[0]?.result?.content).toBe("result-a"); + expect(g.entries[1]?.call.toolCallId).toBe("b"); + expect(g.entries[1]?.result?.content).toBe("result-b"); + }); + + it("positions the batch at the first call and keeps surrounding chunks in order", () => { + const groups = groupRenderedChunks([ + text(1, "assistant", "before"), + call(2, "a", "s1"), + call(3, "b", "s1"), + result(4, "a", "s1"), + result(5, "b", "s1"), + text(6, "assistant", "after"), + ]); + expect(groups.map((g) => g.kind)).toEqual(["single", "tool-batch", "single"]); + }); + + it("marks the batch provisional when any of its calls/results is provisional", () => { + const groups = groupRenderedChunks([call(1, "a", "s1"), call(2, "b", "s1", true)]); + const g = groups[0]; + if (g?.kind !== "tool-batch") throw new Error("expected a tool-batch group"); + expect(g.provisional).toBe(true); + expect(g.entries).toHaveLength(2); + expect(g.entries[1]?.result).toBeNull(); // dangling call (no result yet) + }); + + it("batches one step while leaving a different single-call step ungrouped", () => { + const groups = groupRenderedChunks([ + call(1, "a", "s1"), + call(2, "b", "s1"), + call(3, "c", "s2"), + result(4, "a", "s1"), + result(5, "b", "s1"), + result(6, "c", "s2"), + ]); + expect(groups.map((g) => g.kind)).toEqual(["tool-batch", "single", "single"]); + const batch = groups[0]; + if (batch?.kind !== "tool-batch") throw new Error("expected a tool-batch group"); + expect(batch.entries).toHaveLength(2); + // the s2 single call + its result remain as separate single groups + const singles = groups.slice(1); + expect(singles[0]?.kind === "single" && singles[0].chunk.chunk.type).toBe("tool-call"); + expect(singles[1]?.kind === "single" && singles[1].chunk.chunk.type).toBe("tool-result"); + }); +}); diff --git a/src/core/chunks/groups.ts b/src/core/chunks/groups.ts new file mode 100644 index 0000000..53a2873 --- /dev/null +++ b/src/core/chunks/groups.ts @@ -0,0 +1,95 @@ +import type { ToolCallChunk, ToolResultChunk } from "@dispatch/wire"; +import type { RenderedChunk } from "./types"; + +/** + * One tool call within a batch, paired with its result (matched by `toolCallId`). + * `result` is null while the call is still pending (no result chunk yet). + */ +export interface ToolBatchEntry { + readonly call: ToolCallChunk; + readonly result: ToolResultChunk | null; +} + +/** + * A render group: either a single rendered chunk (rendered as today) or a batch + * of tool calls the model emitted together in one step (shared `stepId`), to be + * rendered as one grouped unit. + */ +export type RenderGroup = + | { readonly kind: "single"; readonly chunk: RenderedChunk } + | { + readonly kind: "tool-batch"; + readonly stepId: string; + readonly entries: readonly ToolBatchEntry[]; + readonly provisional: boolean; + }; + +/** + * Group a flat rendered-chunk stream for display. Tool calls sharing a `stepId` + * (the backend's authoritative batch key) where the step has 2+ calls become one + * `tool-batch` group, positioned at the first call and pairing each call with its + * `tool-result` (by `toolCallId`); the absorbed result chunks are not emitted on + * their own. Single tool calls (one per step, or no `stepId` — e.g. pre-0.2.0 + * replay rows) and every non-tool chunk render as `single` groups, in order. + * + * Pure: input → output, no DOM, no Svelte. + */ +export function groupRenderedChunks(rendered: readonly RenderedChunk[]): readonly RenderGroup[] { + // 1. Steps that batched 2+ tool calls. + const callsPerStep = new Map<string, number>(); + for (const rc of rendered) { + if (rc.chunk.type === "tool-call" && rc.chunk.stepId !== undefined) { + callsPerStep.set(rc.chunk.stepId, (callsPerStep.get(rc.chunk.stepId) ?? 0) + 1); + } + } + const batchSteps = new Set<string>(); + for (const [stepId, count] of callsPerStep) { + if (count >= 2) batchSteps.add(stepId); + } + + // 2. toolCallIds belonging to a batch (so their results are absorbed), and a + // lookup of result chunks by toolCallId for pairing. + const batchCallIds = new Set<string>(); + const resultByCallId = new Map<string, ToolResultChunk>(); + for (const rc of rendered) { + const chunk = rc.chunk; + if (chunk.type === "tool-call" && chunk.stepId !== undefined && batchSteps.has(chunk.stepId)) { + batchCallIds.add(chunk.toolCallId); + } else if (chunk.type === "tool-result" && !resultByCallId.has(chunk.toolCallId)) { + resultByCallId.set(chunk.toolCallId, chunk); + } + } + + // 3. Emit groups in stream order; each batch lands at its first call. + const groups: RenderGroup[] = []; + const emittedSteps = new Set<string>(); + for (const rc of rendered) { + const chunk = rc.chunk; + + if (chunk.type === "tool-call" && chunk.stepId !== undefined && batchSteps.has(chunk.stepId)) { + const stepId = chunk.stepId; + if (emittedSteps.has(stepId)) continue; + emittedSteps.add(stepId); + + const entries: ToolBatchEntry[] = []; + let provisional = false; + for (const inner of rendered) { + if (inner.chunk.type === "tool-call" && inner.chunk.stepId === stepId) { + const result = resultByCallId.get(inner.chunk.toolCallId) ?? null; + entries.push({ call: inner.chunk, result }); + if (inner.provisional) provisional = true; + } + } + groups.push({ kind: "tool-batch", stepId, entries, provisional }); + continue; + } + + if (chunk.type === "tool-result" && batchCallIds.has(chunk.toolCallId)) { + continue; // absorbed into its batch + } + + groups.push({ kind: "single", chunk: rc }); + } + + return groups; +} diff --git a/src/core/chunks/image-url.test.ts b/src/core/chunks/image-url.test.ts new file mode 100644 index 0000000..8a79c09 --- /dev/null +++ b/src/core/chunks/image-url.test.ts @@ -0,0 +1,46 @@ +import { describe, expect, it } from "vitest"; +import { resolveImageUrl } from "./image-url"; + +const BASE = "http://localhost:24203"; + +describe("resolveImageUrl", () => { + it("returns a data URL as-is (the optimistic echo / a pasted image)", () => { + const dataUrl = "data:image/png;base64,iVBORw0KGgo="; + expect(resolveImageUrl(dataUrl, BASE)).toBe(dataUrl); + }); + + it("returns an absolute http URL as-is", () => { + const abs = "https://example.com/img.png"; + expect(resolveImageUrl(abs, BASE)).toBe(abs); + }); + + it("prepends the api base to a relative /images/ path", () => { + expect(resolveImageUrl("/images/conv-123/abc-456.png", BASE)).toBe( + "http://localhost:24203/images/conv-123/abc-456.png", + ); + }); + + it("does not double the slash when the base has a trailing slash", () => { + expect(resolveImageUrl("/images/c/x.png", "http://localhost:24203/")).toBe( + "http://localhost:24203/images/c/x.png", + ); + }); + + it("adds a leading slash to a path-relative url without one", () => { + expect(resolveImageUrl("images/c/x.png", BASE)).toBe("http://localhost:24203/images/c/x.png"); + }); + + it("returns the relative path as-is when apiBase is empty (root-relative)", () => { + // A browser resolves a root-relative `/images/…` against the document origin. + expect(resolveImageUrl("/images/c/x.png", "")).toBe("/images/c/x.png"); + }); + + it("handles a relative path with an empty apiBase (path-relative without slash)", () => { + expect(resolveImageUrl("images/c/x.png", "")).toBe("/images/c/x.png"); + }); + + it("returns a data URL as-is even with an empty apiBase", () => { + const dataUrl = "data:image/jpeg;base64,AAAA"; + expect(resolveImageUrl(dataUrl, "")).toBe(dataUrl); + }); +}); diff --git a/src/core/chunks/image-url.ts b/src/core/chunks/image-url.ts new file mode 100644 index 0000000..e5ec756 --- /dev/null +++ b/src/core/chunks/image-url.ts @@ -0,0 +1,35 @@ +/** + * Resolve an `ImageChunk.url` into a renderable `<img src>` value. + * + * Persisted image chunks now carry a COMPACT HTTP path + * (`/images/<conversationId>/<uuid>.png`) served by the backend — NOT a base64 + * data URL (images are stored on disk under tmp, not in the conversation store, + * to keep SQLite payloads small). The optimistic echo (what the FE just sent in + * `ChatRequest.images`) still carries a data URL, and a chunk could also carry + * an absolute `http(s)://` URL, so the resolution is format-aware: + * + * - `data:` URL → returned as-is (the optimistic echo / a pasted data URL). + * - `http(s)://` → returned as-is (an absolute URL already). + * - anything else (a relative path like `/images/…`) → `apiBase` is prepended + * (with no double slash). An empty `apiBase` leaves a root-relative path, + * which a browser resolves against the document origin. + * + * Pure: input → output, zero DOM, zero Svelte. + * + * @param url The chunk's `url` (data URL, absolute, or relative path). + * @param apiBase The HTTP API base URL (e.g. `http://localhost:24203`). + */ +export function resolveImageUrl(url: string, apiBase: string): string { + if (url.startsWith("data:") || url.startsWith("http://") || url.startsWith("https://")) { + return url; + } + // A relative path (e.g. `/images/…`) — normalize to a leading slash and + // prepend the api base. With an empty base this yields a root-relative path + // (a browser resolves `/images/…` against the document origin). + const path = url.startsWith("/") ? url : `/${url}`; + if (apiBase.length === 0) return path; + // Join without a double slash: strip a trailing slash from the base, then + // append the (leading-slash) path verbatim. + const base = apiBase.endsWith("/") ? apiBase.slice(0, -1) : apiBase; + return `${base}${path}`; +} diff --git a/src/core/chunks/index.ts b/src/core/chunks/index.ts index 67739bc..bdd6ce4 100644 --- a/src/core/chunks/index.ts +++ b/src/core/chunks/index.ts @@ -1,8 +1,31 @@ -export { appendUserMessage, applyHistory, foldEvent, initialState } from "./reducer"; -export { selectChunks, selectMessages } from "./selectors"; +export type { RenderGroup, ToolBatchEntry } from "./groups"; +export { groupRenderedChunks } from "./groups"; +export { resolveImageUrl } from "./image-url"; +export { + appendUserMessage, + applyHistory, + clearGenerating, + foldEvent, + initialState, +} from "./reducer"; +export type { ProviderRetryView } from "./retry-banner"; +export { formatRetryDelay, viewProviderRetry } from "./retry-banner"; +export { selectChunks, selectGenerating, selectMessages, selectProviderRetry } from "./selectors"; +export { + DEFAULT_CHAT_LIMIT, + initialWindowSize, + MAX_CHAT_LIMIT, + MIN_CHAT_LIMIT, + normalizeChatLimit, + restoreEarlier, + selectHasEarlier, + trimTranscript, + unloadCount, + windowTranscript, +} from "./trim"; export type { - AccumulatingChunk, - ProvisionalChunk, - RenderedChunk, - TranscriptState, + AccumulatingChunk, + ProvisionalChunk, + RenderedChunk, + TranscriptState, } from "./types"; diff --git a/src/core/chunks/reducer.test.ts b/src/core/chunks/reducer.test.ts index b7165e4..058552e 100644 --- a/src/core/chunks/reducer.test.ts +++ b/src/core/chunks/reducer.test.ts @@ -1,467 +1,1024 @@ import type { - StoredChunk, - TurnDoneEvent, - TurnErrorEvent, - TurnReasoningDeltaEvent, - TurnSealedEvent, - TurnStartEvent, - TurnTextDeltaEvent, - TurnToolCallEvent, - TurnToolResultEvent, - TurnUsageEvent, + ImageInput, + StepId, + StoredChunk, + TurnDoneEvent, + TurnErrorEvent, + TurnInputEvent, + TurnProviderRetryEvent, + TurnReasoningDeltaEvent, + TurnSealedEvent, + TurnStartEvent, + TurnSteeringEvent, + TurnTextDeltaEvent, + TurnToolCallEvent, + TurnToolResultEvent, + TurnUsageEvent, } from "@dispatch/wire"; import { describe, expect, it } from "vitest"; -import { appendUserMessage, applyHistory, foldEvent, initialState } from "./reducer"; -import { selectChunks, selectMessages } from "./selectors"; +import { + appendUserMessage, + applyHistory, + clearGenerating, + foldEvent, + initialState, +} from "./reducer"; +import { selectChunks, selectGenerating, selectMessages, selectProviderRetry } from "./selectors"; const turnStart = (turnId: string): TurnStartEvent => ({ - type: "turn-start", - conversationId: "c1", - turnId, + type: "turn-start", + conversationId: "c1", + turnId, }); const textDelta = (turnId: string, delta: string): TurnTextDeltaEvent => ({ - type: "text-delta", - conversationId: "c1", - turnId, - delta, + type: "text-delta", + conversationId: "c1", + turnId, + delta, }); const reasoningDelta = (turnId: string, delta: string): TurnReasoningDeltaEvent => ({ - type: "reasoning-delta", - conversationId: "c1", - turnId, - delta, + type: "reasoning-delta", + conversationId: "c1", + turnId, + delta, }); const toolCall = ( - turnId: string, - toolCallId: string, - toolName: string, - input: unknown, + turnId: string, + toolCallId: string, + toolName: string, + input: unknown, + stepId = "s0", ): TurnToolCallEvent => ({ - type: "tool-call", - conversationId: "c1", - turnId, - toolCallId, - toolName, - input, + type: "tool-call", + conversationId: "c1", + turnId, + toolCallId, + toolName, + input, + stepId: stepId as StepId, }); const toolResult = ( - turnId: string, - toolCallId: string, - toolName: string, - content: string, + turnId: string, + toolCallId: string, + toolName: string, + content: string, + stepId = "s0", ): TurnToolResultEvent => ({ - type: "tool-result", - conversationId: "c1", - turnId, - toolCallId, - toolName, - content, - isError: false, + type: "tool-result", + conversationId: "c1", + turnId, + toolCallId, + toolName, + content, + isError: false, + stepId: stepId as StepId, }); const usageEvent = (turnId: string, inputTokens: number, outputTokens: number): TurnUsageEvent => ({ - type: "usage", - conversationId: "c1", - turnId, - usage: { inputTokens, outputTokens }, + type: "usage", + conversationId: "c1", + turnId, + usage: { inputTokens, outputTokens }, }); const errorEvent = (turnId: string, message: string, code?: string): TurnErrorEvent => - code !== undefined - ? { type: "error", conversationId: "c1", turnId, message, code } - : { type: "error", conversationId: "c1", turnId, message }; + code !== undefined + ? { type: "error", conversationId: "c1", turnId, message, code } + : { type: "error", conversationId: "c1", turnId, message }; const doneEvent = (turnId: string): TurnDoneEvent => ({ - type: "done", - conversationId: "c1", - turnId, - reason: "stop", + type: "done", + conversationId: "c1", + turnId, + reason: "stop", }); const turnSealed = (turnId: string): TurnSealedEvent => ({ - type: "turn-sealed", - conversationId: "c1", - turnId, + type: "turn-sealed", + conversationId: "c1", + turnId, }); +const providerRetry = ( + turnId: string, + attempt: number, + delayMs: number, + message = "HTTP 429: overloaded", + code?: string, +): TurnProviderRetryEvent => + code !== undefined + ? { type: "provider-retry", conversationId: "c1", turnId, attempt, delayMs, message, code } + : { type: "provider-retry", conversationId: "c1", turnId, attempt, delayMs, message }; + const storedChunk = ( - seq: number, - role: "user" | "assistant" | "tool" | "system", - chunk: StoredChunk["chunk"], + seq: number, + role: "user" | "assistant" | "tool" | "system", + chunk: StoredChunk["chunk"], ): StoredChunk => ({ - seq, - role, - chunk, + seq, + role, + chunk, }); describe("initialState", () => { - it("initial state is empty", () => { - const s = initialState(); - expect(s.committed).toEqual([]); - expect(s.provisional).toEqual([]); - expect(s.accumulating).toBeNull(); - expect(s.currentTurnId).toBeNull(); - expect(s.latestUsage).toBeNull(); - expect(s.sealedTurnId).toBeNull(); - }); + it("initial state is empty", () => { + const s = initialState(); + expect(s.committed).toEqual([]); + expect(s.provisional).toEqual([]); + expect(s.accumulating).toBeNull(); + expect(s.currentTurnId).toBeNull(); + expect(s.latestUsage).toBeNull(); + expect(s.sealedTurnId).toBeNull(); + expect(s.generating).toBe(false); + }); +}); + +describe("foldEvent — generating (turn-running state)", () => { + it("turn-start sets generating true", () => { + let s = initialState(); + expect(selectGenerating(s)).toBe(false); + s = foldEvent(s, turnStart("t1")); + expect(s.generating).toBe(true); + expect(selectGenerating(s)).toBe(true); + }); + + it("a content delta sets generating true (e.g. a late-joiner replay missing turn-start)", () => { + let s = initialState(); + s = foldEvent(s, textDelta("t1", "hi")); + expect(s.generating).toBe(true); + s = initialState(); + s = foldEvent(s, reasoningDelta("t1", "hmm")); + expect(s.generating).toBe(true); + s = initialState(); + s = foldEvent(s, toolCall("t1", "tc1", "bash", {})); + expect(s.generating).toBe(true); + }); + + it("stays generating across the turn's deltas", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "wor")); + s = foldEvent(s, textDelta("t1", "king")); + expect(s.generating).toBe(true); + }); + + it("done clears generating", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "answer")); + s = foldEvent(s, doneEvent("t1")); + expect(s.generating).toBe(false); + }); + + it("turn-sealed clears generating", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, turnSealed("t1")); + expect(s.generating).toBe(false); + }); + + it("error clears generating", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, errorEvent("t1", "boom")); + expect(s.generating).toBe(false); + }); + + it("a new turn re-asserts generating after the previous one finished", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, doneEvent("t1")); + s = foldEvent(s, turnSealed("t1")); + expect(s.generating).toBe(false); + s = foldEvent(s, turnStart("t2")); + expect(s.generating).toBe(true); + }); + + it("status does not change generating (free-form string, not inferred)", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + const next = foldEvent(s, { type: "status", conversationId: "c1", status: "idle" }); + expect(next.generating).toBe(true); + }); +}); + +describe("clearGenerating", () => { + it("clears a set generating flag", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + expect(s.generating).toBe(true); + const cleared = clearGenerating(s); + expect(cleared.generating).toBe(false); + }); + + it("returns the same object when already not generating (no-op)", () => { + const s = initialState(); + expect(clearGenerating(s)).toBe(s); + }); + + it("preserves transcript content while clearing generating", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "partial")); + const cleared = clearGenerating(s); + expect(cleared.generating).toBe(false); + expect(cleared.accumulating).toEqual({ kind: "text", text: "partial" }); + expect(cleared.currentTurnId).toBe("t1"); + }); }); describe("foldEvent — text-delta", () => { - it("text-delta accumulates into one TextChunk", () => { - let s = initialState(); - s = foldEvent(s, turnStart("t1")); - s = foldEvent(s, textDelta("t1", "hello")); - expect(s.accumulating).toEqual({ kind: "text", text: "hello" }); - expect(s.provisional).toEqual([]); - }); - - it("successive text-deltas extend the same provisional chunk", () => { - let s = initialState(); - s = foldEvent(s, turnStart("t1")); - s = foldEvent(s, textDelta("t1", "hello ")); - s = foldEvent(s, textDelta("t1", "world")); - expect(s.accumulating).toEqual({ kind: "text", text: "hello world" }); - expect(s.provisional).toEqual([]); - }); - - it("text-delta after reasoning-delta flushes thinking and starts text", () => { - let s = initialState(); - s = foldEvent(s, turnStart("t1")); - s = foldEvent(s, reasoningDelta("t1", "thinking...")); - s = foldEvent(s, textDelta("t1", "answer")); - expect(s.accumulating).toEqual({ kind: "text", text: "answer" }); - expect(s.provisional).toHaveLength(1); - expect(s.provisional[0]?.chunk).toEqual({ type: "thinking", text: "thinking..." }); - expect(s.provisional[0]?.role).toBe("assistant"); - }); + it("text-delta accumulates into one TextChunk", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "hello")); + expect(s.accumulating).toEqual({ kind: "text", text: "hello" }); + expect(s.provisional).toEqual([]); + }); + + it("successive text-deltas extend the same provisional chunk", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "hello ")); + s = foldEvent(s, textDelta("t1", "world")); + expect(s.accumulating).toEqual({ kind: "text", text: "hello world" }); + expect(s.provisional).toEqual([]); + }); + + it("text-delta after reasoning-delta flushes thinking and starts text", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, reasoningDelta("t1", "thinking...")); + s = foldEvent(s, textDelta("t1", "answer")); + expect(s.accumulating).toEqual({ kind: "text", text: "answer" }); + expect(s.provisional).toHaveLength(1); + expect(s.provisional[0]?.chunk).toEqual({ type: "thinking", text: "thinking..." }); + expect(s.provisional[0]?.role).toBe("assistant"); + }); }); describe("foldEvent — reasoning-delta", () => { - it("reasoning-delta yields a thinking chunk", () => { - let s = initialState(); - s = foldEvent(s, turnStart("t1")); - s = foldEvent(s, reasoningDelta("t1", "hmm")); - expect(s.accumulating).toEqual({ kind: "thinking", text: "hmm" }); - }); - - it("successive reasoning-deltas extend the same chunk", () => { - let s = initialState(); - s = foldEvent(s, turnStart("t1")); - s = foldEvent(s, reasoningDelta("t1", "hmm ")); - s = foldEvent(s, reasoningDelta("t1", "ok")); - expect(s.accumulating).toEqual({ kind: "thinking", text: "hmm ok" }); - }); + it("reasoning-delta yields a thinking chunk", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, reasoningDelta("t1", "hmm")); + expect(s.accumulating).toEqual({ kind: "thinking", text: "hmm" }); + }); + + it("successive reasoning-deltas extend the same chunk", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, reasoningDelta("t1", "hmm ")); + s = foldEvent(s, reasoningDelta("t1", "ok")); + expect(s.accumulating).toEqual({ kind: "thinking", text: "hmm ok" }); + }); }); describe("foldEvent — tool-call then tool-result", () => { - it("tool-call then tool-result render in order", () => { - let s = initialState(); - s = foldEvent(s, turnStart("t1")); - s = foldEvent(s, toolCall("t1", "tc1", "bash", { cmd: "ls" })); - s = foldEvent(s, toolResult("t1", "tc1", "bash", "file.txt")); - expect(s.provisional).toHaveLength(2); - expect(s.provisional[0]?.role).toBe("assistant"); - expect(s.provisional[0]?.chunk).toEqual({ - type: "tool-call", - toolCallId: "tc1", - toolName: "bash", - input: { cmd: "ls" }, - }); - expect(s.provisional[1]?.role).toBe("tool"); - expect(s.provisional[1]?.chunk).toEqual({ - type: "tool-result", - toolCallId: "tc1", - toolName: "bash", - content: "file.txt", - isError: false, - }); - }); - - it("tool-call flushes accumulating text", () => { - let s = initialState(); - s = foldEvent(s, turnStart("t1")); - s = foldEvent(s, textDelta("t1", "let me check")); - s = foldEvent(s, toolCall("t1", "tc1", "bash", {})); - expect(s.provisional).toHaveLength(2); - expect(s.provisional[0]?.chunk).toEqual({ type: "text", text: "let me check" }); - expect(s.provisional[1]?.chunk).toMatchObject({ type: "tool-call", toolCallId: "tc1" }); - expect(s.accumulating).toBeNull(); - }); + it("tool-call then tool-result render in order", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, toolCall("t1", "tc1", "bash", { cmd: "ls" }, "t1#0")); + s = foldEvent(s, toolResult("t1", "tc1", "bash", "file.txt", "t1#0")); + expect(s.provisional).toHaveLength(2); + expect(s.provisional[0]?.role).toBe("assistant"); + // foldEvent copies the event's stepId onto the chunk (grouping key). + expect(s.provisional[0]?.chunk).toEqual({ + type: "tool-call", + toolCallId: "tc1", + toolName: "bash", + input: { cmd: "ls" }, + stepId: "t1#0", + }); + expect(s.provisional[1]?.role).toBe("tool"); + expect(s.provisional[1]?.chunk).toEqual({ + type: "tool-result", + toolCallId: "tc1", + toolName: "bash", + content: "file.txt", + isError: false, + stepId: "t1#0", + }); + }); + + it("tool-call flushes accumulating text", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "let me check")); + s = foldEvent(s, toolCall("t1", "tc1", "bash", {})); + expect(s.provisional).toHaveLength(2); + expect(s.provisional[0]?.chunk).toEqual({ type: "text", text: "let me check" }); + expect(s.provisional[1]?.chunk).toMatchObject({ type: "tool-call", toolCallId: "tc1" }); + expect(s.accumulating).toBeNull(); + }); }); describe("foldEvent — turn-sealed", () => { - it("turn-sealed sets sealedTurnId", () => { - let s = initialState(); - s = foldEvent(s, turnStart("t1")); - s = foldEvent(s, textDelta("t1", "hi")); - s = foldEvent(s, turnSealed("t1")); - expect(s.sealedTurnId).toBe("t1"); - expect(s.accumulating).toBeNull(); - expect(s.provisional).toHaveLength(1); - expect(s.provisional[0]?.chunk).toEqual({ type: "text", text: "hi" }); - }); + it("turn-sealed sets sealedTurnId", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "hi")); + s = foldEvent(s, turnSealed("t1")); + expect(s.sealedTurnId).toBe("t1"); + expect(s.accumulating).toBeNull(); + expect(s.provisional).toHaveLength(1); + expect(s.provisional[0]?.chunk).toEqual({ type: "text", text: "hi" }); + }); }); describe("foldEvent — usage", () => { - it("stores latest usage", () => { - let s = initialState(); - s = foldEvent(s, usageEvent("t1", 100, 50)); - expect(s.latestUsage).toEqual({ inputTokens: 100, outputTokens: 50 }); - }); - - it("overwrites previous usage", () => { - let s = initialState(); - s = foldEvent(s, usageEvent("t1", 100, 50)); - s = foldEvent(s, usageEvent("t1", 200, 80)); - expect(s.latestUsage).toEqual({ inputTokens: 200, outputTokens: 80 }); - }); + it("stores latest usage", () => { + let s = initialState(); + s = foldEvent(s, usageEvent("t1", 100, 50)); + expect(s.latestUsage).toEqual({ inputTokens: 100, outputTokens: 50 }); + }); + + it("overwrites previous usage", () => { + let s = initialState(); + s = foldEvent(s, usageEvent("t1", 100, 50)); + s = foldEvent(s, usageEvent("t1", 200, 80)); + expect(s.latestUsage).toEqual({ inputTokens: 200, outputTokens: 80 }); + }); }); describe("foldEvent — error", () => { - it("creates error chunk with code", () => { - let s = initialState(); - s = foldEvent(s, turnStart("t1")); - s = foldEvent(s, errorEvent("t1", "bad", "E001")); - expect(s.provisional).toHaveLength(1); - expect(s.provisional[0]?.chunk).toEqual({ type: "error", message: "bad", code: "E001" }); - expect(s.provisional[0]?.role).toBe("assistant"); - }); - - it("creates error chunk without code", () => { - let s = initialState(); - s = foldEvent(s, turnStart("t1")); - s = foldEvent(s, errorEvent("t1", "bad")); - expect(s.provisional).toHaveLength(1); - expect(s.provisional[0]?.chunk).toEqual({ type: "error", message: "bad" }); - }); + it("creates error chunk with code", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, errorEvent("t1", "bad", "E001")); + expect(s.provisional).toHaveLength(1); + expect(s.provisional[0]?.chunk).toEqual({ type: "error", message: "bad", code: "E001" }); + expect(s.provisional[0]?.role).toBe("assistant"); + }); + + it("creates error chunk without code", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, errorEvent("t1", "bad")); + expect(s.provisional).toHaveLength(1); + expect(s.provisional[0]?.chunk).toEqual({ type: "error", message: "bad" }); + }); }); describe("foldEvent — done", () => { - it("flushes accumulating chunk on done", () => { - let s = initialState(); - s = foldEvent(s, turnStart("t1")); - s = foldEvent(s, textDelta("t1", "hello")); - s = foldEvent(s, doneEvent("t1")); - expect(s.accumulating).toBeNull(); - expect(s.provisional).toHaveLength(1); - expect(s.provisional[0]?.chunk).toEqual({ type: "text", text: "hello" }); - }); + it("flushes accumulating chunk on done", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "hello")); + s = foldEvent(s, doneEvent("t1")); + expect(s.accumulating).toBeNull(); + expect(s.provisional).toHaveLength(1); + expect(s.provisional[0]?.chunk).toEqual({ type: "text", text: "hello" }); + }); }); describe("foldEvent — status and tool-output", () => { - it("status is a no-op", () => { - const s = initialState(); - const next = foldEvent(s, { type: "status", conversationId: "c1", status: "running" }); - expect(next).toBe(s); - }); - - it("tool-output is a no-op", () => { - const s = initialState(); - const next = foldEvent(s, { - type: "tool-output", - conversationId: "c1", - turnId: "t1", - toolCallId: "tc1", - data: "output", - stream: "stdout", - }); - expect(next).toBe(s); - }); + it("status is a no-op", () => { + const s = initialState(); + const next = foldEvent(s, { type: "status", conversationId: "c1", status: "running" }); + expect(next).toBe(s); + }); + + it("tool-output is a no-op", () => { + const s = initialState(); + const next = foldEvent(s, { + type: "tool-output", + conversationId: "c1", + turnId: "t1", + toolCallId: "tc1", + data: "output", + stream: "stdout", + }); + expect(next).toBe(s); + }); +}); + +describe("foldEvent — provider-retry (transient retry banner)", () => { + it("sets the provider-retry banner on a provider-retry event", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, providerRetry("t1", 0, 5000, "HTTP 429: overloaded", "429")); + const retry = selectProviderRetry(s); + expect(retry).not.toBeNull(); + expect(retry?.attempt).toBe(0); + expect(retry?.delayMs).toBe(5000); + expect(retry?.code).toBe("429"); + }); + + it("does NOT add a chunk (never persisted — never pollutes the prompt)", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, providerRetry("t1", 0, 5000)); + expect(selectChunks(s)).toHaveLength(0); + expect(s.provisional).toHaveLength(0); + expect(s.accumulating).toBeNull(); + }); + + it("coalesces: the latest attempt + delay replaces the previous", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, providerRetry("t1", 0, 5000, "first", "429")); + s = foldEvent(s, providerRetry("t1", 1, 10000, "second", "429")); + const retry = selectProviderRetry(s); + expect(retry?.attempt).toBe(1); + expect(retry?.delayMs).toBe(10000); + expect(retry?.message).toBe("second"); + }); + + it("keeps generating true (the turn is still in flight, just retrying)", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, providerRetry("t1", 0, 5000)); + expect(s.generating).toBe(true); + }); + + it("clears when content resumes (text-delta)", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, providerRetry("t1", 0, 5000)); + expect(selectProviderRetry(s)).not.toBeNull(); + s = foldEvent(s, textDelta("t1", "here is the reply")); + expect(selectProviderRetry(s)).toBeNull(); + }); + + it("clears when content resumes (reasoning-delta / tool-call / tool-result)", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, providerRetry("t1", 0, 5000)); + s = foldEvent(s, reasoningDelta("t1", "thinking")); + expect(selectProviderRetry(s)).toBeNull(); + + s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, providerRetry("t1", 0, 5000)); + s = foldEvent(s, toolCall("t1", "tc1", "bash", {})); + expect(selectProviderRetry(s)).toBeNull(); + }); + + it("clears when the turn ends (done / turn-sealed / error)", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, providerRetry("t1", 0, 5000)); + s = foldEvent(s, errorEvent("t1", "exhausted")); + expect(selectProviderRetry(s)).toBeNull(); + + s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, providerRetry("t1", 0, 5000)); + s = foldEvent(s, doneEvent("t1")); + expect(selectProviderRetry(s)).toBeNull(); + + s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, providerRetry("t1", 0, 5000)); + s = foldEvent(s, turnSealed("t1")); + expect(selectProviderRetry(s)).toBeNull(); + }); + + it("clears on a new turn (turn-start)", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, providerRetry("t1", 0, 5000)); + expect(selectProviderRetry(s)).not.toBeNull(); + s = foldEvent(s, turnStart("t2")); + expect(selectProviderRetry(s)).toBeNull(); + }); + + it("leaves the banner untouched across metadata events (usage / step-complete / status)", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, providerRetry("t1", 0, 5000)); + s = foldEvent(s, usageEvent("t1", 10, 20)); + expect(selectProviderRetry(s)).not.toBeNull(); + s = foldEvent(s, { + type: "step-complete", + conversationId: "c1", + turnId: "t1", + stepId: "t1#0" as StepId, + ttftMs: 100, + decodeMs: 200, + genTotalMs: 300, + }); + expect(selectProviderRetry(s)).not.toBeNull(); + s = foldEvent(s, { type: "status", conversationId: "c1", status: "running" }); + expect(selectProviderRetry(s)).not.toBeNull(); + }); + + it("is null in the initial state", () => { + expect(selectProviderRetry(initialState())).toBeNull(); + }); +}); + +describe("clearGenerating also clears a stale provider-retry banner (reconnect)", () => { + it("clears the retry banner alongside generating on reconnect", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, providerRetry("t1", 0, 5000)); + expect(s.generating).toBe(true); + expect(selectProviderRetry(s)).not.toBeNull(); + const cleared = clearGenerating(s); + expect(cleared.generating).toBe(false); + expect(selectProviderRetry(cleared)).toBeNull(); + }); + + it("preserves transcript content while clearing the banner", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "partial")); + s = foldEvent(s, providerRetry("t1", 0, 5000)); + const cleared = clearGenerating(s); + expect(cleared.accumulating).toEqual({ kind: "text", text: "partial" }); + expect(selectProviderRetry(cleared)).toBeNull(); + }); +}); + +describe("foldEvent — user-message (the turn's user prompt; backend CR-3)", () => { + const userMessage = (text: string): TurnInputEvent => ({ + type: "user-message", + conversationId: "c1", + turnId: "t1", + text, + }); + + it("a watcher renders the prompt: appends a provisional user chunk + marks generating", () => { + let s = initialState(); + s = foldEvent(s, userMessage("what is 2+2?")); + const chunks = selectChunks(s); + expect(chunks).toHaveLength(1); + expect(chunks[0]?.role).toBe("user"); + expect(chunks[0]?.chunk).toEqual({ type: "text", text: "what is 2+2?" }); + expect(chunks[0]?.provisional).toBe(true); + expect(s.generating).toBe(true); + }); + + it("dedups the SENDER's optimistic echo (no duplicate user bubble)", () => { + let s = initialState(); + s = appendUserMessage(s, "hi"); // optimistic echo from the sender's send() + s = foldEvent(s, userMessage("hi")); // server echo for the same turn + const users = selectChunks(s).filter((c) => c.role === "user"); + expect(users).toHaveLength(1); + }); + + it("appends when the trailing provisional differs (no false dedup)", () => { + let s = initialState(); + s = appendUserMessage(s, "first"); + s = foldEvent(s, userMessage("second")); + const users = selectChunks(s).filter((c) => c.role === "user"); + expect(users).toHaveLength(2); + }); + + it("ignores an empty user-message", () => { + let s = initialState(); + s = foldEvent(s, userMessage("")); + expect(selectChunks(s)).toHaveLength(0); + expect(s.generating).toBe(false); + }); + + it("flushes an accumulating chunk before appending the prompt", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "partial")); + s = foldEvent(s, userMessage("new prompt")); + // the partial assistant text was flushed to provisional, then the user prompt appended + expect(s.accumulating).toBeNull(); + const roles = selectChunks(s).map((c) => c.role); + expect(roles).toEqual(["assistant", "user"]); + }); +}); + +describe("foldEvent — steering (mid-turn steering injection)", () => { + const steering = (text: string): TurnSteeringEvent => ({ + type: "steering", + conversationId: "c1", + turnId: "t1", + text, + }); + + it("appends a provisional user bubble + keeps generating", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, toolResult("t1", "tc1", "read", "output")); + s = foldEvent(s, steering("actually, use a different file")); + const chunks = selectChunks(s); + const last = chunks[chunks.length - 1]; + expect(last?.role).toBe("user"); + expect(last?.chunk).toEqual({ type: "text", text: "actually, use a different file" }); + expect(last?.provisional).toBe(true); + expect(s.generating).toBe(true); + }); + + it("does NOT dedup against the sender's queue (unlike user-message)", () => { + // The sender enqueued the message via `chat.queue` — the queue SURFACE + // showed it. The `steering` event places it in the transcript; the surface + // separately clears on drain. No de-dup here (the transcript never showed + // the queued message). + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, steering("steer once")); + s = foldEvent(s, steering("steer again")); + const users = selectChunks(s).filter((c) => c.role === "user"); + expect(users).toHaveLength(2); + }); + + it("ignores an empty steering event", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, steering("")); + expect(selectChunks(s)).toHaveLength(0); + expect(s.generating).toBe(true); // turn-start already set it + }); + + it("flushes an accumulating chunk before appending the steering bubble", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "partial response")); + s = foldEvent(s, steering("mid-turn correction")); + expect(s.accumulating).toBeNull(); + const roles = selectChunks(s).map((c) => c.role); + expect(roles).toEqual(["assistant", "user"]); + }); }); describe("applyHistory", () => { - it("orders committed chunks by seq", () => { - const s = initialState(); - const chunks = [ - storedChunk(3, "assistant", { type: "text", text: "c" }), - storedChunk(1, "user", { type: "text", text: "a" }), - storedChunk(2, "assistant", { type: "text", text: "b" }), - ]; - const next = applyHistory(s, chunks); - expect(next.committed.map((c) => c.seq)).toEqual([1, 2, 3]); - }); - - it("is idempotent on duplicate seqs", () => { - let s = initialState(); - const batch1 = [ - storedChunk(1, "user", { type: "text", text: "a" }), - storedChunk(2, "assistant", { type: "text", text: "b" }), - ]; - s = applyHistory(s, batch1); - const batch2 = [ - storedChunk(2, "assistant", { type: "text", text: "b" }), - storedChunk(3, "assistant", { type: "text", text: "c" }), - ]; - s = applyHistory(s, batch2); - expect(s.committed.map((c) => c.seq)).toEqual([1, 2, 3]); - expect(s.committed).toHaveLength(3); - }); - - it("supersedes & clears provisional once committed", () => { - let s = initialState(); - s = foldEvent(s, turnStart("t1")); - s = foldEvent(s, textDelta("t1", "hello")); - s = foldEvent(s, turnSealed("t1")); - expect(s.provisional).toHaveLength(1); - expect(s.sealedTurnId).toBe("t1"); - - s = applyHistory(s, [storedChunk(1, "assistant", { type: "text", text: "hello" })]); - expect(s.provisional).toEqual([]); - expect(s.accumulating).toBeNull(); - expect(s.sealedTurnId).toBeNull(); - expect(s.committed).toHaveLength(1); - }); - - it("keeps provisional and accumulating when sealedTurnId is null", () => { - let s = initialState(); - s = foldEvent(s, turnStart("t1")); - s = foldEvent(s, textDelta("t1", "wip")); - s = foldEvent(s, doneEvent("t1")); - s = applyHistory(s, [storedChunk(1, "user", { type: "text", text: "q" })]); - expect(s.provisional).toHaveLength(1); - expect(s.committed).toHaveLength(1); - }); - - it("merges new history into existing committed", () => { - let s = initialState(); - s = applyHistory(s, [storedChunk(1, "user", { type: "text", text: "a" })]); - s = applyHistory(s, [storedChunk(2, "assistant", { type: "text", text: "b" })]); - expect(s.committed).toHaveLength(2); - expect(s.committed.map((c) => c.seq)).toEqual([1, 2]); - }); + it("orders committed chunks by seq", () => { + const s = initialState(); + const chunks = [ + storedChunk(3, "assistant", { type: "text", text: "c" }), + storedChunk(1, "user", { type: "text", text: "a" }), + storedChunk(2, "assistant", { type: "text", text: "b" }), + ]; + const next = applyHistory(s, chunks); + expect(next.committed.map((c) => c.seq)).toEqual([1, 2, 3]); + }); + + it("is idempotent on duplicate seqs", () => { + let s = initialState(); + const batch1 = [ + storedChunk(1, "user", { type: "text", text: "a" }), + storedChunk(2, "assistant", { type: "text", text: "b" }), + ]; + s = applyHistory(s, batch1); + const batch2 = [ + storedChunk(2, "assistant", { type: "text", text: "b" }), + storedChunk(3, "assistant", { type: "text", text: "c" }), + ]; + s = applyHistory(s, batch2); + expect(s.committed.map((c) => c.seq)).toEqual([1, 2, 3]); + expect(s.committed).toHaveLength(3); + }); + + it("supersedes & clears provisional once committed", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "hello")); + s = foldEvent(s, turnSealed("t1")); + expect(s.provisional).toHaveLength(1); + expect(s.sealedTurnId).toBe("t1"); + + s = applyHistory(s, [storedChunk(1, "assistant", { type: "text", text: "hello" })]); + expect(s.provisional).toEqual([]); + expect(s.accumulating).toBeNull(); + expect(s.sealedTurnId).toBeNull(); + expect(s.committed).toHaveLength(1); + }); + + it("keeps provisional and accumulating when sealedTurnId is null", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "wip")); + s = foldEvent(s, doneEvent("t1")); + s = applyHistory(s, [storedChunk(1, "user", { type: "text", text: "q" })]); + expect(s.provisional).toHaveLength(1); + expect(s.committed).toHaveLength(1); + }); + + it("merges new history into existing committed", () => { + let s = initialState(); + s = applyHistory(s, [storedChunk(1, "user", { type: "text", text: "a" })]); + s = applyHistory(s, [storedChunk(2, "assistant", { type: "text", text: "b" })]); + expect(s.committed).toHaveLength(2); + expect(s.committed.map((c) => c.seq)).toEqual([1, 2]); + }); + + it("removes provisional duplicate when committed user message arrives during generation", () => { + // Simulate: send() appends provisional user message, then syncTail + // fetches the same message as committed (CR-6: persisted at turn start). + let s = initialState(); + s = appendUserMessage(s, "hello"); + s = foldEvent(s, turnStart("t1")); + expect(s.provisional).toHaveLength(1); + expect(s.provisional[0]?.role).toBe("user"); + expect(s.generating).toBe(true); + + // syncTail fetches the persisted user message as committed + s = applyHistory(s, [storedChunk(1, "user", { type: "text", text: "hello" })]); + + // The provisional duplicate is removed — no double render + expect(s.provisional).toEqual([]); + expect(s.committed).toHaveLength(1); + expect(s.committed[0]?.role).toBe("user"); + expect(s.committed[0]?.chunk).toEqual({ type: "text", text: "hello" }); + }); }); describe("selectChunks", () => { - it("selectChunks marks provisional with seq null", () => { - let s = initialState(); - s = applyHistory(s, [storedChunk(1, "user", { type: "text", text: "q" })]); - s = foldEvent(s, turnStart("t1")); - s = foldEvent(s, textDelta("t1", "wip")); - const chunks = selectChunks(s); - expect(chunks).toHaveLength(2); - expect(chunks[0]?.seq).toBe(1); - expect(chunks[0]?.provisional).toBe(false); - expect(chunks[1]?.seq).toBeNull(); - expect(chunks[1]?.provisional).toBe(true); - }); - - it("returns empty for empty state", () => { - expect(selectChunks(initialState())).toEqual([]); - }); - - it("includes accumulating chunk as provisional", () => { - let s = initialState(); - s = foldEvent(s, turnStart("t1")); - s = foldEvent(s, textDelta("t1", "building...")); - const chunks = selectChunks(s); - expect(chunks).toHaveLength(1); - expect(chunks[0]?.seq).toBeNull(); - expect(chunks[0]?.provisional).toBe(true); - expect(chunks[0]?.chunk).toEqual({ type: "text", text: "building..." }); - }); + it("selectChunks marks provisional with seq null", () => { + let s = initialState(); + s = applyHistory(s, [storedChunk(1, "user", { type: "text", text: "q" })]); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "wip")); + const chunks = selectChunks(s); + expect(chunks).toHaveLength(2); + expect(chunks[0]?.seq).toBe(1); + expect(chunks[0]?.provisional).toBe(false); + expect(chunks[1]?.seq).toBeNull(); + expect(chunks[1]?.provisional).toBe(true); + }); + + it("returns empty for empty state", () => { + expect(selectChunks(initialState())).toEqual([]); + }); + + it("includes accumulating chunk as provisional", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "building...")); + const chunks = selectChunks(s); + expect(chunks).toHaveLength(1); + expect(chunks[0]?.seq).toBeNull(); + expect(chunks[0]?.provisional).toBe(true); + expect(chunks[0]?.chunk).toEqual({ type: "text", text: "building..." }); + }); + + it("marks ONLY the actively-accumulating chunk as streaming", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + // A flushed-but-still-provisional thinking chunk, then a live accumulating one. + s = foldEvent(s, reasoningDelta("t1", "first thought")); + s = foldEvent(s, toolCall("t1", "tc1", "bash", {})); // flushes the thinking + s = foldEvent(s, textDelta("t1", "now writing")); + const chunks = selectChunks(s); + const thinking = chunks.find((c) => c.chunk.type === "thinking"); + const accumulating = chunks.find((c) => c.streaming === true); + expect(thinking?.streaming).toBeFalsy(); // flushed → not streaming + expect(accumulating?.chunk).toEqual({ type: "text", text: "now writing" }); + expect(chunks.filter((c) => c.streaming === true)).toHaveLength(1); + }); }); describe("selectMessages", () => { - it("selectMessages groups consecutive same-role chunks", () => { - let s = initialState(); - s = applyHistory(s, [ - storedChunk(1, "user", { type: "text", text: "q1" }), - storedChunk(2, "user", { type: "text", text: "q2" }), - storedChunk(3, "assistant", { type: "text", text: "a1" }), - storedChunk(4, "assistant", { type: "text", text: "a2" }), - storedChunk(5, "user", { type: "text", text: "q3" }), - ]); - const msgs = selectMessages(s); - expect(msgs).toHaveLength(3); - expect(msgs[0]?.role).toBe("user"); - expect(msgs[0]?.chunks).toHaveLength(2); - expect(msgs[1]?.role).toBe("assistant"); - expect(msgs[1]?.chunks).toHaveLength(2); - expect(msgs[2]?.role).toBe("user"); - expect(msgs[2]?.chunks).toHaveLength(1); - }); - - it("returns empty for empty state", () => { - expect(selectMessages(initialState())).toEqual([]); - }); - - it("mixes committed and provisional in messages", () => { - let s = initialState(); - s = applyHistory(s, [storedChunk(1, "user", { type: "text", text: "q" })]); - s = foldEvent(s, turnStart("t1")); - s = foldEvent(s, textDelta("t1", "a1")); - s = foldEvent(s, textDelta("t1", "a2")); - const msgs = selectMessages(s); - expect(msgs).toHaveLength(2); - expect(msgs[0]?.role).toBe("user"); - expect(msgs[0]?.chunks).toHaveLength(1); - expect(msgs[1]?.role).toBe("assistant"); - expect(msgs[1]?.chunks).toHaveLength(1); - expect(msgs[1]?.chunks[0]).toEqual({ type: "text", text: "a1a2" }); - }); + it("selectMessages groups consecutive same-role chunks", () => { + let s = initialState(); + s = applyHistory(s, [ + storedChunk(1, "user", { type: "text", text: "q1" }), + storedChunk(2, "user", { type: "text", text: "q2" }), + storedChunk(3, "assistant", { type: "text", text: "a1" }), + storedChunk(4, "assistant", { type: "text", text: "a2" }), + storedChunk(5, "user", { type: "text", text: "q3" }), + ]); + const msgs = selectMessages(s); + expect(msgs).toHaveLength(3); + expect(msgs[0]?.role).toBe("user"); + expect(msgs[0]?.chunks).toHaveLength(2); + expect(msgs[1]?.role).toBe("assistant"); + expect(msgs[1]?.chunks).toHaveLength(2); + expect(msgs[2]?.role).toBe("user"); + expect(msgs[2]?.chunks).toHaveLength(1); + }); + + it("returns empty for empty state", () => { + expect(selectMessages(initialState())).toEqual([]); + }); + + it("mixes committed and provisional in messages", () => { + let s = initialState(); + s = applyHistory(s, [storedChunk(1, "user", { type: "text", text: "q" })]); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "a1")); + s = foldEvent(s, textDelta("t1", "a2")); + const msgs = selectMessages(s); + expect(msgs).toHaveLength(2); + expect(msgs[0]?.role).toBe("user"); + expect(msgs[0]?.chunks).toHaveLength(1); + expect(msgs[1]?.role).toBe("assistant"); + expect(msgs[1]?.chunks).toHaveLength(1); + expect(msgs[1]?.chunks[0]).toEqual({ type: "text", text: "a1a2" }); + }); }); describe("appendUserMessage", () => { - it("adds a provisional user text chunk", () => { - let s = initialState(); - s = appendUserMessage(s, "hello from user"); - const chunks = selectChunks(s); - expect(chunks).toHaveLength(1); - expect(chunks[0]?.seq).toBeNull(); - expect(chunks[0]?.role).toBe("user"); - expect(chunks[0]?.chunk).toEqual({ type: "text", text: "hello from user" }); - expect(chunks[0]?.provisional).toBe(true); - }); - - it("selectMessages includes the optimistic user message", () => { - let s = initialState(); - s = appendUserMessage(s, "what is 2+2?"); - const msgs = selectMessages(s); - expect(msgs).toHaveLength(1); - expect(msgs[0]?.role).toBe("user"); - expect(msgs[0]?.chunks).toHaveLength(1); - expect(msgs[0]?.chunks[0]).toEqual({ type: "text", text: "what is 2+2?" }); - }); - - it("user echo then turn-sealed + applyHistory supersedes the provisional user chunk", () => { - let s = initialState(); - s = appendUserMessage(s, "hi"); - expect(selectChunks(s)).toHaveLength(1); - - s = foldEvent(s, turnStart("t1")); - s = foldEvent(s, textDelta("t1", "hello back")); - s = foldEvent(s, turnSealed("t1")); - s = applyHistory(s, [ - storedChunk(1, "user", { type: "text", text: "hi" }), - storedChunk(2, "assistant", { type: "text", text: "hello back" }), - ]); - const chunks = selectChunks(s); - expect(chunks).toHaveLength(2); - expect(chunks[0]?.seq).toBe(1); - expect(chunks[0]?.role).toBe("user"); - expect(chunks[0]?.chunk).toEqual({ type: "text", text: "hi" }); - expect(chunks[0]?.provisional).toBe(false); - expect(chunks[1]?.seq).toBe(2); - expect(chunks[1]?.role).toBe("assistant"); - expect(chunks[1]?.provisional).toBe(false); - }); - - it("flushes accumulating chunk before appending user message", () => { - let s = initialState(); - s = foldEvent(s, turnStart("t1")); - s = foldEvent(s, textDelta("t1", "partial")); - expect(s.accumulating).toEqual({ kind: "text", text: "partial" }); - - s = appendUserMessage(s, "user msg"); - expect(s.accumulating).toBeNull(); - expect(s.provisional).toHaveLength(2); - expect(s.provisional[0]?.role).toBe("assistant"); - expect(s.provisional[0]?.chunk).toEqual({ type: "text", text: "partial" }); - expect(s.provisional[1]?.role).toBe("user"); - expect(s.provisional[1]?.chunk).toEqual({ type: "text", text: "user msg" }); - }); + it("adds a provisional user text chunk", () => { + let s = initialState(); + s = appendUserMessage(s, "hello from user"); + const chunks = selectChunks(s); + expect(chunks).toHaveLength(1); + expect(chunks[0]?.seq).toBeNull(); + expect(chunks[0]?.role).toBe("user"); + expect(chunks[0]?.chunk).toEqual({ type: "text", text: "hello from user" }); + expect(chunks[0]?.provisional).toBe(true); + }); + + it("selectMessages includes the optimistic user message", () => { + let s = initialState(); + s = appendUserMessage(s, "what is 2+2?"); + const msgs = selectMessages(s); + expect(msgs).toHaveLength(1); + expect(msgs[0]?.role).toBe("user"); + expect(msgs[0]?.chunks).toHaveLength(1); + expect(msgs[0]?.chunks[0]).toEqual({ type: "text", text: "what is 2+2?" }); + }); + + it("user echo then turn-sealed + applyHistory supersedes the provisional user chunk", () => { + let s = initialState(); + s = appendUserMessage(s, "hi"); + expect(selectChunks(s)).toHaveLength(1); + + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "hello back")); + s = foldEvent(s, turnSealed("t1")); + s = applyHistory(s, [ + storedChunk(1, "user", { type: "text", text: "hi" }), + storedChunk(2, "assistant", { type: "text", text: "hello back" }), + ]); + const chunks = selectChunks(s); + expect(chunks).toHaveLength(2); + expect(chunks[0]?.seq).toBe(1); + expect(chunks[0]?.role).toBe("user"); + expect(chunks[0]?.chunk).toEqual({ type: "text", text: "hi" }); + expect(chunks[0]?.provisional).toBe(false); + expect(chunks[1]?.seq).toBe(2); + expect(chunks[1]?.role).toBe("assistant"); + expect(chunks[1]?.provisional).toBe(false); + }); + + it("flushes accumulating chunk before appending user message", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "partial")); + expect(s.accumulating).toEqual({ kind: "text", text: "partial" }); + + s = appendUserMessage(s, "user msg"); + expect(s.accumulating).toBeNull(); + expect(s.provisional).toHaveLength(2); + expect(s.provisional[0]?.role).toBe("assistant"); + expect(s.provisional[0]?.chunk).toEqual({ type: "text", text: "partial" }); + expect(s.provisional[1]?.role).toBe("user"); + expect(s.provisional[1]?.chunk).toEqual({ type: "text", text: "user msg" }); + }); +}); + +const PNG = "data:image/png;base64,AAAA"; +const JPG = "data:image/jpeg;base64,BBBB"; + +describe("appendUserMessage — images (vision handoff)", () => { + it("echoes text then image chunks in order", () => { + const images: ImageInput[] = [ + { url: PNG, mimeType: "image/png" }, + { url: JPG, mimeType: "image/jpeg" }, + ]; + let s = initialState(); + s = appendUserMessage(s, "what's this?", images); + const chunks = selectChunks(s); + expect(chunks).toHaveLength(3); + expect(chunks[0]?.role).toBe("user"); + expect(chunks[0]?.chunk).toEqual({ type: "text", text: "what's this?" }); + expect(chunks[1]?.chunk).toEqual({ type: "image", url: PNG, mimeType: "image/png" }); + expect(chunks[2]?.chunk).toEqual({ type: "image", url: JPG, mimeType: "image/jpeg" }); + expect(chunks.every((c) => c.provisional && c.seq === null)).toBe(true); + }); + + it("omits mimeType when not provided", () => { + let s = initialState(); + s = appendUserMessage(s, "look", [{ url: PNG }]); + const img = selectChunks(s)[1]?.chunk; + expect(img).toEqual({ type: "image", url: PNG }); + expect(img).not.toHaveProperty("mimeType"); + }); + + it("echoes images-only when text is empty (no text chunk)", () => { + let s = initialState(); + s = appendUserMessage(s, "", [{ url: PNG, mimeType: "image/png" }]); + const chunks = selectChunks(s); + expect(chunks).toHaveLength(1); + expect(chunks[0]?.chunk.type).toBe("image"); + }); + + it("echoes nothing for empty text + empty images", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "partial")); + s = appendUserMessage(s, "", []); + // Defensive flush still happened; no user chunk added. + const users = selectChunks(s).filter((c) => c.role === "user"); + expect(users).toHaveLength(0); + expect(s.accumulating).toBeNull(); + }); + + it("skips images with an empty url", () => { + let s = initialState(); + s = appendUserMessage(s, "hi", [ + { url: "", mimeType: "image/png" }, + { url: PNG, mimeType: "image/png" }, + ]); + const chunks = selectChunks(s); + expect(chunks).toHaveLength(2); // text + the one valid image + expect(chunks[1]?.chunk.type).toBe("image"); + }); + + it("groups text + images into one user ChatMessage", () => { + let s = initialState(); + s = appendUserMessage(s, "see this", [{ url: PNG }]); + const msgs = selectMessages(s); + expect(msgs).toHaveLength(1); + expect(msgs[0]?.role).toBe("user"); + expect(msgs[0]?.chunks).toHaveLength(2); + }); +}); + +describe("foldEvent — user-message dedups against a text+image echo", () => { + const userMessage = (text: string): TurnInputEvent => ({ + type: "user-message", + conversationId: "c1", + turnId: "t1", + text, + }); + + it("does not duplicate the text when images follow it in the echo", () => { + let s = initialState(); + s = appendUserMessage(s, "hi", [{ url: PNG, mimeType: "image/png" }]); + expect(selectChunks(s)).toHaveLength(2); // text + image + s = foldEvent(s, userMessage("hi")); + const users = selectChunks(s).filter((c) => c.role === "user"); + expect(users).toHaveLength(2); // unchanged — no duplicate text + expect(users[0]?.chunk.type).toBe("text"); + expect(users[1]?.chunk.type).toBe("image"); + expect(s.generating).toBe(true); + }); + + it("still appends when the echoed text differs", () => { + let s = initialState(); + s = appendUserMessage(s, "first", [{ url: PNG }]); + s = foldEvent(s, userMessage("second")); + const users = selectChunks(s).filter((c) => c.role === "user"); + expect(users.filter((c) => c.chunk.type === "text")).toHaveLength(2); + }); +}); + +describe("applyHistory — multi-chunk image echo is superseded by committed", () => { + it("drops the provisional [text, image] echo when committed arrives during generation", () => { + let s = initialState(); + s = appendUserMessage(s, "hi", [{ url: PNG, mimeType: "image/png" }]); + s = foldEvent(s, turnStart("t1")); + expect(s.provisional).toHaveLength(2); + + s = applyHistory(s, [ + storedChunk(1, "user", { type: "text", text: "hi" }), + storedChunk(2, "user", { type: "image", url: PNG, mimeType: "image/png" }), + ]); + expect(s.provisional).toEqual([]); + expect(s.committed).toHaveLength(2); + expect(s.committed[0]?.chunk).toEqual({ type: "text", text: "hi" }); + expect(s.committed[1]?.chunk).toEqual({ type: "image", url: PNG, mimeType: "image/png" }); + }); + + it("keeps the echo when committed is only a partial match (img not yet persisted)", () => { + let s = initialState(); + s = appendUserMessage(s, "hi", [ + { url: PNG, mimeType: "image/png" }, + { url: JPG, mimeType: "image/jpeg" }, + ]); + s = foldEvent(s, turnStart("t1")); + // Only the text + first image have been persisted so far. + s = applyHistory(s, [ + storedChunk(1, "user", { type: "text", text: "hi" }), + storedChunk(2, "user", { type: "image", url: PNG, mimeType: "image/png" }), + ]); + // Echo is [text, img1, img2]; committed run is [text, img1] — not a full + // match (echo is longer) → keep the echo (turn-seal will drop it wholesale). + expect(s.provisional.length).toBeGreaterThan(0); + }); + + it("renders committed image chunks from history (a non-vision transcription turn)", () => { + // The server persists the original image chunk AND the transcription text + // in the SAME user message: render both (image, then analysis text). + let s = initialState(); + s = applyHistory(s, [ + storedChunk(1, "user", { type: "text", text: "describe this" }), + storedChunk(2, "user", { type: "image", url: PNG, mimeType: "image/png" }), + storedChunk(3, "user", { + type: "text", + text: "[Image analysis (via kimi/k2)]: a red square", + }), + storedChunk(4, "assistant", { type: "text", text: "the square is red" }), + ]); + const msgs = selectMessages(s); + expect(msgs).toHaveLength(2); // one user message (3 chunks) + one assistant + expect(msgs[0]?.role).toBe("user"); + expect(msgs[0]?.chunks).toHaveLength(3); + expect(msgs[0]?.chunks[1]).toEqual({ type: "image", url: PNG, mimeType: "image/png" }); + }); }); diff --git a/src/core/chunks/reducer.ts b/src/core/chunks/reducer.ts index d3b999d..64e41b9 100644 --- a/src/core/chunks/reducer.ts +++ b/src/core/chunks/reducer.ts @@ -1,26 +1,103 @@ -import type { AgentEvent, Chunk, StoredChunk } from "@dispatch/wire"; +import type { AgentEvent, Chunk, ImageInput, Role, StoredChunk } from "@dispatch/wire"; +import { assertChunkExhaustive } from "../wire/conformance"; import type { AccumulatingChunk, ProvisionalChunk, TranscriptState } from "./types"; /** The initial empty transcript state. */ export function initialState(): TranscriptState { - return { - committed: [], - provisional: [], - accumulating: null, - currentTurnId: null, - latestUsage: null, - sealedTurnId: null, - }; + return { + committed: [], + provisional: [], + accumulating: null, + currentTurnId: null, + latestUsage: null, + sealedTurnId: null, + hiddenBeforeSeq: 0, + hiddenThinkingCount: 0, + generating: false, + providerRetry: null, + }; +} + +/** + * Clear the `generating` flag without touching anything else. Used on a WS + * (re)connect: a turn may have sealed while we were disconnected, so the live + * `turn-sealed`/`done` that would have cleared `generating` was missed. The + * caller resets here, then re-subscribes — if the turn is still running the + * server's replay re-asserts `generating` via the replayed `turn-start`. + */ +export function clearGenerating(state: TranscriptState): TranscriptState { + if (!state.generating) return state; + // Also drop a stale `provider-retry` banner — a retry pending at disconnect + // is stale once we re-subscribe (provider-retry events are not replayed), so + // a finished turn must not keep showing a "retrying…" banner forever. + return { ...state, generating: false, providerRetry: null }; } function flushAccumulating( - provisional: readonly ProvisionalChunk[], - acc: AccumulatingChunk | null, + provisional: readonly ProvisionalChunk[], + acc: AccumulatingChunk | null, +): readonly ProvisionalChunk[] { + if (acc === null) return provisional; + const chunk: Chunk = + acc.kind === "text" ? { type: "text", text: acc.text } : { type: "thinking", text: acc.text }; + return [...provisional, { role: "assistant", chunk }]; +} + +/** + * Content equality for two chunks of the SAME role (used to de-dup an + * optimistic echo against the authoritative committed version). Compares the + * discriminating payload only — `stepId` (generation provenance) is ignored + * since it is absent on provisional echoes but present on committed tool chunks. + */ +function chunkContentEquals(a: Chunk, b: Chunk): boolean { + if (a.type !== b.type) return false; + switch (a.type) { + case "text": { + const o = b as Extract<Chunk, { type: "text" }>; + return a.text === o.text; + } + case "thinking": { + const o = b as Extract<Chunk, { type: "thinking" }>; + return a.text === o.text; + } + case "image": { + const o = b as Extract<Chunk, { type: "image" }>; + return a.url === o.url; + } + case "error": { + const o = b as Extract<Chunk, { type: "error" }>; + return a.message === o.message && a.code === o.code; + } + case "system": { + const o = b as Extract<Chunk, { type: "system" }>; + return a.text === o.text; + } + case "tool-call": { + const o = b as Extract<Chunk, { type: "tool-call" }>; + return a.toolCallId === o.toolCallId && a.toolName === o.toolName; + } + case "tool-result": { + const o = b as Extract<Chunk, { type: "tool-result" }>; + return a.toolCallId === o.toolCallId && a.toolName === o.toolName && a.isError === o.isError; + } + default: + return assertChunkExhaustive(a) === assertChunkExhaustive(b); + } +} + +/** + * The trailing run of consecutive same-role provisional chunks at the end of + * `provisional` (the optimistic echo of one message). Returns the start/end + * indices `[start, end)` into `provisional` (empty if none). + */ +function trailingRun( + provisional: readonly ProvisionalChunk[], + role: Role, ): readonly ProvisionalChunk[] { - if (acc === null) return provisional; - const chunk: Chunk = - acc.kind === "text" ? { type: "text", text: acc.text } : { type: "thinking", text: acc.text }; - return [...provisional, { role: "assistant", chunk }]; + const end = provisional.length; + let start = end; + while (start > 0 && provisional[start - 1]?.role === role) start--; + return provisional.slice(start, end); } /** @@ -28,157 +105,342 @@ function flushAccumulating( * Dedupes by seq (new wins), keeps seq-monotonic order, idempotent. * When sealedTurnId is set, drops all provisional chunks (now superseded) * and clears sealedTurnId. + * + * Chunks below the chat-limit unload watermark (`hiddenBeforeSeq`) are + * REJECTED: a full-cache or tail merge must not resurrect what the trim + * unloaded. Restoring earlier history goes through `restoreEarlier` instead. */ export function applyHistory( - state: TranscriptState, - chunks: readonly StoredChunk[], + state: TranscriptState, + chunks: readonly StoredChunk[], ): TranscriptState { - const seqMap = new Map<number, StoredChunk>(); - for (const c of state.committed) seqMap.set(c.seq, c); - for (const c of chunks) seqMap.set(c.seq, c); - const committed = Array.from(seqMap.values()).sort((a, b) => a.seq - b.seq); - - if (state.sealedTurnId !== null) { - return { - ...state, - committed, - provisional: [], - accumulating: null, - sealedTurnId: null, - }; - } - - return { ...state, committed }; + const seqMap = new Map<number, StoredChunk>(); + for (const c of state.committed) seqMap.set(c.seq, c); + let addedNew = false; + for (const c of chunks) { + if (c.seq < state.hiddenBeforeSeq) continue; + if (!seqMap.has(c.seq)) addedNew = true; + seqMap.set(c.seq, c); + } + const committed = Array.from(seqMap.values()).sort((a, b) => a.seq - b.seq); + + if (state.sealedTurnId !== null) { + return { + ...state, + committed, + provisional: [], + accumulating: null, + sealedTurnId: null, + }; + } + + // During generation: if new committed chunks arrived, the provisional + // array may contain duplicates — the optimistic echo from `appendUserMessage` + // is now backed by committed chunks (CR-6: user message persisted at turn + // start). A user message may be multi-chunk (`[text, image, image, …]`), so + // 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) { + const provRun = trailingRun(state.provisional, "user"); + if (provRun.length > 0) { + const commRun = trailingRun(committed, "user"); + // Drop the provisional user echo iff every echoed chunk is content-equal + // to the corresponding committed chunk (the server persisted the same + // message). A partial match (echo longer than committed) keeps the echo + // — the not-yet-committed tail stays until the turn seals. + const fullyBacked = + commRun.length >= provRun.length && + provRun.every((p, i) => { + const c = commRun[i]; + return c !== undefined && chunkContentEquals(p.chunk, c.chunk); + }); + if (fullyBacked) { + const dropStart = state.provisional.length - provRun.length; + const provisional = state.provisional.slice(0, dropStart); + return { ...state, committed, provisional, accumulating: state.accumulating }; + } + } + } + + return { ...state, committed }; } /** * Fold one live AgentEvent into the provisional state. * * - `turn-start` records the turnId. + * - `user-message` appends the turn's user prompt (de-duped vs the sender's + * optimistic echo) so a watcher renders it mid-turn. * - `text-delta` extends the current accumulating TextChunk (or starts one). * - `reasoning-delta` extends the current accumulating ThinkingChunk (or starts one). * - `tool-call` / `tool-result` / `error` finalize any accumulating chunk and * add a new provisional chunk. + * - `steering` appends a user bubble mid-turn (drained from the message queue + * at a tool-result boundary; the queue surface separately clears on drain). * - `usage` stores the latest Usage. * - `done` finalizes any accumulating chunk (turn still provisional). * - `turn-sealed` finalizes any accumulating chunk and sets sealedTurnId. * - `status` and `tool-output` are ignored (best-effort no-ops). + * + * `generating` is folded structurally: a `turn-start` or any content delta sets + * it true; `done` / `turn-sealed` / `error` clear it. This is what a watching + * (or reconnected) client renders as "generating…", with no dependence on the + * free-form `status` event string. + * + * NOTE: this is the inner reducer. The transient `provider-retry` banner is SET + * here (on a `provider-retry` event) but CLEARED by the `foldEvent` wrapper + * below, so the clearing logic stays centralized in one place. */ +function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState { + switch (event.type) { + case "status": + case "tool-output": + return state; + + case "turn-start": + return { ...state, currentTurnId: event.turnId, generating: true }; + + case "user-message": { + // The turn's USER prompt, surfaced on the event stream (backend CR-3) so a + // WATCHER/late-joiner renders it mid-turn instead of waiting for seal. The + // SENDER already echoed its own prompt optimistically (`appendUserMessage`), + // so DE-DUP: skip if the trailing provisional USER run already contains an + // identical user text chunk. The echo may be multi-chunk (`[text, image, + // image, …]` — `appendUserMessage` appends the text first, then images), so + // we scan the whole trailing user run, not just the last chunk (which would + // be an image when images were pasted). A pure watcher has no such echo → + // it appends and renders. The `user-message` event carries ONLY text (never + // images — images arrive via history/loadSince); an images-only send (empty + // 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) { + return { ...state, generating: true }; + } + const provisional = flushAccumulating(state.provisional, state.accumulating); + return { + ...state, + provisional: [...provisional, { role: "user", chunk: { type: "text", text: event.text } }], + accumulating: null, + generating: true, + }; + } + + case "text-delta": { + const acc = state.accumulating; + if (acc !== null && acc.kind === "text") { + return { + ...state, + accumulating: { kind: "text", text: acc.text + event.delta }, + generating: true, + }; + } + const provisional = flushAccumulating(state.provisional, acc); + return { + ...state, + provisional, + accumulating: { kind: "text", text: event.delta }, + generating: true, + }; + } + + case "reasoning-delta": { + const acc = state.accumulating; + if (acc !== null && acc.kind === "thinking") { + return { + ...state, + accumulating: { kind: "thinking", text: acc.text + event.delta }, + generating: true, + }; + } + const provisional = flushAccumulating(state.provisional, acc); + return { + ...state, + provisional, + accumulating: { kind: "thinking", text: event.delta }, + generating: true, + }; + } + + case "tool-call": { + const provisional = flushAccumulating(state.provisional, state.accumulating); + const chunk: Chunk = { + type: "tool-call", + toolCallId: event.toolCallId, + toolName: event.toolName, + input: event.input, + stepId: event.stepId, + }; + return { + ...state, + provisional: [...provisional, { role: "assistant", chunk }], + accumulating: null, + generating: true, + }; + } + + case "tool-result": { + const provisional = flushAccumulating(state.provisional, state.accumulating); + const chunk: Chunk = { + type: "tool-result", + toolCallId: event.toolCallId, + toolName: event.toolName, + content: event.content, + isError: event.isError, + stepId: event.stepId, + }; + return { + ...state, + provisional: [...provisional, { role: "tool", chunk }], + accumulating: null, + generating: true, + }; + } + + case "error": { + const provisional = flushAccumulating(state.provisional, state.accumulating); + const chunk: Chunk = + event.code !== undefined + ? { type: "error", message: event.message, code: event.code } + : { type: "error", message: event.message }; + return { + ...state, + provisional: [...provisional, { role: "assistant", chunk }], + accumulating: null, + generating: false, + }; + } + + case "usage": + return { ...state, latestUsage: event.usage }; + + case "step-complete": + // Timing metadata — no content chunk; handled by the telemetry reducer. + return state; + + case "done": { + const provisional = flushAccumulating(state.provisional, state.accumulating); + return { + ...state, + provisional, + accumulating: null, + generating: false, + }; + } + + case "turn-sealed": { + const provisional = flushAccumulating(state.provisional, state.accumulating); + return { + ...state, + provisional, + accumulating: null, + sealedTurnId: event.turnId, + generating: false, + }; + } + + case "steering": { + // A steering message drained from the queue at a tool-result boundary + // (the model sees it alongside the tool results). Append a user bubble + // to the provisional transcript; the turn is still in flight. The queue + // surface clears separately on drain (a different channel) — no de-dup + // here (unlike `user-message`, steering is never optimistically echoed + // into the transcript by the sender). + if (event.text.length === 0) return state; + const provisional = flushAccumulating(state.provisional, state.accumulating); + return { + ...state, + provisional: [...provisional, { role: "user", chunk: { type: "text", text: event.text } }], + accumulating: null, + generating: true, + }; + } + + case "provider-retry": { + // TRANSIENT: a retryable provider error is being retried with backoff. + // Coalesce — the latest attempt + delay replaces any previous, so a + // single updating "retrying…" banner shows the newest. NOT a chunk: it + // never enters provisional/committed, so it can never pollute the prompt + // or be replayed on a reload. The turn is still in flight, so `generating` + // (already true from `turn-start`) is left untouched. + return { ...state, providerRetry: event }; + } + } +} + +/** + * Fold one live AgentEvent into the transcript state. Wraps `reduceEvent` to + * centralize the TRANSIENT `provider-retry` banner's clearing: the banner is + * SET by `reduceEvent` on a `provider-retry` event (coalescing), and CLEARED + * here when the model's content resumes (the retry succeeded) or the turn ends + * (done/sealed/error) or a new turn starts. Metadata/no-op events + * (`status`/`tool-output`/`usage`/`step-complete`) leave a showing banner + * untouched. The `state.providerRetry !== null` guard keeps the common path + * (no banner pending) identity-stable — no needless new object. + */ +// Events that clear a showing provider-retry banner (content resumed or turn ended). +const RETRY_CLEARING_EVENTS: ReadonlySet<AgentEvent["type"]> = new Set([ + "turn-start", + "text-delta", + "reasoning-delta", + "tool-call", + "tool-result", + "error", + "done", + "turn-sealed", +]); + export function foldEvent(state: TranscriptState, event: AgentEvent): TranscriptState { - switch (event.type) { - case "status": - case "tool-output": - return state; - - case "turn-start": - return { ...state, currentTurnId: event.turnId }; - - case "text-delta": { - const acc = state.accumulating; - if (acc !== null && acc.kind === "text") { - return { ...state, accumulating: { kind: "text", text: acc.text + event.delta } }; - } - const provisional = flushAccumulating(state.provisional, acc); - return { - ...state, - provisional, - accumulating: { kind: "text", text: event.delta }, - }; - } - - case "reasoning-delta": { - const acc = state.accumulating; - if (acc !== null && acc.kind === "thinking") { - return { ...state, accumulating: { kind: "thinking", text: acc.text + event.delta } }; - } - const provisional = flushAccumulating(state.provisional, acc); - return { - ...state, - provisional, - accumulating: { kind: "thinking", text: event.delta }, - }; - } - - case "tool-call": { - const provisional = flushAccumulating(state.provisional, state.accumulating); - const chunk: Chunk = { - type: "tool-call", - toolCallId: event.toolCallId, - toolName: event.toolName, - input: event.input, - }; - return { - ...state, - provisional: [...provisional, { role: "assistant", chunk }], - accumulating: null, - }; - } - - case "tool-result": { - const provisional = flushAccumulating(state.provisional, state.accumulating); - const chunk: Chunk = { - type: "tool-result", - toolCallId: event.toolCallId, - toolName: event.toolName, - content: event.content, - isError: event.isError, - }; - return { - ...state, - provisional: [...provisional, { role: "tool", chunk }], - accumulating: null, - }; - } - - case "error": { - const provisional = flushAccumulating(state.provisional, state.accumulating); - const chunk: Chunk = - event.code !== undefined - ? { type: "error", message: event.message, code: event.code } - : { type: "error", message: event.message }; - return { - ...state, - provisional: [...provisional, { role: "assistant", chunk }], - accumulating: null, - }; - } - - case "usage": - return { ...state, latestUsage: event.usage }; - - case "done": { - const provisional = flushAccumulating(state.provisional, state.accumulating); - return { - ...state, - provisional, - accumulating: null, - }; - } - - case "turn-sealed": { - const provisional = flushAccumulating(state.provisional, state.accumulating); - return { - ...state, - provisional, - accumulating: null, - sealedTurnId: event.turnId, - }; - } - } + 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) { + return { ...next, providerRetry: null }; + } + return next; } /** * Optimistically append a user message to the provisional list. * Flushes any in-progress accumulating chunk first (defensively). - * The provisional user chunk is superseded when applyHistory receives + * The provisional user chunks are superseded when applyHistory receives * the authoritative committed chunks after a turn seals. + * + * When `images` are provided, they are appended AFTER the text chunk (in + * order) as `image` chunks — matching the server's persisted layout + * (`[text, image, image, …]` in one user message). A text chunk is only + * appended when `text` is non-empty (an images-only send echoes just the + * images). The `user-message` event carries only text (never images), so its + * de-dup scans the trailing user run for the echoed text rather than just the + * last chunk. */ -export function appendUserMessage(state: TranscriptState, text: string): TranscriptState { - const provisional = flushAccumulating(state.provisional, state.accumulating); - const userChunk: Chunk = { type: "text", text }; - return { - ...state, - provisional: [...provisional, { role: "user", chunk: userChunk }], - accumulating: null, - }; +export function appendUserMessage( + state: TranscriptState, + text: string, + images?: readonly ImageInput[], +): TranscriptState { + const provisional = flushAccumulating(state.provisional, state.accumulating); + const userChunks: Chunk[] = []; + if (text.length > 0) userChunks.push({ type: "text", text }); + if (images !== undefined) { + for (const img of images) { + if (img.url.length === 0) continue; + const chunk: Chunk = + img.mimeType !== undefined + ? { type: "image", url: img.url, mimeType: img.mimeType } + : { type: "image", url: img.url }; + userChunks.push(chunk); + } + } + if (userChunks.length === 0) { + // Nothing to echo (empty text + no images) — leave state unchanged but + // still flush any accumulating chunk defensively. + return { ...state, provisional, accumulating: null }; + } + return { + ...state, + provisional: [...provisional, ...userChunks.map((chunk) => ({ role: "user", chunk }) as const)], + accumulating: null, + }; } diff --git a/src/core/chunks/retry-banner.test.ts b/src/core/chunks/retry-banner.test.ts new file mode 100644 index 0000000..1d8c1cd --- /dev/null +++ b/src/core/chunks/retry-banner.test.ts @@ -0,0 +1,63 @@ +import type { TurnProviderRetryEvent } from "@dispatch/wire"; +import { describe, expect, it } from "vitest"; +import { formatRetryDelay, viewProviderRetry } from "./retry-banner"; + +const retry = ( + attempt: number, + delayMs: number, + message = "HTTP 429: overloaded", + code?: string, +): TurnProviderRetryEvent => + code !== undefined + ? { + type: "provider-retry", + conversationId: "c1", + turnId: "t1", + attempt, + delayMs, + message, + code, + } + : { type: "provider-retry", conversationId: "c1", turnId: "t1", attempt, delayMs, message }; + +describe("formatRetryDelay", () => { + it("formats sub-minute delays as seconds", () => { + expect(formatRetryDelay(5000)).toBe("5s"); + expect(formatRetryDelay(10000)).toBe("10s"); + expect(formatRetryDelay(30000)).toBe("30s"); + }); + + it("formats minute+ delays as minutes", () => { + expect(formatRetryDelay(60000)).toBe("1m"); + expect(formatRetryDelay(300000)).toBe("5m"); + expect(formatRetryDelay(1800000)).toBe("30m"); + }); + + it("rounds to the nearest whole unit", () => { + expect(formatRetryDelay(5500)).toBe("6s"); // 5.5s -> 6s + expect(formatRetryDelay(90000)).toBe("2m"); // 1.5m -> 2m + }); +}); + +describe("viewProviderRetry", () => { + it("labels the attempt 1-based (attempt 0 = Retry #1)", () => { + expect(viewProviderRetry(retry(0, 5000)).attemptLabel).toBe("Retry #1"); + expect(viewProviderRetry(retry(1, 10000)).attemptLabel).toBe("Retry #2"); + expect(viewProviderRetry(retry(7, 1800000)).attemptLabel).toBe("Retry #8"); + }); + + it("derives the delay label from delayMs", () => { + expect(viewProviderRetry(retry(0, 5000)).delayLabel).toBe("5s"); + expect(viewProviderRetry(retry(4, 300000)).delayLabel).toBe("5m"); + }); + + it("passes the endpoint error verbatim", () => { + const msg = 'HTTP 429: {"error":{"type":"overloaded_error","message":"overloaded"}}'; + expect(viewProviderRetry(retry(0, 5000, msg)).message).toBe(msg); + }); + + it("surfaces the code when present, null when absent", () => { + expect(viewProviderRetry(retry(0, 5000, "msg", "429")).code).toBe("429"); + expect(viewProviderRetry(retry(0, 5000, "msg")).code).toBeNull(); + }); +}); diff --git a/src/core/chunks/retry-banner.ts b/src/core/chunks/retry-banner.ts new file mode 100644 index 0000000..afa6a98 --- /dev/null +++ b/src/core/chunks/retry-banner.ts @@ -0,0 +1,51 @@ +import type { TurnProviderRetryEvent } from "@dispatch/wire"; + +/** + * Pure view-model for the transient `provider-retry` warning banner. Zero DOM, + * zero effects, zero Svelte — mirrors the `core/metrics` view-models the chat UI + * already imports. The banner state itself lives in `TranscriptState.providerRetry` + * (set/cleared by `foldEvent`); this module only formats an event into render data. + * + * The "countdown" is a STATIC label derived from `delayMs` (e.g. "retrying in + * 5s…"), matching the backend's examples — NOT a live ticking timer (which would + * be a component effect and re-render churn). Each new `provider-retry` event + * coalesces over the previous, so the banner always shows the newest attempt + delay. + */ + +/** The display shape for a provider-retry banner. */ +export interface ProviderRetryView { + /** "Retry #N" — `attempt` is 0-based, so +1 (attempt 0 = "Retry #1"). */ + readonly attemptLabel: string; + /** The scheduled sleep as a short duration: "5s" / "30s" / "1m" / "5m" / "30m". */ + readonly delayLabel: string; + /** The endpoint's error verbatim, e.g. "HTTP 429: {…overloaded_error…}". */ + readonly message: string; + /** The HTTP code when known (e.g. "429"), else null. */ + readonly code: string | null; +} + +/** + * Format a millisecond delay as a short, human-friendly duration. Matches the + * backend's backoff schedule (5s→10s→30s→60s→5m→10m→15m→30m): under a minute + * shows seconds, under an hour shows minutes, else hours. + */ +export function formatRetryDelay(ms: number): string { + const totalSeconds = Math.round(ms / 1000); + if (totalSeconds < 60) return `${totalSeconds}s`; + const totalMinutes = Math.round(totalSeconds / 60); + if (totalMinutes < 60) return `${totalMinutes}m`; + return `${Math.round(totalMinutes / 60)}h`; +} + +/** + * Map a `provider-retry` event to its banner view. `attempt` is 0-based (the Nth + * retry about to happen), so the label is 1-based for the user. + */ +export function viewProviderRetry(event: TurnProviderRetryEvent): ProviderRetryView { + return { + attemptLabel: `Retry #${event.attempt + 1}`, + delayLabel: formatRetryDelay(event.delayMs), + message: event.message, + code: event.code ?? null, + }; +} diff --git a/src/core/chunks/selectors.ts b/src/core/chunks/selectors.ts index 8fb832f..e46d5f5 100644 --- a/src/core/chunks/selectors.ts +++ b/src/core/chunks/selectors.ts @@ -1,4 +1,4 @@ -import type { ChatMessage, Chunk } from "@dispatch/wire"; +import type { ChatMessage, Chunk, TurnProviderRetryEvent } from "@dispatch/wire"; import type { RenderedChunk, TranscriptState } from "./types"; /** @@ -6,46 +6,64 @@ import type { RenderedChunk, TranscriptState } from "./types"; * then provisional (seq: null). */ export function selectChunks(state: TranscriptState): readonly RenderedChunk[] { - const result: RenderedChunk[] = []; - for (const c of state.committed) { - result.push({ seq: c.seq, role: c.role, chunk: c.chunk, provisional: false }); - } - for (const p of state.provisional) { - result.push({ seq: null, role: p.role, chunk: p.chunk, provisional: true }); - } - if (state.accumulating !== null) { - const chunk: Chunk = - state.accumulating.kind === "text" - ? { type: "text", text: state.accumulating.text } - : { type: "thinking", text: state.accumulating.text }; - result.push({ seq: null, role: "assistant", chunk, provisional: true }); - } - return result; + const result: RenderedChunk[] = []; + for (const c of state.committed) { + result.push({ seq: c.seq, role: c.role, chunk: c.chunk, provisional: false }); + } + for (const p of state.provisional) { + result.push({ seq: null, role: p.role, chunk: p.chunk, provisional: true }); + } + if (state.accumulating !== null) { + const chunk: Chunk = + state.accumulating.kind === "text" + ? { type: "text", text: state.accumulating.text } + : { type: "thinking", text: state.accumulating.text }; + result.push({ seq: null, role: "assistant", chunk, provisional: true, streaming: true }); + } + return result; +} + +/** + * Whether a turn is currently generating (for a "generating…" indicator). True + * for ANY client watching the conversation — the sender, a second device, or a + * reconnected client whose in-flight turn was replayed. + */ +export function selectGenerating(state: TranscriptState): boolean { + return state.generating; +} + +/** + * The latest `provider-retry` event for the current turn, or `null` when no retry + * is pending. Drives the transient yellow "retrying…" warning banner (rendered + * by ChatView). Never persisted — see `TranscriptState.providerRetry`. + */ +export function selectProviderRetry(state: TranscriptState): TurnProviderRetryEvent | null { + return state.providerRetry; } /** * Group consecutive same-role rendered chunks into ChatMessages. */ export function selectMessages(state: TranscriptState): readonly ChatMessage[] { - const rendered = selectChunks(state); - const first = rendered[0]; - if (first === undefined) return []; + const rendered = selectChunks(state); + const first = rendered[0]; + if (first === undefined) return []; - const messages: ChatMessage[] = []; - let role = first.role; - let chunks: Chunk[] = [first.chunk]; + const messages: ChatMessage[] = []; + let role = first.role; + let chunks: Chunk[] = [first.chunk]; - for (let i = 1; i < rendered.length; i++) { - const rc = rendered[i]; - if (rc === undefined) continue; - if (rc.role === role) { - chunks.push(rc.chunk); - } else { - messages.push({ role, chunks }); - role = rc.role; - chunks = [rc.chunk]; - } - } - messages.push({ role, chunks }); - return messages; + for (let i = 1; i < rendered.length; i++) { + const rc = rendered[i]; + if (rc === undefined) continue; + if (rc.role === role) { + chunks.push(rc.chunk); + } else { + messages.push({ role, chunks }); + role = rc.role; + chunks = [rc.chunk]; + } + } + messages.push({ role, chunks }); + return messages; } diff --git a/src/core/chunks/trim.test.ts b/src/core/chunks/trim.test.ts new file mode 100644 index 0000000..7c4bbef --- /dev/null +++ b/src/core/chunks/trim.test.ts @@ -0,0 +1,235 @@ +import type { StoredChunk } from "@dispatch/wire"; +import { describe, expect, it } from "vitest"; +import { applyHistory, initialState } from "./reducer"; +import { + DEFAULT_CHAT_LIMIT, + initialWindowSize, + MAX_CHAT_LIMIT, + MIN_CHAT_LIMIT, + normalizeChatLimit, + restoreEarlier, + selectHasEarlier, + trimTranscript, + unloadCount, + windowTranscript, +} from "./trim"; +import type { TranscriptState } from "./types"; + +function chunk(seq: number, type: "text" | "thinking" = "text"): StoredChunk { + return { seq, role: "assistant", chunk: { type, text: `c${seq}` } }; +} + +function chunks(from: number, to: number): StoredChunk[] { + const out: StoredChunk[] = []; + for (let seq = from; seq <= to; seq++) out.push(chunk(seq)); + return out; +} + +function stateWith(committed: readonly StoredChunk[]): TranscriptState { + return { ...initialState(), committed }; +} + +describe("normalizeChatLimit", () => { + it("defaults non-numeric / NaN / missing values", () => { + expect(normalizeChatLimit(undefined)).toBe(DEFAULT_CHAT_LIMIT); + expect(normalizeChatLimit(null)).toBe(DEFAULT_CHAT_LIMIT); + expect(normalizeChatLimit("100")).toBe(DEFAULT_CHAT_LIMIT); + expect(normalizeChatLimit(Number.NaN)).toBe(DEFAULT_CHAT_LIMIT); + expect(normalizeChatLimit(Number.POSITIVE_INFINITY)).toBe(DEFAULT_CHAT_LIMIT); + }); + + it("floors and clamps numeric values", () => { + expect(normalizeChatLimit(100.9)).toBe(100); + expect(normalizeChatLimit(0)).toBe(MIN_CHAT_LIMIT); + expect(normalizeChatLimit(-5)).toBe(MIN_CHAT_LIMIT); + expect(normalizeChatLimit(10_000_000)).toBe(MAX_CHAT_LIMIT); + expect(normalizeChatLimit(256)).toBe(256); + }); +}); + +describe("unloadCount / initialWindowSize", () => { + it("unload is a quarter of the limit, rounded up", () => { + expect(unloadCount(100)).toBe(25); + expect(unloadCount(256)).toBe(64); + expect(unloadCount(10)).toBe(3); + }); + + it("initial window is 75% of the limit, rounded down", () => { + expect(initialWindowSize(100)).toBe(75); + expect(initialWindowSize(256)).toBe(192); + expect(initialWindowSize(1)).toBe(1); // never below 1 + }); +}); + +describe("trimTranscript", () => { + it("is the identity at or under the limit", () => { + const at = stateWith(chunks(1, 100)); + expect(trimTranscript(at, 100)).toBe(at); + const under = stateWith(chunks(1, 99)); + expect(trimTranscript(under, 100)).toBe(under); + }); + + it("unloads exactly a quarter when the limit is first exceeded (100 → 101 drops 25)", () => { + const state = stateWith(chunks(1, 101)); + const next = trimTranscript(state, 100); + expect(next.committed).toHaveLength(76); + expect(next.committed[0]?.seq).toBe(26); + expect(next.hiddenBeforeSeq).toBe(26); + }); + + it("unloads multiple quarters when trimming was deferred far past the limit", () => { + const state = stateWith(chunks(1, 130)); + const next = trimTranscript(state, 100); + // 130 → needs 2 quarters (25 each) to get to ≤ 100 → 80 remain. + expect(next.committed).toHaveLength(80); + expect(next.committed[0]?.seq).toBe(51); + expect(next.hiddenBeforeSeq).toBe(51); + }); + + it("counts provisional + accumulating toward the limit (drops committed first)", () => { + const base = stateWith(chunks(1, 98)); + const state: TranscriptState = { + ...base, + provisional: [ + { role: "user", chunk: { type: "text", text: "q" } }, + { role: "assistant", chunk: { type: "text", text: "a" } }, + ], + accumulating: { kind: "text", text: "stream" }, + }; + // 98 + 2 + 1 = 101 > 100 → drop 25 committed. + const next = trimTranscript(state, 100); + expect(next.committed).toHaveLength(73); + expect(next.provisional).toHaveLength(2); + expect(next.accumulating).not.toBeNull(); + }); + + it("drops oldest provisional when committed is exhausted", () => { + const base = stateWith(chunks(1, 2)); + const provisional = Array.from({ length: 20 }, (_, i) => ({ + role: "assistant" as const, + chunk: { type: "text" as const, text: `p${i}` }, + })); + const state: TranscriptState = { ...base, provisional }; + // 2 + 20 = 22 > 10. quarter = 3. Drop 2 committed, then drop + // ceil((20-10)/3)*3 = 12 provisional → 8 remain. + const next = trimTranscript(state, 10); + expect(next.committed).toHaveLength(0); + expect(next.provisional).toHaveLength(8); + expect(next.hiddenBeforeSeq).toBe(3); + }); + + it("accumulates the hidden thinking count for stable render keys", () => { + const committed = [chunk(1, "thinking"), ...chunks(2, 9), chunk(10, "thinking"), chunk(11)]; + const state = stateWith(committed); + const next = trimTranscript(state, 10); // 11 > 10 → drop ceil(10/4)=3 oldest + expect(next.committed[0]?.seq).toBe(4); + expect(next.hiddenThinkingCount).toBe(1); + }); + + it("ignores a nonsensical limit", () => { + const state = stateWith(chunks(1, 50)); + expect(trimTranscript(state, 0)).toBe(state); + expect(trimTranscript(state, Number.NaN)).toBe(state); + }); +}); + +describe("windowTranscript", () => { + it("keeps only the newest maxCommitted chunks and sets the watermark", () => { + const state = stateWith(chunks(1, 1000)); + const next = windowTranscript(state, 75); + expect(next.committed).toHaveLength(75); + expect(next.committed[0]?.seq).toBe(926); + expect(next.hiddenBeforeSeq).toBe(926); + expect(selectHasEarlier(next)).toBe(true); + }); + + it("is the identity within the window", () => { + const state = stateWith(chunks(1, 50)); + expect(windowTranscript(state, 75)).toBe(state); + expect(selectHasEarlier(state)).toBe(false); + }); +}); + +describe("applyHistory respects the watermark", () => { + it("does not resurrect chunks below hiddenBeforeSeq on a full-cache merge", () => { + const trimmed = trimTranscript(stateWith(chunks(1, 101)), 100); + expect(trimmed.hiddenBeforeSeq).toBe(26); + // A later sync merges the FULL cache (seqs 1..101) — the unloaded prefix must stay out. + const merged = applyHistory(trimmed, chunks(1, 101)); + expect(merged.committed[0]?.seq).toBe(26); + expect(merged.committed).toHaveLength(76); + }); + + it("still merges the tail above the watermark", () => { + const trimmed = trimTranscript(stateWith(chunks(1, 101)), 100); + const merged = applyHistory(trimmed, chunks(100, 110)); + expect(merged.committed[merged.committed.length - 1]?.seq).toBe(110); + expect(merged.committed[0]?.seq).toBe(26); + }); +}); + +describe("restoreEarlier", () => { + it("pages the newest `count` earlier chunks back in and lowers the watermark", () => { + const windowed = windowTranscript(stateWith(chunks(1, 1000)), 75); // loaded 926..1000 + const restored = restoreEarlier(windowed, chunks(1, 1000), 64); + expect(restored.committed[0]?.seq).toBe(862); + expect(restored.committed).toHaveLength(75 + 64); + expect(restored.hiddenBeforeSeq).toBe(862); + expect(selectHasEarlier(restored)).toBe(true); + }); + + it("restoring down to seq 1 reaches the contractual origin (hasEarlier clears)", () => { + 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); + }); + + 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. + expect(selectHasEarlier(restored)).toBe(true); + }); + + it("is the identity when the window already starts at seq 1", () => { + const state = stateWith(chunks(1, 10)); + expect(restoreEarlier(state, chunks(1, 10), 5)).toBe(state); + }); + + 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 + // backfilled run merges below it. + const state = stateWith(chunks(809, 1000)); + expect(state.hiddenBeforeSeq).toBe(0); + expect(selectHasEarlier(state)).toBe(true); + const restored = restoreEarlier(state, chunks(745, 808), 64); + expect(restored.committed[0]?.seq).toBe(745); + expect(restored.committed).toHaveLength(192 + 64); + expect(restored.hiddenBeforeSeq).toBe(745); + expect(selectHasEarlier(restored)).toBe(true); + }); + + it("decrements the hidden thinking count by the restored thinking chunks", () => { + const committed = [chunk(1, "thinking"), chunk(2), chunk(3, "thinking"), ...chunks(4, 12)]; + const trimmed = trimTranscript(stateWith(committed), 10); // drops 3: seqs 1..3 (2 thinking) + expect(trimmed.hiddenThinkingCount).toBe(2); + const restored = restoreEarlier(trimmed, committed, 2); // restores seqs 2..3 (1 thinking) + expect(restored.hiddenBeforeSeq).toBe(2); + expect(restored.hiddenThinkingCount).toBe(1); + }); + + it("round-trips with trim: trim → restore-all yields the original committed list", () => { + const original = chunks(1, 101); + const trimmed = trimTranscript(stateWith(original), 100); + const restored = restoreEarlier(trimmed, original, 1000); + expect(restored.committed).toEqual(original); + expect(restored.hiddenBeforeSeq).toBe(1); + expect(selectHasEarlier(restored)).toBe(false); + }); +}); diff --git a/src/core/chunks/trim.ts b/src/core/chunks/trim.ts new file mode 100644 index 0000000..9846357 --- /dev/null +++ b/src/core/chunks/trim.ts @@ -0,0 +1,185 @@ +// Chat-limit windowing for the transcript — PURE policy, zero DOM/Svelte. +// +// In very long conversations an unbounded transcript makes the browser crawl, so +// the FE keeps at most `chat limit` chunks loaded and UNLOADS the oldest ones in +// BULK: a quarter of the limit at a time (limit 100 → at 101 chunks it unloads 25, +// leaving 76). Bulk-on-threshold — NOT one-per-delta like old Dispatch — so a trim +// happens once per ~quarter-limit of new content instead of on every step, which +// was the old scroll-jump-per-step failure mode. A fresh page load shows only the +// newest `floor(0.75 × limit)` chunks, leaving headroom before the first trim. +// +// Unloading drops COMMITTED chunks only (provisional chunks are the in-flight +// turn; they become committed at seal and trimmable then) and records the +// `hiddenBeforeSeq` watermark so history merges can't resurrect them and the +// "Show earlier messages" affordance knows where to page back in from. + +import type { StoredChunk } from "@dispatch/wire"; +import type { TranscriptState } from "./types"; + +/** Default chat limit (max loaded chunks per conversation). */ +export const DEFAULT_CHAT_LIMIT = 256; +/** Hard floor for a configured chat limit (a tiny window would thrash). */ +export const MIN_CHAT_LIMIT = 10; +/** Hard ceiling for a configured chat limit. */ +export const MAX_CHAT_LIMIT = 100_000; + +/** + * Normalize an untrusted configured limit (e.g. parsed from localStorage): + * non-numeric/NaN → the default; otherwise floored + clamped to + * [MIN_CHAT_LIMIT, MAX_CHAT_LIMIT]. + */ +export function normalizeChatLimit(value: unknown): number { + if (typeof value !== "number" || !Number.isFinite(value)) return DEFAULT_CHAT_LIMIT; + const n = Math.floor(value); + if (n < MIN_CHAT_LIMIT) return MIN_CHAT_LIMIT; + if (n > MAX_CHAT_LIMIT) return MAX_CHAT_LIMIT; + return n; +} + +/** The bulk-unload unit: a quarter of the limit, rounded up. */ +export function unloadCount(limit: number): number { + return Math.ceil(limit / 4); +} + +/** The fresh-load window: 75% of the limit, rounded down (≥ 1). */ +export function initialWindowSize(limit: number): number { + return Math.max(1, Math.floor(limit * 0.75)); +} + +/** Total loaded (rendered) chunk count: committed + provisional + accumulating. */ +function totalCount(state: TranscriptState): number { + return state.committed.length + state.provisional.length + (state.accumulating !== null ? 1 : 0); +} + +function countThinking(chunks: readonly StoredChunk[]): number { + let n = 0; + for (const c of chunks) { + if (c.chunk.type === "thinking") n++; + } + return n; +} + +/** Drop the `drop` oldest committed chunks, advancing the watermark + thinking base. */ +function dropOldest(state: TranscriptState, drop: number): TranscriptState { + const dropped = state.committed.slice(0, drop); + const kept = state.committed.slice(drop); + const first = kept[0]; + const lastDropped = dropped[dropped.length - 1]; + let hiddenBeforeSeq = state.hiddenBeforeSeq; + if (first !== undefined) { + hiddenBeforeSeq = first.seq; + } else if (lastDropped !== undefined) { + hiddenBeforeSeq = lastDropped.seq + 1; + } + return { + ...state, + committed: kept, + hiddenBeforeSeq, + hiddenThinkingCount: state.hiddenThinkingCount + countThinking(dropped), + }; +} + +/** + * Enforce the chat limit: when the loaded count EXCEEDS `limit`, unload whole + * quarters (`unloadCount(limit)` each) of the OLDEST committed chunks until back + * at/under the limit — normally exactly one quarter (limit 100: 101 → 76); more + * only when trimming was deferred (e.g. while the reader was scrolled up). + * At/under the limit this is the identity. When committed chunks are + * exhausted, also drops the oldest provisional chunks (the in-flight turn) + * to keep the browser responsive during very long turns. + */ +export function trimTranscript(state: TranscriptState, limit: number): TranscriptState { + if (!Number.isFinite(limit) || limit <= 0) return state; + const total = totalCount(state); + if (total <= limit) return state; + const quarter = unloadCount(limit); + const passes = Math.ceil((total - limit) / quarter); + + // First, drop oldest committed chunks (the usual path). + const committedDrop = Math.min(passes * quarter, state.committed.length); + let next = committedDrop > 0 ? dropOldest(state, committedDrop) : state; + + // If still over the limit and committed is exhausted, drop oldest + // provisional chunks (the in-flight turn). These chunks have no seq + // (not yet persisted) and can't be "Show earlier" — but dropping them + // keeps the browser responsive. They'll come back as committed when + // the turn seals and syncTail fetches them from the server. + const remaining = totalCount(next); + if (remaining > limit && next.provisional.length > 0) { + const provisionalDrop = Math.min( + Math.ceil((remaining - limit) / quarter) * quarter, + next.provisional.length, + ); + if (provisionalDrop > 0) { + next = { + ...next, + provisional: next.provisional.slice(provisionalDrop), + }; + } + } + + return next; +} + +/** + * Window the committed history down to the newest `maxCommitted` chunks (the + * fresh-load path: `maxCommitted = initialWindowSize(limit)`). Identity when + * already within the window. + */ +export function windowTranscript(state: TranscriptState, maxCommitted: number): TranscriptState { + if (!Number.isFinite(maxCommitted) || maxCommitted < 0) return state; + const drop = state.committed.length - maxCommitted; + if (drop <= 0) return state; + return dropOldest(state, drop); +} + +/** + * The oldest LOADED seq — the start of the transcript's loaded window. Usually + * `committed[0].seq`; falls back to the watermark when a trim emptied the + * committed list (all-provisional overflow). 0 = window start unknown/origin. + */ +function oldestLoadedSeq(state: TranscriptState): number { + return state.committed[0]?.seq ?? state.hiddenBeforeSeq; +} + +/** + * Page earlier history back in — the "Show earlier messages" action. + * + * `earlier` is every locally-known chunk older than the loaded window + * (typically the full cached conversation, possibly extended by a CR-5 + * `?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. + */ +export function restoreEarlier( + state: TranscriptState, + earlier: readonly StoredChunk[], + count: number, +): TranscriptState { + const oldest = oldestLoadedSeq(state); + if (oldest <= 1) 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)); + const firstKept = keep[0]; + return { + ...state, + committed: [...keep, ...state.committed], + hiddenBeforeSeq: firstKept?.seq ?? state.hiddenBeforeSeq, + hiddenThinkingCount: Math.max(0, state.hiddenThinkingCount - countThinking(keep)), + }; +} + +/** + * Whether earlier history exists below the loaded window — drives the + * "Show earlier messages" affordance. Derived from the [email protected] 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; +} diff --git a/src/core/chunks/types.ts b/src/core/chunks/types.ts index 3792445..2ced736 100644 --- a/src/core/chunks/types.ts +++ b/src/core/chunks/types.ts @@ -1,31 +1,77 @@ -import type { Chunk, Role, StoredChunk, Usage } from "@dispatch/wire"; +import type { Chunk, Role, StoredChunk, TurnProviderRetryEvent, Usage } from "@dispatch/wire"; /** A chunk being accumulated from streaming deltas (text or thinking). */ export interface AccumulatingChunk { - readonly kind: "text" | "thinking"; - readonly text: string; + readonly kind: "text" | "thinking"; + readonly text: string; } /** A provisional chunk that has no authoritative seq yet. */ export interface ProvisionalChunk { - readonly role: Role; - readonly chunk: Chunk; + readonly role: Role; + readonly chunk: Chunk; } /** The transcript reducer state. Holds committed history + live in-flight turn. */ export interface TranscriptState { - readonly committed: readonly StoredChunk[]; - readonly provisional: readonly ProvisionalChunk[]; - readonly accumulating: AccumulatingChunk | null; - readonly currentTurnId: string | null; - readonly latestUsage: Usage | null; - readonly sealedTurnId: string | null; + readonly committed: readonly StoredChunk[]; + readonly provisional: readonly ProvisionalChunk[]; + readonly accumulating: AccumulatingChunk | null; + readonly currentTurnId: string | null; + readonly latestUsage: Usage | null; + readonly sealedTurnId: string | null; + /** + * The chat-limit UNLOAD watermark: committed chunks with `seq <` this are + * unloaded (not in `committed`, not rendered) to keep long transcripts cheap. + * `0` = nothing unloaded. `applyHistory` refuses chunks below it (a cache/tail + * merge must not resurrect what the trim dropped); "Show earlier messages" + * lowers it via `restoreEarlier`. See `trim.ts`. + */ + readonly hiddenBeforeSeq: number; + /** + * How many thinking-type chunks are currently unloaded below the watermark. + * Pure render-key bookkeeping: the UI keys thinking collapses by ORDINAL (so + * the key survives the provisional→committed seal transition), and this base + * keeps those ordinals stable when a trim removes older thinking chunks — + * otherwise every remaining collapse would shift keys and swap/lose its + * open state mid-stream. + */ + readonly hiddenThinkingCount: number; + /** + * True while a turn is generating on the server — derived STRUCTURALLY from the + * event stream: a `turn-start` (or any turn delta) with no matching `done` / + * `turn-sealed` / `error` yet. A late-joiner that subscribes mid-turn gets the + * in-flight turn replayed from its `turn-start`, so this lights up for any + * watching client. NOT inferred from the free-form `status` event string. + */ + readonly generating: boolean; + /** + * The latest `provider-retry` event for the current turn, or `null` when no + * retry is pending. TRANSIENT UI state (never a Chunk — never committed or + * provisional, so it can NEVER pollute the model's prompt or be replayed on a + * reload/replay of past turns: only committed seq'd chunks are history). Set + * by `foldEvent` on each `provider-retry` (the latest coalesces over previous + * so a single updating "retrying…" banner shows the newest attempt + delay); + * cleared when the model's content resumes (`text-delta`/`reasoning-delta`/ + * `tool-call`/`tool-result`), the turn ends (`done`/`turn-sealed`/`error`), + * or a new turn starts (`turn-start`). Also cleared on a WS reconnect + * (`clearGenerating`) — a retry pending at disconnect is stale once we + * re-subscribe (provider-retry events are not replayed). + */ + readonly providerRetry: TurnProviderRetryEvent | null; } /** A chunk ready for rendering: either committed (with seq) or provisional. */ export interface RenderedChunk { - readonly seq: number | null; - readonly role: Role; - readonly chunk: Chunk; - readonly provisional: boolean; + readonly seq: number | null; + readonly role: Role; + readonly chunk: Chunk; + readonly provisional: boolean; + /** + * True only for the single chunk currently being accumulated from live deltas + * (the in-flight text/thinking the model is actively generating). Absent/false + * once flushed or committed. Lets the UI show a live indicator (e.g. loading + * dots on streaming thinking) and drop it the moment generation moves on. + */ + readonly streaming?: boolean; } diff --git a/src/core/metrics/format.test.ts b/src/core/metrics/format.test.ts new file mode 100644 index 0000000..97170d0 --- /dev/null +++ b/src/core/metrics/format.test.ts @@ -0,0 +1,375 @@ +import type { StepId, StepMetrics, TurnMetrics } from "@dispatch/wire"; +import { describe, expect, it } from "vitest"; +import { + computeCachePct, + computeContextUsage, + computeExpectedCachePct, + computeTps, + formatCompactTokens, + formatContextSize, + viewCacheRate, + viewExpectedCache, + viewStepMetrics, + viewTurnMetrics, +} from "./format"; + +describe("computeTps", () => { + it("null when elapsed missing", () => { + expect(computeTps(100, undefined)).toBeNull(); + }); + + it("null when elapsed is zero", () => { + expect(computeTps(100, 0)).toBeNull(); + }); + + it("null when elapsed is negative", () => { + expect(computeTps(100, -100)).toBeNull(); + }); + + it("computes tokens per second", () => { + expect(computeTps(1000, 2000)).toBe(500); + }); + + it("computes fractional tps", () => { + expect(computeTps(100, 3000)).toBeCloseTo(33.33, 1); + }); +}); + +describe("viewStepMetrics", () => { + it("formats tokens with thousands separator, tps, and durations", () => { + const step: StepMetrics = { + stepId: "s1" as StepId, + usage: { inputTokens: 1234, outputTokens: 567 }, + ttftMs: 820, + decodeMs: 1200, + genTotalMs: 2020, + }; + const view = viewStepMetrics(step, 0); + expect(view.label).toBe("step 1"); + expect(view.tokensLabel).toBe("1,801 tok"); + expect(view.tps).toBe("473 tok/s"); + expect(view.ttft).toBe("820ms"); + expect(view.decode).toBe("1.2s"); + expect(view.genTotal).toBe("2.0s"); + }); + + it("handles missing timing fields", () => { + const step: StepMetrics = { + stepId: "s1" as StepId, + usage: { inputTokens: 100, outputTokens: 50 }, + }; + const view = viewStepMetrics(step, 0); + expect(view.tps).toBeNull(); + expect(view.ttft).toBeNull(); + expect(view.decode).toBeNull(); + expect(view.genTotal).toBeNull(); + }); + + it("formats duration < 1s as ms", () => { + const step: StepMetrics = { + stepId: "s1" as StepId, + usage: { inputTokens: 10, outputTokens: 5 }, + ttftMs: 42, + }; + const view = viewStepMetrics(step, 0); + expect(view.ttft).toBe("42ms"); + }); + + it("formats duration >= 1s as seconds", () => { + const step: StepMetrics = { + stepId: "s1" as StepId, + usage: { inputTokens: 10, outputTokens: 5 }, + genTotalMs: 3200, + }; + const view = viewStepMetrics(step, 0); + expect(view.genTotal).toBe("3.2s"); + }); + + it("uses step index for label", () => { + const step: StepMetrics = { + stepId: "s1" as StepId, + usage: { inputTokens: 10, outputTokens: 5 }, + }; + expect(viewStepMetrics(step, 2).label).toBe("step 3"); + }); + + it("tps uses decodeMs (not genTotalMs)", () => { + const step: StepMetrics = { + stepId: "s1" as StepId, + usage: { inputTokens: 100, outputTokens: 50 }, + decodeMs: 500, + genTotalMs: 800, + }; + const view = viewStepMetrics(step, 0); + // 50 / (500/1000) = 100 tok/s, NOT 50/(800/1000)=62.5 + expect(view.tps).toBe("100 tok/s"); + }); + + it("tps falls back to genTotalMs when decodeMs absent", () => { + const step: StepMetrics = { + stepId: "s1" as StepId, + usage: { inputTokens: 100, outputTokens: 50 }, + genTotalMs: 800, + }; + const view = viewStepMetrics(step, 0); + // 50 / (800/1000) = 62.5 → rounds to 63 + expect(view.tps).toBe("63 tok/s"); + }); +}); + +describe("viewTurnMetrics", () => { + it("formats total tokens and breakdown", () => { + const turn: TurnMetrics = { + turnId: "t1", + usage: { inputTokens: 1000, outputTokens: 234 }, + durationMs: 5000, + steps: [ + { + stepId: "s1" as StepId, + usage: { inputTokens: 1000, outputTokens: 234 }, + decodeMs: 3000, + genTotalMs: 4000, + }, + ], + }; + const view = viewTurnMetrics(turn); + expect(view.tokensLabel).toBe("1,234 tok"); + expect(view.breakdown).toBe("1,000 in / 234 out"); + expect(view.tps).toBe("78 tok/s"); + expect(view.duration).toBe("5.0s"); + }); + + it("breakdown includes cache only when present", () => { + const turn: TurnMetrics = { + turnId: "t1", + usage: { inputTokens: 1000, outputTokens: 234, cacheReadTokens: 500 }, + steps: [], + }; + const view = viewTurnMetrics(turn); + expect(view.breakdown).toBe("1,000 in / 234 out / 500 cache"); + }); + + it("breakdown omits cache when not present", () => { + const turn: TurnMetrics = { + turnId: "t1", + usage: { inputTokens: 100, outputTokens: 50 }, + steps: [], + }; + const view = viewTurnMetrics(turn); + expect(view.breakdown).toBe("100 in / 50 out"); + }); + + it("tps is null when no step has decodeMs or genTotalMs", () => { + const turn: TurnMetrics = { + turnId: "t1", + usage: { inputTokens: 100, outputTokens: 50 }, + steps: [ + { + stepId: "s1" as StepId, + usage: { inputTokens: 100, outputTokens: 50 }, + }, + ], + }; + const view = viewTurnMetrics(turn); + expect(view.tps).toBeNull(); + }); + + it("duration is null when durationMs absent", () => { + const turn: TurnMetrics = { + turnId: "t1", + usage: { inputTokens: 100, outputTokens: 50 }, + steps: [], + }; + const view = viewTurnMetrics(turn); + expect(view.duration).toBeNull(); + }); + + it("sums decodeMs across steps (fallback genTotalMs per step) for tps", () => { + const turn: TurnMetrics = { + turnId: "t1", + usage: { inputTokens: 300, outputTokens: 150 }, + steps: [ + { + stepId: "s1" as StepId, + usage: { inputTokens: 100, outputTokens: 50 }, + decodeMs: 800, + genTotalMs: 1000, + }, + { + stepId: "s2" as StepId, + usage: { inputTokens: 200, outputTokens: 100 }, + genTotalMs: 2000, + }, + ], + }; + const view = viewTurnMetrics(turn); + // step1 uses decodeMs=800, step2 falls back to genTotalMs=2000 → total=2800ms + // 150 / (2800/1000) = 53.57 → rounds to 54 + expect(view.tps).toBe("54 tok/s"); + }); +}); + +describe("computeCachePct", () => { + it("is cacheReadTokens / inputTokens as a rounded percentage", () => { + expect(computeCachePct({ inputTokens: 2737, outputTokens: 10, cacheReadTokens: 2560 })).toBe( + 94, + ); + expect(computeCachePct({ inputTokens: 2669, outputTokens: 10, cacheReadTokens: 384 })).toBe(14); + }); + + it("is 0 when cacheReadTokens absent (legitimate miss, not missing data)", () => { + expect(computeCachePct({ inputTokens: 1000, outputTokens: 50 })).toBe(0); + }); + + it("is 0 when there are no input tokens (guard divide-by-zero)", () => { + expect(computeCachePct({ inputTokens: 0, outputTokens: 0, cacheReadTokens: 5 })).toBe(0); + }); + + it("clamps to 100 if read somehow exceeds input", () => { + expect(computeCachePct({ inputTokens: 100, outputTokens: 0, cacheReadTokens: 250 })).toBe(100); + }); +}); + +describe("viewCacheRate", () => { + it("success level for a high hit rate (>= 66)", () => { + const v = viewCacheRate({ inputTokens: 100, outputTokens: 0, cacheReadTokens: 93 }); + expect(v.pct).toBe(93); + expect(v.level).toBe("success"); + expect(v.isHit).toBe(true); + }); + + it("warning level for a mid hit rate (33..65)", () => { + const v = viewCacheRate({ inputTokens: 100, outputTokens: 0, cacheReadTokens: 54 }); + expect(v.pct).toBe(54); + expect(v.level).toBe("warning"); + }); + + it("error level for a low hit rate (< 33), including a legitimate 0%", () => { + expect(viewCacheRate({ inputTokens: 100, outputTokens: 0, cacheReadTokens: 14 }).level).toBe( + "error", + ); + const miss = viewCacheRate({ inputTokens: 1000, outputTokens: 50 }); + expect(miss.pct).toBe(0); + expect(miss.level).toBe("error"); + expect(miss.isHit).toBe(false); + }); +}); + +describe("computeExpectedCachePct", () => { + it("null when there is no prior turn (first turn has no baseline)", () => { + expect(computeExpectedCachePct({ inputTokens: 100, outputTokens: 0 }, null)).toBeNull(); + }); + + it("null when the prior turn cached nothing (denominator 0)", () => { + const prev = { inputTokens: 100, outputTokens: 0 }; + const current = { inputTokens: 200, outputTokens: 0, cacheReadTokens: 50 }; + expect(computeExpectedCachePct(current, prev)).toBeNull(); + }); + + it("100% when the whole prior cached prefix was read back (backend worked example)", () => { + // turn 1: cacheRead 0, cacheWrite 5146 → prefix 5146; turn 2 reads 5146 back. + const prev = { inputTokens: 5149, outputTokens: 0, cacheReadTokens: 0, cacheWriteTokens: 5146 }; + const current = { + inputTokens: 8462, + outputTokens: 0, + cacheReadTokens: 5146, + cacheWriteTokens: 3313, + }; + expect(computeExpectedCachePct(current, prev)).toBe(100); + }); + + it("drops below 100% when the cache busted (read < prior prefix)", () => { + const prev = { + inputTokens: 1000, + outputTokens: 0, + cacheReadTokens: 100, + cacheWriteTokens: 900, + }; + const current = { inputTokens: 1000, outputTokens: 0, cacheReadTokens: 500 }; + // 500 / (100 + 900) = 50% + expect(computeExpectedCachePct(current, prev)).toBe(50); + }); + + it("clamps to 100 if read somehow exceeds the prior prefix", () => { + const prev = { inputTokens: 100, outputTokens: 0, cacheWriteTokens: 100 }; + const current = { inputTokens: 100, outputTokens: 0, cacheReadTokens: 250 }; + expect(computeExpectedCachePct(current, prev)).toBe(100); + }); +}); + +describe("viewExpectedCache", () => { + it("null view when it cannot be derived (no prior turn)", () => { + expect(viewExpectedCache({ inputTokens: 100, outputTokens: 0 }, null)).toBeNull(); + }); + + it("success level + hit flag for full retention", () => { + const prev = { inputTokens: 5149, outputTokens: 0, cacheWriteTokens: 5146 }; + const current = { inputTokens: 8462, outputTokens: 0, cacheReadTokens: 5146 }; + const v = viewExpectedCache(current, prev); + expect(v?.pct).toBe(100); + expect(v?.level).toBe("success"); + expect(v?.isHit).toBe(true); + }); +}); + +describe("formatContextSize", () => { + it("formats a defined count with thousands separators", () => { + expect(formatContextSize(34102)).toBe("34,102 tokens in context"); + }); + + it("renders a placeholder for undefined (never 0)", () => { + expect(formatContextSize(undefined)).toBe("context size unknown"); + }); + + it("renders an explicit 0 as zero tokens (a real reported value)", () => { + expect(formatContextSize(0)).toBe("0 tokens in context"); + }); +}); + +describe("formatCompactTokens", () => { + it("renders sub-1k counts as-is", () => { + expect(formatCompactTokens(0)).toBe("0"); + expect(formatCompactTokens(812)).toBe("812"); + }); + + it("renders thousands with one decimal (rounded ≥100k)", () => { + expect(formatCompactTokens(12300)).toBe("12.3k"); + expect(formatCompactTokens(150000)).toBe("150k"); + }); + + it("renders millions with one decimal", () => { + expect(formatCompactTokens(1_200_000)).toBe("1.2M"); + expect(formatCompactTokens(1_000_000)).toBe("1.0M"); + }); +}); + +describe("computeContextUsage", () => { + it("computes an unrounded clamped percent against the limit", () => { + const u = computeContextUsage(34102, 1_000_000); + expect(u.current).toBe(34102); + expect(u.max).toBe(1_000_000); + expect(u.percent).toBeCloseTo(3.4102, 4); + }); + + it("treats unknown contextSize as current null (never 0)", () => { + const u = computeContextUsage(undefined, 1_000_000); + expect(u.current).toBeNull(); + expect(u.percent).toBeNull(); + }); + + it("an explicit 0 context size is a real reported value (current 0)", () => { + const u = computeContextUsage(0, 1_000_000); + expect(u.current).toBe(0); + expect(u.percent).toBe(0); + }); + + it("clamps percent to [0,100] and over-limit reads 100", () => { + expect(computeContextUsage(2_000_000, 1_000_000).percent).toBe(100); + }); + + it("max null (no/zero limit) ⇒ percent null", () => { + expect(computeContextUsage(5000, null).percent).toBeNull(); + expect(computeContextUsage(5000, 0).percent).toBeNull(); + expect(computeContextUsage(5000, null).max).toBeNull(); + }); +}); diff --git a/src/core/metrics/format.ts b/src/core/metrics/format.ts new file mode 100644 index 0000000..894bd54 --- /dev/null +++ b/src/core/metrics/format.ts @@ -0,0 +1,179 @@ +import type { StepMetrics, TurnMetrics, Usage } from "@dispatch/wire"; +import type { CacheRateView, StepMetricsView, TurnMetricsView } from "./types"; + +function formatTokens(n: number): string { + return n.toLocaleString("en-US"); +} + +function formatDuration(ms: number | undefined): string | null { + if (ms === undefined || ms <= 0) return null; + if (ms < 1000) return `${Math.round(ms)}ms`; + return `${(ms / 1000).toFixed(1)}s`; +} + +function formatTps(tps: number | null): string | null { + if (tps === null) return null; + if (tps < 10) return `${tps.toFixed(1)} tok/s`; + return `${Math.round(tps)} tok/s`; +} + +/** + * Format the current context size for display. A defined count renders as + * `"<n> tokens in context"` (thousands-separated); `undefined` ("unknown" — no + * per-step usage reported yet) renders the placeholder `"context size unknown"`. + * Never renders `0` for the unknown case. + */ +export function formatContextSize(n: number | undefined): string { + if (n === undefined) return "context size unknown"; + return `${formatTokens(n)} tokens in context`; +} + +/** + * Compact token count for a slim status bar: `812`, `12.3k`, `1.2M`. Full + * thousands-separated numbers live elsewhere; this trades precision for width. + */ +export function formatCompactTokens(n: number): string { + if (n < 1000) return `${n}`; + if (n < 1_000_000) { + const k = n / 1000; + return `${k >= 100 ? Math.round(k) : k.toFixed(1)}k`; + } + const m = n / 1_000_000; + return `${m >= 100 ? Math.round(m) : m.toFixed(1)}M`; +} + +/** + * Context-window occupancy: the current size against a max window limit. + * + * `current` is the latest turn's context size, or `null` when unknown (no + * per-step usage reported yet) — NEVER coerced to `0`, so a consumer cannot + * silently render "0 tokens / 1M"; it must branch on `current === null` and show + * a placeholder instead. `max` is the model's window limit (or `null` when + * unknown). `percent` is `current / max * 100` clamped to [0, 100], UNROUNDED + * (the UI picks the precision) — so a few-thousand-token context against a + * 1,000,000 window still reads non-zero. `percent` is `null` when `current` OR + * `max` is unknown (no bar/denominator). + */ +export interface ContextUsage { + readonly current: number | null; + readonly max: number | null; + readonly percent: number | null; +} + +export function computeContextUsage( + contextSize: number | undefined, + contextLimit: number | null | undefined, +): ContextUsage { + const current = contextSize ?? null; + const max = typeof contextLimit === "number" && contextLimit > 0 ? contextLimit : null; + const percent = + current === null || max === null ? null : Math.max(0, Math.min(100, (current / max) * 100)); + return { current, max, percent }; +} + +/** Compute tokens-per-second. Returns null when elapsed time is absent or zero. */ +export function computeTps(outputTokens: number, elapsedMs: number | undefined): number | null { + if (elapsedMs === undefined || elapsedMs <= 0) return null; + return outputTokens / (elapsedMs / 1000); +} + +function totalTokens(u: Usage): number { + return u.inputTokens + u.outputTokens; +} + +function formatBreakdown(u: Usage): string { + let s = `${formatTokens(u.inputTokens)} in / ${formatTokens(u.outputTokens)} out`; + if (u.cacheReadTokens !== undefined && u.cacheReadTokens > 0) { + s += ` / ${formatTokens(u.cacheReadTokens)} cache`; + } + return s; +} + +/** Build a formatted view of a single step's metrics. */ +export function viewStepMetrics(step: StepMetrics, index: number): StepMetricsView { + const total = totalTokens(step.usage); + const tps = computeTps(step.usage.outputTokens, step.decodeMs ?? step.genTotalMs); + return { + label: `step ${index + 1}`, + tokensLabel: `${formatTokens(total)} tok`, + tps: formatTps(tps), + ttft: formatDuration(step.ttftMs), + decode: formatDuration(step.decodeMs), + genTotal: formatDuration(step.genTotalMs), + }; +} + +/** + * Cache hit rate as a 0..100 integer percentage: `cacheReadTokens / inputTokens`, + * clamped to [0,1]. Absent cache field counts as 0; a 0% rate is legitimate (not + * missing data). Returns 0 when there are no input tokens. + */ +export function computeCachePct(u: Usage): number { + const read = u.cacheReadTokens ?? 0; + if (u.inputTokens <= 0) return 0; + const rate = read / u.inputTokens; + const clamped = rate < 0 ? 0 : rate > 1 ? 1 : rate; + return Math.round(clamped * 100); +} + +/** Colour severity for a cache hit percentage (badge colour). */ +function cacheLevel(pct: number): "success" | "warning" | "error" { + if (pct >= 66) return "success"; + if (pct >= 33) return "warning"; + return "error"; +} + +/** Build a view of a cache hit rate (percentage + colour level + hit flag). */ +export function viewCacheRate(u: Usage): CacheRateView { + const pct = computeCachePct(u); + return { pct, level: cacheLevel(pct), isHit: (u.cacheReadTokens ?? 0) > 0 }; +} + +/** + * Expected cache (retention): of the cache that existed going INTO this turn, how + * much was read back — `clamp01(cacheRead_N / (cacheRead_{N-1} + cacheWrite_{N-1}))`. + * The denominator is the PRIOR turn's cached prefix (what it read + what it wrote). + * Ideally ~100% on every turn after the first; <100% = the cache busted/expired. + * + * Returns `null` when it cannot be derived: no prior turn (`prev === null`) or the + * prior turn cached nothing (denominator <= 0) — distinct from a real 0%. + */ +export function computeExpectedCachePct(current: Usage, prev: Usage | null): number | null { + if (prev === null) return null; + const denom = (prev.cacheReadTokens ?? 0) + (prev.cacheWriteTokens ?? 0); + if (denom <= 0) return null; + const read = current.cacheReadTokens ?? 0; + const rate = read / denom; + const clamped = rate < 0 ? 0 : rate > 1 ? 1 : rate; + return Math.round(clamped * 100); +} + +/** + * Build a view of the cross-turn retention (percentage + colour level + hit flag), + * or `null` when it can't be derived (see `computeExpectedCachePct`). + */ +export function viewExpectedCache(current: Usage, prev: Usage | null): CacheRateView | null { + const pct = computeExpectedCachePct(current, prev); + if (pct === null) return null; + return { pct, level: cacheLevel(pct), isHit: (current.cacheReadTokens ?? 0) > 0 }; +} + +/** Build a formatted view of a turn's aggregate metrics. */ +export function viewTurnMetrics(turn: TurnMetrics, turnNumber?: number): TurnMetricsView { + const total = totalTokens(turn.usage); + let totalGenMs: number | undefined; + for (const step of turn.steps) { + const stepMs = step.decodeMs ?? step.genTotalMs; + if (stepMs !== undefined) { + totalGenMs = (totalGenMs ?? 0) + stepMs; + } + } + const tps = computeTps(turn.usage.outputTokens, totalGenMs); + return { + label: turnNumber !== undefined ? `turn ${turnNumber}` : "turn", + tokensLabel: `${formatTokens(total)} tok`, + breakdown: formatBreakdown(turn.usage), + tps: formatTps(tps), + duration: formatDuration(turn.durationMs), + }; +} diff --git a/src/core/metrics/index.ts b/src/core/metrics/index.ts new file mode 100644 index 0000000..d3c9669 --- /dev/null +++ b/src/core/metrics/index.ts @@ -0,0 +1,31 @@ +export { + type ContextUsage, + computeCachePct, + computeContextUsage, + computeExpectedCachePct, + computeTps, + formatCompactTokens, + formatContextSize, + viewCacheRate, + viewExpectedCache, + viewStepMetrics, + viewTurnMetrics, +} from "./format"; +export { interleaveTurnMetrics } from "./place"; +export { + applyDurableMetrics, + foldMetricsEvent, + initialMetricsState, + selectCurrentContextSize, + selectOrderedTurnMetrics, +} from "./reducer"; +export type { + CacheRateView, + MetricsRow, + MetricsState, + StepMetrics, + StepMetricsView, + TurnMetrics, + TurnMetricsEntry, + TurnMetricsView, +} from "./types"; diff --git a/src/core/metrics/place.test.ts b/src/core/metrics/place.test.ts new file mode 100644 index 0000000..9c925a3 --- /dev/null +++ b/src/core/metrics/place.test.ts @@ -0,0 +1,621 @@ +import type { StepId, StepMetrics, TurnMetrics } from "@dispatch/wire"; +import { describe, expect, it } from "vitest"; +import type { RenderGroup } from "../chunks"; +import { interleaveTurnMetrics } from "./place"; +import type { MetricsRow, TurnMetricsEntry } from "./types"; + +function userGroup(seq: number, text: string): RenderGroup { + return { + kind: "single", + chunk: { + seq, + role: "user", + chunk: { type: "text", text }, + provisional: false, + }, + }; +} + +function assistantGroup(seq: number, text: string): RenderGroup { + return { + kind: "single", + chunk: { + seq, + role: "assistant", + chunk: { type: "text", text }, + provisional: false, + }, + }; +} + +function toolCallGroup(seq: number, stepId: string, toolCallId: string): RenderGroup { + return { + kind: "single", + chunk: { + seq, + role: "assistant", + chunk: { + type: "tool-call", + toolCallId, + toolName: "test", + input: {}, + stepId: stepId as StepId, + }, + provisional: false, + }, + }; +} + +function toolResultGroup(seq: number, stepId: string, toolCallId: string): RenderGroup { + return { + kind: "single", + chunk: { + seq, + role: "tool", + chunk: { + type: "tool-result", + toolCallId, + toolName: "test", + content: "", + isError: false, + stepId: stepId as StepId, + }, + provisional: false, + }, + }; +} + +function toolBatchGroup(stepId: string, toolCallIds: string[]): RenderGroup { + return { + kind: "tool-batch", + stepId, + entries: toolCallIds.map((id) => ({ + call: { + type: "tool-call" as const, + toolCallId: id, + toolName: "test", + input: {}, + stepId: stepId as StepId, + }, + result: null, + })), + provisional: false, + }; +} + +function makeStep(stepId: string, inputTokens: number, outputTokens: number): StepMetrics { + return { + stepId: stepId as StepId, + usage: { inputTokens, outputTokens }, + }; +} + +function makeTurn( + turnId: string, + inputTokens: number, + outputTokens: number, + steps: StepMetrics[] = [], +): TurnMetrics { + return { + turnId, + usage: { inputTokens, outputTokens }, + steps, + }; +} + +function makeEntry( + turnId: string, + inputTokens: number, + outputTokens: number, + steps: StepMetrics[] = [], +): TurnMetricsEntry { + return { + turnId, + steps, + total: makeTurn(turnId, inputTokens, outputTokens, steps), + }; +} + +function makeProgressiveEntry(turnId: string, steps: StepMetrics[]): TurnMetricsEntry { + return { + turnId, + steps, + total: null, + }; +} + +function expectGroupAt( + rows: readonly { readonly kind: string }[], + index: number, + expected: RenderGroup, +): void { + const row = rows[index]; + expect(row?.kind).toBe("group"); + expect((row as { readonly group: RenderGroup } | undefined)?.group).toBe(expected); +} + +function expectStepMetricsAt( + rows: readonly { readonly kind: string }[], + index: number, + expectedStepId: string, + expectedIndex: number, +): void { + const row = rows[index]; + expect(row?.kind).toBe("step-metrics"); + const sm = row as { readonly step: StepMetrics; readonly index: number } | undefined; + expect(sm?.step.stepId).toBe(expectedStepId); + expect(sm?.index).toBe(expectedIndex); +} + +function expectTurnMetricsAt( + rows: readonly { readonly kind: string }[], + index: number, + expectedTurnId: string, +): void { + const row = rows[index]; + expect(row?.kind).toBe("turn-metrics"); + expect((row as { readonly turn: TurnMetrics } | undefined)?.turn.turnId).toBe(expectedTurnId); +} + +describe("interleaveTurnMetrics", () => { + it("no metrics: rows are all groups, unchanged order", () => { + const g1 = userGroup(1, "q"); + const g2 = assistantGroup(2, "a"); + const rows = interleaveTurnMetrics([g1, g2], []); + expect(rows).toHaveLength(2); + expectGroupAt(rows, 0, g1); + expectGroupAt(rows, 1, g2); + }); + + it("head-aligned: segment i gets entries[i]", () => { + const g1 = userGroup(1, "q1"); + const g2 = toolCallGroup(2, "s1", "c1"); + const g3 = userGroup(3, "q2"); + const g4 = toolCallGroup(4, "s2", "c2"); + const step1 = makeStep("s1", 100, 50); + const step2 = makeStep("s2", 200, 80); + const rows = interleaveTurnMetrics( + [g1, g2, g3, g4], + [makeEntry("t1", 100, 50, [step1]), makeEntry("t2", 200, 80, [step2])], + ); + + expect(rows).toHaveLength(8); + expectGroupAt(rows, 0, g1); + expectGroupAt(rows, 1, g2); + expectStepMetricsAt(rows, 2, "s1", 0); + expectTurnMetricsAt(rows, 3, "t1"); + expectGroupAt(rows, 4, g3); + expectGroupAt(rows, 5, g4); + expectStepMetricsAt(rows, 6, "s2", 0); + expectTurnMetricsAt(rows, 7, "t2"); + }); + + it("a trailing segment with no entry (in-flight turn) renders no metrics", () => { + const g1 = userGroup(1, "q1"); + const g2 = toolCallGroup(2, "s1", "c1"); + const g3 = userGroup(3, "q2"); + const g4 = assistantGroup(4, "a2"); + const step = makeStep("s1", 100, 50); + const rows = interleaveTurnMetrics([g1, g2, g3, g4], [makeEntry("t1", 100, 50, [step])]); + + expect(rows).toHaveLength(6); + expectGroupAt(rows, 0, g1); + expectGroupAt(rows, 1, g2); + expectStepMetricsAt(rows, 2, "s1", 0); + expectTurnMetricsAt(rows, 3, "t1"); + expectGroupAt(rows, 4, g3); + expectGroupAt(rows, 5, g4); + }); + + it("single text-only turn: no step row (unanchored), turn-metrics at tail", () => { + const g1 = userGroup(1, "q1"); + const g2 = assistantGroup(2, "a1"); + const step = makeStep("s1", 100, 50); + const turn = makeEntry("t1", 100, 50, [step]); + const rows = interleaveTurnMetrics([g1, g2], [turn]); + + expect(rows).toHaveLength(3); + expectGroupAt(rows, 0, g1); + expectGroupAt(rows, 1, g2); + expectTurnMetricsAt(rows, 2, "t1"); + }); + + it("tool step anchors inline after its tool-batch group", () => { + const g1 = userGroup(1, "q1"); + const g2 = toolBatchGroup("t#0", ["c1", "c2"]); + const g3 = assistantGroup(3, "a1"); + const step0 = makeStep("t#0", 100, 50); + const step1 = makeStep("t#1", 200, 80); + const turn = makeEntry("t1", 300, 130, [step0, step1]); + const rows = interleaveTurnMetrics([g1, g2, g3], [turn]); + + expect(rows).toHaveLength(5); + expectGroupAt(rows, 0, g1); + expectGroupAt(rows, 1, g2); + expectStepMetricsAt(rows, 2, "t#0", 0); + expectGroupAt(rows, 3, g3); + expectTurnMetricsAt(rows, 4, "t1"); + }); + + it("single tool-call group anchors its step", () => { + const g1 = userGroup(1, "q1"); + const g2 = toolCallGroup(2, "s1", "c1"); + const g3 = assistantGroup(3, "a1"); + const step = makeStep("s1", 100, 50); + const turn = makeEntry("t1", 100, 50, [step]); + const rows = interleaveTurnMetrics([g1, g2, g3], [turn]); + + expect(rows).toHaveLength(5); + expectGroupAt(rows, 0, g1); + expectGroupAt(rows, 1, g2); + expectStepMetricsAt(rows, 2, "s1", 0); + expectGroupAt(rows, 3, g3); + expectTurnMetricsAt(rows, 4, "t1"); + }); + + it("single tool-result group anchors its step", () => { + const g1 = userGroup(1, "q1"); + const g2 = toolResultGroup(2, "s1", "c1"); + const g3 = assistantGroup(3, "a1"); + const step = makeStep("s1", 100, 50); + const turn = makeEntry("t1", 100, 50, [step]); + const rows = interleaveTurnMetrics([g1, g2, g3], [turn]); + + expect(rows).toHaveLength(5); + expectGroupAt(rows, 0, g1); + expectGroupAt(rows, 1, g2); + expectStepMetricsAt(rows, 2, "s1", 0); + expectGroupAt(rows, 3, g3); + expectTurnMetricsAt(rows, 4, "t1"); + }); + + it("multi-step: each tool step inline, unanchored text step skipped", () => { + const g1 = userGroup(1, "q1"); + const g2 = toolBatchGroup("t#0", ["c1"]); + const g3 = assistantGroup(2, "thinking"); + const g4 = toolBatchGroup("t#1", ["c2", "c3"]); + const g5 = assistantGroup(3, "a1"); + const step0 = makeStep("t#0", 100, 50); + const step1 = makeStep("t#1", 200, 80); + const step2 = makeStep("t#2", 50, 20); + const turn = makeEntry("t1", 350, 150, [step0, step1, step2]); + const rows = interleaveTurnMetrics([g1, g2, g3, g4, g5], [turn]); + + expect(rows).toHaveLength(8); + expectGroupAt(rows, 0, g1); + expectGroupAt(rows, 1, g2); + expectStepMetricsAt(rows, 2, "t#0", 0); + expectGroupAt(rows, 3, g3); + expectGroupAt(rows, 4, g4); + expectStepMetricsAt(rows, 5, "t#1", 1); + expectGroupAt(rows, 6, g5); + expectTurnMetricsAt(rows, 7, "t1"); + }); + + it("multiple turns head-aligned with inline steps", () => { + const g1 = userGroup(1, "q1"); + const g2 = toolBatchGroup("s1", ["c1"]); + const g3 = assistantGroup(2, "a1"); + const g4 = userGroup(3, "q2"); + const g5 = toolCallGroup(4, "s2", "c2"); + const step1 = makeStep("s1", 100, 50); + const step2 = makeStep("s2", 200, 80); + const rows = interleaveTurnMetrics( + [g1, g2, g3, g4, g5], + [makeEntry("t1", 100, 50, [step1]), makeEntry("t2", 200, 80, [step2])], + ); + + expect(rows).toHaveLength(9); + expectGroupAt(rows, 0, g1); + expectGroupAt(rows, 1, g2); + expectStepMetricsAt(rows, 2, "s1", 0); + expectGroupAt(rows, 3, g3); + expectTurnMetricsAt(rows, 4, "t1"); + expectGroupAt(rows, 5, g4); + expectGroupAt(rows, 6, g5); + expectStepMetricsAt(rows, 7, "s2", 0); + expectTurnMetricsAt(rows, 8, "t2"); + }); + + it("unanchored step (stepId not in groups) is skipped — only turn-metrics", () => { + const g1 = userGroup(1, "q1"); + const g2 = assistantGroup(2, "a1"); + const step0 = makeStep("orphan", 100, 50); + const turn = makeEntry("t1", 100, 50, [step0]); + const rows = interleaveTurnMetrics([g1, g2], [turn]); + + expect(rows).toHaveLength(3); + expectGroupAt(rows, 0, g1); + expectGroupAt(rows, 1, g2); + expectTurnMetricsAt(rows, 2, "t1"); + }); + + it("fewer metrics than segments: trailing segments are bare", () => { + const g1 = userGroup(1, "q1"); + const g2 = toolCallGroup(2, "s1", "c1"); + const g3 = userGroup(3, "q2"); + const g4 = assistantGroup(4, "a2"); + const g5 = userGroup(5, "q3"); + const g6 = assistantGroup(6, "a3"); + const step = makeStep("s1", 300, 120); + const rows = interleaveTurnMetrics( + [g1, g2, g3, g4, g5, g6], + [makeEntry("t1", 300, 120, [step])], + ); + + expect(rows).toHaveLength(8); + expectGroupAt(rows, 0, g1); + expectGroupAt(rows, 1, g2); + expectStepMetricsAt(rows, 2, "s1", 0); + expectTurnMetricsAt(rows, 3, "t1"); + expectGroupAt(rows, 4, g3); + expectGroupAt(rows, 5, g4); + expectGroupAt(rows, 6, g5); + expectGroupAt(rows, 7, g6); + }); + + it("trimmed leading turns: a mixed tool+text transcript tail-aligns text-only turns to their OWN (newest) entries, not stale trimmed ones", () => { + // A long conversation where the chat limit unloaded the oldest turn (t1). + // Metrics still hold all three turns; the loaded transcript is turns 2-3. + // Turn 2 is a tool turn (matched by stepId); turn 3 is text-only (no + // stepId groups) — the failure case. The text-only turn MUST get its OWN + // entry (t3), NOT the trimmed t1's stale metrics. + const g3 = userGroup(3, "q2"); + const g4 = toolBatchGroup("s2", ["c2"]); + const g5 = assistantGroup(4, "tool-reply"); + const g6 = userGroup(5, "q3"); + const g7 = assistantGroup(6, "text-reply"); + const step1 = makeStep("s1", 11, 1); // t1 (trimmed) + const step2 = makeStep("s2", 22, 2); // t2 (loaded, tool) + const step3 = makeStep("s3", 33, 3); // t3 (loaded, text-only — unanchored) + const entries = [ + makeEntry("t1", 11, 1, [step1]), + makeEntry("t2", 22, 2, [step2]), + makeEntry("t3", 33, 3, [step3]), + ]; + const rows = interleaveTurnMetrics([g3, g4, g5, g6, g7], entries); + + const tmRows = rows.filter( + (r): r is Extract<MetricsRow, { kind: "turn-metrics" }> => r.kind === "turn-metrics", + ); + // Two loaded turns → two turn-metrics rows. The trimmed t1 does NOT render. + expect(tmRows).toHaveLength(2); + // CRITICAL: the text-only turn (segment 1) got t3 (its own newest entry), + // not t1 (the stale trimmed one). A misaligned head-align would show t1. + expect(tmRows[1]?.turn.turnId).toBe("t3"); + expect(tmRows[0]?.turn.turnId).toBe("t2"); + // And t1 never appears as a rendered row. + expect(tmRows.some((r) => r.turn.turnId === "t1")).toBe(false); + }); + + it("trimmed turn still counts toward the cumulative 'chat total' on the first visible turn", () => { + // t1 is trimmed (no segment) but finalized; t2 is the loaded visible turn. + // t2's "Chat Total" cumulative must INCLUDE t1's usage (the whole chat), + // even though t1 renders no row of its own. + const g1 = userGroup(2, "q2"); + const g2 = assistantGroup(3, "a2"); + const entries = [ + { + turnId: "t1", + steps: [], + total: { + turnId: "t1", + usage: { inputTokens: 1000, outputTokens: 10, cacheReadTokens: 500 }, + steps: [], + }, + }, + { + turnId: "t2", + steps: [], + total: { + turnId: "t2", + usage: { inputTokens: 2000, outputTokens: 20, cacheReadTokens: 1600 }, + steps: [], + }, + }, + ]; + const rows = interleaveTurnMetrics([g1, g2], entries); + const tmRows = rows.filter( + (r): r is Extract<MetricsRow, { kind: "turn-metrics" }> => r.kind === "turn-metrics", + ); + // Only the loaded turn renders a row; the trimmed t1 does not. + expect(tmRows).toHaveLength(1); + expect(tmRows[0]?.turn.turnId).toBe("t2"); + // Cumulative includes BOTH turns (t1 + t2): input 3000, cacheRead 2100. + expect(tmRows[0]?.cumulativeUsage.inputTokens).toBe(3000); + expect(tmRows[0]?.cumulativeUsage.cacheReadTokens).toBe(2100); + // Retention baseline is the prior finalized turn (t1, even though trimmed). + expect(tmRows[0]?.prevTurnUsage?.inputTokens).toBe(1000); + expect(tmRows[0]?.prevTurnUsage?.cacheReadTokens).toBe(500); + }); + + it("in-flight turn (no durationMs) still produces turn row", () => { + const g1 = userGroup(1, "q1"); + const g2 = toolCallGroup(2, "s1", "c1"); + const step = makeStep("s1", 100, 50); + const turn: TurnMetricsEntry = { + turnId: "t1", + steps: [step], + total: { + turnId: "t1", + usage: { inputTokens: 100, outputTokens: 50 }, + steps: [step], + }, + }; + const rows = interleaveTurnMetrics([g1, g2], [turn]); + + expect(rows).toHaveLength(4); + expectStepMetricsAt(rows, 2, "s1", 0); + expectTurnMetricsAt(rows, 3, "t1"); + const metricsRow = rows[3] as { readonly turn: TurnMetrics } | undefined; + expect(metricsRow?.turn.durationMs).toBeUndefined(); + }); + + it("leading non-turn groups emit as plain group rows", () => { + const g0 = assistantGroup(1, "system msg"); + const g1 = userGroup(2, "q1"); + const g2 = toolCallGroup(3, "s1", "c1"); + const step = makeStep("s1", 100, 50); + const rows = interleaveTurnMetrics([g0, g1, g2], [makeEntry("t1", 100, 50, [step])]); + + expect(rows).toHaveLength(5); + expectGroupAt(rows, 0, g0); + expect(rows[1]?.kind).toBe("group"); + expect(rows[2]?.kind).toBe("group"); + expectStepMetricsAt(rows, 3, "s1", 0); + expectTurnMetricsAt(rows, 4, "t1"); + }); + + it("trimmed turn (more metrics than segments) does NOT emit a standalone row at the top", () => { + const g1 = userGroup(1, "q1"); + const g2 = toolCallGroup(2, "s1", "c1"); + const step1 = makeStep("s1", 100, 50); + const step2 = makeStep("s2", 200, 80); + const rows = interleaveTurnMetrics( + [g1, g2], + [makeEntry("t1", 100, 50, [step1]), makeEntry("t2", 200, 80, [step2])], + ); + + // t2's content was unloaded by the chat limit (no segment for it); its + // metrics must NOT render a standalone row piled at the top. Only the + // loaded turn's content + its matched metrics appear. (t2 still counts + // toward the cumulative "chat total" — see the cache-total tests.) + expect(rows).toHaveLength(4); + expectGroupAt(rows, 0, g1); + expectGroupAt(rows, 1, g2); + expectStepMetricsAt(rows, 2, "s1", 0); + expectTurnMetricsAt(rows, 3, "t1"); + // No standalone turn-metrics row for t2 anywhere. + const tmRows = rows.filter((r) => r.kind === "turn-metrics"); + expect(tmRows).toHaveLength(1); + expect((tmRows[0] as { readonly turn: TurnMetrics }).turn.turnId).toBe("t1"); + }); + + it("turn with no steps emits only turn-metrics (no step-metrics)", () => { + const g1 = userGroup(1, "q1"); + const g2 = assistantGroup(2, "a1"); + const rows = interleaveTurnMetrics([g1, g2], [makeEntry("t1", 100, 50)]); + + expect(rows).toHaveLength(3); + expectGroupAt(rows, 0, g1); + expectGroupAt(rows, 1, g2); + expectTurnMetricsAt(rows, 2, "t1"); + }); + + it("progressive: entry with steps but total=null emits step rows and NO turn-metrics row", () => { + const g1 = userGroup(1, "q1"); + const g2 = toolBatchGroup("s1", ["c1"]); + const g3 = assistantGroup(2, "a1"); + const step1 = makeStep("s1", 100, 50); + const entry = makeProgressiveEntry("t1", [step1]); + const rows = interleaveTurnMetrics([g1, g2, g3], [entry]); + + expect(rows).toHaveLength(4); + expectGroupAt(rows, 0, g1); + expectGroupAt(rows, 1, g2); + expectStepMetricsAt(rows, 2, "s1", 0); + expectGroupAt(rows, 3, g3); + }); + + it("entry with total emits step rows + a turn-metrics row", () => { + const g1 = userGroup(1, "q1"); + const g2 = toolBatchGroup("s1", ["c1"]); + const g3 = assistantGroup(2, "a1"); + const step1 = makeStep("s1", 100, 50); + const entry = makeEntry("t1", 100, 50, [step1]); + const rows = interleaveTurnMetrics([g1, g2, g3], [entry]); + + expect(rows).toHaveLength(5); + expectGroupAt(rows, 0, g1); + expectGroupAt(rows, 1, g2); + expectStepMetricsAt(rows, 2, "s1", 0); + expectGroupAt(rows, 3, g3); + expectTurnMetricsAt(rows, 4, "t1"); + }); + + it("progressive multi-step: unanchored steps skipped, no turn-metrics", () => { + const g1 = userGroup(1, "q1"); + const g2 = assistantGroup(2, "a1"); + const step0 = makeStep("s1", 100, 50); + const step1 = makeStep("s2", 200, 80); + const entry = makeProgressiveEntry("t1", [step0, step1]); + const rows = interleaveTurnMetrics([g1, g2], [entry]); + + expect(rows).toHaveLength(2); + expectGroupAt(rows, 0, g1); + expectGroupAt(rows, 1, g2); + }); +}); + +describe("interleaveTurnMetrics — cumulative usage (cache total)", () => { + function turnMetricsRows(rows: readonly MetricsRow[]) { + return rows.filter((r): r is Extract<MetricsRow, { kind: "turn-metrics" }> => { + return r.kind === "turn-metrics"; + }); + } + + function cacheEntry( + turnId: string, + inputTokens: number, + outputTokens: number, + cacheReadTokens: number, + ): TurnMetricsEntry { + const total: TurnMetrics = { + turnId, + usage: { inputTokens, outputTokens, cacheReadTokens }, + steps: [], + }; + return { turnId, steps: [], total }; + } + + it("turn-metrics row carries this turn's usage and the running cumulative", () => { + const rows = interleaveTurnMetrics( + [userGroup(1, "q1"), assistantGroup(2, "a1")], + [makeEntry("t1", 1000, 100)], + ); + const tm = turnMetricsRows(rows); + expect(tm).toHaveLength(1); + expect(tm[0]?.turn.turnId).toBe("t1"); + expect(tm[0]?.cumulativeUsage).toEqual({ inputTokens: 1000, outputTokens: 100 }); + }); + + it("accumulates cache read + input across turns (chat total)", () => { + const rows = interleaveTurnMetrics( + [userGroup(1, "q1"), assistantGroup(2, "a1"), userGroup(3, "q2"), assistantGroup(4, "a2")], + [cacheEntry("t1", 2669, 10, 384), cacheEntry("t2", 2737, 10, 2560)], + ); + const tm = turnMetricsRows(rows); + expect(tm).toHaveLength(2); + // turn 1: only its own usage + expect(tm[0]?.cumulativeUsage.inputTokens).toBe(2669); + expect(tm[0]?.cumulativeUsage.cacheReadTokens).toBe(384); + // turn 2: sum of both (input 5406, cacheRead 2944 → matches the backend's 54% example) + expect(tm[1]?.cumulativeUsage.inputTokens).toBe(5406); + expect(tm[1]?.cumulativeUsage.cacheReadTokens).toBe(2944); + }); + + it("an in-flight (total=null) turn does not contribute to the cumulative", () => { + const rows = interleaveTurnMetrics( + [userGroup(1, "q1"), assistantGroup(2, "a1"), userGroup(3, "q2"), assistantGroup(4, "a2")], + [cacheEntry("t1", 1000, 10, 500), makeProgressiveEntry("t2", [makeStep("s1", 200, 5)])], + ); + const tm = turnMetricsRows(rows); + // only the finalized turn emits a turn-metrics row; its cumulative is just itself + expect(tm).toHaveLength(1); + expect(tm[0]?.cumulativeUsage.inputTokens).toBe(1000); + expect(tm[0]?.cumulativeUsage.cacheReadTokens).toBe(500); + }); + + it("carries the prior finalized turn's usage as the retention baseline", () => { + const rows = interleaveTurnMetrics( + [userGroup(1, "q1"), assistantGroup(2, "a1"), userGroup(3, "q2"), assistantGroup(4, "a2")], + [cacheEntry("t1", 2669, 10, 384), cacheEntry("t2", 2737, 10, 2560)], + ); + const tm = turnMetricsRows(rows); + // first finalized turn has no earlier baseline + expect(tm[0]?.prevTurnUsage).toBeNull(); + // second turn's baseline is the first turn's usage + expect(tm[1]?.prevTurnUsage?.inputTokens).toBe(2669); + expect(tm[1]?.prevTurnUsage?.cacheReadTokens).toBe(384); + }); +}); diff --git a/src/core/metrics/place.ts b/src/core/metrics/place.ts new file mode 100644 index 0000000..7122b09 --- /dev/null +++ b/src/core/metrics/place.ts @@ -0,0 +1,298 @@ +import type { Usage } from "@dispatch/wire"; +import type { RenderGroup } from "../chunks"; +import type { MetricsRow, TurnMetricsEntry } from "./types"; + +function groupStepId(g: RenderGroup): string | undefined { + if (g.kind === "tool-batch") return g.stepId; + const c = g.chunk.chunk; + return c.type === "tool-call" || c.type === "tool-result" ? c.stepId : undefined; +} + +/** Element-wise sum of two token usages (cache fields included only when nonzero). */ +function addUsage(a: Usage, b: Usage): Usage { + const out: Usage = { + inputTokens: a.inputTokens + b.inputTokens, + outputTokens: a.outputTokens + b.outputTokens, + }; + const read = (a.cacheReadTokens ?? 0) + (b.cacheReadTokens ?? 0); + const write = (a.cacheWriteTokens ?? 0) + (b.cacheWriteTokens ?? 0); + if (read > 0) (out as { cacheReadTokens?: number }).cacheReadTokens = read; + if (write > 0) (out as { cacheWriteTokens?: number }).cacheWriteTokens = write; + return out; +} + +/** + * Interleave turn metrics into the rendered transcript. + * + * Splits groups into per-turn segments: a new segment begins at each `single` + * group with `group.chunk.role === "user"`. Segments are matched to entries + * by `stepId` presence when possible (robust against chat-limit trimming: when + * a turn's user message is trimmed, positional alignment would be off, but + * stepId matching still finds the right entry). Segments with no stepId-bearing + * groups (text-only turns) fall back to POSITIONAL tail-alignment: since the + * loaded transcript is always a SUFFIX of the full turn history (the chat limit + * keeps the newest and unloads the oldest), segment `seg` ↔ entry `K - T + seg`. + * + * Within a segment that has a matched entry, each completed step's metrics + * are placed INLINE right after the last group bearing that step's `stepId`. + * Steps whose `stepId` does not appear in any group ("unanchored"): + * - If the segment HAS stepId-bearing groups (tool chunks exist but this step's + * were trimmed): SKIPPED (no blank "step N · 0 tok" bubbles). + * - If the segment has NO stepId-bearing groups (text-only turn): placed at the + * segment tail before the turn-metrics row (the original behavior). + * + * A `turn-metrics` row is emitted ONLY when `entry.total !== null` (i.e. the turn + * is finalized via `done` or durable data). A still-generating turn emits no + * turn-total row. + * + * Fully trimmed turns (entries whose content was unloaded by the chat limit and + * which match no segment) are NOT rendered as standalone rows — that previously + * piled a wall of stale cache badges at the top of a long, trimmed transcript. + * Their usage still counts toward the per-turn "chat total" cumulative (computed + * across ALL finalized turns in entry-array order), so the running cache rate + * stays correct regardless of which turns were trimmed; paging earlier history + * back in ("Show earlier messages") re-matches them and re-renders their rows. + */ +export function interleaveTurnMetrics( + groups: readonly RenderGroup[], + entries: readonly TurnMetricsEntry[], +): readonly MetricsRow[] { + if (entries.length === 0) { + return groups.map((g) => ({ kind: "group" as const, group: g })); + } + + const segmentStarts: number[] = []; + for (let i = 0; i < groups.length; i++) { + const g = groups[i]; + if (g !== undefined && g.kind === "single" && g.chunk.role === "user") { + segmentStarts.push(i); + } + } + + let T = segmentStarts.length; + + // No user messages — e.g. a compacted conversation whose history starts + // with a system summary. Treat the entire transcript as one segment so + // turn/step metrics can still be placed. + if (T === 0 && entries.length > 0) { + segmentStarts.push(0); + T = 1; + } + + if (T === 0) { + return groups.map((g) => ({ kind: "group" as const, group: g })); + } + + const K = entries.length; + + // Build stepId → entry-index lookup for matching. + const entryStepIds: Set<string>[] = entries.map((e) => new Set(e.steps.map((s) => s.stepId))); + + // Match segments to entries. Pass 1: match by stepId overlap (handles + // trimming where positional alignment alone could be ambiguous). Pass 2: + // positional tail-alignment fallback for unmatched segments (text-only turns + // with no stepId-bearing groups). + const usedEntries = new Set<number>(); + const segmentEntry = new Map<number, TurnMetricsEntry>(); + const segmentEntryIndex = new Map<number, number>(); + + // Pass 1: stepId matching. + for (let seg = 0; seg < T; seg++) { + const start = segmentStarts[seg] ?? 0; + const end = seg + 1 < T ? (segmentStarts[seg + 1] ?? groups.length) : groups.length; + + const segStepIds = new Set<string>(); + for (let i = start; i < end; i++) { + const g = groups[i]; + if (g === undefined) continue; + const sid = groupStepId(g); + if (sid !== undefined) segStepIds.add(sid); + } + if (segStepIds.size === 0) continue; // text-only — defer to pass 2 + + let bestEntry = -1; + let bestMatch = 0; + for (let i = 0; i < K; i++) { + if (usedEntries.has(i)) continue; + let match = 0; + for (const sid of segStepIds) { + if (entryStepIds[i]?.has(sid)) match++; + } + if (match > bestMatch) { + bestMatch = match; + bestEntry = i; + } + } + if (bestEntry >= 0) { + usedEntries.add(bestEntry); + const e = entries[bestEntry]; + if (e !== undefined) { + segmentEntry.set(seg, e); + segmentEntryIndex.set(seg, bestEntry); + } + } + } + + // Pass 2: positional fallback for segments pass 1 left unmatched + // (text-only turns with no stepId-bearing groups to anchor on). + // + // The loaded transcript is always a SUFFIX of the full turn history — + // chat-limit/windowing keeps the NEWEST chunks and unloads the OLDEST — so + // the T loaded segments correspond to the LAST T entries. TAIL-ALIGNMENT + // (segment `seg` ↔ entry `K - T + seg`) is therefore correct whenever the + // metrics hold at least as many turns as there are loaded segments + // (`K >= T`): the leading `K - T` entries are TRIMMED turns (their content + // was unloaded) and must be skipped, never matched to a newer segment. + // + // This MUST run even when pass 1 matched SOME segments (tool turns). The + // earlier code only tail-aligned when pass 1 matched NONE, falling back to + // HEAD-alignment otherwise — which, with leading trimmed entries, matched a + // brand-new text-only turn to an old (trimmed) entry's STALE metrics (the + // "new steps show no / wrong cache" failure). Tail-aligning by position is + // safe alongside pass 1: stepIds are unique per turn, so pass 1 already + // grabbed each tool turn's positionally-correct entry, leaving the right + // entry free for each text-only turn. + // + // Only when `K < T` (fewer entries than segments — some loaded turns have no + // metrics yet, e.g. a metrics sync still pending or a freshly loaded + // transcript) do we head-align, assigning the first K entries to the first K + // unmatched segments (the turns that DO have metrics sit at the front). + if (K >= T) { + // Tail-align: skip the first K-T entries (trimmed turns). + for (let seg = 0; seg < T; seg++) { + if (segmentEntry.has(seg)) continue; + const entryIdx = K - T + seg; + if (entryIdx >= 0 && entryIdx < K && !usedEntries.has(entryIdx)) { + usedEntries.add(entryIdx); + const e = entries[entryIdx]; + if (e !== undefined) { + segmentEntry.set(seg, e); + segmentEntryIndex.set(seg, entryIdx); + } + } + } + } else { + // Head-align fallback (K < T): first K entries to first K unmatched segments. + let nextUnused = 0; + for (let seg = 0; seg < T; seg++) { + if (segmentEntry.has(seg)) continue; + while (nextUnused < K && usedEntries.has(nextUnused)) nextUnused++; + if (nextUnused < K) { + usedEntries.add(nextUnused); + const e = entries[nextUnused]; + if (e !== undefined) { + segmentEntry.set(seg, e); + segmentEntryIndex.set(seg, nextUnused); + } + nextUnused++; + } + } + } + + // Running cumulative usage across ALL finalized turns (in entry order), for + // the per-turn "chat total" cache rate. Alongside it, the previous finalized + // turn's usage at each index — the baseline for cross-turn retention. + const cumulativeByEntry: Usage[] = []; + const prevUsageByEntry: (Usage | null)[] = []; + let runningUsage: Usage = { inputTokens: 0, outputTokens: 0 }; + let lastFinalizedUsage: Usage | null = null; + for (const e of entries) { + prevUsageByEntry.push(lastFinalizedUsage); + if (e.total !== null) { + runningUsage = addUsage(runningUsage, e.total.usage); + lastFinalizedUsage = e.total.usage; + } + cumulativeByEntry.push(runningUsage); + } + + const rows: MetricsRow[] = []; + + const firstUserIdx = segmentStarts[0] ?? 0; + + for (let i = 0; i < firstUserIdx; i++) { + const g = groups[i]; + if (g !== undefined) { + rows.push({ kind: "group", group: g }); + } + } + + for (let seg = 0; seg < T; seg++) { + const start = segmentStarts[seg] ?? 0; + const end = seg + 1 < T ? (segmentStarts[seg + 1] ?? groups.length) : groups.length; + + const entry = segmentEntry.get(seg); + + if (entry === undefined) { + for (let i = start; i < end; i++) { + const g = groups[i]; + if (g !== undefined) { + rows.push({ kind: "group", group: g }); + } + } + continue; + } + + const entryIdx = segmentEntryIndex.get(seg) ?? 0; + + // Build anchor map: for each stepId, the LAST group index in this segment. + const anchorByStepId = new Map<string, number>(); + for (let i = start; i < end; i++) { + const g = groups[i]; + if (g === undefined) continue; + const sid = groupStepId(g); + if (sid !== undefined) { + anchorByStepId.set(sid, i); + } + } + + // Classify each step as anchored or unanchored. Unanchored steps + // (content trimmed, or text-only steps with no tool chunks) are SKIPPED — + // step-metrics are only shown inline next to the content they describe. + const anchored: Map<number, { stepIndex: number; step: (typeof entry.steps)[number] }[]> = + new Map(); + + for (let i = 0; i < entry.steps.length; i++) { + const step = entry.steps[i]; + if (step === undefined) continue; + const anchorGroupIdx = anchorByStepId.get(step.stepId); + if (anchorGroupIdx !== undefined) { + let arr = anchored.get(anchorGroupIdx); + if (arr === undefined) { + arr = []; + anchored.set(anchorGroupIdx, arr); + } + arr.push({ stepIndex: i, step }); + } + // Unanchored steps (no matching group) are skipped — no tail bubbles. + } + + // Emit groups; after each anchored group, emit its step-metrics rows. + for (let i = start; i < end; i++) { + const g = groups[i]; + if (g !== undefined) { + rows.push({ kind: "group", group: g }); + } + const stepsHere = anchored.get(i); + if (stepsHere !== undefined) { + stepsHere.sort((a, b) => a.stepIndex - b.stepIndex); + for (const { step, stepIndex } of stepsHere) { + rows.push({ kind: "step-metrics", step, index: stepIndex }); + } + } + } + + // Turn-metrics row (only when the turn is finalized). Unanchored steps + // are skipped — no tail bubbles. + if (entry.total !== null) { + rows.push({ + kind: "turn-metrics", + turn: entry.total, + turnNumber: entryIdx + 1, + cumulativeUsage: cumulativeByEntry[entryIdx] ?? entry.total.usage, + prevTurnUsage: prevUsageByEntry[entryIdx] ?? null, + }); + } + } + + return rows; +} diff --git a/src/core/metrics/reducer.test.ts b/src/core/metrics/reducer.test.ts new file mode 100644 index 0000000..581a8b7 --- /dev/null +++ b/src/core/metrics/reducer.test.ts @@ -0,0 +1,689 @@ +import type { StepId, TurnDoneEvent, TurnStepCompleteEvent, TurnUsageEvent } from "@dispatch/wire"; +import { describe, expect, it } from "vitest"; +import { + applyDurableMetrics, + foldMetricsEvent, + initialMetricsState, + selectCurrentContextSize, + selectOrderedTurnMetrics, +} from "./reducer"; + +const usageEvent = ( + turnId: string, + inputTokens: number, + outputTokens: number, + stepId?: string, +): TurnUsageEvent => { + const base = { + type: "usage" as const, + conversationId: "c1", + turnId, + usage: { inputTokens, outputTokens }, + }; + if (stepId !== undefined) { + return { ...base, stepId: stepId as StepId }; + } + return base; +}; + +const stepCompleteEvent = ( + turnId: string, + stepId: string, + timing: { ttftMs?: number; decodeMs?: number; genTotalMs?: number } = {}, +): TurnStepCompleteEvent => ({ + type: "step-complete", + conversationId: "c1", + turnId, + stepId: stepId as StepId, + ...timing, +}); + +const doneEvent = ( + turnId: string, + extra: { + durationMs?: number; + usage?: { inputTokens: number; outputTokens: number }; + contextSize?: number; + } = {}, +): TurnDoneEvent => ({ + type: "done", + conversationId: "c1", + turnId, + reason: "stop", + ...extra, +}); + +describe("initialMetricsState", () => { + it("starts empty", () => { + const s = initialMetricsState(); + expect(s.live.size).toBe(0); + expect(s.liveOrder).toEqual([]); + expect(s.durable.size).toBe(0); + expect(s.durableOrder).toEqual([]); + }); +}); + +describe("foldMetricsEvent", () => { + it("folds per-step usage by stepId into a turn", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, usageEvent("t1", 100, 50, "s1")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s1")); + s = foldMetricsEvent(s, usageEvent("t1", 200, 80, "s2")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s2")); + s = foldMetricsEvent(s, doneEvent("t1")); + + const ordered = selectOrderedTurnMetrics(s); + expect(ordered).toHaveLength(1); + expect(ordered[0]?.turnId).toBe("t1"); + expect(ordered[0]?.steps).toHaveLength(2); + expect(ordered[0]?.steps[0]?.stepId).toBe("s1"); + expect(ordered[0]?.steps[0]?.usage).toEqual({ inputTokens: 100, outputTokens: 50 }); + expect(ordered[0]?.steps[1]?.stepId).toBe("s2"); + expect(ordered[0]?.steps[1]?.usage).toEqual({ inputTokens: 200, outputTokens: 80 }); + }); + + it("folds step-complete timing and merges with same-step usage", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, usageEvent("t1", 100, 50, "s1")); + s = foldMetricsEvent( + s, + stepCompleteEvent("t1", "s1", { ttftMs: 200, decodeMs: 800, genTotalMs: 1000 }), + ); + s = foldMetricsEvent(s, doneEvent("t1")); + + const ordered = selectOrderedTurnMetrics(s); + expect(ordered).toHaveLength(1); + const step = ordered[0]?.steps[0]; + expect(step?.usage).toEqual({ inputTokens: 100, outputTokens: 50 }); + expect(step?.ttftMs).toBe(200); + expect(step?.decodeMs).toBe(800); + expect(step?.genTotalMs).toBe(1000); + }); + + it("step-complete before usage defaults usage to zeros", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s1", { genTotalMs: 500 })); + s = foldMetricsEvent(s, doneEvent("t1")); + + const ordered = selectOrderedTurnMetrics(s); + const step = ordered[0]?.steps[0]; + expect(step?.usage).toEqual({ inputTokens: 0, outputTokens: 0 }); + expect(step?.genTotalMs).toBe(500); + }); + + it("done sets durationMs and aggregate usage", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, usageEvent("t1", 100, 50, "s1")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s1")); + s = foldMetricsEvent( + s, + doneEvent("t1", { + durationMs: 5000, + usage: { inputTokens: 300, outputTokens: 150 }, + }), + ); + + const ordered = selectOrderedTurnMetrics(s); + expect(ordered[0]?.total?.durationMs).toBe(5000); + expect(ordered[0]?.total?.usage).toEqual({ inputTokens: 300, outputTokens: 150 }); + }); + + it("aggregate usage sums steps when done.usage absent", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, usageEvent("t1", 100, 50, "s1")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s1")); + s = foldMetricsEvent(s, usageEvent("t1", 200, 80, "s2")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s2")); + s = foldMetricsEvent(s, doneEvent("t1")); + + const ordered = selectOrderedTurnMetrics(s); + expect(ordered[0]?.total?.usage).toEqual({ inputTokens: 300, outputTokens: 130 }); + }); + + it("aggregate usage includes cache only when a step had cache", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, { + type: "usage", + conversationId: "c1", + turnId: "t1", + stepId: "s1" as StepId, + usage: { inputTokens: 100, outputTokens: 50, cacheReadTokens: 30 }, + }); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s1")); + s = foldMetricsEvent(s, usageEvent("t1", 200, 80, "s2")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s2")); + s = foldMetricsEvent(s, doneEvent("t1")); + + const ordered = selectOrderedTurnMetrics(s); + expect(ordered[0]?.total?.usage.cacheReadTokens).toBe(30); + expect(ordered[0]?.total?.usage.cacheWriteTokens).toBeUndefined(); + }); + + it("tolerates missing clock (no genTotalMs/ttft/decode)", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, usageEvent("t1", 100, 50, "s1")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s1")); + s = foldMetricsEvent(s, doneEvent("t1")); + + const ordered = selectOrderedTurnMetrics(s); + const step = ordered[0]?.steps[0]; + expect(step?.ttftMs).toBeUndefined(); + expect(step?.decodeMs).toBeUndefined(); + expect(step?.genTotalMs).toBeUndefined(); + expect(ordered[0]?.total?.durationMs).toBeUndefined(); + }); + + it("usage without stepId does not create a turn", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, usageEvent("t1", 100, 50)); + + const ordered = selectOrderedTurnMetrics(s); + expect(ordered).toHaveLength(0); + }); + + it("ignores non-metrics events", () => { + const s = initialMetricsState(); + const next = foldMetricsEvent(s, { + type: "status", + conversationId: "c1", + status: "running", + }); + expect(next).toBe(s); + }); + + it("preserves first-seen order of steps", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, usageEvent("t1", 10, 5, "s2")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s2")); + s = foldMetricsEvent(s, usageEvent("t1", 20, 8, "s1")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s1")); + s = foldMetricsEvent(s, doneEvent("t1")); + + const ordered = selectOrderedTurnMetrics(s); + expect(ordered[0]?.steps[0]?.stepId).toBe("s2"); + expect(ordered[0]?.steps[1]?.stepId).toBe("s1"); + }); + + it("preserves first-seen order of turns", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, usageEvent("t2", 10, 5, "s1")); + s = foldMetricsEvent(s, usageEvent("t1", 20, 8, "s1")); + s = foldMetricsEvent(s, doneEvent("t2")); + s = foldMetricsEvent(s, doneEvent("t1")); + + const ordered = selectOrderedTurnMetrics(s); + expect(ordered[0]?.turnId).toBe("t2"); + expect(ordered[1]?.turnId).toBe("t1"); + }); +}); + +describe("selectOrderedTurnMetrics", () => { + it("durable wins over live by turnId, live-done appended last", () => { + let s = initialMetricsState(); + + s = foldMetricsEvent(s, usageEvent("t1", 100, 50, "s1")); + s = foldMetricsEvent(s, usageEvent("t2", 200, 80, "s1")); + s = foldMetricsEvent(s, stepCompleteEvent("t2", "s1")); + s = foldMetricsEvent(s, doneEvent("t2")); + + s = applyDurableMetrics(s, [ + { + turnId: "t1", + usage: { inputTokens: 999, outputTokens: 999 }, + durationMs: 3000, + steps: [ + { + stepId: "s1" as StepId, + usage: { inputTokens: 999, outputTokens: 999 }, + genTotalMs: 3000, + }, + ], + }, + ]); + + const ordered = selectOrderedTurnMetrics(s); + expect(ordered).toHaveLength(2); + expect(ordered[0]?.turnId).toBe("t1"); + expect(ordered[0]?.total?.usage.inputTokens).toBe(999); + expect(ordered[0]?.total?.durationMs).toBe(3000); + expect(ordered[1]?.turnId).toBe("t2"); + expect(ordered[1]?.total?.durationMs).toBeUndefined(); + }); + + it("empty state returns empty", () => { + const s = initialMetricsState(); + expect(selectOrderedTurnMetrics(s)).toEqual([]); + }); + + it("selectOrderedTurnMetrics: in-flight turn exposes only completed steps and total=null", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, usageEvent("t1", 100, 50, "s1")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s1", { genTotalMs: 1000 })); + s = foldMetricsEvent(s, usageEvent("t1", 200, 80, "s2")); + + const ordered = selectOrderedTurnMetrics(s); + expect(ordered).toHaveLength(1); + expect(ordered[0]?.turnId).toBe("t1"); + expect(ordered[0]?.steps).toHaveLength(1); + expect(ordered[0]?.steps[0]?.stepId).toBe("s1"); + expect(ordered[0]?.total).toBeNull(); + }); + + it("selectOrderedTurnMetrics: a turn with no complete step and not done is omitted", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, usageEvent("t1", 100, 50, "s1")); + s = foldMetricsEvent(s, usageEvent("t1", 200, 80, "s2")); + + const ordered = selectOrderedTurnMetrics(s); + expect(ordered).toHaveLength(0); + }); + + it("selectOrderedTurnMetrics: after done, total is present", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, usageEvent("t1", 100, 50, "s1")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s1", { genTotalMs: 1000 })); + s = foldMetricsEvent(s, doneEvent("t1", { durationMs: 2000 })); + + const ordered = selectOrderedTurnMetrics(s); + expect(ordered).toHaveLength(1); + expect(ordered[0]?.turnId).toBe("t1"); + expect(ordered[0]?.total?.durationMs).toBe(2000); + expect(ordered[0]?.steps).toHaveLength(1); + }); + + it("step-complete marks the step complete", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, usageEvent("t1", 100, 50, "s1")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s1", { genTotalMs: 500 })); + + const ordered = selectOrderedTurnMetrics(s); + expect(ordered).toHaveLength(1); + expect(ordered[0]?.steps).toHaveLength(1); + expect(ordered[0]?.steps[0]?.stepId).toBe("s1"); + expect(ordered[0]?.steps[0]?.genTotalMs).toBe(500); + }); + + it("selectOrderedTurnMetrics: durable turn → steps + total present", () => { + let s = initialMetricsState(); + s = applyDurableMetrics(s, [ + { + turnId: "t1", + usage: { inputTokens: 300, outputTokens: 150 }, + durationMs: 5000, + steps: [ + { + stepId: "s1" as StepId, + usage: { inputTokens: 100, outputTokens: 50 }, + genTotalMs: 1000, + }, + { + stepId: "s2" as StepId, + usage: { inputTokens: 200, outputTokens: 100 }, + genTotalMs: 2000, + }, + ], + }, + ]); + + const ordered = selectOrderedTurnMetrics(s); + expect(ordered).toHaveLength(1); + expect(ordered[0]?.turnId).toBe("t1"); + expect(ordered[0]?.steps).toHaveLength(2); + expect(ordered[0]?.steps[0]?.stepId).toBe("s1"); + expect(ordered[0]?.steps[1]?.stepId).toBe("s2"); + expect(ordered[0]?.total?.usage.inputTokens).toBe(300); + expect(ordered[0]?.total?.durationMs).toBe(5000); + }); +}); + +describe("applyDurableMetrics", () => { + it("stores durable turns in order", () => { + let s = initialMetricsState(); + s = applyDurableMetrics(s, [ + { turnId: "t1", usage: { inputTokens: 10, outputTokens: 5 }, steps: [] }, + { turnId: "t2", usage: { inputTokens: 20, outputTokens: 8 }, steps: [] }, + ]); + expect(s.durableOrder).toEqual(["t1", "t2"]); + expect(s.durable.size).toBe(2); + }); + + it("is idempotent for same turnId", () => { + let s = initialMetricsState(); + const turn = { + turnId: "t1", + usage: { inputTokens: 10, outputTokens: 5 }, + steps: [], + }; + s = applyDurableMetrics(s, [turn]); + s = applyDurableMetrics(s, [turn]); + expect(s.durableOrder).toEqual(["t1"]); + expect(s.durable.size).toBe(1); + }); + + it("overwrites durable turn data for same turnId", () => { + let s = initialMetricsState(); + s = applyDurableMetrics(s, [ + { turnId: "t1", usage: { inputTokens: 10, outputTokens: 5 }, steps: [] }, + ]); + s = applyDurableMetrics(s, [ + { turnId: "t1", usage: { inputTokens: 99, outputTokens: 99 }, steps: [] }, + ]); + expect(s.durable.get("t1")?.usage.inputTokens).toBe(99); + }); +}); + +describe("contextSize / selectCurrentContextSize", () => { + it("live done carries contextSize onto the turn total", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, usageEvent("t1", 100, 50, "s1")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s1")); + s = foldMetricsEvent(s, doneEvent("t1", { contextSize: 1234 })); + + const ordered = selectOrderedTurnMetrics(s); + expect(ordered[0]?.total?.contextSize).toBe(1234); + expect(selectCurrentContextSize(s)).toBe(1234); + }); + + it("contextSize is NOT the aggregate usage sum (multi-step turn)", () => { + let s = initialMetricsState(); + // Two steps: usage sums to 300 in / 130 out = 430, but contextSize is the + // backend-stamped final-step occupancy, independent of the sum. + s = foldMetricsEvent(s, usageEvent("t1", 100, 50, "s1")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s1")); + s = foldMetricsEvent(s, usageEvent("t1", 200, 80, "s2")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s2")); + s = foldMetricsEvent(s, doneEvent("t1", { contextSize: 250 })); + + const ordered = selectOrderedTurnMetrics(s); + expect(ordered[0]?.total?.usage).toEqual({ inputTokens: 300, outputTokens: 130 }); + expect(ordered[0]?.total?.contextSize).toBe(250); + expect(selectCurrentContextSize(s)).toBe(250); + }); + + it("persisted (durable) contextSize is preserved and selected", () => { + let s = initialMetricsState(); + s = applyDurableMetrics(s, [ + { turnId: "t1", usage: { inputTokens: 10, outputTokens: 5 }, steps: [], contextSize: 4096 }, + ]); + expect(s.durable.get("t1")?.contextSize).toBe(4096); + expect(selectCurrentContextSize(s)).toBe(4096); + }); + + it("selectCurrentContextSize returns the LATEST turn's value", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, doneEvent("t1", { contextSize: 100 })); + s = foldMetricsEvent(s, doneEvent("t2", { contextSize: 900 })); + expect(selectCurrentContextSize(s)).toBe(900); + }); + + it("selectCurrentContextSize skips a later turn that lacks contextSize", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, doneEvent("t1", { contextSize: 700 })); + // t2 finishes but the provider reported no per-step usage → no contextSize. + s = foldMetricsEvent(s, doneEvent("t2")); + expect(selectCurrentContextSize(s)).toBe(700); + }); + + it("selectCurrentContextSize is undefined (not 0) when nothing reported", () => { + let s = initialMetricsState(); + expect(selectCurrentContextSize(s)).toBeUndefined(); + s = foldMetricsEvent(s, doneEvent("t1")); + expect(selectCurrentContextSize(s)).toBeUndefined(); + }); + + it("durable contextSize wins over live for a shared turnId", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, doneEvent("t1", { contextSize: 111 })); + s = applyDurableMetrics(s, [ + { turnId: "t1", usage: { inputTokens: 1, outputTokens: 1 }, steps: [], contextSize: 222 }, + ]); + expect(selectCurrentContextSize(s)).toBe(222); + }); + + it("in-flight turn updates context size after the first step completes", () => { + // Before the requirement: an in-flight turn had total=null so its step usage + // was ignored until `done`. Now the latest step's input+output is used. + let s = initialMetricsState(); + s = foldMetricsEvent(s, usageEvent("t1", 5000, 200, "s1")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s1")); + + // Still generating (no done) — context = step 1 input+output = 5200. + expect(selectCurrentContextSize(s)).toBe(5200); + }); + + it("in-flight turn updates progressively as each step reports usage", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, usageEvent("t1", 5000, 200, "s1")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s1")); + expect(selectCurrentContextSize(s)).toBe(5200); + + // Step 2 reports usage mid-stream (before its step-complete): each step's + // input already includes all prior context, so the last step's input+output + // is the current occupancy. + s = foldMetricsEvent(s, usageEvent("t1", 5200, 150, "s2")); + expect(selectCurrentContextSize(s)).toBe(5350); + + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s2")); + expect(selectCurrentContextSize(s)).toBe(5350); + }); + + it("in-flight context size is the latest step with usage, NOT the aggregate sum", () => { + // Mirrors the finalized-turn test: contextSize is the FINAL step's + // input+output, not the sum across steps (which would overcount a + // multi-step turn because every step re-prefills the growing prompt). + let s = initialMetricsState(); + s = foldMetricsEvent(s, usageEvent("t1", 100, 50, "s1")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s1")); + s = foldMetricsEvent(s, usageEvent("t1", 200, 80, "s2")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s2")); + // Aggregate would be 300+130=430; the latest step is 200+80=280. + expect(selectCurrentContextSize(s)).toBe(280); + }); + + it("in-flight turn with a step-complete but no usage falls back to older turn", () => { + // step-complete before usage → the step has no usage yet, so the in-flight + // turn exposes no context size and the display falls back to the prior + // finalized turn's value (never 0). + let s = initialMetricsState(); + s = foldMetricsEvent(s, doneEvent("t1", { contextSize: 700 })); + s = foldMetricsEvent(s, stepCompleteEvent("t2", "s1", { genTotalMs: 500 })); + + expect(selectCurrentContextSize(s)).toBe(700); + }); + + it("in-flight turn with no steps/usage returns undefined (falls back)", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, doneEvent("t1", { contextSize: 700 })); + // t2 just started — no usage, no complete step — omitted entirely. + s = foldMetricsEvent(s, { type: "turn-start", conversationId: "c1", turnId: "t2" }); + expect(selectCurrentContextSize(s)).toBe(700); + + // t2's first step reports usage → the display jumps to t2's live value. + s = foldMetricsEvent(s, usageEvent("t2", 800, 10, "s1")); + s = foldMetricsEvent(s, stepCompleteEvent("t2", "s1")); + expect(selectCurrentContextSize(s)).toBe(810); + }); + + it("done finalizes the in-flight progressive value with the authoritative contextSize", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, usageEvent("t1", 5000, 200, "s1")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s1")); + expect(selectCurrentContextSize(s)).toBe(5200); + + s = foldMetricsEvent(s, usageEvent("t1", 5200, 150, "s2")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s2")); + expect(selectCurrentContextSize(s)).toBe(5350); + + // done stamps the authoritative contextSize (the final step's input+output). + s = foldMetricsEvent(s, doneEvent("t1", { contextSize: 5350 })); + expect(selectCurrentContextSize(s)).toBe(5350); + }); + + it("in-flight context size excludes cache tokens (they are a subset of inputTokens)", () => { + // cacheReadTokens / cacheWriteTokens are portions of inputTokens already + // counted — adding them would double-count. Only input+output is occupancy. + let s = initialMetricsState(); + s = foldMetricsEvent(s, { + type: "usage", + conversationId: "c1", + turnId: "t1", + stepId: "s1" as StepId, + usage: { + inputTokens: 5000, + outputTokens: 200, + cacheReadTokens: 4000, + cacheWriteTokens: 1000, + }, + }); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s1")); + // 5000+200=5200, NOT 9200 (with cacheRead) or 10200 (with both). + expect(selectCurrentContextSize(s)).toBe(5200); + }); + + it("multiple in-flight turns: the newest turn's live value wins", () => { + let s = initialMetricsState(); + // t1 (older) in-flight with one completed step → 5200. + s = foldMetricsEvent(s, usageEvent("t1", 5000, 200, "s1")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s1")); + // t2 (newer, seen later → last in liveOrder) in-flight → 8000. + s = foldMetricsEvent(s, usageEvent("t2", 7800, 200, "s1")); + s = foldMetricsEvent(s, stepCompleteEvent("t2", "s1")); + expect(selectCurrentContextSize(s)).toBe(8000); + }); + + it("out-of-order step IDs: usage for step 2 before step 1's step-complete still scans newest-first", () => { + // stepOrder is FIRST-SEEN: s1 (its usage arrived first), then s2. So s2 is + // the newest step regardless of when each step's step-complete arrives. + let s = initialMetricsState(); + s = foldMetricsEvent(s, usageEvent("t1", 5000, 200, "s1")); + s = foldMetricsEvent(s, usageEvent("t1", 5200, 150, "s2")); + // Neither step complete yet → the turn is omitted (no complete step), so the + // display can't update until the first step completes. + expect(selectCurrentContextSize(s)).toBeUndefined(); + + // s1 completes AFTER s2's usage was reported. The turn is now visible; the + // newest-first scan picks s2 (the later step), not s1 (the just-completed one). + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s1")); + expect(selectCurrentContextSize(s)).toBe(5350); + + // s2 completes — still s2, unchanged. + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s2")); + expect(selectCurrentContextSize(s)).toBe(5350); + }); + + it("done turn without contextSize falls back to an older turn (even with step usage)", () => { + // Contract lock-in: a done turn's step usage is NOT consulted for the + // context display — only its authoritative total.contextSize is. When that + // is absent, the display falls back to the next older finalized turn rather + // than synthesizing a value from the step usage. + let s = initialMetricsState(); + s = foldMetricsEvent(s, doneEvent("t1", { contextSize: 700 })); + // t2 done WITH step usage but NO done.contextSize (edge case: the done event + // omitted contextSize despite per-step usage). + s = foldMetricsEvent(s, usageEvent("t2", 800, 10, "s1")); + s = foldMetricsEvent(s, stepCompleteEvent("t2", "s1")); + s = foldMetricsEvent(s, doneEvent("t2")); + expect(selectCurrentContextSize(s)).toBe(700); + }); + + it("in-flight context size skips a step with unsafe usage (NaN / negative)", () => { + // A corrupt provider report must never reach the status bar. The newest + // step with invalid counters is skipped, falling back to the prior valid one. + let s = initialMetricsState(); + s = foldMetricsEvent(s, usageEvent("t1", 5000, 200, "s1")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s1")); + // s2 reports NaN input (e.g. a non-numeric provider field coerced). + s = foldMetricsEvent(s, { + type: "usage", + conversationId: "c1", + turnId: "t1", + stepId: "s2" as StepId, + usage: { inputTokens: Number.NaN, outputTokens: 150 }, + }); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s2")); + // s2 skipped (NaN) → falls back to s1's 5200, NOT NaN. + expect(selectCurrentContextSize(s)).toBe(5200); + + // Negative tokens are likewise skipped. + s = foldMetricsEvent(s, { + type: "usage", + conversationId: "c1", + turnId: "t1", + stepId: "s3" as StepId, + usage: { inputTokens: -10, outputTokens: 5 }, + }); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s3")); + expect(selectCurrentContextSize(s)).toBe(5200); + }); +}); + +describe("applyDurableMetrics pruning", () => { + it("prunes a live turn once durable data covers it (no unbounded growth)", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, usageEvent("t1", 100, 50, "s1")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s1")); + s = foldMetricsEvent(s, doneEvent("t1", { contextSize: 150 })); + expect(s.live.has("t1")).toBe(true); + expect(s.liveOrder).toContain("t1"); + + s = applyDurableMetrics(s, [ + { + turnId: "t1", + usage: { inputTokens: 100, outputTokens: 50 }, + steps: [{ stepId: "s1" as StepId, usage: { inputTokens: 100, outputTokens: 50 } }], + contextSize: 150, + }, + ]); + // The live copy is gone; the durable (authoritative) entry replaces it. + expect(s.live.has("t1")).toBe(false); + expect(s.liveOrder).not.toContain("t1"); + expect(s.durable.has("t1")).toBe(true); + // The display still reads the durable value atomically (no gap). + expect(selectCurrentContextSize(s)).toBe(150); + }); + + it("prunes only the turns present in the durable batch (leaves other live turns)", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, usageEvent("t1", 100, 50, "s1")); + s = foldMetricsEvent(s, stepCompleteEvent("t1", "s1")); + s = foldMetricsEvent(s, doneEvent("t1", { contextSize: 150 })); + // t2 still in flight — must NOT be pruned when only t1 seals. + s = foldMetricsEvent(s, usageEvent("t2", 800, 10, "s1")); + s = foldMetricsEvent(s, stepCompleteEvent("t2", "s1")); + + s = applyDurableMetrics(s, [ + { turnId: "t1", usage: { inputTokens: 100, outputTokens: 50 }, steps: [], contextSize: 150 }, + ]); + expect(s.live.has("t1")).toBe(false); + expect(s.live.has("t2")).toBe(true); + expect(s.liveOrder).toEqual(["t2"]); + // The newest (in-flight) turn's live value still wins. + expect(selectCurrentContextSize(s)).toBe(810); + }); + + it("is a no-op when no incoming turn is live (no live mutation)", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, usageEvent("t2", 800, 10, "s1")); + s = foldMetricsEvent(s, stepCompleteEvent("t2", "s1")); + const before = s; + s = applyDurableMetrics(s, [ + { turnId: "t1", usage: { inputTokens: 1, outputTokens: 1 }, steps: [] }, + ]); + // t1 was never live → the live map/order are unchanged (same reference). + expect(s.live).toBe(before.live); + expect(s.liveOrder).toBe(before.liveOrder); + // t1 (durable) is older; the in-flight t2 still wins. + expect(selectCurrentContextSize(s)).toBe(810); + }); + + it("durable wins over live for a shared turnId (pruned live no longer consulted)", () => { + let s = initialMetricsState(); + s = foldMetricsEvent(s, doneEvent("t1", { contextSize: 111 })); + s = applyDurableMetrics(s, [ + { turnId: "t1", usage: { inputTokens: 1, outputTokens: 1 }, steps: [], contextSize: 222 }, + ]); + // The live (111) copy is pruned; only durable (222) remains. + expect(s.live.has("t1")).toBe(false); + expect(selectCurrentContextSize(s)).toBe(222); + }); +}); diff --git a/src/core/metrics/reducer.ts b/src/core/metrics/reducer.ts new file mode 100644 index 0000000..39fc5ee --- /dev/null +++ b/src/core/metrics/reducer.ts @@ -0,0 +1,345 @@ +import type { AgentEvent, StepId, StepMetrics, TurnMetrics, Usage } from "@dispatch/wire"; +import type { BuildingStep, LiveTurn, MetricsState, TurnMetricsEntry } from "./types"; + +function sumStepUsages(steps: readonly BuildingStep[]): Usage { + let inputTokens = 0; + let outputTokens = 0; + let hasCacheRead = false; + let hasCacheWrite = false; + let cacheReadTokens = 0; + let cacheWriteTokens = 0; + + for (const step of steps) { + if (step.usage === undefined) continue; + inputTokens += step.usage.inputTokens; + outputTokens += step.usage.outputTokens; + if (step.usage.cacheReadTokens !== undefined && step.usage.cacheReadTokens > 0) { + hasCacheRead = true; + cacheReadTokens += step.usage.cacheReadTokens; + } + if (step.usage.cacheWriteTokens !== undefined && step.usage.cacheWriteTokens > 0) { + hasCacheWrite = true; + cacheWriteTokens += step.usage.cacheWriteTokens; + } + } + + const base: Usage = { inputTokens, outputTokens }; + if (hasCacheRead) { + (base as { cacheReadTokens?: number }).cacheReadTokens = cacheReadTokens; + } + if (hasCacheWrite) { + (base as { cacheWriteTokens?: number }).cacheWriteTokens = cacheWriteTokens; + } + return base; +} + +function buildingStepToMetrics(bs: BuildingStep): StepMetrics { + const usage: Usage = bs.usage ?? { inputTokens: 0, outputTokens: 0 }; + const base: StepMetrics = { stepId: bs.stepId as StepId, usage }; + if (bs.ttftMs !== undefined) { + (base as { ttftMs?: number }).ttftMs = bs.ttftMs; + } + if (bs.decodeMs !== undefined) { + (base as { decodeMs?: number }).decodeMs = bs.decodeMs; + } + if (bs.genTotalMs !== undefined) { + (base as { genTotalMs?: number }).genTotalMs = bs.genTotalMs; + } + return base; +} + +function getStep(lt: LiveTurn, id: string): BuildingStep { + const step = lt.stepMap.get(id); + if (step === undefined) throw new Error(`Missing step ${id} in live turn`); + return step; +} + +function liveTurnToMetrics(lt: LiveTurn): TurnMetrics { + const buildingSteps = lt.stepOrder.map((id) => getStep(lt, id)); + const steps = buildingSteps.map((bs) => buildingStepToMetrics(bs)); + const usage = lt.doneUsage ?? sumStepUsages(buildingSteps); + const base: TurnMetrics = { turnId: lt.turnId, usage, steps }; + if (lt.durationMs !== undefined) { + (base as { durationMs?: number }).durationMs = lt.durationMs; + } + if (lt.doneContextSize !== undefined) { + (base as { contextSize?: number }).contextSize = lt.doneContextSize; + } + return base; +} + +/** + * A step's contribution to the live context size: `inputTokens + outputTokens`, + * or `undefined` when the step has no usage yet OR its counters are not safe to + * sum (non-finite / negative — defensive: a corrupt provider report must never + * reach the status bar as NaN/Infinity). Cache tokens are deliberately NOT + * included: `cacheReadTokens` / `cacheWriteTokens` are a SUBSET of + * `inputTokens`, so adding them would double-count. + */ +function stepContextSize(usage: Usage | undefined): number | undefined { + if (usage === undefined) return undefined; + const { inputTokens, outputTokens } = usage; + if (!Number.isFinite(inputTokens) || !Number.isFinite(outputTokens)) return undefined; + if (inputTokens < 0 || outputTokens < 0) return undefined; + return inputTokens + outputTokens; +} + +/** + * The context size an IN-FLIGHT (not-done) turn occupies right now — for + * progressive display DURING a turn (before it seals), so the indicator updates + * after each step instead of waiting for `done`. + * + * CONTRACT: only call this on a turn whose `done` event has NOT arrived (the + * caller, `selectCurrentContextSize`, reaches it solely for entries with + * `total === null`, i.e. `lt.done === false`). Finalized turns use their + * authoritative `contextSize` instead; `doneContextSize` is read on the + * `total` path, never here. + * + * Returns the most recent step WITH USABLE USAGE's `inputTokens + outputTokens` + * (scanning newest → oldest by first-seen step order): each step's input + * already includes all prior context (the prompt is re-prefilled every step), so + * the last step's input+output is the true occupancy — the same definition + * `TurnDoneEvent.contextSize` stamps at turn end. A just-reported step's usage + * wins immediately, even mid-stream. Steps with no usage or unsafe usage are + * skipped, falling back to the next older usable step. `undefined` when no step + * has reported usable usage yet. + */ +function liveTurnContextSize(lt: LiveTurn): number | undefined { + for (let i = lt.stepOrder.length - 1; i >= 0; i--) { + const step = lt.stepMap.get(lt.stepOrder[i] ?? ""); + const ctx = stepContextSize(step?.usage); + if (ctx !== undefined) return ctx; + } + return undefined; +} + +function ensureLiveTurn(state: MetricsState, turnId: string): [MetricsState, LiveTurn] { + const existing = state.live.get(turnId); + if (existing !== undefined) return [state, existing]; + + const newTurn: LiveTurn = { + turnId, + done: false, + durationMs: undefined, + doneUsage: undefined, + doneContextSize: undefined, + stepMap: new Map(), + stepOrder: [], + }; + const newLive = new Map(state.live); + newLive.set(turnId, newTurn); + return [{ ...state, live: newLive, liveOrder: [...state.liveOrder, turnId] }, newTurn]; +} + +function upsertStep(lt: LiveTurn, stepId: string, update: Partial<BuildingStep>): LiveTurn { + const existing = lt.stepMap.get(stepId); + if (existing !== undefined) { + const merged: BuildingStep = { + stepId, + usage: update.usage ?? existing.usage, + ttftMs: update.ttftMs ?? existing.ttftMs, + decodeMs: update.decodeMs ?? existing.decodeMs, + genTotalMs: update.genTotalMs ?? existing.genTotalMs, + complete: update.complete ?? existing.complete, + }; + const newMap = new Map(lt.stepMap); + newMap.set(stepId, merged); + return { ...lt, stepMap: newMap }; + } + + const fresh: BuildingStep = { + stepId, + usage: update.usage, + ttftMs: update.ttftMs, + decodeMs: update.decodeMs, + genTotalMs: update.genTotalMs, + complete: update.complete ?? false, + }; + const newMap = new Map(lt.stepMap); + newMap.set(stepId, fresh); + return { ...lt, stepMap: newMap, stepOrder: [...lt.stepOrder, stepId] }; +} + +/** The initial empty metrics state. */ +export function initialMetricsState(): MetricsState { + return { + live: new Map(), + liveOrder: [], + durable: new Map(), + durableOrder: [], + }; +} + +/** + * Fold one live AgentEvent into the metrics state. + * + * - `usage` with `stepId`: upsert that step's usage. + * - `usage` without `stepId`: ignored. + * - `step-complete`: upsert that step's timing; default usage to zeros if absent. + * - `done`: set turn's `durationMs`, optional aggregate `usage`, and optional `contextSize`. + * - All other event types: return state unchanged. + */ +export function foldMetricsEvent(state: MetricsState, event: AgentEvent): MetricsState { + switch (event.type) { + case "usage": { + if (event.stepId === undefined) return state; + const [s1, lt] = ensureLiveTurn(state, event.turnId); + const updated = upsertStep(lt, event.stepId, { usage: event.usage }); + const newLive = new Map(s1.live); + newLive.set(event.turnId, updated); + return { ...s1, live: newLive }; + } + + case "step-complete": { + const [s1, lt] = ensureLiveTurn(state, event.turnId); + const updated = upsertStep(lt, event.stepId, { + ttftMs: event.ttftMs, + decodeMs: event.decodeMs, + genTotalMs: event.genTotalMs, + complete: true, + }); + const newLive = new Map(s1.live); + newLive.set(event.turnId, updated); + return { ...s1, live: newLive }; + } + + case "done": { + const [s1, lt] = ensureLiveTurn(state, event.turnId); + const updated: LiveTurn = { + ...lt, + done: true, + durationMs: event.durationMs ?? lt.durationMs, + doneUsage: event.usage ?? lt.doneUsage, + doneContextSize: event.contextSize ?? lt.doneContextSize, + }; + const newLive = new Map(s1.live); + newLive.set(event.turnId, updated); + return { ...s1, live: newLive }; + } + + default: + return state; + } +} + +/** + * Store durable (sealed) metrics from the backend. These win over live data + * for any shared `turnId`. + * + * Once durable (authoritative) data covers a turn, its live (in-memory) copy + * is REDUNDANT and is pruned from `state.live` / `liveOrder` so the live map + * doesn't grow unbounded over a long conversation. There is no display gap: + * the durable entry replaces the live one atomically in the same fold, and + * `selectOrderedTurnMetrics` / `selectCurrentContextSize` read durable for it. + */ +export function applyDurableMetrics( + state: MetricsState, + turns: readonly TurnMetrics[], +): MetricsState { + const newDurable = new Map(state.durable); + const newDurableOrder = [...state.durableOrder]; + const prunedIds = new Set<string>(); + for (const turn of turns) { + if (!newDurable.has(turn.turnId)) { + newDurableOrder.push(turn.turnId); + } + newDurable.set(turn.turnId, turn); + if (state.live.has(turn.turnId)) prunedIds.add(turn.turnId); + } + + if (prunedIds.size === 0) { + return { ...state, durable: newDurable, durableOrder: newDurableOrder }; + } + + const newLive = new Map(state.live); + for (const id of prunedIds) newLive.delete(id); + const newLiveOrder = state.liveOrder.filter((id) => !prunedIds.has(id)); + + return { + ...state, + live: newLive, + liveOrder: newLiveOrder, + durable: newDurable, + durableOrder: newDurableOrder, + }; +} + +/** + * Select the merged ordered list of turn metrics entries. + * Durable turns come first (in their order), then any live turns whose + * `turnId` is not in durable (in live first-seen order). + * + * Each entry contains the completed steps so far and an optional total + * (null until the turn is finalized via `done` or durable data). + * Live turns with no completed steps and not done are omitted. + */ +export function selectOrderedTurnMetrics(state: MetricsState): readonly TurnMetricsEntry[] { + const result: TurnMetricsEntry[] = []; + const seen = new Set<string>(); + + for (const turnId of state.durableOrder) { + const tm = state.durable.get(turnId); + if (tm !== undefined) { + result.push({ turnId, steps: tm.steps, total: tm }); + seen.add(turnId); + } + } + + for (const turnId of state.liveOrder) { + if (seen.has(turnId)) continue; + const lt = state.live.get(turnId); + if (lt === undefined) continue; + + const completeSteps = lt.stepOrder + .map((id) => lt.stepMap.get(id)) + .filter((s): s is BuildingStep => s?.complete === true) + .map((s) => buildingStepToMetrics(s)); + + if (completeSteps.length === 0 && !lt.done) continue; + + result.push({ + turnId, + steps: completeSteps, + total: lt.done ? liveTurnToMetrics(lt) : null, + }); + } + + return result; +} + +/** + * Select the conversation's CURRENT context size — the tokens it occupies right + * now. Per the wire contract a client reads the LATEST turn's `contextSize`; we + * scan the merged ordered turns NEWEST → OLDEST and return the first DEFINED + * value. + * + * For a FINALIZED turn (`done` event or durable data) we use its authoritative + * `contextSize`. For an IN-FLIGHT (not-done) turn we compute it PROGRESSIVELY + * from the most recent step WITH USAGE — its `inputTokens + outputTokens` is the + * current occupancy (mirroring `TurnDoneEvent.contextSize`'s definition) — so + * the indicator updates after each step completes instead of waiting for the + * turn to seal. An in-flight turn with no step usage yet is skipped, falling + * back to the next older finalized turn. + * + * Returns `undefined` ("unknown") when no turn carries a context size — the + * caller renders a placeholder, NEVER `0`. Durable (sealed) data wins over + * live for a shared `turnId` (it is the persisted, authoritative value). + */ +export function selectCurrentContextSize(state: MetricsState): number | undefined { + const ordered = selectOrderedTurnMetrics(state); + for (let i = ordered.length - 1; i >= 0; i--) { + const entry = ordered[i]; + if (entry === undefined) continue; + if (entry.total !== null) { + if (entry.total.contextSize !== undefined) return entry.total.contextSize; + continue; + } + // In-flight turn: progressive context size from the latest step with usage. + const lt = state.live.get(entry.turnId); + if (lt !== undefined) { + const live = liveTurnContextSize(lt); + if (live !== undefined) return live; + } + } + return undefined; +} diff --git a/src/core/metrics/types.ts b/src/core/metrics/types.ts new file mode 100644 index 0000000..84d1904 --- /dev/null +++ b/src/core/metrics/types.ts @@ -0,0 +1,97 @@ +import type { StepMetrics, TurnMetrics, Usage } from "@dispatch/wire"; +import type { RenderGroup } from "../chunks"; + +export type { StepMetrics, TurnMetrics }; + +/** A step being built from live events (may be incomplete). */ +export interface BuildingStep { + readonly stepId: string; + readonly usage: Usage | undefined; + readonly ttftMs: number | undefined; + readonly decodeMs: number | undefined; + readonly genTotalMs: number | undefined; + readonly complete: boolean; +} + +/** A turn being built from live events (in-flight). */ +export interface LiveTurn { + readonly turnId: string; + readonly done: boolean; + readonly durationMs: number | undefined; + readonly doneUsage: Usage | undefined; + /** + * Context size carried on the turn's `done` event (the turn's FINAL step + * `inputTokens + outputTokens` — current context occupancy). `undefined` when + * the provider reported no per-step usage; never coerced to `0`. + */ + readonly doneContextSize: number | undefined; + readonly stepMap: ReadonlyMap<string, BuildingStep>; + readonly stepOrder: readonly string[]; +} + +/** + * Reducer state for per-turn / per-step token + timing metrics. + * + * - `live`: in-flight turns keyed by `turnId` in FIRST-SEEN order. + * - `durable`: sealed turns keyed by `turnId` in the order they arrived. + */ +export interface MetricsState { + readonly live: ReadonlyMap<string, LiveTurn>; + readonly liveOrder: readonly string[]; + readonly durable: ReadonlyMap<string, TurnMetrics>; + readonly durableOrder: readonly string[]; +} + +/** Per-turn placement entry: completed steps so far + optional turn total. */ +export interface TurnMetricsEntry { + readonly turnId: string; + readonly steps: readonly StepMetrics[]; + readonly total: TurnMetrics | null; +} + +/** A row in the interleaved transcript: a render group, per-step metrics, or turn metrics. */ +export type MetricsRow = + | { readonly kind: "group"; readonly group: RenderGroup } + | { readonly kind: "step-metrics"; readonly step: StepMetrics; readonly index: number } + | { + readonly kind: "turn-metrics"; + readonly turn: TurnMetrics; + /** 1-based turn number (the entry's position in the metrics array + 1). */ + readonly turnNumber: number; + /** Cumulative usage across all finalized turns up to and including this one. */ + readonly cumulativeUsage: Usage; + /** + * Usage of the most recent EARLIER finalized turn, or `null` when this is the + * first finalized turn. The baseline for cross-turn retention (expected cache). + */ + readonly prevTurnUsage: Usage | null; + }; + +/** Formatted cache hit-rate view: percentage + colour severity + hit flag. */ +export interface CacheRateView { + /** Cache hit rate as a 0..100 integer percentage (`cacheReadTokens / inputTokens`). */ + readonly pct: number; + /** Colour severity for a badge (maps to DaisyUI `badge-{level}`). */ + readonly level: "success" | "warning" | "error"; + /** Whether any input tokens were served from cache. */ + readonly isHit: boolean; +} + +/** Formatted per-step view for display. */ +export interface StepMetricsView { + readonly label: string; + readonly tokensLabel: string; + readonly tps: string | null; + readonly ttft: string | null; + readonly decode: string | null; + readonly genTotal: string | null; +} + +/** Formatted per-turn view for display. */ +export interface TurnMetricsView { + readonly label: string; + readonly tokensLabel: string; + readonly breakdown: string; + readonly tps: string | null; + readonly duration: string | null; +} diff --git a/src/core/protocol/index.ts b/src/core/protocol/index.ts index 25174ea..2c1c290 100644 --- a/src/core/protocol/index.ts +++ b/src/core/protocol/index.ts @@ -1,2 +1,9 @@ -export { applyServerMessage, initialState, invoke, subscribe, unsubscribe } from "./reducer"; -export type { ProtocolResult, ProtocolState } from "./types"; +export { + applyServerMessage, + getSurfaceSpec, + initialState, + invoke, + subscribe, + unsubscribe, +} from "./reducer"; +export type { ProtocolResult, ProtocolState, Subscription } from "./types"; diff --git a/src/core/protocol/reducer.test.ts b/src/core/protocol/reducer.test.ts index 57e12f2..d42dce7 100644 --- a/src/core/protocol/reducer.test.ts +++ b/src/core/protocol/reducer.test.ts @@ -1,151 +1,245 @@ import { describe, expect, it } from "vitest"; -import { applyServerMessage, initialState, invoke, subscribe, unsubscribe } from "./reducer"; +import { + applyServerMessage, + getSurfaceSpec, + initialState, + invoke, + subscribe, + unsubscribe, +} from "./reducer"; const makeSpec = (id: string, title = id) => ({ - id, - region: "test", - title, - fields: [], + id, + region: "test", + title, + fields: [], }); describe("initialState", () => { - it("returns empty catalog, no subscriptions, no error", () => { - const s = initialState(); - expect(s.catalog).toEqual([]); - expect(s.subscriptions.size).toBe(0); - expect(s.lastError).toBeNull(); - }); + it("returns empty catalog, no subscriptions, no error", () => { + const s = initialState(); + expect(s.catalog).toEqual([]); + expect(s.subscriptions.size).toBe(0); + expect(s.lastError).toBeNull(); + }); }); describe("applyServerMessage — catalog", () => { - it("replaces the catalog", () => { - const s = initialState(); - const catalog = [ - { id: "a", region: "r", title: "A" }, - { id: "b", region: "r", title: "B" }, - ]; - const next = applyServerMessage(s, { type: "catalog", catalog }); - expect(next.catalog).toEqual(catalog); - }); + it("replaces the catalog", () => { + const s = initialState(); + const catalog = [ + { id: "a", region: "r", title: "A" }, + { id: "b", region: "r", title: "B" }, + ]; + const next = applyServerMessage(s, { type: "catalog", catalog }); + expect(next.catalog).toEqual(catalog); + }); }); describe("applyServerMessage — surface", () => { - it("sets the spec for a subscribed surface", () => { - let s = initialState(); - const result = subscribe(s, "s1"); - s = result.state; - const spec = makeSpec("s1", "Surface 1"); - const next = applyServerMessage(s, { type: "surface", spec }); - expect(next.subscriptions.get("s1")).toEqual(spec); - }); - - it("ignores a surface message for a non-subscribed surface", () => { - const s = initialState(); - const spec = makeSpec("unknown"); - const next = applyServerMessage(s, { type: "surface", spec }); - expect(next.subscriptions.has("unknown")).toBe(false); - }); + it("sets the spec for a subscribed surface", () => { + let s = initialState(); + s = subscribe(s, "s1").state; + const spec = makeSpec("s1", "Surface 1"); + const next = applyServerMessage(s, { type: "surface", spec }); + expect(getSurfaceSpec(next, "s1")).toEqual(spec); + }); + + it("ignores a surface message for a non-subscribed surface", () => { + const s = initialState(); + const spec = makeSpec("unknown"); + const next = applyServerMessage(s, { type: "surface", spec }); + expect(next.subscriptions.has("unknown")).toBe(false); + }); }); describe("applyServerMessage — update", () => { - it("replaces spec for a subscribed surface", () => { - let s = initialState(); - s = subscribe(s, "s1").state; - s = applyServerMessage(s, { type: "surface", spec: makeSpec("s1", "V1") }); - const next = applyServerMessage(s, { - type: "update", - update: { surfaceId: "s1", spec: makeSpec("s1", "V2") }, - }); - expect(next.subscriptions.get("s1")?.title).toBe("V2"); - }); - - it("ignores an update for a non-subscribed surface", () => { - const s = initialState(); - const next = applyServerMessage(s, { - type: "update", - update: { surfaceId: "nope", spec: makeSpec("nope") }, - }); - expect(next.subscriptions.has("nope")).toBe(false); - }); + it("replaces spec for a subscribed surface", () => { + let s = initialState(); + s = subscribe(s, "s1").state; + s = applyServerMessage(s, { type: "surface", spec: makeSpec("s1", "V1") }); + const next = applyServerMessage(s, { + type: "update", + update: { surfaceId: "s1", spec: makeSpec("s1", "V2") }, + }); + expect(getSurfaceSpec(next, "s1")?.title).toBe("V2"); + }); + + it("ignores an update for a non-subscribed surface", () => { + const s = initialState(); + const next = applyServerMessage(s, { + type: "update", + update: { surfaceId: "nope", spec: makeSpec("nope") }, + }); + expect(next.subscriptions.has("nope")).toBe(false); + }); }); describe("applyServerMessage — error", () => { - it("records the error without throwing", () => { - const s = initialState(); - const err = { type: "error" as const, surfaceId: "s1", message: "boom" }; - const next = applyServerMessage(s, err); - expect(next.lastError).toEqual(err); - }); - - it("records error without surfaceId", () => { - const s = initialState(); - const err = { type: "error" as const, message: "global boom" }; - const next = applyServerMessage(s, err); - expect(next.lastError).toEqual(err); - }); + it("records the error without throwing", () => { + const s = initialState(); + const err = { type: "error" as const, surfaceId: "s1", message: "boom" }; + const next = applyServerMessage(s, err); + expect(next.lastError).toEqual(err); + }); + + it("records error without surfaceId", () => { + const s = initialState(); + const err = { type: "error" as const, message: "global boom" }; + const next = applyServerMessage(s, err); + expect(next.lastError).toEqual(err); + }); }); describe("subscribe", () => { - it("emits exactly one subscribe message", () => { - const s = initialState(); - const result = subscribe(s, "s1"); - expect(result.outgoing).toEqual([{ type: "subscribe", surfaceId: "s1" }]); - expect(result.outgoing).toHaveLength(1); - }); - - it("adds the surface to subscriptions with null spec", () => { - const s = initialState(); - const result = subscribe(s, "s1"); - expect(result.state.subscriptions.get("s1")).toBeNull(); - }); - - it("is idempotent — second subscribe is a no-op", () => { - let s = initialState(); - s = subscribe(s, "s1").state; - const result = subscribe(s, "s1"); - expect(result.outgoing).toEqual([]); - expect(result.state).toBe(s); - }); + it("emits exactly one subscribe message (global, no conversationId)", () => { + const s = initialState(); + const result = subscribe(s, "s1"); + expect(result.outgoing).toEqual([{ type: "subscribe", surfaceId: "s1" }]); + expect(result.outgoing).toHaveLength(1); + }); + + it("adds the surface to subscriptions with null spec", () => { + const s = initialState(); + const result = subscribe(s, "s1"); + expect(result.state.subscriptions.get("s1")).toEqual({ + conversationId: undefined, + spec: null, + }); + expect(getSurfaceSpec(result.state, "s1")).toBeNull(); + }); + + it("is idempotent — second subscribe with the same scope is a no-op", () => { + let s = initialState(); + s = subscribe(s, "s1").state; + const result = subscribe(s, "s1"); + expect(result.outgoing).toEqual([]); + expect(result.state).toBe(s); + }); +}); + +describe("subscribe — conversation-scoped", () => { + it("includes conversationId in the subscribe message", () => { + const s = initialState(); + const result = subscribe(s, "cache-warming", "conv-A"); + expect(result.outgoing).toEqual([ + { type: "subscribe", surfaceId: "cache-warming", conversationId: "conv-A" }, + ]); + expect(result.state.subscriptions.get("cache-warming")?.conversationId).toBe("conv-A"); + }); + + it("re-scopes on conversation switch: unsubscribe old pair then subscribe new", () => { + let s = initialState(); + s = subscribe(s, "cw", "conv-A").state; + s = applyServerMessage(s, { + type: "surface", + spec: makeSpec("cw", "A-spec"), + conversationId: "conv-A", + }); + const result = subscribe(s, "cw", "conv-B"); + expect(result.outgoing).toEqual([ + { type: "unsubscribe", surfaceId: "cw", conversationId: "conv-A" }, + { type: "subscribe", surfaceId: "cw", conversationId: "conv-B" }, + ]); + // previous spec retained until the new one arrives (no flicker) + expect(getSurfaceSpec(result.state, "cw")?.title).toBe("A-spec"); + expect(result.state.subscriptions.get("cw")?.conversationId).toBe("conv-B"); + }); + + it("drops a stale update echoing the previous conversationId", () => { + let s = initialState(); + s = subscribe(s, "cw", "conv-A").state; + s = subscribe(s, "cw", "conv-B").state; // re-scoped to B + const next = applyServerMessage(s, { + type: "update", + update: { surfaceId: "cw", spec: makeSpec("cw", "STALE-A"), conversationId: "conv-A" }, + }); + expect(getSurfaceSpec(next, "cw")).toBeNull(); // stale ignored, no spec yet for B + }); + + it("accepts an update echoing the current conversationId", () => { + let s = initialState(); + s = subscribe(s, "cw", "conv-B").state; + const next = applyServerMessage(s, { + type: "update", + update: { surfaceId: "cw", spec: makeSpec("cw", "B-spec"), conversationId: "conv-B" }, + }); + expect(getSurfaceSpec(next, "cw")?.title).toBe("B-spec"); + }); + + it("accepts a global (no-echo) surface message even when subscribed with a conversationId", () => { + // loaded-extensions is global: server ignores our conversationId and echoes none. + let s = initialState(); + s = subscribe(s, "loaded-extensions", "conv-A").state; + const next = applyServerMessage(s, { + type: "surface", + spec: makeSpec("loaded-extensions", "Ext"), + }); + expect(getSurfaceSpec(next, "loaded-extensions")?.title).toBe("Ext"); + }); }); describe("unsubscribe", () => { - it("emits unsubscribe and drops the spec", () => { - let s = initialState(); - s = subscribe(s, "s1").state; - s = applyServerMessage(s, { type: "surface", spec: makeSpec("s1") }); - const result = unsubscribe(s, "s1"); - expect(result.outgoing).toEqual([{ type: "unsubscribe", surfaceId: "s1" }]); - expect(result.state.subscriptions.has("s1")).toBe(false); - }); - - it("is a no-op if not subscribed", () => { - const s = initialState(); - const result = unsubscribe(s, "nope"); - expect(result.outgoing).toEqual([]); - expect(result.state).toBe(s); - }); + it("emits unsubscribe and drops the spec", () => { + let s = initialState(); + s = subscribe(s, "s1").state; + s = applyServerMessage(s, { type: "surface", spec: makeSpec("s1") }); + const result = unsubscribe(s, "s1"); + expect(result.outgoing).toEqual([{ type: "unsubscribe", surfaceId: "s1" }]); + expect(result.state.subscriptions.has("s1")).toBe(false); + }); + + it("includes conversationId for a scoped subscription", () => { + let s = initialState(); + s = subscribe(s, "cw", "conv-A").state; + const result = unsubscribe(s, "cw"); + expect(result.outgoing).toEqual([ + { type: "unsubscribe", surfaceId: "cw", conversationId: "conv-A" }, + ]); + }); + + it("is a no-op if not subscribed", () => { + const s = initialState(); + const result = unsubscribe(s, "nope"); + expect(result.outgoing).toEqual([]); + expect(result.state).toBe(s); + }); }); describe("invoke", () => { - it("emits the correct InvokeMessage", () => { - const s = initialState(); - const result = invoke(s, "s1", "toggle", true); - expect(result.outgoing).toEqual([ - { type: "invoke", surfaceId: "s1", actionId: "toggle", payload: true }, - ]); - }); - - it("omits payload when not provided", () => { - const s = initialState(); - const result = invoke(s, "s1", "click"); - expect(result.outgoing).toEqual([ - { type: "invoke", surfaceId: "s1", actionId: "click", payload: undefined }, - ]); - }); - - it("does not mutate state", () => { - const s = initialState(); - const result = invoke(s, "s1", "a1"); - expect(result.state).toBe(s); - }); + it("emits the correct InvokeMessage", () => { + const s = initialState(); + const result = invoke(s, "s1", "toggle", true); + expect(result.outgoing).toEqual([ + { type: "invoke", surfaceId: "s1", actionId: "toggle", payload: true }, + ]); + }); + + it("omits payload when not provided", () => { + const s = initialState(); + const result = invoke(s, "s1", "click"); + expect(result.outgoing).toEqual([ + { type: "invoke", surfaceId: "s1", actionId: "click", payload: undefined }, + ]); + }); + + it("includes conversationId when provided", () => { + const s = initialState(); + const result = invoke(s, "cw", "cache-warming/set-interval", 120, "conv-A"); + expect(result.outgoing).toEqual([ + { + type: "invoke", + surfaceId: "cw", + actionId: "cache-warming/set-interval", + payload: 120, + conversationId: "conv-A", + }, + ]); + }); + + it("does not mutate state", () => { + const s = initialState(); + const result = invoke(s, "s1", "a1"); + expect(result.state).toBe(s); + }); }); diff --git a/src/core/protocol/reducer.ts b/src/core/protocol/reducer.ts index 992a918..976eb88 100644 --- a/src/core/protocol/reducer.ts +++ b/src/core/protocol/reducer.ts @@ -1,82 +1,143 @@ import type { - InvokeMessage, - SubscribeMessage, - SurfaceServerMessage, - UnsubscribeMessage, + InvokeMessage, + SubscribeMessage, + SurfaceServerMessage, + SurfaceSpec, + UnsubscribeMessage, } from "@dispatch/ui-contract"; import type { ProtocolResult, ProtocolState } from "./types"; /** The initial protocol state: empty catalog, no subscriptions, no error. */ export function initialState(): ProtocolState { - return { - catalog: [], - subscriptions: new Map(), - lastError: null, - }; + return { + catalog: [], + subscriptions: new Map(), + lastError: null, + }; +} + +// ── Message builders (respect exactOptionalPropertyTypes: omit `conversationId` +// entirely for a global subscription rather than setting it to `undefined`). ── + +function subMsg(surfaceId: string, conversationId: string | undefined): SubscribeMessage { + return conversationId === undefined + ? { type: "subscribe", surfaceId } + : { type: "subscribe", surfaceId, conversationId }; +} + +function unsubMsg(surfaceId: string, conversationId: string | undefined): UnsubscribeMessage { + return conversationId === undefined + ? { type: "unsubscribe", surfaceId } + : { type: "unsubscribe", surfaceId, conversationId }; +} + +/** + * Is an inbound spec/update (which echoes `echoedId`) current for the + * subscription whose desired scope is `desiredId`? A scoped surface echoes its + * conversationId, so it must match the one we last subscribed with; a GLOBAL + * surface echoes nothing (`undefined`) and is always current. + */ +function isCurrent(desiredId: string | undefined, echoedId: string | undefined): boolean { + return echoedId === undefined || echoedId === desiredId; } /** Fold an inbound server message into the next protocol state. */ export function applyServerMessage(state: ProtocolState, msg: SurfaceServerMessage): ProtocolState { - switch (msg.type) { - case "catalog": - return { ...state, catalog: msg.catalog }; + switch (msg.type) { + case "catalog": + return { ...state, catalog: msg.catalog }; - case "surface": { - const surfaceId = msg.spec.id; - if (!state.subscriptions.has(surfaceId)) return state; - const subs = new Map(state.subscriptions); - subs.set(surfaceId, msg.spec); - return { ...state, subscriptions: subs }; - } + case "surface": { + const sub = state.subscriptions.get(msg.spec.id); + if (sub === undefined) return state; + if (!isCurrent(sub.conversationId, msg.conversationId)) return state; + const subs = new Map(state.subscriptions); + subs.set(msg.spec.id, { conversationId: sub.conversationId, spec: msg.spec }); + return { ...state, subscriptions: subs }; + } - case "update": { - const surfaceId = msg.update.surfaceId; - if (!state.subscriptions.has(surfaceId)) return state; - const subs = new Map(state.subscriptions); - subs.set(surfaceId, msg.update.spec); - return { ...state, subscriptions: subs }; - } + case "update": { + const { surfaceId, spec, conversationId } = msg.update; + const sub = state.subscriptions.get(surfaceId); + if (sub === undefined) return state; + if (!isCurrent(sub.conversationId, conversationId)) return state; + const subs = new Map(state.subscriptions); + subs.set(surfaceId, { conversationId: sub.conversationId, spec }); + return { ...state, subscriptions: subs }; + } - case "error": - return { ...state, lastError: msg }; - } + case "error": + return { ...state, lastError: msg }; + } } /** - * Subscribe to a surface. Idempotent: if already subscribed, returns the same - * state with no outgoing message. + * Subscribe to a surface for a given conversation (omit `conversationId` for a + * GLOBAL surface / when no conversation is focused). + * + * - Not yet subscribed → emits one `subscribe`. + * - Already subscribed with the SAME scope → idempotent no-op. + * - Already subscribed with a DIFFERENT conversation (a re-scope on conversation + * switch) → emits `unsubscribe` for the old pair then `subscribe` for the new + * one, retaining the previous spec until the new one arrives (no flicker). */ -export function subscribe(state: ProtocolState, surfaceId: string): ProtocolResult { - if (state.subscriptions.has(surfaceId)) { - return { state, outgoing: [] }; - } - const subs = new Map(state.subscriptions); - subs.set(surfaceId, null); - const outgoing: SubscribeMessage = { type: "subscribe", surfaceId }; - return { state: { ...state, subscriptions: subs }, outgoing: [outgoing] }; +export function subscribe( + state: ProtocolState, + surfaceId: string, + conversationId?: string, +): ProtocolResult { + const existing = state.subscriptions.get(surfaceId); + if (existing !== undefined && existing.conversationId === conversationId) { + return { state, outgoing: [] }; + } + const subs = new Map(state.subscriptions); + const outgoing: (SubscribeMessage | UnsubscribeMessage)[] = []; + const priorSpec: SurfaceSpec | null = existing?.spec ?? null; + if (existing !== undefined) { + outgoing.push(unsubMsg(surfaceId, existing.conversationId)); + } + subs.set(surfaceId, { conversationId, spec: priorSpec }); + outgoing.push(subMsg(surfaceId, conversationId)); + return { state: { ...state, subscriptions: subs }, outgoing }; } /** - * Unsubscribe from a surface. Drops the local spec and emits one unsubscribe. - * If not subscribed, returns the same state with no outgoing. + * Unsubscribe from a surface. Drops the local subscription and emits one + * `unsubscribe` (for the conversation pair it was subscribed under). No-op if + * not subscribed. */ export function unsubscribe(state: ProtocolState, surfaceId: string): ProtocolResult { - if (!state.subscriptions.has(surfaceId)) { - return { state, outgoing: [] }; - } - const subs = new Map(state.subscriptions); - subs.delete(surfaceId); - const outgoing: UnsubscribeMessage = { type: "unsubscribe", surfaceId }; - return { state: { ...state, subscriptions: subs }, outgoing: [outgoing] }; + const existing = state.subscriptions.get(surfaceId); + if (existing === undefined) { + return { state, outgoing: [] }; + } + const subs = new Map(state.subscriptions); + subs.delete(surfaceId); + return { + state: { ...state, subscriptions: subs }, + outgoing: [unsubMsg(surfaceId, existing.conversationId)], + }; } -/** Invoke a field's action on a surface. Emits an InvokeMessage; no state change. */ +/** + * Invoke a field's action on a surface. Emits an InvokeMessage (carrying + * `conversationId` for a scoped surface); no state change. + */ export function invoke( - state: ProtocolState, - surfaceId: string, - actionId: string, - payload?: unknown, + state: ProtocolState, + surfaceId: string, + actionId: string, + payload?: unknown, + conversationId?: string, ): ProtocolResult { - const outgoing: InvokeMessage = { type: "invoke", surfaceId, actionId, payload }; - return { state, outgoing: [outgoing] }; + const outgoing: InvokeMessage = + conversationId === undefined + ? { type: "invoke", surfaceId, actionId, payload } + : { type: "invoke", surfaceId, actionId, payload, conversationId }; + return { state, outgoing: [outgoing] }; +} + +/** The current spec for a subscribed surface, or `null` if absent/unsubscribed. */ +export function getSurfaceSpec(state: ProtocolState, surfaceId: string): SurfaceSpec | null { + return state.subscriptions.get(surfaceId)?.spec ?? null; } diff --git a/src/core/protocol/types.ts b/src/core/protocol/types.ts index effec0d..6debb1d 100644 --- a/src/core/protocol/types.ts +++ b/src/core/protocol/types.ts @@ -1,22 +1,37 @@ import type { - SurfaceCatalog, - SurfaceClientMessage, - SurfaceErrorMessage, - SurfaceSpec, + SurfaceCatalog, + SurfaceClientMessage, + SurfaceErrorMessage, + SurfaceSpec, } from "@dispatch/ui-contract"; +/** + * One surface subscription's local state. + * + * `conversationId` is the conversation we last subscribed this surface WITH + * (`undefined` = subscribed globally, no conversation in focus). It is the + * "desired" scope: an inbound `surface`/`update` that echoes a DIFFERENT + * conversation is stale (we have since re-scoped) and is dropped. A GLOBAL + * surface ignores the id server-side and echoes none — that (`undefined` echo) + * is always accepted. `spec` is `null` until the first `surface` arrives. + */ +export interface Subscription { + readonly conversationId: string | undefined; + readonly spec: SurfaceSpec | null; +} + /** The client-side view of the surface protocol state. */ export interface ProtocolState { - /** The latest catalog received from the server (empty until first CatalogMessage). */ - readonly catalog: SurfaceCatalog; - /** Surfaces the client intends to be subscribed to; null = subscribed but no spec yet. */ - readonly subscriptions: ReadonlyMap<string, SurfaceSpec | null>; - /** The last error received from the server, if any. */ - readonly lastError: SurfaceErrorMessage | null; + /** The latest catalog received from the server (empty until first CatalogMessage). */ + readonly catalog: SurfaceCatalog; + /** Surfaces the client intends to be subscribed to, keyed by surfaceId. */ + readonly subscriptions: ReadonlyMap<string, Subscription>; + /** The last error received from the server, if any. */ + readonly lastError: SurfaceErrorMessage | null; } /** A state transition result: the next state plus any outgoing messages to send. */ export interface ProtocolResult { - readonly state: ProtocolState; - readonly outgoing: readonly SurfaceClientMessage[]; + readonly state: ProtocolState; + readonly outgoing: readonly SurfaceClientMessage[]; } diff --git a/src/core/wire/conformance.test.ts b/src/core/wire/conformance.test.ts index c0f276f..0d955d5 100644 --- a/src/core/wire/conformance.test.ts +++ b/src/core/wire/conformance.test.ts @@ -1,181 +1,280 @@ import type { ChatSendMessage, ConversationHistoryResponse } from "@dispatch/transport-contract"; -import type { AgentEvent, StoredChunk } from "@dispatch/wire"; +import type { AgentEvent, StepId, StoredChunk } from "@dispatch/wire"; import { describe, expect, it } from "vitest"; import { - assertAgentEventExhaustive, - assertChunkExhaustive, - assertWsClientMessageExhaustive, - assertWsServerMessageExhaustive, + assertAgentEventExhaustive, + assertChunkExhaustive, + assertWsClientMessageExhaustive, + assertWsServerMessageExhaustive, } from "./conformance"; describe("StoredChunk round-trips JSON", () => { - it("preserves shape through JSON serialize/deserialize", () => { - const original: StoredChunk = { - seq: 42, - role: "assistant", - chunk: { type: "text", text: "hello" }, - }; - const roundTripped: StoredChunk = JSON.parse(JSON.stringify(original)) as StoredChunk; - expect(roundTripped).toEqual(original); - expect(roundTripped.seq).toBe(42); - expect(roundTripped.role).toBe("assistant"); - expect(roundTripped.chunk.type).toBe("text"); - }); + it("preserves shape through JSON serialize/deserialize", () => { + const original: StoredChunk = { + seq: 42, + role: "assistant", + chunk: { type: "text", text: "hello" }, + }; + const roundTripped: StoredChunk = JSON.parse(JSON.stringify(original)) as StoredChunk; + expect(roundTripped).toEqual(original); + expect(roundTripped.seq).toBe(42); + expect(roundTripped.role).toBe("assistant"); + expect(roundTripped.chunk.type).toBe("text"); + }); }); describe("classifies every AgentEvent type", () => { - const samples: AgentEvent[] = [ - { type: "status", conversationId: "c1", status: "idle" }, - { type: "turn-start", conversationId: "c1", turnId: "t1" }, - { type: "text-delta", conversationId: "c1", turnId: "t1", delta: "hi" }, - { type: "reasoning-delta", conversationId: "c1", turnId: "t1", delta: "thinking" }, - { - type: "tool-call", - conversationId: "c1", - turnId: "t1", - toolCallId: "tc1", - toolName: "read", - input: {}, - }, - { - type: "tool-result", - conversationId: "c1", - turnId: "t1", - toolCallId: "tc1", - toolName: "read", - content: "ok", - isError: false, - }, - { - type: "tool-output", - conversationId: "c1", - turnId: "t1", - toolCallId: "tc1", - data: "out", - stream: "stdout", - }, - { - type: "usage", - conversationId: "c1", - turnId: "t1", - usage: { inputTokens: 10, outputTokens: 20 }, - }, - { type: "error", conversationId: "c1", turnId: "t1", message: "oops" }, - { type: "done", conversationId: "c1", turnId: "t1", reason: "complete" }, - { type: "turn-sealed", conversationId: "c1", turnId: "t1" }, - ]; + const samples: AgentEvent[] = [ + { type: "status", conversationId: "c1", status: "idle" }, + { type: "turn-start", conversationId: "c1", turnId: "t1" }, + { type: "user-message", conversationId: "c1", turnId: "t1", text: "hi" }, + { type: "text-delta", conversationId: "c1", turnId: "t1", delta: "hi" }, + { type: "reasoning-delta", conversationId: "c1", turnId: "t1", delta: "thinking" }, + { + type: "tool-call", + conversationId: "c1", + turnId: "t1", + toolCallId: "tc1", + toolName: "read", + input: {}, + stepId: "t1#0" as StepId, + }, + { + type: "tool-result", + conversationId: "c1", + turnId: "t1", + toolCallId: "tc1", + toolName: "read", + content: "ok", + isError: false, + stepId: "t1#0" as StepId, + }, + { + type: "tool-output", + conversationId: "c1", + turnId: "t1", + toolCallId: "tc1", + data: "out", + stream: "stdout", + }, + { + type: "usage", + conversationId: "c1", + turnId: "t1", + usage: { inputTokens: 10, outputTokens: 20 }, + }, + { + type: "step-complete", + conversationId: "c1", + turnId: "t1", + stepId: "t1#0" as StepId, + ttftMs: 300, + decodeMs: 700, + genTotalMs: 1000, + }, + { type: "error", conversationId: "c1", turnId: "t1", message: "oops" }, + { type: "done", conversationId: "c1", turnId: "t1", reason: "complete" }, + { type: "turn-sealed", conversationId: "c1", turnId: "t1" }, + { type: "steering", conversationId: "c1", turnId: "t1", text: "steer mid-turn" }, + { + type: "provider-retry", + conversationId: "c1", + turnId: "t1", + attempt: 0, + delayMs: 5000, + message: "HTTP 429: overloaded", + code: "429", + }, + ]; - it("returns a stable label for every AgentEvent.type variant", () => { - const labels = samples.map(assertAgentEventExhaustive); - expect(labels).toEqual([ - "status", - "turn-start", - "text-delta", - "reasoning-delta", - "tool-call", - "tool-result", - "tool-output", - "usage", - "error", - "done", - "turn-sealed", - ]); - }); + it("returns a stable label for every AgentEvent.type variant", () => { + const labels = samples.map(assertAgentEventExhaustive); + expect(labels).toEqual([ + "status", + "turn-start", + "user-message", + "text-delta", + "reasoning-delta", + "tool-call", + "tool-result", + "tool-output", + "usage", + "step-complete", + "error", + "done", + "turn-sealed", + "steering", + "provider-retry", + ]); + }); - it("covers all 11 AgentEvent variants", () => { - expect(samples).toHaveLength(11); - }); + it("covers all 15 AgentEvent variants", () => { + expect(samples).toHaveLength(15); + }); }); describe("classifies every Chunk type", () => { - it("returns a stable label for each Chunk.type variant", () => { - const chunks = [ - { type: "text" as const, text: "a" }, - { type: "thinking" as const, text: "b" }, - { type: "tool-call" as const, toolCallId: "tc", toolName: "n", input: null }, - { - type: "tool-result" as const, - toolCallId: "tc", - toolName: "n", - content: "c", - isError: false, - }, - { type: "error" as const, message: "e" }, - { type: "system" as const, text: "s" }, - ]; - const labels = chunks.map(assertChunkExhaustive); - expect(labels).toEqual(["text", "thinking", "tool-call", "tool-result", "error", "system"]); - }); + it("returns a stable label for each Chunk.type variant", () => { + const chunks = [ + { type: "text" as const, text: "a" }, + { type: "thinking" as const, text: "b" }, + { type: "tool-call" as const, toolCallId: "tc", toolName: "n", input: null }, + { + type: "tool-result" as const, + toolCallId: "tc", + toolName: "n", + content: "c", + isError: false, + }, + { type: "error" as const, message: "e" }, + { type: "system" as const, text: "s" }, + { type: "image" as const, url: "data:image/png;base64,AAAA", mimeType: "image/png" }, + ]; + const labels = chunks.map(assertChunkExhaustive); + expect(labels).toEqual([ + "text", + "thinking", + "tool-call", + "tool-result", + "error", + "system", + "image", + ]); + }); + + it("covers all 7 Chunk variants", () => { + // Keeps the exhaustive guard honest: a new Chunk.type variant must be added + // both here and to `assertChunkExhaustive` or the `satisfies never` errors. + expect([ + "text", + "thinking", + "tool-call", + "tool-result", + "error", + "system", + "image", + ] as const).toHaveLength(7); + }); }); describe("classifies every WsServerMessage type", () => { - it("returns a stable label for each variant", () => { - const msgs = [ - { type: "catalog" as const, catalog: [] }, - { type: "surface" as const, spec: { id: "s", region: "r", title: "S", fields: [] } }, - { - type: "update" as const, - update: { surfaceId: "s", spec: { id: "s", region: "r", title: "S", fields: [] } }, - }, - { type: "error" as const, message: "e" }, - { - type: "chat.delta" as const, - event: { type: "done" as const, conversationId: "c", turnId: "t", reason: "r" }, - }, - { type: "chat.error" as const, message: "e" }, - ]; - const labels = msgs.map(assertWsServerMessageExhaustive); - expect(labels).toEqual(["catalog", "surface", "update", "error", "chat.delta", "chat.error"]); - }); + it("returns a stable label for each variant", () => { + const msgs = [ + { type: "catalog" as const, catalog: [] }, + { type: "surface" as const, spec: { id: "s", region: "r", title: "S", fields: [] } }, + { + type: "update" as const, + update: { surfaceId: "s", spec: { id: "s", region: "r", title: "S", fields: [] } }, + }, + { type: "error" as const, message: "e" }, + { + type: "chat.delta" as const, + event: { type: "done" as const, conversationId: "c", turnId: "t", reason: "r" }, + }, + { type: "chat.error" as const, message: "e" }, + { type: "conversation.open" as const, conversationId: "c1", workspaceId: "w1" }, + { + type: "conversation.statusChanged" as const, + conversationId: "c1", + status: "active" as const, + workspaceId: "w1", + }, + { + type: "conversation.compacted" as const, + conversationId: "c1", + newConversationId: "c2", + messagesSummarized: 10, + messagesKept: 5, + }, + ]; + const labels = msgs.map(assertWsServerMessageExhaustive); + expect(labels).toEqual([ + "catalog", + "surface", + "update", + "error", + "chat.delta", + "chat.error", + "conversation.open", + "conversation.statusChanged", + "conversation.compacted", + ]); + }); }); describe("classifies every WsClientMessage type", () => { - it("returns a stable label for each variant", () => { - const msgs = [ - { type: "subscribe" as const, surfaceId: "s" }, - { type: "unsubscribe" as const, surfaceId: "s" }, - { type: "invoke" as const, surfaceId: "s", actionId: "a" }, - { type: "chat.send" as const, message: "hi" }, - ]; - const labels = msgs.map(assertWsClientMessageExhaustive); - expect(labels).toEqual(["subscribe", "unsubscribe", "invoke", "chat.send"]); - }); + it("returns a stable label for each variant", () => { + const msgs = [ + { type: "subscribe" as const, surfaceId: "s" }, + { type: "unsubscribe" as const, surfaceId: "s" }, + { type: "invoke" as const, surfaceId: "s", actionId: "a" }, + { type: "chat.send" as const, message: "hi" }, + { type: "chat.subscribe" as const, conversationId: "c1" }, + { type: "chat.unsubscribe" as const, conversationId: "c1" }, + { type: "chat.queue" as const, conversationId: "c1", text: "steer" }, + ]; + const labels = msgs.map(assertWsClientMessageExhaustive); + expect(labels).toEqual([ + "subscribe", + "unsubscribe", + "invoke", + "chat.send", + "chat.subscribe", + "chat.unsubscribe", + "chat.queue", + ]); + }); }); describe("ChatSendMessage shape is constructible", () => { - it("constructs a minimal ChatSendMessage", () => { - const msg: ChatSendMessage = { type: "chat.send", message: "hello" }; - expect(msg.type).toBe("chat.send"); - expect(msg.message).toBe("hello"); - }); + it("constructs a minimal ChatSendMessage", () => { + const msg: ChatSendMessage = { type: "chat.send", message: "hello" }; + expect(msg.type).toBe("chat.send"); + expect(msg.message).toBe("hello"); + }); + + it("constructs a full ChatSendMessage", () => { + const msg: ChatSendMessage = { + type: "chat.send", + conversationId: "c1", + message: "hello", + model: "default/gpt-4", + cwd: "/tmp", + }; + expect(msg.conversationId).toBe("c1"); + expect(msg.model).toBe("default/gpt-4"); + expect(msg.cwd).toBe("/tmp"); + }); - it("constructs a full ChatSendMessage", () => { - const msg: ChatSendMessage = { - type: "chat.send", - conversationId: "c1", - message: "hello", - model: "default/gpt-4", - cwd: "/tmp", - }; - expect(msg.conversationId).toBe("c1"); - expect(msg.model).toBe("default/gpt-4"); - expect(msg.cwd).toBe("/tmp"); - }); + it("constructs a ChatSendMessage with pasted images", () => { + const msg: ChatSendMessage = { + type: "chat.send", + conversationId: "c1", + message: "what's in this image?", + images: [ + { url: "data:image/png;base64,AAAA", mimeType: "image/png" }, + { url: "https://example.com/cat.jpg" }, + ], + }; + expect(msg.images).toHaveLength(2); + expect(msg.images?.[0]?.mimeType).toBe("image/png"); + expect(msg.images?.[1]?.mimeType).toBeUndefined(); + }); }); describe("ConversationHistoryResponse shape is constructible", () => { - it("constructs a response with chunks", () => { - const resp: ConversationHistoryResponse = { - chunks: [{ seq: 1, role: "user", chunk: { type: "text", text: "hi" } }], - latestSeq: 1, - }; - expect(resp.chunks).toHaveLength(1); - expect(resp.latestSeq).toBe(1); - }); + it("constructs a response with chunks", () => { + const resp: ConversationHistoryResponse = { + chunks: [{ seq: 1, role: "user", chunk: { type: "text", text: "hi" } }], + latestSeq: 1, + }; + expect(resp.chunks).toHaveLength(1); + expect(resp.latestSeq).toBe(1); + }); - it("constructs an empty (caught-up) response", () => { - const resp: ConversationHistoryResponse = { chunks: [], latestSeq: 5 }; - expect(resp.chunks).toHaveLength(0); - expect(resp.latestSeq).toBe(5); - }); + it("constructs an empty (caught-up) response", () => { + const resp: ConversationHistoryResponse = { chunks: [], latestSeq: 5 }; + expect(resp.chunks).toHaveLength(0); + expect(resp.latestSeq).toBe(5); + }); }); diff --git a/src/core/wire/conformance.ts b/src/core/wire/conformance.ts index 5d75a60..bfa67dc 100644 --- a/src/core/wire/conformance.ts +++ b/src/core/wire/conformance.ts @@ -7,54 +7,64 @@ import type { AgentEvent, Chunk } from "@dispatch/wire"; * default branch becomes reachable → TypeScript error at build time. */ export function assertAgentEventExhaustive(event: AgentEvent): string { - switch (event.type) { - case "status": - return "status"; - case "turn-start": - return "turn-start"; - case "text-delta": - return "text-delta"; - case "reasoning-delta": - return "reasoning-delta"; - case "tool-call": - return "tool-call"; - case "tool-result": - return "tool-result"; - case "tool-output": - return "tool-output"; - case "usage": - return "usage"; - case "error": - return "error"; - case "done": - return "done"; - case "turn-sealed": - return "turn-sealed"; - default: - return event satisfies never; - } + switch (event.type) { + case "status": + return "status"; + case "turn-start": + return "turn-start"; + case "user-message": + return "user-message"; + case "text-delta": + return "text-delta"; + case "reasoning-delta": + return "reasoning-delta"; + case "tool-call": + return "tool-call"; + case "tool-result": + return "tool-result"; + case "tool-output": + return "tool-output"; + case "usage": + return "usage"; + case "error": + return "error"; + case "done": + return "done"; + case "turn-sealed": + return "turn-sealed"; + case "step-complete": + return "step-complete"; + case "steering": + return "steering"; + case "provider-retry": + return "provider-retry"; + default: + return event satisfies never; + } } /** * Compile-time exhaustiveness guard for `Chunk.type`. */ export function assertChunkExhaustive(chunk: Chunk): string { - switch (chunk.type) { - case "text": - return "text"; - case "thinking": - return "thinking"; - case "tool-call": - return "tool-call"; - case "tool-result": - return "tool-result"; - case "error": - return "error"; - case "system": - return "system"; - default: - return chunk satisfies never; - } + switch (chunk.type) { + case "text": + return "text"; + case "thinking": + return "thinking"; + case "tool-call": + return "tool-call"; + case "tool-result": + return "tool-result"; + case "error": + return "error"; + case "system": + return "system"; + case "image": + return "image"; + default: + return chunk satisfies never; + } } /** @@ -62,22 +72,28 @@ export function assertChunkExhaustive(chunk: Chunk): string { * Covers both surface ops and chat ops. */ export function assertWsServerMessageExhaustive(msg: WsServerMessage): string { - switch (msg.type) { - case "catalog": - return "catalog"; - case "surface": - return "surface"; - case "update": - return "update"; - case "error": - return "error"; - case "chat.delta": - return "chat.delta"; - case "chat.error": - return "chat.error"; - default: - return msg satisfies never; - } + switch (msg.type) { + case "catalog": + return "catalog"; + case "surface": + return "surface"; + case "update": + return "update"; + case "error": + return "error"; + case "chat.delta": + return "chat.delta"; + case "chat.error": + return "chat.error"; + case "conversation.open": + return "conversation.open"; + case "conversation.statusChanged": + return "conversation.statusChanged"; + case "conversation.compacted": + return "conversation.compacted"; + default: + return msg satisfies never; + } } /** @@ -85,16 +101,22 @@ export function assertWsServerMessageExhaustive(msg: WsServerMessage): string { * Covers both surface ops and chat ops. */ export function assertWsClientMessageExhaustive(msg: WsClientMessage): string { - switch (msg.type) { - case "subscribe": - return "subscribe"; - case "unsubscribe": - return "unsubscribe"; - case "invoke": - return "invoke"; - case "chat.send": - return "chat.send"; - default: - return msg satisfies never; - } + switch (msg.type) { + case "subscribe": + return "subscribe"; + case "unsubscribe": + return "unsubscribe"; + case "invoke": + return "invoke"; + case "chat.send": + return "chat.send"; + case "chat.subscribe": + return "chat.subscribe"; + case "chat.unsubscribe": + return "chat.unsubscribe"; + case "chat.queue": + return "chat.queue"; + default: + return msg satisfies never; + } } diff --git a/src/core/wire/index.ts b/src/core/wire/index.ts index ae6b3e6..d4215cf 100644 --- a/src/core/wire/index.ts +++ b/src/core/wire/index.ts @@ -1,6 +1,6 @@ export { - assertAgentEventExhaustive, - assertChunkExhaustive, - assertWsClientMessageExhaustive, - assertWsServerMessageExhaustive, + assertAgentEventExhaustive, + assertChunkExhaustive, + assertWsClientMessageExhaustive, + assertWsServerMessageExhaustive, } from "./conformance"; |