diff options
Diffstat (limited to 'src/core')
32 files changed, 4113 insertions, 597 deletions
diff --git a/src/core/chunks/groups.test.ts b/src/core/chunks/groups.test.ts index c8b0fa2..1196590 100644 --- a/src/core/chunks/groups.test.ts +++ b/src/core/chunks/groups.test.ts @@ -3,16 +3,16 @@ 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 => ({ +const text = (seq: number, role: Role, t: string): RenderedChunk => ({ seq, role, chunk: { type: "text", text: t }, - provisional, + provisional: false, }); - const call = (seq: number, id: string, stepId?: string, provisional = false): RenderedChunk => ({ seq, role: "assistant", + provisional, chunk: { type: "tool-call", toolCallId: id, @@ -20,12 +20,11 @@ const call = (seq: number, id: string, stepId?: string, provisional = false): Re input: { id }, ...(stepId !== undefined ? { stepId: stepId as StepId } : {}), }, - provisional, }); - const result = (seq: number, id: string, stepId?: string, provisional = false): RenderedChunk => ({ seq, role: "tool", + provisional, chunk: { type: "tool-result", toolCallId: id, @@ -34,92 +33,142 @@ const result = (seq: number, id: string, stepId?: string, provisional = false): isError: false, ...(stepId !== undefined ? { stepId: stepId as StepId } : {}), }, - provisional, }); +function toolGroups(rows: readonly RenderedChunk[]) { + return groupRenderedChunks(rows).map((g) => { + if (g.kind !== "tool") throw new Error("expected a tool group"); + return g; + }); +} + 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("passes non-tool chunks through unchanged and in order", () => { + const rows = [text(1, "user", "hi"), text(2, "assistant", "hello")]; + expect(groupRenderedChunks(rows)).toEqual(rows.map((chunk) => ({ kind: "single", chunk }))); }); - - 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"), + it("pairs a single call and result into one group", () => { + const c = call(1, "a", "s1"); + const r = result(2, "a", "s1"); + expect(toolGroups([c, r])).toEqual([ + { + kind: "tool", + entry: { toolCallId: "a", toolName: "tool-a", call: c.chunk, result: r.chunk }, + stepId: "s1", + provisional: false, + }, ]); - 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([ + it("renders parallel calls independently in call order, with out-of-order results", () => { + const groups = toolGroups([ call(1, "a", "s1"), call(2, "b", "s1"), - result(3, "a", "s1"), - result(4, "b", "s1"), + result(3, "b", "s1"), + result(4, "a", "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"); + expect(groups.map((g) => g.entry.toolCallId)).toEqual(["a", "b"]); + expect(groups.map((g) => g.entry.result?.content)).toEqual(["result-a", "result-b"]); }); - - it("positions the batch at the first call and keeps surrounding chunks in order", () => { + it("pairs legacy chunks without stepId and mismatching stepIds by toolCallId", () => { + const groups = toolGroups([ + call(1, "a"), + result(2, "a"), + call(3, "b", "s1"), + result(4, "b", "s2"), + ]); + expect(groups).toHaveLength(2); + expect(groups[0]?.entry.result).not.toBeNull(); + expect(groups[1]?.stepId).toBe("s1"); + }); + it("keeps surrounding text at its position and absorbs results", () => { 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"), + text(3, "assistant", "between"), + call(4, "b", "s1"), + result(5, "a"), + result(6, "b"), + text(7, "assistant", "after"), ]); - expect(groups.map((g) => g.kind)).toEqual(["single", "tool-batch", "single"]); + expect(groups.map((g) => g.kind)).toEqual(["single", "tool", "single", "tool", "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("keeps pending calls and orphan results", () => { + const groups = toolGroups([call(1, "pending"), result(2, "orphan", "s2", true)]); + expect(groups[0]?.entry.result).toBeNull(); + expect(groups[1]?.entry.call).toBeNull(); + expect(groups[1]?.entry.toolName).toBe("tool-orphan"); + expect(groups[1]?.provisional).toBe(true); + expect(groups[1]?.stepId).toBe("s2"); }); - - 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"), + it.each([ + [true, false], + [false, true], + [false, false], + ])("combines call/result provisional flags (%s, %s)", (cp, rp) => { + expect(toolGroups([call(1, "a", "s1", cp), result(2, "a", "s1", rp)])[0]?.provisional).toBe( + cp || rp, + ); + }); + it("prefers call provenance, falling back to result provenance", () => { + const groups = toolGroups([ + { ...call(1, "a", "s1"), turnId: "t1" }, + { ...result(2, "a", "s2"), turnId: "t2" }, + call(3, "b"), + { ...result(4, "b", "s3"), turnId: "t3" }, + ]); + expect(groups.map((g) => [g.stepId, g.turnId])).toEqual([ + ["s1", "t1"], + ["s3", "t3"], + ]); + }); + it("dedupes calls/results across steps, first occurrence wins", () => { + const first = call(1, "a", "s1", true); + const firstResult = result(4, "a", "s1"); + const groups = toolGroups([ + first, + call(2, "a", "s2"), + call(3, "b"), + firstResult, + result(5, "a", "s2"), + result(6, "b"), + result(7, "b"), ]); - 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"); + expect(groups).toHaveLength(2); + expect(groups[0]?.entry.call).toBe(first.chunk); + expect(groups[0]?.entry.result).toBe(firstResult.chunk); + expect(groups[0]?.provisional).toBe(true); + }); + it("dedupes orphan results too", () => { + expect(toolGroups([result(1, "a"), result(2, "a")])).toHaveLength(1); + }); + it("positions a card at its call even when its result came first", () => { + const c = call(3, "a"); + const groups = groupRenderedChunks([result(1, "a"), text(2, "assistant", "before call"), c]); + expect(groups.map((g) => g.kind)).toEqual(["single", "tool"]); }); }); + +it("keeps windowed committed orphan calls in Show earlier even while another payload recovers", () => { + const groups = groupRenderedChunks( + [ + { + seq: 50, + role: "tool", + provisional: false, + chunk: { + type: "tool-result", + toolCallId: "older", + toolName: "run_shell", + content: "output", + isError: false, + }, + }, + ], + true, + ); + expect(groups[0]).toMatchObject({ kind: "tool", entry: { call: null } }); + if (groups[0]?.kind === "tool") expect(groups[0].entry.recovering).toBeUndefined(); +}); diff --git a/src/core/chunks/groups.ts b/src/core/chunks/groups.ts index 53a2873..6e0f8ea 100644 --- a/src/core/chunks/groups.ts +++ b/src/core/chunks/groups.ts @@ -1,95 +1,75 @@ 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; +/** One tool call and its response. Null call = unloaded history; null result = pending. */ +export interface ToolEntry { + readonly recovering?: boolean; + readonly toolCallId: string; + readonly toolName: string; + readonly call: ToolCallChunk | null; 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 kind: "tool"; + readonly entry: ToolEntry; readonly provisional: boolean; + readonly stepId?: string; + readonly turnId?: string; }; /** - * 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. + * One independent group per call, at the call's position, absorbing its result + * by toolCallId (including legacy chunks without stepId). Orphan results stay + * visible at their own position. First occurrence wins for duplicate ids, so + * the stable t<toolCallId> render keys are always unique, even on corrupt replay. + * Pure: input → output; no DOM, effects, or Svelte. */ -export function groupRenderedChunks(rendered: readonly RenderedChunk[]): readonly RenderGroup[] { - // 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>(); +export function groupRenderedChunks( + rendered: readonly RenderedChunk[], + recovering = false, +): readonly RenderGroup[] { + const calls = new Map<string, RenderedChunk>(); + const results = new Map<string, RenderedChunk>(); 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); - } + const c = rc.chunk; + if (c.type === "tool-call" && !calls.has(c.toolCallId)) calls.set(c.toolCallId, rc); + if (c.type === "tool-result" && !results.has(c.toolCallId)) results.set(c.toolCallId, rc); } - // 3. Emit groups in stream order; each batch lands at its first call. const groups: RenderGroup[] = []; - const emittedSteps = new Set<string>(); + const emitted = 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 }); + const c = rc.chunk; + if (c.type !== "tool-call" && c.type !== "tool-result") { + groups.push({ kind: "single", chunk: rc }); continue; } - - if (chunk.type === "tool-result" && batchCallIds.has(chunk.toolCallId)) { - continue; // absorbed into its batch - } - - groups.push({ kind: "single", chunk: rc }); + if (emitted.has(c.toolCallId)) continue; + // A matching call owns the card's position, even if the result came first. + if (c.type === "tool-result" && calls.has(c.toolCallId)) continue; + emitted.add(c.toolCallId); + const callRow = calls.get(c.toolCallId); + const resultRow = results.get(c.toolCallId); + const call = callRow?.chunk.type === "tool-call" ? callRow.chunk : null; + const result = resultRow?.chunk.type === "tool-result" ? resultRow.chunk : null; + const stepId = call?.stepId ?? result?.stepId; + const turnId = callRow?.turnId ?? resultRow?.turnId; + groups.push({ + kind: "tool", + entry: { + toolCallId: c.toolCallId, + toolName: call?.toolName ?? c.toolName, + call, + result, + ...(recovering && call === null && resultRow?.provisional ? { recovering: true } : {}), + }, + provisional: Boolean(callRow?.provisional || resultRow?.provisional), + ...(stepId !== undefined ? { stepId } : {}), + ...(turnId !== undefined ? { turnId } : {}), + }); } - return groups; } diff --git a/src/core/chunks/index.ts b/src/core/chunks/index.ts index bdd6ce4..03d687c 100644 --- a/src/core/chunks/index.ts +++ b/src/core/chunks/index.ts @@ -1,4 +1,4 @@ -export type { RenderGroup, ToolBatchEntry } from "./groups"; +export type { RenderGroup, ToolEntry } from "./groups"; export { groupRenderedChunks } from "./groups"; export { resolveImageUrl } from "./image-url"; export { @@ -7,6 +7,8 @@ export { clearGenerating, foldEvent, initialState, + markHistoryRecovering, + removeLocalEcho, } from "./reducer"; export type { ProviderRetryView } from "./retry-banner"; export { formatRetryDelay, viewProviderRetry } from "./retry-banner"; diff --git a/src/core/chunks/local-echo.test.ts b/src/core/chunks/local-echo.test.ts new file mode 100644 index 0000000..565017c --- /dev/null +++ b/src/core/chunks/local-echo.test.ts @@ -0,0 +1,45 @@ +import { describe, expect, it } from "vitest"; +import { appendUserMessage, applyHistory, foldEvent, initialState, removeLocalEcho } from "./index"; + +describe("request-local optimistic echo", () => { + it("retains identity through turn-start tags without conflating identical sends", () => { + let state = appendUserMessage(initialState(), "same", [{ url: "data:first" }], "r1"); + state = appendUserMessage(state, "same", [{ url: "data:second" }], "r2"); + state = foldEvent(state, { type: "turn-start", conversationId: "c1", turnId: "t1" }); + state = removeLocalEcho(state, "r1"); + expect(state.provisional.map((p) => p.localRequestId)).toEqual(["r2", "r2"]); + expect(state.provisional.map((p) => p.chunk)).toEqual([ + { type: "text", text: "same" }, + { type: "image", url: "data:second" }, + ]); + expect(state.generating).toBe(true); + expect(removeLocalEcho(state, "r1")).toBe(state); + }); + + it("a live user-message confirms only its deduplicated text, not every identical local send", () => { + let state = appendUserMessage(initialState(), "same", [{ url: "data:first" }], "r1"); + state = appendUserMessage(state, "same", [{ url: "data:second" }], "r2"); + state = foldEvent(state, { type: "turn-start", conversationId: "c1", turnId: "t1" }); + state = foldEvent(state, { + type: "user-message", + conversationId: "c1", + turnId: "t1", + text: "same", + }); + state = removeLocalEcho(removeLocalEcho(state, "r1"), "r2"); + expect(state.provisional).toEqual([ + { role: "user", turnId: "t1", opening: true, chunk: { type: "text", text: "same" } }, + ]); + }); + + it("does not remove committed rows even if their content is identical", () => { + let state = appendUserMessage(initialState(), "same", undefined, "r1"); + state = applyHistory(state, [{ seq: 1, role: "user", chunk: { type: "text", text: "same" } }]); + const committed = state.committed; + state = removeLocalEcho(state, "r1"); + expect(state.committed).toBe(committed); + expect(state.provisional).toEqual([]); + expect(removeLocalEcho(state, "r1")).toBe(state); + expect(removeLocalEcho(initialState(), "r1")).toEqual(initialState()); + }); +}); diff --git a/src/core/chunks/reconcile.test.ts b/src/core/chunks/reconcile.test.ts new file mode 100644 index 0000000..3d90a51 --- /dev/null +++ b/src/core/chunks/reconcile.test.ts @@ -0,0 +1,143 @@ +import type { AgentEvent, StepId, StoredChunk } from "@dispatch/wire"; +import { describe, expect, it } from "vitest"; +import { applyHistory, foldEvent, initialState, markHistoryRecovering } from "./reducer"; +import { selectChunks } from "./selectors"; +import type { TranscriptState } from "./types"; + +const stepId = "t1#1" as StepId; +function fold(s: TranscriptState, e: AgentEvent) { + return foldEvent(s, e); +} +const call = (toolCallId = "a", turnId = "t1", step = stepId): AgentEvent => ({ + type: "tool-call", + conversationId: "c1", + turnId, + stepId: step, + toolCallId, + toolName: "run_shell", + input: "partial", +}); +const result = (toolCallId = "a"): AgentEvent => ({ + type: "tool-result", + conversationId: "c1", + turnId: "t1", + stepId, + toolCallId, + toolName: "run_shell", + content: "partial", + isError: false, +}); +const stored: StoredChunk[] = [ + { seq: 50, role: "assistant", turnId: "t1", chunk: { type: "text", text: "full" } }, + { + seq: 51, + role: "assistant", + turnId: "t1", + chunk: { + type: "tool-call", + stepId, + toolCallId: "a", + toolName: "run_shell", + input: { command: "full" }, + }, + }, + { + seq: 52, + role: "tool", + turnId: "t1", + chunk: { + type: "tool-result", + stepId, + toolCallId: "a", + toolName: "run_shell", + content: "full result", + isError: false, + }, + }, +]; + +describe("history recovery identities", () => { + it.each([ + "before-live", + "after-live", + ])("reconciles committed tool/text identity %s, not payload equality", (order) => { + let s = markHistoryRecovering(initialState(), "t1"); + if (order === "before-live") s = applyHistory(s, stored); + s = fold(s, { type: "text-delta", conversationId: "c1", turnId: "t1", delta: "short" }); + s = fold(s, call()); + s = fold(s, result()); + if (order === "after-live") s = applyHistory(s, stored); + expect(selectChunks(s).map((c) => c.chunk)).toEqual(stored.map((c) => c.chunk)); + // Idempotent refresh; different step's identical text must remain live. + s = fold(s, { type: "text-delta", conversationId: "c1", turnId: "t1", delta: "full" }); + s = applyHistory(s, stored); + expect(selectChunks(s).at(-1)).toMatchObject({ + provisional: true, + streaming: true, + chunk: { type: "text", text: "full" }, + }); + }); + + it("does not conflate call/result, another turn, or another step's tool identity", () => { + let s = markHistoryRecovering(initialState(), "t1"); + s = fold(s, call("a", "other")); + s = fold(s, call("a", "t1", "other-step" as StepId)); + s = fold(s, result("a")); + s = applyHistory(s, stored); + expect(s.provisional).toHaveLength(2); + expect(s.provisional.map((p) => p.turnId)).toEqual(["other", "t1"]); + expect(s.provisional.every((p) => p.chunk.type === "tool-call")).toBe(true); + }); + + it("terminal sync removes only that turn and keeps optimistic/unrelated/carried content", () => { + let s = markHistoryRecovering(initialState(), "t1"); + s = fold(s, call()); + s = fold(s, { type: "turn-sealed", conversationId: "c1", turnId: "t1" }); + s = fold(s, { type: "turn-start", conversationId: "c1", turnId: "t2" }); + s = fold(s, { type: "text-delta", conversationId: "c1", turnId: "t2", delta: "queued tail" }); + s = applyHistory(s, stored, ["t1"]); + expect(s.generating).toBe(true); + expect(s.currentTurnId).toBe("t2"); + expect(s.accumulating?.text).toBe("queued tail"); + expect(s.historyRecoveryTurnIds).toEqual([]); + }); + + it("does not resurrect watermark-hidden history; older content stays available to Show earlier", () => { + let s = markHistoryRecovering({ ...initialState(), hiddenBeforeSeq: 51 }, "t1"); + s = fold(s, call()); + s = applyHistory(s, stored); + expect(s.committed.map((c) => c.seq)).toEqual([51, 52]); + expect(s.provisional).toEqual([]); + expect(s.hiddenBeforeSeq).toBe(51); + }); +}); + +it("keeps identical Stop→carry openings as distinct turn identities", () => { + let s = markHistoryRecovering(initialState(), "t1"); + s = fold(s, { type: "user-message", conversationId: "c1", turnId: "t1", text: "same" }); + s = fold(s, { type: "turn-ended", conversationId: "c1", turnId: "t1", reason: "aborted" }); + s = fold(s, { type: "user-message", conversationId: "c1", turnId: "t2", text: "same" }); + s = fold(s, { type: "turn-start", conversationId: "c1", turnId: "t2" }); + s = applyHistory( + s, + [{ seq: 1, role: "user", turnId: "t1", chunk: { type: "text", text: "same" } }], + ["t1"], + ); + expect(selectChunks(s).map((c) => [c.turnId, c.provisional])).toEqual([ + ["t1", false], + ["t2", true], + ]); + expect(s.generating).toBe(true); +}); + +it("reconciles steering by its completed tool batch, not repeated user text", () => { + let s = markHistoryRecovering(initialState(), "t1"); + s = fold(s, result()); + s = fold(s, { type: "steering", conversationId: "c1", turnId: "t1", text: "short" }); + s = applyHistory(s, [ + ...stored, + { seq: 53, role: "user", turnId: "t1", chunk: { type: "text", text: "full steering" } }, + ]); + expect(s.provisional).toEqual([]); + expect(selectChunks(s).at(-1)?.chunk).toEqual({ type: "text", text: "full steering" }); +}); diff --git a/src/core/chunks/reconcile.ts b/src/core/chunks/reconcile.ts new file mode 100644 index 0000000..277d25b --- /dev/null +++ b/src/core/chunks/reconcile.ts @@ -0,0 +1,54 @@ +import type { StoredChunk } from "@dispatch/wire"; +import type { ProvisionalChunk, TranscriptState } from "./types"; + +/** + * A persisted tool call proves its entire assistant step committed atomically. + * Text/thinking wire chunks have no stepId: their provisional provenance is + * learned from tool-call/step-complete, never from text equality. This also + * handles a window starting midway through the step and an omitted prefix. + * Also usable when HTTP finished before replay/live flush. + */ +export function historyBackedProvisional( + state: TranscriptState, + committed: readonly StoredChunk[] = state.committed, +): ReadonlySet<ProvisionalChunk> { + const backed = new Set<ProvisionalChunk>(); + if (!state.historyRecoveryTurnIds?.length) return backed; + const steps = new Set<string>(); + const steeringAnchors = new Set<string>(); + let lastResult: StoredChunk | undefined; + const toolRows = new Map<string, Set<string | undefined>>(); + for (const row of committed) { + const c = row.chunk; + if (row.role === "assistant" || row.turnId !== lastResult?.turnId) lastResult = undefined; + if (c.type === "tool-result") lastResult = row; + if (row.role === "user" && c.type === "text" && lastResult?.chunk.type === "tool-result") { + steeringAnchors.add(`${row.turnId}\0${lastResult.chunk.toolCallId}`); + } + if (c.type === "tool-call" || c.type === "tool-result") { + const key = `${row.turnId}\0${c.type}\0${c.toolCallId}`; + const ids = toolRows.get(key) ?? new Set<string | undefined>(); + ids.add(c.stepId); + toolRows.set(key, ids); + if (c.type === "tool-call" && c.stepId !== undefined) steps.add(`${row.turnId}\0${c.stepId}`); + } + } + for (const p of state.provisional) { + if (p.turnId === undefined || !state.historyRecoveryTurnIds?.includes(p.turnId)) continue; + const c = p.chunk; + if (c.type === "tool-call" || c.type === "tool-result") { + const ids = toolRows.get(`${p.turnId}\0${c.type}\0${c.toolCallId}`); + if (ids !== undefined && (c.stepId === undefined || ids.has(undefined) || ids.has(c.stepId))) + backed.add(p); + } else if (p.afterToolCallId !== undefined && p.role === "user") { + if (steeringAnchors.has(`${p.turnId}\0${p.afterToolCallId}`)) backed.add(p); + } else if (p.opening && c.type === "text") { + // The opening is the first user row of this turn; steering is NOT opening. + const first = committed.find((row) => row.turnId === p.turnId); + if (first?.role === "user" && first.chunk.type === "text") backed.add(p); + } else if (p.stepId !== undefined && p.role === "assistant") { + if (steps.has(`${p.turnId}\0${p.stepId}`)) backed.add(p); + } + } + return backed; +} diff --git a/src/core/chunks/reducer.test.ts b/src/core/chunks/reducer.test.ts index 058552e..10d662e 100644 --- a/src/core/chunks/reducer.test.ts +++ b/src/core/chunks/reducer.test.ts @@ -7,6 +7,7 @@ import type { TurnInputEvent, TurnProviderRetryEvent, TurnReasoningDeltaEvent, + TurnReasoningEndEvent, TurnSealedEvent, TurnStartEvent, TurnSteeringEvent, @@ -45,6 +46,12 @@ const reasoningDelta = (turnId: string, delta: string): TurnReasoningDeltaEvent delta, }); +const reasoningEnd = (turnId: string): TurnReasoningEndEvent => ({ + type: "reasoning-end", + conversationId: "c1", + turnId, +}); + const toolCall = ( turnId: string, toolCallId: string, @@ -226,7 +233,7 @@ describe("clearGenerating", () => { 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.accumulating).toEqual({ kind: "text", text: "partial", turnId: "t1" }); expect(cleared.currentTurnId).toBe("t1"); }); }); @@ -236,7 +243,7 @@ describe("foldEvent — text-delta", () => { let s = initialState(); s = foldEvent(s, turnStart("t1")); s = foldEvent(s, textDelta("t1", "hello")); - expect(s.accumulating).toEqual({ kind: "text", text: "hello" }); + expect(s.accumulating).toEqual({ kind: "text", text: "hello", turnId: "t1" }); expect(s.provisional).toEqual([]); }); @@ -245,7 +252,7 @@ describe("foldEvent — text-delta", () => { 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.accumulating).toEqual({ kind: "text", text: "hello world", turnId: "t1" }); expect(s.provisional).toEqual([]); }); @@ -254,7 +261,7 @@ describe("foldEvent — text-delta", () => { 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.accumulating).toEqual({ kind: "text", text: "answer", turnId: "t1" }); expect(s.provisional).toHaveLength(1); expect(s.provisional[0]?.chunk).toEqual({ type: "thinking", text: "thinking..." }); expect(s.provisional[0]?.role).toBe("assistant"); @@ -266,7 +273,7 @@ describe("foldEvent — reasoning-delta", () => { let s = initialState(); s = foldEvent(s, turnStart("t1")); s = foldEvent(s, reasoningDelta("t1", "hmm")); - expect(s.accumulating).toEqual({ kind: "thinking", text: "hmm" }); + expect(s.accumulating).toEqual({ kind: "thinking", text: "hmm", turnId: "t1" }); }); it("successive reasoning-deltas extend the same chunk", () => { @@ -274,7 +281,73 @@ describe("foldEvent — reasoning-delta", () => { 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" }); + expect(s.accumulating).toEqual({ kind: "thinking", text: "hmm ok", turnId: "t1" }); + }); +}); + +describe("foldEvent — reasoning-end (thinking block boundary)", () => { + it("closes each block so consecutive deltas become separate chunks", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, reasoningDelta("t1", "a")); + s = foldEvent(s, reasoningEnd("t1")); + s = foldEvent(s, reasoningDelta("t1", "b")); + s = foldEvent(s, reasoningEnd("t1")); + expect(s.provisional.map((p) => p.chunk)).toEqual([ + { type: "thinking", text: "a" }, + { type: "thinking", text: "b" }, + ]); + expect(s.accumulating).toBeNull(); + }); + + it("still merges legacy deltas with no boundary (no reasoning-end)", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, reasoningDelta("t1", "a")); + s = foldEvent(s, reasoningDelta("t1", "b")); + expect(s.accumulating).toEqual({ kind: "thinking", text: "ab", turnId: "t1" }); + }); + + it("end with nothing open appends an empty thinking chunk for the turn", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, reasoningEnd("t1")); + expect(s.provisional).toHaveLength(1); + expect(s.provisional[0]?.chunk).toEqual({ type: "thinking", text: "" }); + expect(s.provisional[0]?.turnId).toBe("t1"); + expect(s.accumulating).toBeNull(); + }); + + it("text-delta then end flushes the text, then appends an empty thinking chunk", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "x")); + s = foldEvent(s, reasoningEnd("t1")); + expect(s.provisional.map((p) => p.chunk)).toEqual([ + { type: "text", text: "x" }, + { type: "thinking", text: "" }, + ]); + }); + + it("end for another turn flushes the open thinking and appends an empty one", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, reasoningDelta("t1", "open")); + s = foldEvent(s, reasoningEnd("t2")); + expect(s.provisional.map((p) => [p.turnId, p.chunk])).toEqual([ + ["t1", { type: "thinking", text: "open" }], + ["t2", { type: "thinking", text: "" }], + ]); + expect(s.accumulating).toBeNull(); + }); + + it("clears a pending provider-retry banner", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, providerRetry("t1", 0, 5000)); + expect(selectProviderRetry(s)).not.toBeNull(); + s = foldEvent(s, reasoningEnd("t1")); + expect(selectProviderRetry(s)).toBeNull(); }); }); @@ -531,7 +604,7 @@ describe("clearGenerating also clears a stale provider-retry banner (reconnect)" 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(cleared.accumulating).toEqual({ kind: "text", text: "partial", turnId: "t1" }); expect(selectProviderRetry(cleared)).toBeNull(); }); }); @@ -859,7 +932,7 @@ describe("appendUserMessage", () => { let s = initialState(); s = foldEvent(s, turnStart("t1")); s = foldEvent(s, textDelta("t1", "partial")); - expect(s.accumulating).toEqual({ kind: "text", text: "partial" }); + expect(s.accumulating).toEqual({ kind: "text", text: "partial", turnId: "t1" }); s = appendUserMessage(s, "user msg"); expect(s.accumulating).toBeNull(); @@ -1022,3 +1095,115 @@ describe("applyHistory — multi-chunk image echo is superseded by committed", ( expect(msgs[0]?.chunks[1]).toEqual({ type: "image", url: PNG, mimeType: "image/png" }); }); }); + +describe("turnId provenance on live chunks", () => { + it("tags every live chunk with its event's turnId, and the optimistic echo on turn-start", () => { + let s = initialState(); + s = appendUserMessage(s, "q", [{ url: "data:image/png;base64,AA", mimeType: "image/png" }]); + expect(selectChunks(s).map((c) => c.turnId)).toEqual([undefined, undefined]); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "a")); + const chunks = selectChunks(s); + expect(chunks.map((c) => c.turnId)).toEqual(["t1", "t1", "t1"]); + }); + + it("turn-start does not re-tag an echo already owned by an earlier turn", () => { + let s = initialState(); + s = appendUserMessage(s, "q1"); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "a1")); + s = foldEvent(s, turnStart("t2")); + expect(selectChunks(s).map((c) => c.turnId)).toEqual(["t1", "t1"]); + }); + + it("committed chunks expose StoredChunk.turnId", () => { + const s = applyHistory(initialState(), [ + { seq: 1, role: "user", chunk: { type: "text", text: "q" }, turnId: "t9" }, + { seq: 2, role: "assistant", chunk: { type: "text", text: "a" } }, + ]); + expect(selectChunks(s).map((c) => c.turnId)).toEqual(["t9", undefined]); + }); +}); + +describe("foldEvent — additive wire compatibility", () => { + it.each(["future-event"])("preserves transcript identity for %s", (type) => { + const state = foldEvent(foldEvent(initialState(), turnStart("t1")), textDelta("t1", "partial")); + const before = structuredClone(state); + const event = { type, conversationId: "c1", turnId: "t1", reason: "sealed" }; + expect(foldEvent(state, event as Parameters<typeof foldEvent>[1])).toBe(state); + expect(state).toEqual(before); + }); +}); + +describe("foldEvent — authoritative turn terminal", () => { + it.each([ + "sealed", + "aborted", + "error", + "rejected", + ] as const)("ends generation on %s", (reason) => { + const before = foldEvent( + foldEvent(initialState(), turnStart("t1")), + textDelta("t1", "partial"), + ); + const next = foldEvent(before, { + type: "turn-ended", + conversationId: "c1", + turnId: "t1", + reason, + }); + expect(next.generating).toBe(false); + expect(next.accumulating).toBeNull(); + expect(selectChunks(next).map((c) => c.chunk)).toEqual([{ type: "text", text: "partial" }]); + expect(before.accumulating?.text).toBe("partial"); + }); +}); + +describe("foldEvent — re-replay of an in-flight turn (re-subscribe after tab switch / reconnect)", () => { + const userMessage = (turnId: string, text: string): TurnInputEvent => ({ + type: "user-message", + conversationId: "c1", + turnId, + text, + }); + const replay = (upTo: number) => + [ + userMessage("t1", "go"), + turnStart("t1"), + reasoningDelta("t1", "plan"), + toolCall("t1", "tc1", "read", {}), + toolResult("t1", "tc1", "read", "out"), + { type: "steering", conversationId: "c1", turnId: "t1", text: "also this" } as const, + reasoningDelta("t1", "more"), + toolCall("t1", "tc2", "read", {}), + ].slice(0, upTo); + + it("replaces the stale copy instead of appending a duplicate turn", () => { + let s = replay(7).reduce(foldEvent, initialState()); // first view, mid-thinking + s = replay(8).reduce(foldEvent, s); // came back: the whole turn replays again + const once = replay(8).reduce(foldEvent, initialState()); + expect(selectChunks(s)).toEqual(selectChunks(once)); + expect(s.accumulating).toBeNull(); + }); + + it("drops a stale steering bubble that ended the old copy", () => { + let s = replay(6).reduce(foldEvent, initialState()); // old copy ends in steering + s = replay(8).reduce(foldEvent, s); + expect(selectChunks(s).filter((c) => c.role === "user")).toHaveLength(2); // opening + steer + }); + + it("keeps other turns' provisional chunks", () => { + let s = [turnStart("t0"), textDelta("t0", "earlier"), doneEvent("t0")].reduce( + foldEvent, + initialState(), + ); + s = replay(4).reduce(foldEvent, s); + s = replay(4).reduce(foldEvent, s); + expect(selectChunks(s).map((c) => c.chunk.type)).toEqual([ + "text", + "text", + "thinking", + "tool-call", + ]); + }); +}); diff --git a/src/core/chunks/reducer.ts b/src/core/chunks/reducer.ts index 64e41b9..71df4dd 100644 --- a/src/core/chunks/reducer.ts +++ b/src/core/chunks/reducer.ts @@ -1,5 +1,6 @@ import type { AgentEvent, Chunk, ImageInput, Role, StoredChunk } from "@dispatch/wire"; import { assertChunkExhaustive } from "../wire/conformance"; +import { historyBackedProvisional } from "./reconcile"; import type { AccumulatingChunk, ProvisionalChunk, TranscriptState } from "./types"; /** The initial empty transcript state. */ @@ -40,7 +41,59 @@ function flushAccumulating( if (acc === null) return provisional; const chunk: Chunk = acc.kind === "text" ? { type: "text", text: acc.text } : { type: "thinking", text: acc.text }; - return [...provisional, { role: "assistant", chunk }]; + return [ + ...provisional, + withTurn( + { role: "assistant", chunk, ...(acc.stepId !== undefined ? { stepId: acc.stepId } : {}) }, + acc.turnId, + ), + ]; +} + +/** Attach a `turnId` to a provisional chunk (omitted when unknown). */ +function withTurn(p: ProvisionalChunk, turnId: string | undefined): ProvisionalChunk { + return turnId !== undefined ? { ...p, turnId } : p; +} + +/** + * Tag the trailing UNTAGGED user run (the sender's optimistic echo, appended + * before the server assigned a turn id) with the turn that just started. + */ +function tagTrailingUserEcho( + provisional: readonly ProvisionalChunk[], + turnId: string, +): readonly ProvisionalChunk[] { + let start = provisional.length; + while (start > 0) { + const p = provisional[start - 1]; + if (p === undefined || p.role !== "user" || p.turnId !== undefined) break; + start--; + } + if (start === provisional.length) return provisional; + return [ + ...provisional.slice(0, start), + ...provisional.slice(start).map((p) => ({ ...p, turnId })), + ]; +} + +/** + * Drop the previous copy of a turn that is being replayed from its start. + * Keeps the trailing user run — the opening this replay's `user-message` just + * appended (or the sender's echo that `turn-start` just tagged) — and every + * other turn's chunks. + */ +function dropReplayedTurn( + provisional: readonly ProvisionalChunk[], + turnId: string, +): readonly ProvisionalChunk[] { + let tail = provisional.length; + while (tail > 0 && provisional[tail - 1]?.role === "user") tail--; + // A stale steering bubble (replayed again as its own `steering` event) can + // end the old copy and so sit inside the trailing user run. + const stale = (p: ProvisionalChunk, i: number): boolean => + p.turnId === turnId && (i < tail || p.afterToolCallId !== undefined); + if (!provisional.some(stale)) return provisional; + return provisional.filter((p, i) => !stale(p, i)); } /** @@ -103,8 +156,8 @@ function trailingRun( /** * Merge authoritative seq-keyed chunks into the committed history. * Dedupes by seq (new wins), keeps seq-monotonic order, idempotent. - * When sealedTurnId is set, drops all provisional chunks (now superseded) - * and clears sealedTurnId. + * A sealed/finally-synced turn supersedes only its own provisional chunks; + * unrelated turns and an already-running queued carry remain live. * * Chunks below the chat-limit unload watermark (`hiddenBeforeSeq`) are * REJECTED: a full-cache or tail merge must not resurrect what the trim @@ -113,6 +166,7 @@ function trailingRun( export function applyHistory( state: TranscriptState, chunks: readonly StoredChunk[], + finishedTurnIds: readonly string[] = [], ): TranscriptState { const seqMap = new Map<number, StoredChunk>(); for (const c of state.committed) seqMap.set(c.seq, c); @@ -123,14 +177,49 @@ export function applyHistory( seqMap.set(c.seq, c); } const committed = Array.from(seqMap.values()).sort((a, b) => a.seq - b.seq); + const identityRows = new Map(seqMap); + for (const c of chunks) identityRows.set(c.seq, c); + const backed = historyBackedProvisional( + state, + [...identityRows.values()].sort((a, b) => a.seq - b.seq), + ); + const finished = new Set(finishedTurnIds); + if (state.sealedTurnId !== null) finished.add(state.sealedTurnId); + state = { + ...state, + provisional: state.provisional.filter((p) => !backed.has(p) && !finished.has(p.turnId ?? "")), + accumulating: + state.accumulating?.turnId !== undefined && finished.has(state.accumulating.turnId) + ? null + : state.accumulating, + }; if (state.sealedTurnId !== null) { + // Legacy (non-recovery) histories may carry no turn provenance at all. + // Retain their seal cleanup, but never erase an already-live queued carry. + const legacySeal = + (state.historyRecoveryTurnIds?.length ?? 0) === 0 && finishedTurnIds.length === 0; return { ...state, committed, - provisional: [], - accumulating: null, + provisional: legacySeal + ? state.provisional.filter( + (p) => + state.generating && + p.turnId === state.currentTurnId && + p.turnId !== state.sealedTurnId, + ) + : state.provisional, + accumulating: + legacySeal && state.accumulating?.turnId === state.sealedTurnId ? null : state.accumulating, sealedTurnId: null, + ...(state.historyRecoveryTurnIds !== undefined + ? { + historyRecoveryTurnIds: state.historyRecoveryTurnIds.filter( + (id) => !finishedTurnIds.includes(id), + ), + } + : {}), }; } @@ -141,7 +230,14 @@ export function applyHistory( // match the trailing provisional user-run against the trailing committed // user-run by content equality and drop the whole echo when it is fully // backed. Leaves the accumulating (streaming) chunk untouched. - if (addedNew && state.generating && state.provisional.length > 0) { + if ( + addedNew && + state.generating && + state.provisional.length > 0 && + !state.provisional.some( + (p) => p.turnId !== undefined && state.historyRecoveryTurnIds?.includes(p.turnId), + ) + ) { const provRun = trailingRun(state.provisional, "user"); if (provRun.length > 0) { const commRun = trailingRun(committed, "user"); @@ -153,7 +249,11 @@ export function applyHistory( commRun.length >= provRun.length && provRun.every((p, i) => { const c = commRun[i]; - return c !== undefined && chunkContentEquals(p.chunk, c.chunk); + return ( + c !== undefined && + (c.turnId === undefined || p.turnId === undefined || c.turnId === p.turnId) && + chunkContentEquals(p.chunk, c.chunk) + ); }); if (fullyBacked) { const dropStart = state.provisional.length - provRun.length; @@ -163,7 +263,17 @@ export function applyHistory( } } - return { ...state, committed }; + return { + ...state, + committed, + ...(state.historyRecoveryTurnIds !== undefined + ? { + historyRecoveryTurnIds: state.historyRecoveryTurnIds.filter( + (id) => !finishedTurnIds.includes(id), + ), + } + : {}), + }; } /** @@ -174,6 +284,8 @@ export function applyHistory( * 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). + * - `reasoning-end` closes the open ThinkingChunk as its own chunk; with none open + * (a redacted/empty block) it appends an empty thinking chunk, mirroring history. * - `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 @@ -198,8 +310,26 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState case "tool-output": return state; - case "turn-start": - return { ...state, currentTurnId: event.turnId, generating: true }; + case "turn-start": { + // A turn starts exactly once live, so a `turn-start` for a turn that + // already has provisional content is a late-join REPLAY restarting from + // the top (re-subscribe after a tab switch / WS reconnect): the server + // re-delivers the whole in-flight turn. Replace the stale copy instead of + // appending a second one (a duplicate copy rendered as a tail of thinking + // blocks with its tool cards collapsed onto the first copy's). + const stale = state.accumulating?.turnId === event.turnId; + const provisional = tagTrailingUserEcho( + flushAccumulating(state.provisional, stale ? null : state.accumulating), + event.turnId, + ); + return { + ...state, + provisional: dropReplayedTurn(provisional, event.turnId), + accumulating: null, + currentTurnId: event.turnId, + generating: true, + }; + } case "user-message": { // The turn's USER prompt, surfaced on the event stream (backend CR-3) so a @@ -215,14 +345,37 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState // text) emits no `user-message` and is not de-duped here. if (event.text.length === 0) return state; const run = trailingRun(state.provisional, "user"); - const alreadyEchoed = run.some((p) => p.chunk.type === "text" && p.chunk.text === event.text); - if (alreadyEchoed) { - return { ...state, generating: true }; + const echoed = run.find( + (p) => + (p.turnId === undefined || p.turnId === event.turnId) && + p.chunk.type === "text" && + p.chunk.text === event.text, + ); + if (echoed !== undefined) { + return { + ...state, + // Preserve the live text independently of the local send. A delayed + // queued ack may remove its images, but cannot erase this server event. + provisional: tagTrailingUserEcho(state.provisional, event.turnId).map((p) => { + if (p.chunk !== echoed.chunk) return p; + const { localRequestId: _local, ...confirmed } = p; + return { ...confirmed, opening: true }; + }), + generating: true, + }; } const provisional = flushAccumulating(state.provisional, state.accumulating); return { ...state, - provisional: [...provisional, { role: "user", chunk: { type: "text", text: event.text } }], + provisional: [ + ...provisional, + { + role: "user", + chunk: { type: "text", text: event.text }, + turnId: event.turnId, + opening: true, + }, + ], accumulating: null, generating: true, }; @@ -230,10 +383,10 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState case "text-delta": { const acc = state.accumulating; - if (acc !== null && acc.kind === "text") { + if (acc !== null && acc.kind === "text" && acc.turnId === event.turnId) { return { ...state, - accumulating: { kind: "text", text: acc.text + event.delta }, + accumulating: { ...acc, text: acc.text + event.delta }, generating: true, }; } @@ -241,17 +394,17 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState return { ...state, provisional, - accumulating: { kind: "text", text: event.delta }, + accumulating: { kind: "text", text: event.delta, turnId: event.turnId }, generating: true, }; } case "reasoning-delta": { const acc = state.accumulating; - if (acc !== null && acc.kind === "thinking") { + if (acc !== null && acc.kind === "thinking" && acc.turnId === event.turnId) { return { ...state, - accumulating: { kind: "thinking", text: acc.text + event.delta }, + accumulating: { ...acc, text: acc.text + event.delta }, generating: true, }; } @@ -259,7 +412,31 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState return { ...state, provisional, - accumulating: { kind: "thinking", text: event.delta }, + accumulating: { kind: "thinking", text: event.delta, turnId: event.turnId }, + generating: true, + }; + } + + case "reasoning-end": { + const acc = state.accumulating; + // Close the open thinking block of this turn: flush it as its own provisional chunk. + if (acc !== null && acc.kind === "thinking" && acc.turnId === event.turnId) { + return { + ...state, + provisional: flushAccumulating(state.provisional, acc), + accumulating: null, + generating: true, + }; + } + // No open thinking (a redacted/empty block with no deltas): history persists an + // EMPTY thinking chunk for it, so mirror that 1:1. + return { + ...state, + provisional: [ + ...flushAccumulating(state.provisional, acc), + { role: "assistant", chunk: { type: "thinking", text: "" }, turnId: event.turnId }, + ], + accumulating: null, generating: true, }; } @@ -275,7 +452,7 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState }; return { ...state, - provisional: [...provisional, { role: "assistant", chunk }], + provisional: [...provisional, { role: "assistant", chunk, turnId: event.turnId }], accumulating: null, generating: true, }; @@ -293,7 +470,7 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState }; return { ...state, - provisional: [...provisional, { role: "tool", chunk }], + provisional: [...provisional, { role: "tool", chunk, turnId: event.turnId }], accumulating: null, generating: true, }; @@ -307,7 +484,7 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState : { type: "error", message: event.message }; return { ...state, - provisional: [...provisional, { role: "assistant", chunk }], + provisional: [...provisional, { role: "assistant", chunk, turnId: event.turnId }], accumulating: null, generating: false, }; @@ -317,8 +494,24 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState return { ...state, latestUsage: event.usage }; case "step-complete": - // Timing metadata — no content chunk; handled by the telemetry reducer. - return state; + // A persistence boundary may lag this notification. Stamp provenance and + // finish the generation chunk, but keep it until history really backs it. + return { + ...state, + provisional: flushAccumulating(state.provisional, state.accumulating), + accumulating: null, + }; + + case "turn-ended": + return { + ...state, + provisional: + state.accumulating?.turnId === event.turnId + ? flushAccumulating(state.provisional, state.accumulating) + : state.provisional, + accumulating: state.accumulating?.turnId === event.turnId ? null : state.accumulating, + generating: state.currentTurnId === event.turnId ? false : state.generating, + }; case "done": { const provisional = flushAccumulating(state.provisional, state.accumulating); @@ -350,9 +543,22 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState // into the transcript by the sender). if (event.text.length === 0) return state; const provisional = flushAccumulating(state.provisional, state.accumulating); + const preceding = [...state.committed, ...provisional] + .reverse() + .find((p) => p.turnId === event.turnId && p.chunk.type === "tool-result"); + const afterToolCallId = + preceding?.chunk.type === "tool-result" ? preceding.chunk.toolCallId : undefined; return { ...state, - provisional: [...provisional, { role: "user", chunk: { type: "text", text: event.text } }], + provisional: [ + ...provisional, + { + role: "user", + chunk: { type: "text", text: event.text }, + turnId: event.turnId, + ...(afterToolCallId !== undefined ? { afterToolCallId } : {}), + }, + ], accumulating: null, generating: true, }; @@ -367,6 +573,9 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState // (already true from `turn-start`) is left untouched. return { ...state, providerRetry: event }; } + + default: + return state; // Forward-compatible with additive wire events. } } @@ -385,14 +594,37 @@ const RETRY_CLEARING_EVENTS: ReadonlySet<AgentEvent["type"]> = new Set([ "turn-start", "text-delta", "reasoning-delta", + "reasoning-end", "tool-call", "tool-result", "error", "done", "turn-sealed", + "turn-ended", ]); export function foldEvent(state: TranscriptState, event: AgentEvent): TranscriptState { + if ( + event.type === "tool-call" || + event.type === "tool-result" || + event.type === "step-complete" + ) { + state = { + ...state, + provisional: state.provisional.map((p) => + p.turnId === event.turnId && + p.role === "assistant" && + p.stepId === undefined && + (p.chunk.type === "text" || p.chunk.type === "thinking") + ? { ...p, stepId: event.stepId } + : p, + ), + accumulating: + state.accumulating?.turnId === event.turnId + ? { ...state.accumulating, stepId: event.stepId } + : state.accumulating, + }; + } const next = reduceEvent(state, event); if (event.type === "provider-retry") return next; // set by reduceEvent; not a clearing event if (RETRY_CLEARING_EVENTS.has(event.type) && state.providerRetry !== null) { @@ -419,6 +651,7 @@ export function appendUserMessage( state: TranscriptState, text: string, images?: readonly ImageInput[], + localRequestId?: string, ): TranscriptState { const provisional = flushAccumulating(state.provisional, state.accumulating); const userChunks: Chunk[] = []; @@ -440,7 +673,35 @@ export function appendUserMessage( } return { ...state, - provisional: [...provisional, ...userChunks.map((chunk) => ({ role: "user", chunk }) as const)], + provisional: [ + ...provisional, + ...userChunks.map( + (chunk) => + ({ + role: "user", + chunk, + ...(localRequestId !== undefined ? { localRequestId } : {}), + }) as const, + ), + ], accumulating: null, }; } + +/** Remove only still-local artifacts of one accepted queued send; never history/live text. */ +export function removeLocalEcho(state: TranscriptState, requestId: string): TranscriptState { + const provisional = state.provisional.filter((p) => p.localRequestId !== requestId); + return provisional.length === state.provisional.length ? state : { ...state, provisional }; +} + +/** A resync notice is lifecycle evidence, never fabricated transcript content. */ +export function markHistoryRecovering(state: TranscriptState, turnId: string): TranscriptState { + return { + ...state, + currentTurnId: turnId, + generating: true, + historyRecoveryTurnIds: state.historyRecoveryTurnIds?.includes(turnId) + ? state.historyRecoveryTurnIds + : [...(state.historyRecoveryTurnIds ?? []), turnId], + }; +} diff --git a/src/core/chunks/selectors.test.ts b/src/core/chunks/selectors.test.ts new file mode 100644 index 0000000..90bb5cc --- /dev/null +++ b/src/core/chunks/selectors.test.ts @@ -0,0 +1,373 @@ +import type { AgentEvent, StepId, StoredChunk } from "@dispatch/wire"; +import { describe, expect, it } from "vitest"; +import { applyHistory, foldEvent, initialState, markHistoryRecovering } from "./reducer"; +import { selectChunks } from "./selectors"; +import { trimTranscript } from "./trim"; +import type { TranscriptState } from "./types"; + +const C = "conv"; +const T = "turn-1"; + +function fold(state: TranscriptState, events: readonly AgentEvent[]): TranscriptState { + return events.reduce(foldEvent, state); +} + +const turnStart: AgentEvent = { type: "turn-start", conversationId: C, turnId: T }; +const userMsg: AgentEvent = { type: "user-message", conversationId: C, turnId: T, text: "hi" }; +const think = (delta: string): AgentEvent => ({ + type: "reasoning-delta", + conversationId: C, + turnId: T, + delta, +}); +const thinkEnd: AgentEvent = { type: "reasoning-end", conversationId: C, turnId: T }; +const text = (delta: string): AgentEvent => ({ + type: "text-delta", + conversationId: C, + turnId: T, + delta, +}); +const call = (id: string, step = 0): AgentEvent => ({ + type: "tool-call", + conversationId: C, + turnId: T, + stepId: `${T}#${step}` as StepId, + toolCallId: id, + toolName: "run_shell", + input: {}, +}); +const result = (id: string, step = 0): AgentEvent => ({ + type: "tool-result", + conversationId: C, + turnId: T, + stepId: `${T}#${step}` as StepId, + toolCallId: id, + toolName: "run_shell", + content: "ok", + isError: false, +}); + +/** The persisted chunks for steps 0 (user + think + call/result). */ +const committedStep0: StoredChunk[] = [ + { seq: 1, role: "user", chunk: { type: "text", text: "hi" }, turnId: T }, + { seq: 2, role: "assistant", chunk: { type: "thinking", text: "plan" }, turnId: T }, + { + seq: 3, + role: "assistant", + chunk: { + type: "tool-call", + toolCallId: "a", + toolName: "run_shell", + input: {}, + stepId: `${T}#0` as StepId, + }, + turnId: T, + }, + { + seq: 4, + role: "tool", + chunk: { + type: "tool-result", + toolCallId: "a", + toolName: "run_shell", + content: "ok", + isError: false, + stepId: `${T}#0` as StepId, + }, + turnId: T, + }, +]; + +describe("selectChunks — de-dup of a late-join replay against committed history", () => { + it("hides replayed provisional chunks already persisted for the in-flight turn", () => { + // Fresh load mid-turn: history has step 0; the replay re-delivers step 0 + // and then step 1 (still streaming). + let s = applyHistory(initialState(), committedStep0); + s = fold(s, [ + userMsg, + turnStart, + think("pl"), + think("an"), + call("a"), + result("a"), + text("now"), + ]); + + const rendered = selectChunks(s); + expect(rendered.map((r) => [r.seq, r.chunk.type])).toEqual([ + [1, "text"], + [2, "thinking"], + [3, "tool-call"], + [4, "tool-result"], + [null, "text"], + ]); + expect(rendered[4]?.streaming).toBe(true); + }); + + it("keeps provisional chunks beyond the committed run", () => { + let s = applyHistory(initialState(), committedStep0); + s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("a"), call("b", 1)]); + const rendered = selectChunks(s); + expect(rendered).toHaveLength(5); + expect(rendered[4]?.chunk).toMatchObject({ type: "tool-call", toolCallId: "b" }); + expect(rendered[4]?.provisional).toBe(true); + }); + + it("hides the accumulating chunk once its content is committed", () => { + const committed: StoredChunk[] = [ + ...committedStep0, + { seq: 5, role: "assistant", chunk: { type: "text", text: "final" }, turnId: T }, + ]; + let s = applyHistory(initialState(), committed); + s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("a"), text("final")]); + const rendered = selectChunks(s); + expect(rendered).toHaveLength(5); + expect(rendered.every((r) => !r.provisional)).toBe(true); + }); + + it("stops de-duping at the first mismatch (never hides divergent content)", () => { + let s = applyHistory(initialState(), committedStep0); + s = fold(s, [userMsg, turnStart, think("DIFFERENT"), call("a"), result("a")]); + const rendered = selectChunks(s); + // user echo matched (1 hidden); everything from the mismatch on is shown. + expect(rendered.filter((r) => r.provisional)).toHaveLength(3); + }); + + it("preserves an uncommitted accumulating tail even when its text repeats history", () => { + const committed: StoredChunk[] = [ + ...committedStep0, + { seq: 5, role: "assistant", chunk: { type: "text", text: "final" }, turnId: T }, + ]; + let s = applyHistory(initialState(), committed); + s = markHistoryRecovering(s, T); + s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("a"), text("final")]); + const rendered = selectChunks(s); + expect(rendered).toHaveLength(6); + expect(rendered.slice(0, 5).every((r) => !r.provisional)).toBe(true); + expect(rendered[5]).toMatchObject({ + provisional: true, + streaming: true, + chunk: { type: "text", text: "final" }, + }); + }); + + it("reconciles the same step identity even when provisional payload differs", () => { + let s = applyHistory(initialState(), committedStep0); + s = markHistoryRecovering(s, T); + s = fold(s, [userMsg, turnStart, think("DIFFERENT"), call("a"), result("a")]); + const rendered = selectChunks(s); + // Authoritative identity wins; payload equality cannot identify omitted/truncated content. + expect(rendered.filter((r) => r.provisional)).toHaveLength(0); + expect(rendered.map((r) => r.chunk)).toEqual(committedStep0.map((r) => r.chunk)); + }); + + it("does nothing when the committed tail belongs to another turn", () => { + const other = committedStep0.map((c) => ({ ...c, turnId: "older" })); + let s = applyHistory(initialState(), other); + s = fold(s, [userMsg, turnStart, think("plan")]); + expect(selectChunks(s).filter((r) => r.provisional)).toHaveLength(2); + }); + + it("de-dups the replay when the committed turn's head was unloaded (windowed/trimmed)", () => { + // A long in-flight turn: the chat-limit window/trim unloaded the turn's + // head (seq 1–2), so the committed run starts mid-turn while the late-join + // replay re-delivers the WHOLE turn from its opening. Positional alignment + // from the turn start fails at index 0; without re-anchoring, the whole + // replay rendered as a duplicate whose tool cards collapsed onto the + // committed ones — leaving a tail of thinking blocks with no tool calls. + let s = trimTranscript(applyHistory(initialState(), committedStep0), 2); + expect(s.committed.map((c) => c.seq)).toEqual([3, 4]); + s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("a"), think("next")]); + const rendered = selectChunks(s); + expect(rendered.map((r) => [r.seq, r.chunk.type])).toEqual([ + [3, "tool-call"], + [4, "tool-result"], + [null, "thinking"], + ]); + expect(rendered[2]).toMatchObject({ streaming: true, chunk: { text: "next" } }); + }); + + it("de-dups when the provisional replay starts after the committed run's head", () => { + // Provisional trimmed past the committed head (e.g. committed restored via + // "Show earlier" after an all-provisional overflow): re-anchor on the + // provisional head instead. + let s = applyHistory(initialState(), committedStep0); + s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("a"), call("b", 1)]); + s = { ...s, provisional: s.provisional.slice(2) }; // drop user + thinking + const rendered = selectChunks(s); + expect(rendered.map((r) => [r.seq, r.chunk.type])).toEqual([ + [1, "text"], + [2, "thinking"], + [3, "tool-call"], + [4, "tool-result"], + [null, "tool-call"], + ]); + }); + + it("re-anchors on a persisted tool call even if the replay diverges after it", () => { + // A toolCallId is a unique identity: everything the replay delivered before + // it was persisted earlier in the turn (append-only), so it is unloaded + // history — not divergent content. The divergence itself stays visible. + let s = applyHistory(initialState(), committedStep0.slice(2)); + s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("OTHER")]); + const provisional = selectChunks(s).filter((r) => r.provisional); + expect(provisional.map((r) => r.chunk)).toMatchObject([ + { type: "tool-result", toolCallId: "OTHER" }, + ]); + }); + + it("does not re-anchor on an unverified text match (never hides divergent content)", () => { + const committed: StoredChunk[] = [ + { seq: 7, role: "assistant", chunk: { type: "thinking", text: "plan" }, turnId: T }, + { seq: 8, role: "assistant", chunk: { type: "text", text: "persisted" }, turnId: T }, + ]; + let s = applyHistory(initialState(), committed); + s = fold(s, [userMsg, turnStart, think("plan"), text("diverged"), call("z")]); + // "plan" matches but the next chunk diverges: no anchor → nothing hidden. + expect(selectChunks(s).filter((r) => r.provisional)).toHaveLength(4); + }); + + it("steps over persisted images the replay can never carry", () => { + // A tool attached an image mid-turn: history persists it as a user image + // chunk, but no live event carries it, so the replay lacks it. + const image: StoredChunk = { + seq: 5, + role: "user", + chunk: { type: "image", url: "/images/conv/x.png" }, + turnId: T, + }; + const thinking: StoredChunk = { + seq: 6, + role: "assistant", + chunk: { type: "thinking", text: "look" }, + turnId: T, + }; + let s = applyHistory(initialState(), [...committedStep0, image, thinking]); + s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("a"), think("look")]); + s = fold(s, [call("b", 1)]); + const provisional = selectChunks(s).filter((r) => r.provisional); + expect(provisional.map((r) => r.chunk)).toMatchObject([{ type: "tool-call", toolCallId: "b" }]); + }); + + it("matches one merged thinking block against its split committed blocks (Claude)", () => { + // Claude may return several thinking blocks in one step; history persists + // each as its own chunk, but reasoning-deltas carry no block boundary, so + // the replay delivers them merged into one thinking chunk. + const split: StoredChunk[] = [ + committedStep0[0] as StoredChunk, + { seq: 2, role: "assistant", chunk: { type: "thinking", text: "first. " }, turnId: T }, + { seq: 3, role: "assistant", chunk: { type: "thinking", text: "second." }, turnId: T }, + { ...(committedStep0[2] as StoredChunk), seq: 4 }, + { ...(committedStep0[3] as StoredChunk), seq: 5 }, + ]; + let s = applyHistory(initialState(), split); + s = fold(s, [userMsg, turnStart, think("first. "), think("second."), call("a"), result("a")]); + s = fold(s, [think("next")]); + const rendered = selectChunks(s); + expect(rendered.map((r) => [r.seq, r.chunk.type])).toEqual([ + [1, "text"], + [2, "thinking"], + [3, "thinking"], + [4, "tool-call"], + [5, "tool-result"], + [null, "thinking"], + ]); + }); + + it("hides an accumulating merged block once its split blocks are committed", () => { + const split: StoredChunk[] = [ + ...committedStep0, + { seq: 5, role: "assistant", chunk: { type: "thinking", text: "a" }, turnId: T }, + { seq: 6, role: "assistant", chunk: { type: "thinking", text: "b" }, turnId: T }, + ]; + let s = applyHistory(initialState(), split); + s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("a"), think("ab")]); + expect(selectChunks(s).every((r) => !r.provisional)).toBe(true); + }); + + it("re-anchors on a tool chunk when the window cut between split blocks", () => { + // The committed head is the SECOND of two split blocks: it equals no + // provisional chunk, so alignment anchors on the run's first tool chunk. + const committed: StoredChunk[] = [ + { seq: 3, role: "assistant", chunk: { type: "thinking", text: "second." }, turnId: T }, + { ...(committedStep0[2] as StoredChunk), seq: 4 }, + { ...(committedStep0[3] as StoredChunk), seq: 5 }, + ]; + let s = applyHistory(initialState(), committed); + s = fold(s, [userMsg, turnStart, think("first. second."), call("a"), result("a")]); + s = fold(s, [call("b", 1)]); + const provisional = selectChunks(s).filter((r) => r.provisional); + expect(provisional.map((r) => r.chunk)).toMatchObject([{ type: "tool-call", toolCallId: "b" }]); + }); + + it("aligns a boundary-carrying replay 1:1 against split committed blocks", () => { + const split: StoredChunk[] = [ + committedStep0[0] as StoredChunk, + { seq: 2, role: "assistant", chunk: { type: "thinking", text: "first. " }, turnId: T }, + { seq: 3, role: "assistant", chunk: { type: "thinking", text: "second." }, turnId: T }, + { ...(committedStep0[2] as StoredChunk), seq: 4 }, + { ...(committedStep0[3] as StoredChunk), seq: 5 }, + ]; + let s = applyHistory(initialState(), split); + s = fold(s, [ + userMsg, + turnStart, + think("first. "), + thinkEnd, + think("second."), + thinkEnd, + call("a"), + result("a"), + text("next"), + ]); + const rendered = selectChunks(s); + expect(rendered.map((r) => [r.seq, r.chunk.type])).toEqual([ + [1, "text"], + [2, "thinking"], + [3, "thinking"], + [4, "tool-call"], + [5, "tool-result"], + [null, "text"], + ]); + expect(rendered[5]?.streaming).toBe(true); + }); + + it("aligns a redacted empty block with a bare reasoning-end", () => { + const split: StoredChunk[] = [ + committedStep0[0] as StoredChunk, + { seq: 2, role: "assistant", chunk: { type: "thinking", text: "first. " }, turnId: T }, + { seq: 3, role: "assistant", chunk: { type: "thinking", text: "" }, turnId: T }, + { seq: 4, role: "assistant", chunk: { type: "thinking", text: "second." }, turnId: T }, + { ...(committedStep0[2] as StoredChunk), seq: 5 }, + { ...(committedStep0[3] as StoredChunk), seq: 6 }, + ]; + let s = applyHistory(initialState(), split); + s = fold(s, [ + userMsg, + turnStart, + think("first. "), + thinkEnd, + thinkEnd, + think("second."), + thinkEnd, + call("a"), + result("a"), + text("next"), + ]); + const rendered = selectChunks(s); + expect(rendered.map((r) => [r.seq, r.chunk.type])).toEqual([ + [1, "text"], + [2, "thinking"], + [3, "thinking"], + [4, "thinking"], + [5, "tool-call"], + [6, "tool-result"], + [null, "text"], + ]); + }); + + it("is the identity when there is no committed history", () => { + const s = fold(initialState(), [userMsg, turnStart, text("hello")]); + expect(selectChunks(s)).toHaveLength(2); + }); +}); diff --git a/src/core/chunks/selectors.ts b/src/core/chunks/selectors.ts index e46d5f5..8d32fcf 100644 --- a/src/core/chunks/selectors.ts +++ b/src/core/chunks/selectors.ts @@ -1,24 +1,278 @@ -import type { ChatMessage, Chunk, TurnProviderRetryEvent } from "@dispatch/wire"; -import type { RenderedChunk, TranscriptState } from "./types"; +import type { ChatMessage, Chunk, StoredChunk, TurnProviderRetryEvent } from "@dispatch/wire"; +import { historyBackedProvisional } from "./reconcile"; +import type { ProvisionalChunk, RenderedChunk, TranscriptState } from "./types"; + +/** + * Whether a provisional chunk is the same content as a committed chunk at the + * same position of the same turn (identity for de-dup, not deep equality). + * Tool chunks match by `toolCallId`; text-bearing chunks by exact text. + */ +function sameChunk(p: ProvisionalChunk, c: StoredChunk): boolean { + if (p.role !== c.role || p.chunk.type !== c.chunk.type) return false; + const a = p.chunk; + const b = c.chunk; + switch (a.type) { + case "tool-call": + case "tool-result": + return a.toolCallId === (b as typeof a).toolCallId; + case "text": + case "thinking": + case "system": + return a.text === (b as typeof a).text; + case "error": + return a.message === (b as typeof a).message; + case "image": + return true; + } +} + +/** + * A committed chunk with no live-event counterpart, so a late-join replay can + * never contain it: an image persisted into the turn's history (pasted-image + * openings, images a tool attaches mid-turn). Alignment steps over it. + */ +function replayless(c: StoredChunk): boolean { + return c.chunk.type === "image"; +} + +/** Where an alignment walk stopped, and whether it stopped on a mismatch. */ +interface Walk { + /** First provisional index NOT matched. */ + readonly p: number; + /** First committed index NOT matched (replayless chunks stepped over). */ + readonly c: number; + readonly matched: number; + /** Ended because a side ran out — no mismatch. */ + readonly complete: boolean; +} + +/** + * How many consecutive committed chunks starting at `c` one provisional text / + * thinking chunk stands for: their texts must concatenate EXACTLY to its text + * (≥ 2 chunks; 0 = no such run). A provider may return several blocks in one + * step (Claude emits multiple thinking blocks, each with its own signature) + * which history persists as separate chunks, while the delta stream carries + * no block boundary — so live (and in the replay) they arrive merged. + */ +function mergedRun( + state: TranscriptState, + text: string, + kind: string, + role: string, + c: number, +): number { + let joined = ""; + for (let k = c; k < state.committed.length; k++) { + const row = state.committed[k]; + if (row === undefined || row.role !== role || row.chunk.type !== kind) return 0; + if (row.chunk.type !== "text" && row.chunk.type !== "thinking") return 0; + joined += row.chunk.text; + if (joined === text) return k - c + 1 >= 2 ? k - c + 1 : 0; + if (!text.startsWith(joined)) return 0; + } + return 0; +} + +/** + * Walk provisional (from `p`) and committed (from `c`) of turn `turnId` in + * lockstep, stepping over committed chunks the replay cannot carry and + * matching one merged provisional block against its split committed run. + * Stops at the first mismatch or when either side runs out. + */ +function walk(state: TranscriptState, turnId: string, p: number, c: number): Walk { + let matched = 0; + for (;;) { + const prov = state.provisional[p]; + let row = state.committed[c]; + while (row !== undefined && replayless(row) && prov?.chunk.type !== "image") { + row = state.committed[++c]; + } + if (prov === undefined || row === undefined) return { p, c, matched, complete: true }; + if (prov.turnId !== turnId) return { p, c, matched, complete: false }; + let span = sameChunk(prov, row) ? 1 : 0; + if (span === 0 && (prov.chunk.type === "text" || prov.chunk.type === "thinking")) { + span = mergedRun(state, prov.chunk.text, prov.chunk.type, prov.role, c); + } + if (span === 0) return { p, c, matched, complete: false }; + p++; + c += span; + matched++; + } +} + +/** + * Accept a re-anchored alignment. A tool chunk is a unique identity + * (`toolCallId`), so a match on one is trusted outright; a text-bearing match + * must be confirmed by the WHOLE overlap aligning (no mismatch), so a repeated + * text can never hide divergent content. + */ +function anchored(w: Walk, anchor: StoredChunk): boolean { + if (w.matched === 0) return false; + return w.complete || anchor.chunk.type === "tool-call" || anchor.chunk.type === "tool-result"; +} + +/** + * Align the provisional run with the committed run of turn T (which begins at + * committed index `start`). Returns how many leading provisional chunks are + * hidden as already committed, and the walk's end. + * + * Normally both start at the turn's opening. But the chat-limit window (fresh + * load `?limit=` / `windowTranscript`) and `trimTranscript` unload the OLDEST + * committed chunks, so in a long in-flight turn the committed run is a SUFFIX + * of the turn while the late-join replay re-delivers the whole turn from its + * `turn-start`: re-anchor on the committed head inside the replay, and hide + * the replayed chunks before it too (persisted history, unloaded — pageable + * via "Show earlier"). Conversely, provisional overflow trimming can leave the + * replay starting after the committed head: re-anchor on the replay's head + * inside the committed run. Unverifiable → the plain prefix walk (fail-safe: + * a misalignment shows a duplicate, never hides content). + */ +function align(state: TranscriptState, turnId: string, start: number): Walk { + const direct = walk(state, turnId, 0, start); + // Aligned from the turn's head (or fully consistent): plain prefix semantics. + if (direct.complete || direct.matched > 0) return direct; + let headAt = start; + while ( + state.committed[headAt] !== undefined && + replayless(state.committed[headAt] as StoredChunk) + ) + headAt++; + // Anchor on the committed head — or, when the head is unmatchable (e.g. the + // window cut between two blocks a merged provisional chunk spans), on the + // run's first tool chunk, a unique identity. + const anchorsAt = [headAt]; + for (let i = headAt + 1; i < state.committed.length; i++) { + const t = state.committed[i]?.chunk.type; + if (t === "tool-call" || t === "tool-result") { + anchorsAt.push(i); + break; + } + } + for (const at of anchorsAt) { + const head = state.committed[at]; + if (head === undefined) continue; + for (let k = 1; k < state.provisional.length; k++) { + const p = state.provisional[k]; + if (p?.turnId !== turnId || !sameChunk(p, head)) continue; + const w = walk(state, turnId, k, at); + if (anchored(w, head)) return w; + } + } + const p0 = state.provisional[0]; + if (p0 !== undefined) { + for (let j = headAt + 1; j < state.committed.length; j++) { + const row = state.committed[j]; + if (row === undefined || !sameChunk(p0, row)) continue; + const w = walk(state, turnId, 0, j); + if (anchored(w, row)) return w; + } + } + return direct; +} + +/** + * How many LEADING provisional chunks are already backed by committed chunks. + * + * The backend persists each step of an in-flight turn as it completes (CR-6), + * so a client that opens a conversation mid-turn gets those steps from history + * AND again from the late-join event replay (as provisional chunks). Both + * carry the turn's chunks in the same order, so the provisional run of turn T + * is aligned chunk-by-chunk with the committed run of turn T (see `align`). + * Skip the aligned prefix; stop at the first mismatch. + * + * A failed alignment renders the ENTIRE replay as a duplicate — and because + * tool cards de-dup by `toolCallId` onto the first (committed) occurrence, + * that duplicate tail shows only thinking/text blocks ("the tool calls + * disappear"). Hence the re-anchoring and the replayless-image stepping. + * + * Also reports whether the ACCUMULATING chunk is already committed (the step + * completed + history synced, but the next live event hasn't flushed it yet). + */ +function committedOverlap(state: TranscriptState): { skip: number; hideAccumulating: boolean } { + const first = state.provisional[0]; + const turnId = first?.turnId ?? state.accumulating?.turnId; + if (turnId === undefined || state.historyRecoveryTurnIds?.includes(turnId)) + return { skip: 0, hideAccumulating: false }; + // The committed run for this turn (contiguous at the tail of `committed`). + let start = state.committed.length; + while (start > 0 && state.committed[start - 1]?.turnId === turnId) start--; + if (start === state.committed.length) return { skip: 0, hideAccumulating: false }; + + const end = align(state, turnId, start); + const skip = end.p; + + let hideAccumulating = false; + const acc = state.accumulating; + if (acc !== null && acc.turnId === turnId && skip === state.provisional.length) { + let next = end.c; + while (state.committed[next] !== undefined && replayless(state.committed[next] as StoredChunk)) + next++; + const c = state.committed[next]; + hideAccumulating = + (c !== undefined && + c.role === "assistant" && + c.chunk.type === acc.kind && + c.chunk.text === acc.text) || + mergedRun(state, acc.text, acc.kind, "assistant", next) > 0; + } + return { skip, hideAccumulating }; +} + +/** + * How many chunks `selectChunks` would render, without materializing them. + * Provisional chunks hidden as history-backed / replay duplicates of committed + * chunks are NOT counted — the chat limit bounds what is on screen, and + * counting invisible duplicates would unload real history to make room for them. + */ +export function selectRenderedCount(state: TranscriptState): number { + const backed = historyBackedProvisional(state); + const { skip, hideAccumulating } = committedOverlap(state); + let count = state.committed.length; + for (let i = skip; i < state.provisional.length; i++) { + const p = state.provisional[i]; + if (p !== undefined && !backed.has(p)) count++; + } + if (state.accumulating !== null && !hideAccumulating) count++; + return count; +} /** * Select all chunks for rendering: committed first (seq order), - * then provisional (seq: null). + * then provisional (seq: null), excluding history-backed identities. Unknown + * accumulating content stays live, even if its text equals an earlier step. */ export function selectChunks(state: TranscriptState): readonly RenderedChunk[] { const result: RenderedChunk[] = []; + const backed = historyBackedProvisional(state); + const { skip, hideAccumulating } = committedOverlap(state); for (const c of state.committed) { - result.push({ seq: c.seq, role: c.role, chunk: c.chunk, provisional: false }); + result.push( + c.turnId !== undefined + ? { seq: c.seq, role: c.role, chunk: c.chunk, provisional: false, turnId: c.turnId } + : { 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 }); + for (let i = skip; i < state.provisional.length; i++) { + const p = state.provisional[i]; + if (p === undefined || backed.has(p)) continue; + result.push( + p.turnId !== undefined + ? { seq: null, role: p.role, chunk: p.chunk, provisional: true, turnId: p.turnId } + : { seq: null, role: p.role, chunk: p.chunk, provisional: true }, + ); } - if (state.accumulating !== null) { + const acc = state.accumulating; + if (acc !== null && !hideAccumulating) { 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 }); + acc.kind === "text" ? { type: "text", text: acc.text } : { type: "thinking", text: acc.text }; + result.push({ + seq: null, + role: "assistant", + chunk, + provisional: true, + streaming: true, + ...(acc.turnId !== undefined ? { turnId: acc.turnId } : {}), + }); } return result; } diff --git a/src/core/chunks/trim.test.ts b/src/core/chunks/trim.test.ts index 7c4bbef..dcc1755 100644 --- a/src/core/chunks/trim.test.ts +++ b/src/core/chunks/trim.test.ts @@ -103,6 +103,19 @@ describe("trimTranscript", () => { expect(next.accumulating).not.toBeNull(); }); + it("does not count a late-join replay's duplicates of committed chunks", () => { + // An in-flight turn of 8 persisted chunks + its late-join replay (8 more + // provisional copies). Only 8 are rendered, so a limit of 10 keeps them all + // instead of unloading real history to make room for invisible duplicates. + const T = "turn"; + const committed = chunks(1, 8).map((c) => ({ ...c, turnId: T })); + const state: TranscriptState = { + ...stateWith(committed), + provisional: committed.map((c) => ({ role: c.role, chunk: c.chunk, turnId: T })), + }; + expect(trimTranscript(state, 10)).toBe(state); + }); + it("drops oldest provisional when committed is exhausted", () => { const base = stateWith(chunks(1, 2)); const provisional = Array.from({ length: 20 }, (_, i) => ({ @@ -146,7 +159,7 @@ describe("windowTranscript", () => { it("is the identity within the window", () => { const state = stateWith(chunks(1, 50)); expect(windowTranscript(state, 75)).toBe(state); - expect(selectHasEarlier(state)).toBe(false); + expect(selectHasEarlier(state, 1)).toBe(false); }); }); @@ -178,21 +191,21 @@ describe("restoreEarlier", () => { expect(selectHasEarlier(restored)).toBe(true); }); - it("restoring down to seq 1 reaches the contractual origin (hasEarlier clears)", () => { + it("restoring down to a known boundary clears hasEarlier", () => { const windowed = windowTranscript(stateWith(chunks(1, 100)), 75); // hidden: 1..25 const restored = restoreEarlier(windowed, chunks(1, 100), 64); expect(restored.committed).toHaveLength(100); expect(restored.committed[0]?.seq).toBe(1); expect(restored.hiddenBeforeSeq).toBe(1); // floor at the origin — inert expect(restored.hiddenThinkingCount).toBe(0); - expect(selectHasEarlier(restored)).toBe(false); + expect(selectHasEarlier(restored, 1)).toBe(false); }); it("is the identity when nothing older is known locally (server may still hold more)", () => { const windowed = windowTranscript(stateWith(chunks(50, 200)), 75); const restored = restoreEarlier(windowed, [], 64); expect(restored).toBe(windowed); - // seqs are 1-based gap-free: window starts at 126 ⇒ older chunks DO exist. + // The server boundary is unknown, so another page may exist. expect(selectHasEarlier(restored)).toBe(true); }); @@ -203,7 +216,7 @@ describe("restoreEarlier", () => { it("works on a server-windowed transcript (no local watermark)", () => { // A cold-cache fresh load with `?limit=` commits a suffix (seq 809..1000) - // with hiddenBeforeSeq still 0 — hasEarlier derives from seq > 1, and a + // with hiddenBeforeSeq still 0 — the server boundary is unknown, and a // backfilled run merges below it. const state = stateWith(chunks(809, 1000)); expect(state.hiddenBeforeSeq).toBe(0); @@ -230,6 +243,15 @@ describe("restoreEarlier", () => { const restored = restoreEarlier(trimmed, original, 1000); expect(restored.committed).toEqual(original); expect(restored.hiddenBeforeSeq).toBe(1); - expect(selectHasEarlier(restored)).toBe(false); + expect(selectHasEarlier(restored, 1)).toBe(false); }); }); + +it("uses a non-1 history boundary rather than the seq origin", () => { + const state = stateWith(chunks(101, 110)); + expect(selectHasEarlier(state, 101)).toBe(false); + expect(selectHasEarlier(state, null)).toBe(true); + const windowed = windowTranscript(state, 5); + expect(selectHasEarlier(windowed, 101)).toBe(true); + expect(selectHasEarlier(restoreEarlier(windowed, state.committed, 5), 101)).toBe(false); +}); diff --git a/src/core/chunks/trim.ts b/src/core/chunks/trim.ts index 9846357..6612bfe 100644 --- a/src/core/chunks/trim.ts +++ b/src/core/chunks/trim.ts @@ -14,6 +14,7 @@ // "Show earlier messages" affordance knows where to page back in from. import type { StoredChunk } from "@dispatch/wire"; +import { selectRenderedCount } from "./selectors"; import type { TranscriptState } from "./types"; /** Default chat limit (max loaded chunks per conversation). */ @@ -46,9 +47,14 @@ export function initialWindowSize(limit: number): number { return Math.max(1, Math.floor(limit * 0.75)); } -/** Total loaded (rendered) chunk count: committed + provisional + accumulating. */ +/** + * Total loaded (rendered) chunk count: committed + provisional + accumulating, + * EXCLUDING provisional replay duplicates of committed chunks (a late-join + * replay re-delivers an in-flight turn's already-persisted steps). Counting + * those invisible duplicates unloaded real history to make room for them. + */ function totalCount(state: TranscriptState): number { - return state.committed.length + state.provisional.length + (state.accumulating !== null ? 1 : 0); + return selectRenderedCount(state); } function countThinking(chunks: readonly StoredChunk[]): number { @@ -150,8 +156,7 @@ function oldestLoadedSeq(state: TranscriptState): number { * `?beforeSeq=` backfill; chunks at/inside the window are ignored). The newest * `count` of them are merged back in front of `committed`, and the watermark * follows the new window start so history merges still can't resurrect what - * remains unloaded. Identity when the window already starts at seq 1 (the - * contractual origin) or nothing older is known locally. + * remains unloaded. Identity when nothing older is known locally. */ export function restoreEarlier( state: TranscriptState, @@ -159,7 +164,7 @@ export function restoreEarlier( count: number, ): TranscriptState { const oldest = oldestLoadedSeq(state); - if (oldest <= 1) return state; + if (oldest === 0) return state; const below = earlier.filter((c) => c.seq < oldest).sort((a, b) => a.seq - b.seq); if (below.length === 0) return state; const keep = below.slice(-Math.max(1, count)); @@ -172,14 +177,11 @@ export function restoreEarlier( }; } -/** - * 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; +/** Earlier availability uses the known server boundary, never a seq origin. */ +export function selectHasEarlier( + state: TranscriptState, + earliestSeq: number | null = null, +): boolean { + const oldest = oldestLoadedSeq(state); + return oldest > 0 && (earliestSeq === null || oldest > earliestSeq); } diff --git a/src/core/chunks/types.ts b/src/core/chunks/types.ts index 2ced736..2e2ee0c 100644 --- a/src/core/chunks/types.ts +++ b/src/core/chunks/types.ts @@ -2,18 +2,36 @@ import type { Chunk, Role, StoredChunk, TurnProviderRetryEvent, Usage } from "@d /** A chunk being accumulated from streaming deltas (text or thinking). */ export interface AccumulatingChunk { + readonly stepId?: string; readonly kind: "text" | "thinking"; readonly text: string; + /** The turn generating this chunk (from the live event's `turnId`). */ + readonly turnId?: string; } /** A provisional chunk that has no authoritative seq yet. */ export interface ProvisionalChunk { + /** Local provenance learned from tool-call/step-complete, not content. */ + readonly stepId?: string; + readonly opening?: boolean; + /** Steering is persisted immediately after this completed tool batch. */ + readonly afterToolCallId?: string; + /** Local send identity only; turn-start tagging does NOT confirm this content. */ + readonly localRequestId?: string; readonly role: Role; readonly chunk: Chunk; + /** + * The turn this chunk belongs to (from the live event's `turnId`). Absent + * only on the sender's optimistic user echo until its `turn-start` arrives + * (which tags it). + */ + readonly turnId?: string; } /** The transcript reducer state. Holds committed history + live in-flight turn. */ export interface TranscriptState { + /** Turns with omitted delivery: never use payload equality as replay identity. */ + readonly historyRecoveryTurnIds?: readonly string[]; readonly committed: readonly StoredChunk[]; readonly provisional: readonly ProvisionalChunk[]; readonly accumulating: AccumulatingChunk | null; @@ -74,4 +92,11 @@ export interface RenderedChunk { * dots on streaming thinking) and drop it the moment generation moves on. */ readonly streaming?: boolean; + /** + * The turn that produced this chunk — `StoredChunk.turnId` for committed + * chunks, the live event's `turnId` for provisional ones. The exact join key + * for placing per-turn metrics. Absent on legacy rows (persisted before the + * backend stamped it) and on the optimistic echo before `turn-start`. + */ + readonly turnId?: string; } diff --git a/src/core/metrics/cache-timer.test.ts b/src/core/metrics/cache-timer.test.ts new file mode 100644 index 0000000..d450135 --- /dev/null +++ b/src/core/metrics/cache-timer.test.ts @@ -0,0 +1,359 @@ +import type { AgentEvent, StepId, TurnMetrics } from "@dispatch/wire"; +import { describe, expect, it } from "vitest"; +import type { RenderGroup } from "../chunks"; +import { + CACHE_TIMER_MAX_MS, + CACHE_TTL_MS, + type CacheTimerState, + foldCacheTimerEvent, + initialCacheTimerState, + lastTurnRowIndex, + seedTurnGap, + stopTurnGap, + viewCacheTimer, +} from "./cache-timer"; +import type { MetricsRow } from "./types"; + +const SEC = 1000; +const MIN = 60 * SEC; +const C = "conv"; + +const ev = { + turnStart: (turnId: string): AgentEvent => ({ type: "turn-start", conversationId: C, turnId }), + userMessage: (turnId: string): AgentEvent => ({ + type: "user-message", + conversationId: C, + turnId, + text: "hi", + }), + text: (turnId: string): AgentEvent => ({ + type: "text-delta", + conversationId: C, + turnId, + delta: "x", + }), + reasoningEnd: (turnId: string): AgentEvent => ({ + type: "reasoning-end", + conversationId: C, + turnId, + }), + call: (turnId: string, stepId: string, toolCallId: string): AgentEvent => ({ + type: "tool-call", + conversationId: C, + turnId, + stepId: stepId as StepId, + toolCallId, + toolName: "t", + input: {}, + }), + result: ( + turnId: string, + stepId: string, + toolCallId: string, + durationMs?: number, + ): AgentEvent => ({ + type: "tool-result", + conversationId: C, + turnId, + stepId: stepId as StepId, + toolCallId, + toolName: "t", + content: "ok", + isError: false, + ...(durationMs !== undefined ? { durationMs } : {}), + }), + stepComplete: (turnId: string, stepId: string): AgentEvent => ({ + type: "step-complete", + conversationId: C, + turnId, + stepId: stepId as StepId, + }), + usage: (turnId: string, stepId: string): AgentEvent => ({ + type: "usage", + conversationId: C, + turnId, + stepId: stepId as StepId, + usage: { inputTokens: 1, outputTokens: 1 }, + }), + done: (turnId: string): AgentEvent => ({ + type: "done", + conversationId: C, + turnId, + reason: "end-turn", + }), + error: (turnId: string): AgentEvent => ({ + type: "error", + conversationId: C, + turnId, + message: "boom", + }), +}; + +const noTurn = () => undefined; + +/** Fold `[time, event]` pairs from the initial state. */ +function run( + steps: readonly (readonly [number, AgentEvent])[], + start: CacheTimerState = initialCacheTimerState(), + lastTurnId: () => string | undefined = noTurn, +): CacheTimerState { + return steps.reduce((s, [at, e]) => foldCacheTimerEvent(s, e, at, lastTurnId), start); +} + +describe("foldCacheTimerEvent — step gaps", () => { + it("runs from step-complete while tools execute, and stamps when the last result returns", () => { + const s1 = run([ + [0, ev.turnStart("t1")], + [100, ev.call("t1", "t1#0", "a")], + [110, ev.call("t1", "t1#0", "b")], + [1_000, ev.stepComplete("t1", "t1#0")], + ]); + expect(s1.running).toMatchObject({ + kind: "step", + stepId: "t1#0", + startedAt: 1_000, + toolCount: 2, + }); + + const s2 = run([[4_000, ev.result("t1", "t1#0", "a")]], s1); + expect(s2.running?.kind).toBe("step"); // still waiting on "b" + expect(s2.stepStamps.size).toBe(0); + + const s3 = run([[13_000, ev.result("t1", "t1#0", "b")]], s2); + expect(s3.running).toBeNull(); + expect(s3.stepStamps.get("t1#0")).toBe(12_000); + }); + + it("stamps each step of a multi-step turn separately", () => { + const s = run([ + [0, ev.call("t1", "t1#0", "a")], + [1_000, ev.stepComplete("t1", "t1#0")], + [3_000, ev.result("t1", "t1#0", "a")], + [5_000, ev.call("t1", "t1#1", "b")], + [6_000, ev.stepComplete("t1", "t1#1")], + [66_000, ev.result("t1", "t1#1", "b")], + ]); + expect(s.stepStamps.get("t1#0")).toBe(2_000); + expect(s.stepStamps.get("t1#1")).toBe(60_000); + }); + + it("uses the backend tool duration when it exceeds the wall clock (late-join replay burst)", () => { + const s = run([ + [5_000, ev.call("t1", "t1#0", "a")], + [5_000, ev.stepComplete("t1", "t1#0")], + [5_000, ev.result("t1", "t1#0", "a", 42_000)], + ]); + expect(s.stepStamps.get("t1#0")).toBe(42_000); + }); + + it("stops at the next step's output when a result never matched (fallback)", () => { + const s = run([ + [0, ev.call("t1", "t1#0", "a")], + [1_000, ev.stepComplete("t1", "t1#0")], + [9_000, ev.text("t1")], + ]); + expect(s.running).toBeNull(); + expect(s.stepStamps.get("t1#0")).toBe(8_000); + }); + + it("treats a reasoning-end boundary as resumed output too", () => { + const s = run([ + [0, ev.call("t1", "t1#0", "a")], + [1_000, ev.stepComplete("t1", "t1#0")], + [9_000, ev.reasoningEnd("t1")], + ]); + expect(s.running).toBeNull(); + expect(s.stepStamps.get("t1#0")).toBe(8_000); + }); + + it("stops on a tool-call or usage from a DIFFERENT step, not its own", () => { + const base = run([ + [0, ev.call("t1", "t1#0", "a")], + [1_000, ev.stepComplete("t1", "t1#0")], + ]); + expect(run([[2_000, ev.usage("t1", "t1#0")]], base).running?.kind).toBe("step"); + expect(run([[3_000, ev.usage("t1", "t1#1")]], base).stepStamps.get("t1#0")).toBe(2_000); + expect(run([[4_000, ev.call("t1", "t1#1", "z")]], base).stepStamps.get("t1#0")).toBe(3_000); + }); + + it("discards a step with no tool calls (the final step of a turn)", () => { + const s = run([ + [1_000, ev.stepComplete("t1", "t1#0")], + [1_050, ev.done("t1")], + ]); + expect(s.stepStamps.size).toBe(0); + expect(s.running).toEqual({ kind: "turn", turnId: "t1", startedAt: 1_050 }); + }); + + it("stamps a step whose tools were still running when the turn ended (stopped)", () => { + const s = run([ + [0, ev.call("t1", "t1#0", "a")], + [1_000, ev.stepComplete("t1", "t1#0")], + [21_000, ev.done("t1")], + ]); + expect(s.stepStamps.get("t1#0")).toBe(20_000); + expect(s.running?.kind).toBe("turn"); + }); +}); + +describe("foldCacheTimerEvent — turn gaps", () => { + it("starts at done and is stamped on that turn when the next turn starts", () => { + const s = run([ + [1_000, ev.done("t1")], + [91_000, ev.userMessage("t2")], + [91_500, ev.turnStart("t2")], + ]); + expect(s.running).toBeNull(); + expect(s.turnStamps.get("t1")).toBe(90_000); + }); + + it("an error also ends the turn", () => { + expect(run([[7, ev.error("t1")]]).running).toEqual({ + kind: "turn", + turnId: "t1", + startedAt: 7, + }); + }); + + it("stopTurnGap stamps on a local send", () => { + const s = stopTurnGap(run([[1_000, ev.done("t1")]]), 4_000, noTurn); + expect(s.running).toBeNull(); + expect(s.turnStamps.get("t1")).toBe(3_000); + }); + + it("stopTurnGap is a no-op while a step gap runs", () => { + const base = run([ + [0, ev.call("t1", "t1#0", "a")], + [1_000, ev.stepComplete("t1", "t1#0")], + ]); + expect(stopTurnGap(base, 2_000, noTurn)).toBe(base); + }); +}); + +describe("seedTurnGap", () => { + it("seeds a turn gap with no turnId, stamped on the most recent turn when stopped", () => { + const seeded = seedTurnGap(initialCacheTimerState(), 1_000); + expect(seeded.running).toEqual({ kind: "turn", turnId: null, startedAt: 1_000 }); + const s = stopTurnGap(seeded, 61_000, () => "t9"); + expect(s.turnStamps.get("t9")).toBe(60_000); + }); + + it("drops a seeded gap when the most recent turn is unknown", () => { + const s = stopTurnGap(seedTurnGap(initialCacheTimerState(), 1_000), 2_000, noTurn); + expect(s.running).toBeNull(); + expect(s.turnStamps.size).toBe(0); + }); + + it("keeps the later start, never overrides a step gap, ignores non-finite", () => { + const live = run([[5_000, ev.done("t1")]]); + expect(seedTurnGap(live, 4_000).running).toEqual({ + kind: "turn", + turnId: "t1", + startedAt: 5_000, + }); + expect(seedTurnGap(live, 6_000).running).toEqual({ + kind: "turn", + turnId: "t1", + startedAt: 6_000, + }); + const step = run([ + [0, ev.call("t1", "t1#0", "a")], + [1_000, ev.stepComplete("t1", "t1#0")], + ]); + expect(seedTurnGap(step, 9_000)).toBe(step); + expect(seedTurnGap(initialCacheTimerState(), Number.NaN).running).toBeNull(); + }); +}); + +describe("viewCacheTimer", () => { + it("formats m:ss up to 10 minutes", () => { + expect(viewCacheTimer(0).label).toBe("0:00"); + expect(viewCacheTimer(9 * SEC + 999).label).toBe("0:09"); + expect(viewCacheTimer(MIN + 5 * SEC).label).toBe("1:05"); + expect(viewCacheTimer(5 * MIN + 30 * SEC).label).toBe("5:30"); + expect(viewCacheTimer(9 * MIN + 59 * SEC).label).toBe("9:59"); + expect(viewCacheTimer(9 * MIN + 59 * SEC).capped).toBe(false); + }); + + it("reads '10+ min' and is capped at and past 10 minutes", () => { + expect(viewCacheTimer(CACHE_TIMER_MAX_MS)).toEqual({ + label: "10+ min", + level: "error", + capped: true, + }); + expect(viewCacheTimer(3 * 60 * MIN).label).toBe("10+ min"); + }); + + it("colours by the 5-minute cache TTL", () => { + expect(viewCacheTimer(3 * MIN + 59 * SEC).level).toBe("success"); + expect(viewCacheTimer(4 * MIN).level).toBe("warning"); + expect(viewCacheTimer(CACHE_TTL_MS).level).toBe("error"); + expect(viewCacheTimer(7 * MIN).level).toBe("error"); + }); + + it("clamps negative / non-finite elapsed (clock skew) to 0:00", () => { + expect(viewCacheTimer(-5 * SEC).label).toBe("0:00"); + expect(viewCacheTimer(Number.NaN).label).toBe("0:00"); + }); +}); + +function userRow(text = "hi"): MetricsRow { + const group: RenderGroup = { + kind: "single", + chunk: { seq: null, role: "user", chunk: { type: "text", text }, provisional: true }, + }; + return { kind: "group", group }; +} + +function assistantRow(turnId: string): MetricsRow { + const group: RenderGroup = { + kind: "single", + chunk: { + seq: 1, + role: "assistant", + chunk: { type: "text", text: "ok" }, + provisional: false, + turnId, + }, + }; + return { kind: "group", group }; +} + +function turnRow(turnId: string, turnNumber: number): MetricsRow { + const usage = { inputTokens: 10, outputTokens: 5 }; + const turn: TurnMetrics = { turnId, usage, steps: [{ stepId: `${turnId}#0` as StepId, usage }] }; + return { + kind: "turn-metrics", + turn, + turnNumber, + cumulativeUsage: usage, + prevLastStepUsage: null, + }; +} + +describe("lastTurnRowIndex", () => { + it("picks the LAST turn-metrics row", () => { + const rows = [ + userRow(), + assistantRow("t1"), + turnRow("t1", 1), + userRow(), + assistantRow("t2"), + turnRow("t2", 2), + ]; + expect(lastTurnRowIndex(rows)).toBe(5); + }); + + it("is null once a user message follows the last turn", () => { + expect(lastTurnRowIndex([userRow(), turnRow("t1", 1), userRow("next")])).toBeNull(); + }); + + it("skips trailing non-user rows after the turn row", () => { + expect(lastTurnRowIndex([userRow(), turnRow("t1", 1), assistantRow("t1")])).toBe(1); + }); + + it("is null with no turn-metrics rows", () => { + expect(lastTurnRowIndex([])).toBeNull(); + expect(lastTurnRowIndex([userRow(), assistantRow("t1")])).toBeNull(); + }); +}); diff --git a/src/core/metrics/cache-timer.ts b/src/core/metrics/cache-timer.ts new file mode 100644 index 0000000..3d7270d --- /dev/null +++ b/src/core/metrics/cache-timer.ts @@ -0,0 +1,286 @@ +import type { AgentEvent } from "@dispatch/wire"; +import type { MetricsRow } from "./types"; + +/** + * Pure core for the **cache timer** — how long the prompt cache has sat idle + * between model requests. Anthropic's prompt cache (as used by the Claude Code + * subscription) expires 5 minutes after its last hit, so the gaps BETWEEN + * requests decide whether the next request reads from cache: + * + * - **step gap**: a step (one LLM round-trip) finished and its tools are + * executing. Starts at `step-complete`; stops when the step's last tool + * result arrives (the next request fires right after) or, failing that, when + * anything from the next step shows up. Its final value is **stamped** on + * that step (keyed by `stepId`). + * - **turn gap**: the turn ended (`done` / `error`) and the conversation is + * waiting on the user. Stops when a new message is sent (locally, or a + * `user-message` / `turn-start` from another client) and is stamped on that + * turn (keyed by `turnId`). + * + * At most ONE gap runs at a time (`running`). Zero DOM, zero clock: every fold + * takes the current time as an argument. + */ + +/** Prompt-cache time-to-live: 5 minutes (drives the badge colour). */ +export const CACHE_TTL_MS = 5 * 60 * 1000; + +/** The timer counts up to this, then reads "10+ min". */ +export const CACHE_TIMER_MAX_MS = 10 * 60 * 1000; + +/** Within this much of the TTL the timer warns (colour only). */ +const WARN_AT_MS = CACHE_TTL_MS - 60 * 1000; + +/** A step gap in progress: the step's tools are executing. */ +export interface RunningStepGap { + readonly kind: "step"; + readonly turnId: string; + readonly stepId: string; + /** Epoch ms the step completed. */ + readonly startedAt: number; + /** How many tool calls the step made (`0` → nothing to show; never stamped). */ + readonly toolCount: number; + /** Tool calls still awaiting their result. */ + readonly pending: ReadonlySet<string>; + /** Longest backend-reported tool `durationMs` seen so far for this step. */ + readonly maxToolMs: number; +} + +/** A turn gap in progress: the turn ended and nothing has been sent since. */ +export interface RunningTurnGap { + readonly kind: "turn"; + /** + * The turn that ended, or `null` when seeded from the backend's + * `lastActivityAt` after a reload (no live `done` seen) — then it belongs to + * the most recent turn, resolved positionally / via `lastTurnId` on stop. + */ + readonly turnId: string | null; + /** Epoch ms the turn ended. */ + readonly startedAt: number; +} + +export type RunningGap = RunningStepGap | RunningTurnGap; + +export interface CacheTimerState { + readonly running: RunningGap | null; + /** Stamped step gaps: `stepId` → elapsed ms. */ + readonly stepStamps: ReadonlyMap<string, number>; + /** Stamped turn gaps: `turnId` → elapsed ms. */ + readonly turnStamps: ReadonlyMap<string, number>; + /** Tool calls seen per step in the CURRENT turn (`stepId` → call ids). */ + readonly calls: ReadonlyMap<string, ReadonlySet<string>>; + /** Tool results seen per step in the CURRENT turn (`stepId` → call ids). */ + readonly results: ReadonlyMap<string, ReadonlySet<string>>; + /** Longest tool `durationMs` per step in the CURRENT turn. */ + readonly toolMs: ReadonlyMap<string, number>; +} + +export function initialCacheTimerState(): CacheTimerState { + return { + running: null, + stepStamps: new Map(), + turnStamps: new Map(), + calls: new Map(), + results: new Map(), + toolMs: new Map(), + }; +} + +function addTo( + map: ReadonlyMap<string, ReadonlySet<string>>, + key: string, + value: string, +): ReadonlyMap<string, ReadonlySet<string>> { + const next = new Map(map); + next.set(key, new Set([...(map.get(key) ?? []), value])); + return next; +} + +function withEntry(map: ReadonlyMap<string, number>, key: string, value: number) { + const next = new Map(map); + next.set(key, value); + return next; +} + +/** + * Stop the running STEP gap (if any). A step that made tool calls is stamped + * with `max(wall-clock elapsed, longest backend tool duration)` — the backend + * figure keeps a late-join replay (whole step delivered in one burst, ~0 + * elapsed) honest. A step with no tool calls is discarded. + */ +function stopStep(state: CacheTimerState, now: number): CacheTimerState { + const r = state.running; + if (r === null || r.kind !== "step") return state; + if (r.toolCount === 0) return { ...state, running: null }; + const elapsed = Math.max(0, now - r.startedAt, r.maxToolMs); + return { ...state, running: null, stepStamps: withEntry(state.stepStamps, r.stepId, elapsed) }; +} + +/** + * Stop the running TURN gap (if any) and stamp it on its turn. A seeded gap + * (`turnId === null`) is stamped on `lastTurnId()` — the most recent turn's + * id — or dropped when that is unknown. + */ +export function stopTurnGap( + state: CacheTimerState, + now: number, + lastTurnId: () => string | undefined, +): CacheTimerState { + const r = state.running; + if (r === null || r.kind !== "turn") return state; + const turnId = r.turnId ?? lastTurnId(); + if (turnId === undefined) return { ...state, running: null }; + const elapsed = Math.max(0, now - r.startedAt); + return { ...state, running: null, turnStamps: withEntry(state.turnStamps, turnId, elapsed) }; +} + +/** A turn ended: close any step gap, forget the turn's tool bookkeeping, start the turn gap. */ +function endTurn(state: CacheTimerState, turnId: string, now: number): CacheTimerState { + const s = stopStep(state, now); + return { + ...s, + running: { kind: "turn", turnId, startedAt: now }, + calls: new Map(), + results: new Map(), + toolMs: new Map(), + }; +} + +/** + * Fold one live event into the cache-timer state at time `now`. `lastTurnId` + * resolves a seeded turn gap's turn when a new turn starts (see + * `stopTurnGap`); it is only called in that case. + */ +export function foldCacheTimerEvent( + state: CacheTimerState, + event: AgentEvent, + now: number, + lastTurnId: () => string | undefined, +): CacheTimerState { + const r = state.running; + switch (event.type) { + case "user-message": + case "turn-start": + // A new turn is starting — the user (here or elsewhere) sent a message. + return stopTurnGap(state, now, lastTurnId); + + case "tool-call": { + // A call from a DIFFERENT step means the next step is already streaming. + const s = r?.kind === "step" && r.stepId !== event.stepId ? stopStep(state, now) : state; + return { ...s, calls: addTo(s.calls, event.stepId, event.toolCallId) }; + } + + case "tool-result": { + const results = addTo(state.results, event.stepId, event.toolCallId); + const dur = event.durationMs ?? 0; + const toolMs = + dur > (state.toolMs.get(event.stepId) ?? 0) + ? withEntry(state.toolMs, event.stepId, dur) + : state.toolMs; + let s: CacheTimerState = { ...state, results, toolMs }; + if (r?.kind === "step" && r.stepId === event.stepId) { + const pending = new Set(r.pending); + pending.delete(event.toolCallId); + s = { ...s, running: { ...r, pending, maxToolMs: Math.max(r.maxToolMs, dur) } }; + // Every tool of the step has returned → the next request fires now. + if (pending.size === 0) s = stopStep(s, now); + } + return s; + } + + case "step-complete": { + const s = stopStep(stopTurnGap(state, now, lastTurnId), now); + const calls = s.calls.get(event.stepId) ?? new Set<string>(); + const done = s.results.get(event.stepId) ?? new Set<string>(); + const pending = new Set([...calls].filter((id) => !done.has(id))); + return { + ...s, + running: { + kind: "step", + turnId: event.turnId, + stepId: event.stepId, + startedAt: now, + toolCount: calls.size, + pending, + maxToolMs: s.toolMs.get(event.stepId) ?? 0, + }, + }; + } + + case "text-delta": + case "reasoning-delta": + case "reasoning-end": + case "provider-retry": + // Output (or a retry) from the next request: the gap is over. + return r?.kind === "step" ? stopStep(state, now) : state; + + case "usage": + return r?.kind === "step" && event.stepId !== undefined && event.stepId !== r.stepId + ? stopStep(state, now) + : state; + + case "done": + case "error": + return endTurn(state, event.turnId, now); + + default: + return state; + } +} + +/** + * Seed a turn gap from an authoritative "last activity" timestamp (the + * backend's `lastActivityAt`, for a conversation that is idle — e.g. after a + * reload, when no live `done` was seen). Never overrides a running step gap; + * a running turn gap keeps the LATER start. Non-finite input is ignored. + */ +export function seedTurnGap(state: CacheTimerState, at: number): CacheTimerState { + if (!Number.isFinite(at)) return state; + const r = state.running; + if (r === null) return { ...state, running: { kind: "turn", turnId: null, startedAt: at } }; + if (r.kind === "turn" && at > r.startedAt) return { ...state, running: { ...r, startedAt: at } }; + return state; +} + +// ── View ────────────────────────────────────────────────────────────────────── + +/** Formatted cache-timer view: label + colour severity + capped flag. */ +export interface CacheTimerView { + /** `"m:ss"` up to 10 minutes, then `"10+ min"`. */ + readonly label: string; + /** Colour by the 5-minute cache TTL (maps to DaisyUI `badge-{level}`). */ + readonly level: "success" | "warning" | "error"; + /** Whether the 10-minute display cap was reached (a live timer may stop ticking). */ + readonly capped: boolean; +} + +/** + * Build the view for `elapsedMs`. Colour tracks the cache TTL: success under + * 4 min, warning in the last minute, error once the cache has likely expired. + * Negative / non-finite elapsed (clock skew) clamps to `0:00`. + */ +export function viewCacheTimer(elapsedMs: number): CacheTimerView { + const ms = Number.isFinite(elapsedMs) && elapsedMs > 0 ? elapsedMs : 0; + const level = ms >= CACHE_TTL_MS ? "error" : ms >= WARN_AT_MS ? "warning" : "success"; + if (ms >= CACHE_TIMER_MAX_MS) return { label: "10+ min", level, capped: true }; + const totalSeconds = Math.floor(ms / 1000); + const minutes = Math.floor(totalSeconds / 60); + const seconds = totalSeconds % 60; + return { label: `${minutes}:${seconds.toString().padStart(2, "0")}`, level, capped: false }; +} + +/** + * Index of the most recent turn's metrics row — the LAST `turn-metrics` row, + * provided no user message follows it. Where a SEEDED turn gap (no `turnId`) + * renders. `null` when there is no such row. + */ +export function lastTurnRowIndex(rows: readonly MetricsRow[]): number | null { + for (let i = rows.length - 1; i >= 0; i--) { + const row = rows[i]; + if (row === undefined) continue; + if (row.kind === "turn-metrics") return i; + if (row.kind === "group" && row.group.kind === "single" && row.group.chunk.role === "user") { + return null; + } + } + return null; +} diff --git a/src/core/metrics/format.test.ts b/src/core/metrics/format.test.ts index 97170d0..3e8ebdd 100644 --- a/src/core/metrics/format.test.ts +++ b/src/core/metrics/format.test.ts @@ -46,7 +46,7 @@ describe("viewStepMetrics", () => { }; const view = viewStepMetrics(step, 0); expect(view.label).toBe("step 1"); - expect(view.tokensLabel).toBe("1,801 tok"); + expect(view.tokensLabel).toBe("1,234 in / 567 out"); expect(view.tps).toBe("473 tok/s"); expect(view.ttft).toBe("820ms"); expect(view.decode).toBe("1.2s"); @@ -105,15 +105,14 @@ describe("viewStepMetrics", () => { expect(view.tps).toBe("100 tok/s"); }); - it("tps falls back to genTotalMs when decodeMs absent", () => { + it("tps is null when decodeMs is absent (never falls back to genTotalMs, which includes TTFT)", () => { 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"); + expect(view.tps).toBeNull(); }); }); @@ -133,8 +132,8 @@ describe("viewTurnMetrics", () => { ], }; const view = viewTurnMetrics(turn); - expect(view.tokensLabel).toBe("1,234 tok"); - expect(view.breakdown).toBe("1,000 in / 234 out"); + expect(view.tokensLabel).toBe("234 tok out"); + expect(view.breakdown).toBe("1,000 in"); expect(view.tps).toBe("78 tok/s"); expect(view.duration).toBe("5.0s"); }); @@ -146,7 +145,7 @@ describe("viewTurnMetrics", () => { steps: [], }; const view = viewTurnMetrics(turn); - expect(view.breakdown).toBe("1,000 in / 234 out / 500 cache"); + expect(view.breakdown).toBe("1,000 in (500 cached)"); }); it("breakdown omits cache when not present", () => { @@ -156,7 +155,7 @@ describe("viewTurnMetrics", () => { steps: [], }; const view = viewTurnMetrics(turn); - expect(view.breakdown).toBe("100 in / 50 out"); + expect(view.breakdown).toBe("100 in"); }); it("tps is null when no step has decodeMs or genTotalMs", () => { @@ -184,7 +183,7 @@ describe("viewTurnMetrics", () => { expect(view.duration).toBeNull(); }); - it("sums decodeMs across steps (fallback genTotalMs per step) for tps", () => { + it("turn tps counts only steps that report decodeMs (both tokens and time)", () => { const turn: TurnMetrics = { turnId: "t1", usage: { inputTokens: 300, outputTokens: 150 }, @@ -203,9 +202,9 @@ describe("viewTurnMetrics", () => { ], }; 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"); + // only step1 is timed: 50 / (800/1000) = 62.5 → 63. step2 (no decodeMs) + // contributes neither tokens nor time — never its TTFT-inflated genTotalMs. + expect(view.tps).toBe("63 tok/s"); }); }); @@ -260,40 +259,47 @@ describe("computeExpectedCachePct", () => { 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("null when this turn has no first step", () => { + expect(computeExpectedCachePct(undefined, { inputTokens: 100, outputTokens: 0 })).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("null when the prior prompt was empty (denominator 0)", () => { + const prev = { inputTokens: 0, outputTokens: 0 }; + expect(computeExpectedCachePct({ inputTokens: 10, outputTokens: 0 }, prev)).toBeNull(); }); - 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("100% when the first step reads back the whole prior final prompt", () => { + // prev turn's last step sent 5149 tokens; this turn's first step (a + // superset prompt) reads 5149 of them from cache. + const prevLast = { inputTokens: 5149, outputTokens: 40, cacheReadTokens: 3000 }; + const first = { inputTokens: 8462, outputTokens: 0, cacheReadTokens: 5149 }; + expect(computeExpectedCachePct(first, prevLast)).toBe(100); + }); + + it("drops below 100% when the cache busted", () => { + const prevLast = { inputTokens: 1000, outputTokens: 0, cacheReadTokens: 900 }; + const first = { inputTokens: 1200, outputTokens: 0, cacheReadTokens: 500 }; + expect(computeExpectedCachePct(first, prevLast)).toBe(50); + }); + + it("0% (not null) on a full miss", () => { + const prevLast = { inputTokens: 1000, outputTokens: 0 }; + const first = { inputTokens: 1200, outputTokens: 0 }; + expect(computeExpectedCachePct(first, prevLast)).toBe(0); + }); + + it("is NOT inflated by multi-step turns (step-level, not turn sums)", () => { + // A 10-step turn reads the same prefix 10x; the old turn-sum ratio read + // ~10x the baseline and pinned at 100%. Step-level reflects the bust. + const prevLast = { inputTokens: 50_000, outputTokens: 0, cacheReadTokens: 49_000 }; + const first = { inputTokens: 51_000, outputTokens: 0, cacheReadTokens: 10_000 }; + expect(computeExpectedCachePct(first, prevLast)).toBe(20); }); - 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); + it("clamps to 100", () => { + const prevLast = { inputTokens: 100, outputTokens: 0 }; + const first = { inputTokens: 300, outputTokens: 0, cacheReadTokens: 250 }; + expect(computeExpectedCachePct(first, prevLast)).toBe(100); }); }); @@ -303,9 +309,9 @@ describe("viewExpectedCache", () => { }); 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); + const prevLast = { inputTokens: 5146, outputTokens: 0 }; + const first = { inputTokens: 8462, outputTokens: 0, cacheReadTokens: 5146 }; + const v = viewExpectedCache(first, prevLast); expect(v?.pct).toBe(100); expect(v?.level).toBe("success"); expect(v?.isHit).toBe(true); diff --git a/src/core/metrics/format.ts b/src/core/metrics/format.ts index 894bd54..5a13a9a 100644 --- a/src/core/metrics/format.ts +++ b/src/core/metrics/format.ts @@ -77,26 +77,31 @@ export function computeTps(outputTokens: number, elapsedMs: number | undefined): 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`; + let s = `${formatTokens(u.inputTokens)} in`; if (u.cacheReadTokens !== undefined && u.cacheReadTokens > 0) { - s += ` / ${formatTokens(u.cacheReadTokens)} cache`; + s += ` (${formatTokens(u.cacheReadTokens)} cached)`; } return s; } +/** + * A step's tokens-per-second: `outputTokens / decodeMs`. Decode time is the + * model's generation window (first output token → stream end) with tool + * execution excluded by the backend. `genTotalMs` is deliberately NOT a + * fallback: it includes time-to-first-token (prompt processing / queueing), + * which would understate the rate by multiples. No decode time → no TPS. + */ +function stepTps(step: StepMetrics): number | null { + return computeTps(step.usage.outputTokens, step.decodeMs); +} + /** 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), + tokensLabel: `${formatTokens(step.usage.inputTokens)} in / ${formatTokens(step.usage.outputTokens)} out`, + tps: formatTps(stepTps(step)), ttft: formatDuration(step.ttftMs), decode: formatDuration(step.decodeMs), genTotal: formatDuration(step.genTotalMs), @@ -130,19 +135,25 @@ export function viewCacheRate(u: Usage): CacheRateView { } /** - * 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. + * Expected cache (retention): of the prompt the PREVIOUS turn last sent (its + * final step's input), how much this turn's FIRST step read back from cache — + * `clamp01(firstStep.cacheRead / prevLastStep.input)`. Everything in that + * prompt is a prefix of this turn's first prompt, so ~100% means the cache + * survived between turns; lower means it busted/expired. Step-level on both + * sides: comparing whole-turn sums is meaningless (a multi-step turn reads + * the same prefix once per step). * - * 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%. + * Returns `null` when it cannot be derived: no earlier turn, no first step, + * or the previous prompt was empty — 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); +export function computeExpectedCachePct( + firstStep: Usage | undefined, + prevLastStep: Usage | null, +): number | null { + if (prevLastStep === null || firstStep === undefined) return null; + const denom = prevLastStep.inputTokens; if (denom <= 0) return null; - const read = current.cacheReadTokens ?? 0; + const read = firstStep.cacheReadTokens ?? 0; const rate = read / denom; const clamped = rate < 0 ? 0 : rate > 1 ? 1 : rate; return Math.round(clamped * 100); @@ -152,28 +163,36 @@ export function computeExpectedCachePct(current: Usage, prev: Usage | null): num * 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); +export function viewExpectedCache( + firstStep: Usage | undefined, + prevLastStep: Usage | null, +): CacheRateView | null { + const pct = computeExpectedCachePct(firstStep, prevLastStep); if (pct === null) return null; - return { pct, level: cacheLevel(pct), isHit: (current.cacheReadTokens ?? 0) > 0 }; + return { pct, level: cacheLevel(pct), isHit: (firstStep?.cacheReadTokens ?? 0) > 0 }; } -/** Build a formatted view of a turn's aggregate metrics. */ +/** + * Build a formatted view of a turn's aggregate metrics. + * + * Turn TPS = output tokens ÷ decode time, summed over ONLY the steps that + * report a decode time (both numerator and denominator), so a step without + * timing can neither inflate nor deflate it. Tool execution between steps is + * never part of any step's decode time. + */ export function viewTurnMetrics(turn: TurnMetrics, turnNumber?: number): TurnMetricsView { - const total = totalTokens(turn.usage); - let totalGenMs: number | undefined; + let timedOut = 0; + let decodeMs: number | undefined; for (const step of turn.steps) { - const stepMs = step.decodeMs ?? step.genTotalMs; - if (stepMs !== undefined) { - totalGenMs = (totalGenMs ?? 0) + stepMs; - } + if (step.decodeMs === undefined || step.decodeMs <= 0) continue; + timedOut += step.usage.outputTokens; + decodeMs = (decodeMs ?? 0) + step.decodeMs; } - const tps = computeTps(turn.usage.outputTokens, totalGenMs); return { label: turnNumber !== undefined ? `turn ${turnNumber}` : "turn", - tokensLabel: `${formatTokens(total)} tok`, + tokensLabel: `${formatTokens(turn.usage.outputTokens)} tok out`, breakdown: formatBreakdown(turn.usage), - tps: formatTps(tps), + tps: formatTps(computeTps(timedOut, decodeMs)), duration: formatDuration(turn.durationMs), }; } diff --git a/src/core/metrics/index.ts b/src/core/metrics/index.ts index d3c9669..d9df7f5 100644 --- a/src/core/metrics/index.ts +++ b/src/core/metrics/index.ts @@ -1,4 +1,19 @@ export { + CACHE_TIMER_MAX_MS, + CACHE_TTL_MS, + type CacheTimerState, + type CacheTimerView, + foldCacheTimerEvent, + initialCacheTimerState, + lastTurnRowIndex, + type RunningGap, + type RunningStepGap, + type RunningTurnGap, + seedTurnGap, + stopTurnGap, + viewCacheTimer, +} from "./cache-timer"; +export { type ContextUsage, computeCachePct, computeContextUsage, diff --git a/src/core/metrics/place.test.ts b/src/core/metrics/place.test.ts index 9c925a3..7695e21 100644 --- a/src/core/metrics/place.test.ts +++ b/src/core/metrics/place.test.ts @@ -28,31 +28,36 @@ function assistantGroup(seq: number, text: string): RenderGroup { }; } -function toolCallGroup(seq: number, stepId: string, toolCallId: string): RenderGroup { +function toolCallGroup(stepId: string, toolCallId: string): RenderGroup { return { - kind: "single", - chunk: { - seq, - role: "assistant", - chunk: { + kind: "tool", + stepId, + provisional: false, + entry: { + toolCallId, + toolName: "test", + result: null, + call: { type: "tool-call", toolCallId, toolName: "test", input: {}, stepId: stepId as StepId, }, - provisional: false, }, }; } -function toolResultGroup(seq: number, stepId: string, toolCallId: string): RenderGroup { +function toolResultGroup(stepId: string, toolCallId: string): RenderGroup { return { - kind: "single", - chunk: { - seq, - role: "tool", - chunk: { + kind: "tool", + stepId, + provisional: false, + entry: { + toolCallId, + toolName: "test", + call: null, + result: { type: "tool-result", toolCallId, toolName: "test", @@ -60,29 +65,10 @@ function toolResultGroup(seq: number, stepId: string, toolCallId: string): Rende 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, @@ -169,9 +155,9 @@ describe("interleaveTurnMetrics", () => { it("head-aligned: segment i gets entries[i]", () => { const g1 = userGroup(1, "q1"); - const g2 = toolCallGroup(2, "s1", "c1"); + const g2 = toolCallGroup("s1", "c1"); const g3 = userGroup(3, "q2"); - const g4 = toolCallGroup(4, "s2", "c2"); + const g4 = toolCallGroup("s2", "c2"); const step1 = makeStep("s1", 100, 50); const step2 = makeStep("s2", 200, 80); const rows = interleaveTurnMetrics( @@ -192,7 +178,7 @@ describe("interleaveTurnMetrics", () => { 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 g2 = toolCallGroup("s1", "c1"); const g3 = userGroup(3, "q2"); const g4 = assistantGroup(4, "a2"); const step = makeStep("s1", 100, 50); @@ -220,26 +206,28 @@ describe("interleaveTurnMetrics", () => { expectTurnMetricsAt(rows, 2, "t1"); }); - it("tool step anchors inline after its tool-batch group", () => { + it("tool step anchors inline after its LAST independent tool card", () => { const g1 = userGroup(1, "q1"); - const g2 = toolBatchGroup("t#0", ["c1", "c2"]); + const g2 = toolCallGroup("t#0", "c1"); + const parallel = toolCallGroup("t#0", "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]); + const rows = interleaveTurnMetrics([g1, g2, parallel, g3], [turn]); - expect(rows).toHaveLength(5); + expect(rows).toHaveLength(6); expectGroupAt(rows, 0, g1); expectGroupAt(rows, 1, g2); - expectStepMetricsAt(rows, 2, "t#0", 0); - expectGroupAt(rows, 3, g3); - expectTurnMetricsAt(rows, 4, "t1"); + expectGroupAt(rows, 2, parallel); + expectStepMetricsAt(rows, 3, "t#0", 0); + expectGroupAt(rows, 4, g3); + expectTurnMetricsAt(rows, 5, "t1"); }); it("single tool-call group anchors its step", () => { const g1 = userGroup(1, "q1"); - const g2 = toolCallGroup(2, "s1", "c1"); + const g2 = toolCallGroup("s1", "c1"); const g3 = assistantGroup(3, "a1"); const step = makeStep("s1", 100, 50); const turn = makeEntry("t1", 100, 50, [step]); @@ -255,7 +243,7 @@ describe("interleaveTurnMetrics", () => { it("single tool-result group anchors its step", () => { const g1 = userGroup(1, "q1"); - const g2 = toolResultGroup(2, "s1", "c1"); + const g2 = toolResultGroup("s1", "c1"); const g3 = assistantGroup(3, "a1"); const step = makeStep("s1", 100, 50); const turn = makeEntry("t1", 100, 50, [step]); @@ -271,9 +259,9 @@ describe("interleaveTurnMetrics", () => { it("multi-step: each tool step inline, unanchored text step skipped", () => { const g1 = userGroup(1, "q1"); - const g2 = toolBatchGroup("t#0", ["c1"]); + const g2 = toolCallGroup("t#0", "c1"); const g3 = assistantGroup(2, "thinking"); - const g4 = toolBatchGroup("t#1", ["c2", "c3"]); + const g4 = toolCallGroup("t#1", "c2"); const g5 = assistantGroup(3, "a1"); const step0 = makeStep("t#0", 100, 50); const step1 = makeStep("t#1", 200, 80); @@ -294,10 +282,10 @@ describe("interleaveTurnMetrics", () => { it("multiple turns head-aligned with inline steps", () => { const g1 = userGroup(1, "q1"); - const g2 = toolBatchGroup("s1", ["c1"]); + const g2 = toolCallGroup("s1", "c1"); const g3 = assistantGroup(2, "a1"); const g4 = userGroup(3, "q2"); - const g5 = toolCallGroup(4, "s2", "c2"); + const g5 = toolCallGroup("s2", "c2"); const step1 = makeStep("s1", 100, 50); const step2 = makeStep("s2", 200, 80); const rows = interleaveTurnMetrics( @@ -332,7 +320,7 @@ describe("interleaveTurnMetrics", () => { it("fewer metrics than segments: trailing segments are bare", () => { const g1 = userGroup(1, "q1"); - const g2 = toolCallGroup(2, "s1", "c1"); + const g2 = toolCallGroup("s1", "c1"); const g3 = userGroup(3, "q2"); const g4 = assistantGroup(4, "a2"); const g5 = userGroup(5, "q3"); @@ -361,7 +349,7 @@ describe("interleaveTurnMetrics", () => { // 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 g4 = toolCallGroup("s2", "c2"); const g5 = assistantGroup(4, "tool-reply"); const g6 = userGroup(5, "q3"); const g7 = assistantGroup(6, "text-reply"); @@ -401,7 +389,12 @@ describe("interleaveTurnMetrics", () => { total: { turnId: "t1", usage: { inputTokens: 1000, outputTokens: 10, cacheReadTokens: 500 }, - steps: [], + steps: [ + { + stepId: "t1#0" as StepId, + usage: { inputTokens: 1000, outputTokens: 10, cacheReadTokens: 500 }, + }, + ], }, }, { @@ -424,14 +417,14 @@ describe("interleaveTurnMetrics", () => { // 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); + // Retention baseline is the prior finalized turn's LAST step (t1, even though trimmed). + expect(tmRows[0]?.prevLastStepUsage?.inputTokens).toBe(1000); + expect(tmRows[0]?.prevLastStepUsage?.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 g2 = toolCallGroup("s1", "c1"); const step = makeStep("s1", 100, 50); const turn: TurnMetricsEntry = { turnId: "t1", @@ -454,7 +447,7 @@ describe("interleaveTurnMetrics", () => { 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 g2 = toolCallGroup("s1", "c1"); const step = makeStep("s1", 100, 50); const rows = interleaveTurnMetrics([g0, g1, g2], [makeEntry("t1", 100, 50, [step])]); @@ -468,7 +461,7 @@ describe("interleaveTurnMetrics", () => { 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 g2 = toolCallGroup("s1", "c1"); const step1 = makeStep("s1", 100, 50); const step2 = makeStep("s2", 200, 80); const rows = interleaveTurnMetrics( @@ -504,7 +497,7 @@ describe("interleaveTurnMetrics", () => { 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 g2 = toolCallGroup("s1", "c1"); const g3 = assistantGroup(2, "a1"); const step1 = makeStep("s1", 100, 50); const entry = makeProgressiveEntry("t1", [step1]); @@ -519,7 +512,7 @@ describe("interleaveTurnMetrics", () => { it("entry with total emits step rows + a turn-metrics row", () => { const g1 = userGroup(1, "q1"); - const g2 = toolBatchGroup("s1", ["c1"]); + const g2 = toolCallGroup("s1", "c1"); const g3 = assistantGroup(2, "a1"); const step1 = makeStep("s1", 100, 50); const entry = makeEntry("t1", 100, 50, [step1]); @@ -606,16 +599,36 @@ describe("interleaveTurnMetrics — cumulative usage (cache total)", () => { expect(tm[0]?.cumulativeUsage.cacheReadTokens).toBe(500); }); - it("carries the prior finalized turn's usage as the retention baseline", () => { + it("carries the prior finalized turn's LAST STEP usage as the retention baseline", () => { + const steps = (id: string): StepMetrics[] => [ + { stepId: `${id}#0` as StepId, usage: { inputTokens: 1000, outputTokens: 5 } }, + { + stepId: `${id}#1` as StepId, + usage: { inputTokens: 1669, outputTokens: 5, cacheReadTokens: 384 }, + }, + ]; + const t1: TurnMetrics = { + turnId: "t1", + usage: { inputTokens: 2669, outputTokens: 10, cacheReadTokens: 384 }, + steps: steps("t1"), + }; + const t2: TurnMetrics = { + turnId: "t2", + usage: { inputTokens: 2737, outputTokens: 10, cacheReadTokens: 2560 }, + steps: steps("t2"), + }; 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)], + [ + { turnId: "t1", steps: t1.steps, total: t1 }, + { turnId: "t2", steps: t2.steps, total: t2 }, + ], ); 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); + expect(tm[0]?.prevLastStepUsage).toBeNull(); + // second turn's baseline is the first turn's LAST step (not the turn sum) + expect(tm[1]?.prevLastStepUsage?.inputTokens).toBe(1669); + expect(tm[1]?.prevLastStepUsage?.cacheReadTokens).toBe(384); }); }); diff --git a/src/core/metrics/place.ts b/src/core/metrics/place.ts index 7122b09..d4c343a 100644 --- a/src/core/metrics/place.ts +++ b/src/core/metrics/place.ts @@ -1,11 +1,17 @@ import type { Usage } from "@dispatch/wire"; import type { RenderGroup } from "../chunks"; -import type { MetricsRow, TurnMetricsEntry } from "./types"; +import type { MetricsRow, StepMetrics, 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; + return g.kind === "tool" ? g.stepId : undefined; +} + +function groupTurnId(g: RenderGroup): string | undefined { + return g.kind === "tool" ? g.turnId : g.chunk.turnId; +} + +function isUser(g: RenderGroup): boolean { + return g.kind === "single" && g.chunk.role === "user"; } /** Element-wise sum of two token usages (cache fields included only when nonzero). */ @@ -21,277 +27,255 @@ function addUsage(a: Usage, b: Usage): Usage { return out; } +/** A run of consecutive groups that belong to one (possibly unknown) turn. */ +interface Segment { + readonly start: number; + readonly end: number; // exclusive + /** Turn resolved from the segment's own groups (turnId / stepId), if any. */ + readonly known: string | undefined; + /** Whether the segment holds model output (anything but user / error chunks). */ + readonly hasResponse: boolean; +} + /** - * Interleave turn metrics into the rendered transcript. + * Assign every group to the turn that produced it. EXACT whenever the data + * carries provenance — which is always for anything produced since the backend + * began stamping `StoredChunk.turnId` (and always for live chunks, which carry + * the live event's `turnId`): * - * 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`. + * 1. A group's own `turnId` (committed `StoredChunk.turnId` / live event). + * 2. Else the turn whose metrics contain the group's `stepId`. * - * 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. + * Only LEGACY groups (persisted before turn provenance existed, with no + * stepId) are left; they are resolved per segment. A segment boundary is a + * user message start (a user chunk after a non-user one, or any user TEXT + * chunk — a message carries at most one, first; image chunks continue it) or a + * change of known turn. A segment with a known turn assigns it to its legacy + * groups. Runs of still-unknown (text-only legacy) segments are matched to the + * UNCLAIMED metrics entries that lie strictly between their resolved + * neighbours' entries — order-preserving, and only when the count is + * unambiguous (leading runs tail-align, since chat-limit trimming removes the + * oldest turns). An ambiguous run gets NO metrics rather than wrong ones. */ -export function interleaveTurnMetrics( +function assignTurns( groups: readonly RenderGroup[], entries: readonly TurnMetricsEntry[], -): readonly MetricsRow[] { - if (entries.length === 0) { - return groups.map((g) => ({ kind: "group" as const, group: g })); - } +): (string | undefined)[] { + const turnOfStep = new Map<string, string>(); + const entryIndex = new Map<string, number>(); + entries.forEach((e, i) => { + entryIndex.set(e.turnId, i); + for (const s of e.steps) turnOfStep.set(s.stepId, e.turnId); + }); + + const direct: (string | undefined)[] = groups.map((g) => { + const t = groupTurnId(g); + if (t !== undefined) return t; + const sid = groupStepId(g); + return sid !== undefined ? turnOfStep.get(sid) : undefined; + }); - const segmentStarts: number[] = []; + // Fast path: everything resolved directly (the normal case). + if (direct.every((t) => t !== undefined)) return direct; + + // Segment the transcript (legacy resolution). + const segments: Segment[] = []; + let segStart = 0; + let segKnown: string | undefined; + let segHasResponse = false; + const close = (end: number) => { + if (end > segStart) { + segments.push({ start: segStart, end, known: segKnown, hasResponse: segHasResponse }); + } + }; 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); + if (g === undefined) continue; + const prev = i > 0 ? groups[i - 1] : undefined; + const t = direct[i]; + const userStart = + isUser(g) && + (prev === undefined || + !isUser(prev) || + (g.kind === "single" && g.chunk.chunk.type === "text")); + const turnChange = t !== undefined && segKnown !== undefined && t !== segKnown; + if (i > segStart && (userStart || turnChange)) { + close(i); + segStart = i; + segKnown = undefined; + segHasResponse = false; + } + if (t !== undefined && segKnown === undefined) segKnown = t; + if (!isUser(g) && !(g.kind === "single" && g.chunk.chunk.type === "error")) { + segHasResponse = true; } } + close(groups.length); - 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 segTurn: (string | undefined)[] = segments.map((s) => s.known); + const claimed = new Set<number>(); + for (const t of segTurn) { + const idx = t !== undefined ? entryIndex.get(t) : undefined; + if (idx !== undefined) claimed.add(idx); } + // Entry index of a resolved segment; a known turn with no entry yet (an + // in-flight turn whose first step hasn't completed) sits after every entry. + const segEntryIdx = (k: number): number => { + const t = segTurn[k]; + if (t === undefined) return -1; + return entryIndex.get(t) ?? entries.length; + }; - 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); + let k = 0; + while (k < segments.length) { + if (segTurn[k] !== undefined) { + k++; + continue; } - if (segStepIds.size === 0) continue; // text-only — defer to pass 2 + let runEnd = k; + while (runEnd < segments.length && segTurn[runEnd] === undefined) runEnd++; + const leading = k === 0; + const lower = leading ? -1 : segEntryIdx(k - 1); + const upper = runEnd < segments.length ? segEntryIdx(runEnd) : entries.length; + const candidates: number[] = []; + for (let e = lower + 1; e < upper; e++) if (!claimed.has(e)) candidates.push(e); - 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; - } + let run: number[] = []; + for (let s = k; s < runEnd; s++) run.push(s); + if (run.length !== candidates.length) { + // A segment with no model output (only the user message, or an error) is + // a turn that failed before producing anything — or one just sent, whose + // turn hasn't started yet — and legitimately has no metrics. + const responsive = run.filter((s) => segments[s]?.hasResponse === true); + if (responsive.length <= candidates.length) run = responsive; } - if (bestEntry >= 0) { - usedEntries.add(bestEntry); - const e = entries[bestEntry]; - if (e !== undefined) { - segmentEntry.set(seg, e); - segmentEntryIndex.set(seg, bestEntry); - } + let pick: number[] | null = null; + if (run.length === candidates.length) pick = candidates; + else if (leading && run.length < candidates.length) pick = candidates.slice(-run.length); + if (pick !== null) { + run.forEach((s, i) => { + const e = pick?.[i]; + const entry = e !== undefined ? entries[e] : undefined; + if (entry !== undefined && e !== undefined) { + segTurn[s] = entry.turnId; + claimed.add(e); + } + }); } + k = runEnd; } - // 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++; - } - } + const out = [...direct]; + segments.forEach((s, si) => { + const t = segTurn[si]; + if (t === undefined) return; + for (let i = s.start; i < s.end; i++) if (out[i] === undefined) out[i] = t; + }); + return out; +} + +/** + * Interleave per-turn / per-step metrics into the rendered transcript. + * + * Every group is attributed to the turn that produced it (see `assignTurns` — + * exact via `turnId` provenance; a guarded order-preserving fallback only for + * legacy rows). Then: + * - each completed step's metrics row goes right after the LAST group bearing + * that step's `stepId` (steps with no visible content — text-only steps, or + * content unloaded by the chat limit — get no row); + * - a finalized turn's (`entry.total !== null`) metrics row goes right after + * the LAST group of that turn — so a turn whose user message was unloaded + * still shows its total under its visible tail, and a mid-turn steering + * message never pulls the total up. + * + * Because attribution never depends on counting user messages, sending a new + * message, a new turn streaming, steering, image attachments, errored turns, or + * chat-limit unloading can never shift one turn's metrics onto another. + * + * Turns whose content is fully unloaded are not rendered (no wall of stale + * rows at the top) but still count toward the running "chat total". + */ +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 })); } - // 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 turnOf = assignTurns(groups, entries); + + const entryByTurn = new Map<string, { entry: TurnMetricsEntry; index: number }>(); + entries.forEach((entry, index) => { + entryByTurn.set(entry.turnId, { entry, index }); + }); + + // Running cumulative usage across ALL finalized turns (entry order), and the + // previous finalized turn's LAST step usage (the retention baseline: the + // prompt that turn last sent, which the next turn's first step should hit in + // cache). const cumulativeByEntry: Usage[] = []; - const prevUsageByEntry: (Usage | null)[] = []; + const prevLastStepByEntry: (Usage | null)[] = []; let runningUsage: Usage = { inputTokens: 0, outputTokens: 0 }; - let lastFinalizedUsage: Usage | null = null; + let lastFinalizedStep: Usage | null = null; for (const e of entries) { - prevUsageByEntry.push(lastFinalizedUsage); + prevLastStepByEntry.push(lastFinalizedStep); if (e.total !== null) { runningUsage = addUsage(runningUsage, e.total.usage); - lastFinalizedUsage = e.total.usage; + const last = e.total.steps[e.total.steps.length - 1]; + if (last !== undefined) lastFinalizedStep = last.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); + // Anchor positions: last group of each step (within its turn) / each turn. + const lastGroupOfStep = new Map<string, number>(); + const lastGroupOfTurn = new Map<string, number>(); + groups.forEach((g, i) => { + const t = turnOf[i]; + if (t === undefined) return; + lastGroupOfTurn.set(t, i); + const sid = groupStepId(g); + if (sid !== undefined) lastGroupOfStep.set(sid, i); + }); - 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); + const stepsAt = new Map<number, { step: StepMetrics; index: number }[]>(); + for (const entry of entries) { + entry.steps.forEach((step, index) => { + const at = lastGroupOfStep.get(step.stepId); + if (at === undefined || turnOf[at] !== entry.turnId) return; + let arr = stepsAt.get(at); + if (arr === undefined) { + arr = []; + stepsAt.set(at, arr); } - } - - // 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(); + arr.push({ step, index }); + }); + } - 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. - } + const rows: MetricsRow[] = []; + for (let i = 0; i < groups.length; i++) { + const g = groups[i]; + if (g === undefined) continue; + rows.push({ kind: "group", group: g }); - // 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 }); - } - } + const steps = stepsAt.get(i); + if (steps !== undefined) { + steps.sort((a, b) => a.index - b.index); + for (const { step, index } of steps) rows.push({ kind: "step-metrics", step, index }); } - // 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, - }); - } + const t = turnOf[i]; + if (t === undefined || lastGroupOfTurn.get(t) !== i) continue; + const found = entryByTurn.get(t); + if (found === undefined || found.entry.total === null) continue; + rows.push({ + kind: "turn-metrics", + turn: found.entry.total, + turnNumber: found.index + 1, + cumulativeUsage: cumulativeByEntry[found.index] ?? found.entry.total.usage, + prevLastStepUsage: prevLastStepByEntry[found.index] ?? null, + }); } return rows; diff --git a/src/core/metrics/placement-scenarios.test.ts b/src/core/metrics/placement-scenarios.test.ts new file mode 100644 index 0000000..ce3ed5b --- /dev/null +++ b/src/core/metrics/placement-scenarios.test.ts @@ -0,0 +1,376 @@ +// End-to-end placement scenarios: real transcript + metrics reducers fed real +// AgentEvents, rendered through groupRenderedChunks → interleaveTurnMetrics. +// The invariant under test: at EVERY point of a turn (not just after it ends), +// each turn's metrics sit under that turn — never under a neighbour. +import type { AgentEvent, StepId, StoredChunk, TurnMetrics, Usage } from "@dispatch/wire"; +import { describe, expect, it } from "vitest"; +import { + appendUserMessage, + applyHistory, + foldEvent, + groupRenderedChunks, + initialState, + selectChunks, + type TranscriptState, + trimTranscript, +} from "../chunks"; +import { interleaveTurnMetrics } from "./place"; +import { + applyDurableMetrics, + foldMetricsEvent, + initialMetricsState, + selectOrderedTurnMetrics, +} from "./reducer"; +import type { MetricsRow, MetricsState } from "./types"; + +const C = "c1"; +const u = (i: number, o: number): Usage => ({ inputTokens: i, outputTokens: o }); + +/** Ordered list of "what follows each user text bubble / which turn rows appear where". */ +function layout(rows: readonly MetricsRow[]): string[] { + const out: string[] = []; + for (const r of rows) { + if (r.kind === "turn-metrics") out.push(`[turn ${r.turn.turnId}]`); + else if (r.kind === "step-metrics") out.push(`[step ${r.step.stepId}]`); + else if (r.group.kind === "single") { + const c = r.group.chunk.chunk; + if (c.type === "text") out.push(`${r.group.chunk.role}:${c.text}`); + else if (c.type === "image") out.push(`${r.group.chunk.role}:<img>`); + else if (c.type === "error") out.push(`error:${c.message}`); + } else { + out.push(`tool:${r.group.entry.toolCallId}`); + } + } + return out; +} + +class Sim { + t: TranscriptState = initialState(); + m: MetricsState = initialMetricsState(); + seq = 0; + /** Durable server-side log (what loadSince returns). */ + log: StoredChunk[] = []; + durable: TurnMetrics[] = []; + + constructor(private readonly legacy = false) {} + + private ev(e: AgentEvent): void { + this.t = foldEvent(this.t, e); + this.m = foldMetricsEvent(this.m, e); + } + + private persist(role: StoredChunk["role"], chunk: StoredChunk["chunk"], turnId: string): void { + this.seq++; + this.log.push( + this.legacy ? { seq: this.seq, role, chunk } : { seq: this.seq, role, chunk, turnId }, + ); + } + + /** The sender types + sends (optimistic echo), optionally with an image. */ + send(text: string, image = false): void { + this.t = appendUserMessage( + this.t, + text, + image ? [{ url: "data:image/png;base64,AA", mimeType: "image/png" }] : undefined, + ); + } + + start(turnId: string, text: string, image = false): void { + this.ev({ type: "turn-start", conversationId: C, turnId } as AgentEvent); + this.ev({ type: "user-message", conversationId: C, turnId, text } as AgentEvent); + this.persist("user", { type: "text", text }, turnId); + if (image) this.persist("user", { type: "image", url: "/images/c1/x.png" }, turnId); + } + + text(turnId: string, text: string): void { + this.ev({ type: "text-delta", conversationId: C, turnId, delta: text } as AgentEvent); + } + + toolStep(turnId: string, step: number, id: string): void { + const stepId = `${turnId}#${step}` as StepId; + this.ev({ + type: "tool-call", + conversationId: C, + turnId, + stepId, + toolCallId: id, + toolName: "t", + input: {}, + } as AgentEvent); + this.completeStep(turnId, step); + this.ev({ + type: "tool-result", + conversationId: C, + turnId, + stepId, + toolCallId: id, + toolName: "t", + content: "ok", + isError: false, + } as AgentEvent); + this.persist( + "assistant", + { type: "tool-call", toolCallId: id, toolName: "t", input: {}, stepId }, + turnId, + ); + this.persist( + "tool", + { type: "tool-result", toolCallId: id, toolName: "t", content: "ok", isError: false, stepId }, + turnId, + ); + } + + completeStep(turnId: string, step: number): void { + const stepId = `${turnId}#${step}` as StepId; + this.ev({ type: "usage", conversationId: C, turnId, stepId, usage: u(100, 10) } as AgentEvent); + this.ev({ + type: "step-complete", + conversationId: C, + turnId, + stepId, + ttftMs: 100, + decodeMs: 100, + genTotalMs: 200, + } as AgentEvent); + } + + steer(turnId: string, text: string): void { + this.ev({ type: "steering", conversationId: C, turnId, text } as AgentEvent); + this.persist("user", { type: "text", text }, turnId); + } + + /** Finish a turn: final text step, done, seal, then history + metrics sync. */ + finish(turnId: string, text: string, finalStep: number): void { + this.text(turnId, text); + this.completeStep(turnId, finalStep); + this.ev({ + type: "done", + conversationId: C, + turnId, + reason: "stop", + durationMs: 500, + usage: u(100, 10), + } as AgentEvent); + this.persist("assistant", { type: "text", text }, turnId); + this.ev({ type: "turn-sealed", conversationId: C, turnId } as AgentEvent); + this.sync(turnId, finalStep); + } + + /** A turn that errors (never seals); backend persists its partial metrics. */ + fail(turnId: string, message: string, steps: number): void { + this.ev({ type: "error", conversationId: C, turnId, message } as AgentEvent); + this.persist("assistant", { type: "error", message }, turnId); + this.t = applyHistory(this.t, this.log); + if (steps > 0) this.syncMetricsFor(turnId, steps - 1); + } + + private syncMetricsFor(turnId: string, lastStep: number): void { + const steps = []; + for (let i = 0; i <= lastStep; i++) { + steps.push({ stepId: `${turnId}#${i}` as StepId, usage: u(100, 10), decodeMs: 100 }); + } + this.durable.push({ turnId, usage: u(100 * steps.length, 10 * steps.length), steps }); + this.m = applyDurableMetrics(this.m, this.durable); + } + + private sync(turnId: string, lastStep: number): void { + this.t = applyHistory(this.t, this.log); + this.syncMetricsFor(turnId, lastStep); + } + + trim(limit: number): void { + this.t = trimTranscript(this.t, limit); + } + + layout(): string[] { + const groups = groupRenderedChunks(selectChunks(this.t)); + return layout(interleaveTurnMetrics(groups, selectOrderedTurnMetrics(this.m))); + } +} + +/** Build a history of N text-only turns (t1..tN), each "qK" → "aK". */ +function textHistory(sim: Sim, n: number): void { + for (let k = 1; k <= n; k++) { + sim.send(`q${k}`); + sim.start(`t${k}`, `q${k}`); + sim.finish(`t${k}`, `a${k}`, 0); + } +} + +for (const legacy of [false, true]) { + // `legacy`: the persisted rows carry NO turnId (history written before the + // backend stamped it). Live chunks always carry it. + describe(`placement stays exact through a turn (${legacy ? "legacy rows" : "turnId rows"})`, () => { + it("plain text turns", () => { + const sim = new Sim(legacy); + textHistory(sim, 3); + expect(sim.layout()).toEqual([ + "user:q1", + "assistant:a1", + "[turn t1]", + "user:q2", + "assistant:a2", + "[turn t2]", + "user:q3", + "assistant:a3", + "[turn t3]", + ]); + }); + + it("never shifts while a new turn is sent and streams — even after older turns were unloaded", () => { + const sim = new Sim(legacy); + textHistory(sim, 4); + sim.trim(5); // 8 chunks > 5 → unloads whole quarters: q1 a1 q2 a2 + const before = sim.layout(); + expect(before).toEqual([ + "user:q3", + "assistant:a3", + "[turn t3]", + "user:q4", + "assistant:a4", + "[turn t4]", + ]); + + const expectStable = () => expect(sim.layout().slice(0, before.length)).toEqual(before); + + sim.send("q5"); + expectStable(); // just sent (optimistic echo, no turn yet) + sim.start("t5", "q5"); + expectStable(); // turn started + sim.toolStep("t5", 0, "k1"); + expectStable(); // tool step completed (metrics entry exists, not done) + sim.text("t5", "partial…"); + expectStable(); // streaming text + sim.finish("t5", "a5", 1); + expect(sim.layout()).toEqual([ + ...before, + "user:q5", + "tool:k1", + "[step t5#0]", + "assistant:a5", + "[turn t5]", + ]); + }); + + it("image attachments never shift other turns' metrics", () => { + const sim = new Sim(legacy); + sim.send("q1"); + sim.start("t1", "q1"); + sim.finish("t1", "a1", 0); + sim.send("q2", true); + sim.start("t2", "q2", true); + sim.finish("t2", "a2", 0); + sim.send("q3"); + sim.start("t3", "q3"); + sim.finish("t3", "a3", 0); + expect(sim.layout()).toEqual([ + "user:q1", + "assistant:a1", + "[turn t1]", + "user:q2", + "user:<img>", + "assistant:a2", + "[turn t2]", + "user:q3", + "assistant:a3", + "[turn t3]", + ]); + }); + + it("a mid-turn steering message keeps the turn total at the END of the turn", () => { + const sim = new Sim(legacy); + textHistory(sim, 1); + sim.send("q2"); + sim.start("t2", "q2"); + sim.toolStep("t2", 0, "k1"); + sim.steer("t2", "steer!"); + // live, mid-turn: t1 untouched, steering bubble shows no totals + expect(sim.layout()).toEqual([ + "user:q1", + "assistant:a1", + "[turn t1]", + "user:q2", + "tool:k1", + "[step t2#0]", + "user:steer!", + ]); + sim.toolStep("t2", 1, "k2"); + sim.finish("t2", "a2", 2); + textHistoryFrom(sim, 3); + expect(sim.layout()).toEqual([ + "user:q1", + "assistant:a1", + "[turn t1]", + "user:q2", + "tool:k1", + "[step t2#0]", + "user:steer!", + "tool:k2", + "[step t2#1]", + "assistant:a2", + "[turn t2]", + "user:q3", + "assistant:a3", + "[turn t3]", + ]); + }); + + it("an errored turn gets its own metrics and never shifts the others", () => { + const sim = new Sim(legacy); + textHistory(sim, 1); + sim.send("q2"); + sim.start("t2", "q2"); + sim.text("t2", "partial"); + sim.completeStep("t2", 0); + sim.fail("t2", "boom", 1); + textHistoryFrom(sim, 3); + const l = sim.layout(); + expect(l.indexOf("[turn t1]")).toBe(l.indexOf("assistant:a1") + 1); + expect(l.indexOf("[turn t2]")).toBeGreaterThan(l.indexOf("error:boom")); + expect(l.indexOf("[turn t2]")).toBeLessThan(l.indexOf("user:q3")); + expect(l.indexOf("[turn t3]")).toBe(l.indexOf("assistant:a3") + 1); + }); + + it("an errored turn with NO metrics never steals another turn's", () => { + const sim = new Sim(legacy); + textHistory(sim, 1); + sim.send("q2"); + sim.start("t2", "q2"); + sim.fail("t2", "boom", 0); // failed before any step → no metrics row + textHistoryFrom(sim, 3); + expect(sim.layout()).toEqual([ + "user:q1", + "assistant:a1", + "[turn t1]", + "user:q2", + "error:boom", + "user:q3", + "assistant:a3", + "[turn t3]", + ]); + }); + + it("a turn cut in half by the chat limit still shows its total under its visible tail", () => { + const sim = new Sim(legacy); + sim.send("q1"); + sim.start("t1", "q1"); + sim.toolStep("t1", 0, "k1"); + sim.toolStep("t1", 1, "k2"); + sim.finish("t1", "a1", 2); + textHistoryFrom(sim, 2); + // log: q1 call result call result a1 | q2 a2 → keep newest 6 + sim.trim(7); + const l = sim.layout(); + expect(l).not.toContain("user:q1"); + expect(l.indexOf("[turn t1]")).toBe(l.indexOf("assistant:a1") + 1); + expect(l.indexOf("[turn t2]")).toBe(l.indexOf("assistant:a2") + 1); + }); + }); +} + +/** Continue a text history from turn `from` (one more turn). */ +function textHistoryFrom(sim: Sim, from: number): void { + sim.send(`q${from}`); + sim.start(`t${from}`, `q${from}`); + sim.finish(`t${from}`, `a${from}`, 0); +} diff --git a/src/core/metrics/types.ts b/src/core/metrics/types.ts index 84d1904..95b5b02 100644 --- a/src/core/metrics/types.ts +++ b/src/core/metrics/types.ts @@ -61,10 +61,11 @@ export type MetricsRow = /** 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). + * Usage of the LAST step of the most recent EARLIER finalized turn, or + * `null` when there is none. The baseline for cross-turn retention: that + * step's prompt is what this turn's first step should find in cache. */ - readonly prevTurnUsage: Usage | null; + readonly prevLastStepUsage: Usage | null; }; /** Formatted cache hit-rate view: percentage + colour severity + hit flag. */ @@ -80,6 +81,7 @@ export interface CacheRateView { /** Formatted per-step view for display. */ export interface StepMetricsView { readonly label: string; + /** `"<in> in / <out> out"` — this step's prompt and generated tokens. */ readonly tokensLabel: string; readonly tps: string | null; readonly ttft: string | null; @@ -90,7 +92,13 @@ export interface StepMetricsView { /** Formatted per-turn view for display. */ export interface TurnMetricsView { readonly label: string; + /** `"<out> tok out"` — tokens the model generated this turn. */ readonly tokensLabel: string; + /** + * Prompt-side breakdown: input tokens processed (summed over the turn's + * steps — each step re-sends the whole prompt, so this is NOT the context + * size) and how many of them were cache reads. + */ readonly breakdown: string; readonly tps: string | null; readonly duration: string | null; diff --git a/src/core/plugin-abi/driver.test.ts b/src/core/plugin-abi/driver.test.ts new file mode 100644 index 0000000..43d0af5 --- /dev/null +++ b/src/core/plugin-abi/driver.test.ts @@ -0,0 +1,210 @@ +import { describe, expect, it, vi } from "vitest"; +import { createFrontendDriver, type FrontendDriverDeps } from "./driver.js"; +import type { FrontendCandidate } from "./load-diff.js"; + +/** + * Asymmetric testing: the ONLY fake is the OUTERMOST edge (the dynamic + * import + the host-API builder — injected effects). The driver's internals, + * the load diff, and the registry-facing sweep contract are real. + */ + +const c = (id: string, bust = "1"): FrontendCandidate => ({ + id, + url: `http://x/plugins/${id}/frontend?workspace=w&v=${bust}`, + apiVersion: "^1", +}); + +interface Harness { + readonly deps: FrontendDriverDeps & { + // The race test swaps the import edge before the driver exists — the + // driver reads the dep object per call, so the mutation is the swap. + importModule: (url: string) => Promise<unknown>; + }; + readonly imports: string[]; + readonly sweeps: string[]; + /** The modules the fake import() returns, keyed by the next id. */ + readonly modules: Map<string, unknown>; + readonly logs: string[]; +} + +function harness(): Harness { + const imports: string[] = []; + const sweeps: string[] = []; + const logs: string[] = []; + const modules = new Map<string, unknown>(); + const deps: FrontendDriverDeps = { + importModule: async (url) => { + imports.push(url); + const id = new URL(url).pathname.split("/")[2] ?? ""; + const mod = modules.get(id); + if (mod === undefined) throw new Error(`404: ${url}`); + return mod; + }, + buildHostApi: (pluginId) => ({ + abiVersion: "1.1", + registerView: () => () => {}, + surfaces: { + subscribe: () => {}, + unsubscribe: () => {}, + invoke: () => {}, + getSpec: () => null, + }, + logger: { info: () => {}, warn: () => {} }, + layout: { + applyOps: async () => ({ ok: false, error: "no layout provider active" }), + getTree: () => null, + }, + ...(pluginId === undefined ? {} : {}), + }), + unregisterViewsByPlugin: (pluginId) => sweeps.push(pluginId), + logger: { + info: (msg) => logs.push(msg), + warn: (msg) => logs.push(msg), + }, + }; + return { deps, imports, sweeps, modules, logs }; +} + +describe("createFrontendDriver — the load lifecycle", () => { + it("boot: imports newcomers and calls activate with the plugin's own host API", async () => { + const h = harness(); + const activate = vi.fn(); + h.modules.set("a", { default: { activate } }); + const driver = createFrontendDriver(h.deps); + await driver.sync([c("a")]); + expect(h.imports).toEqual([c("a").url]); + expect(activate).toHaveBeenCalledOnce(); + expect(driver.states().get("a")).toEqual({ kind: "loaded" }); + }); + + it("extensions-changed: a vanished plugin runs its deactivate + the registry sweep", async () => { + const h = harness(); + const deactivate = vi.fn(); + h.modules.set("a", { default: { activate: () => deactivate } }); + const driver = createFrontendDriver(h.deps); + await driver.sync([c("a")]); + await driver.sync([]); // the plugin is gone from the desired set + expect(deactivate).toHaveBeenCalledOnce(); + expect(h.sweeps).toEqual(["a"]); + expect(driver.states().has("a")).toBe(false); + }); + + it("unload disposes EXACTLY that plugin (the sweep is per-pluginId)", async () => { + const h = harness(); + h.modules.set("a", { default: { activate: () => () => {} } }); + h.modules.set("b", { default: { activate: () => () => {} } }); + const driver = createFrontendDriver(h.deps); + await driver.sync([c("a"), c("b")]); + await driver.sync([c("b")]); // only `a` vanishes + expect(h.sweeps).toEqual(["a"]); + }); + + it("workspace.set / re-enable: the SAME url re-activates the cached module", async () => { + const h = harness(); + const activate = vi.fn(); + h.modules.set("a", { default: { activate } }); + const driver = createFrontendDriver(h.deps); + await driver.sync([c("a")]); + await driver.sync([]); // disable → unload (disposer + sweep) + await driver.sync([c("a")]); // re-enable → activate re-runs + expect(h.imports).toEqual([c("a").url, c("a").url]); // re-fetched URL + expect(activate).toHaveBeenCalledTimes(2); + expect(h.sweeps).toEqual(["a"]); + }); + + it("a changed URL (backend reload) unloads then re-imports fresh code", async () => { + const h = harness(); + const activate = vi.fn(() => () => {}); + h.modules.set("a", { default: { activate } }); + const driver = createFrontendDriver(h.deps); + await driver.sync([c("a", "old")]); + await driver.sync([c("a", "new")]); + expect(h.sweeps).toEqual(["a"]); + expect(h.imports).toEqual([c("a", "old").url, c("a", "new").url]); + expect(activate).toHaveBeenCalledTimes(2); + }); + + it("a throwing activate is LOUD: failed state, no crash, others still load", async () => { + const h = harness(); + h.modules.set("bad", { + default: { + activate: () => { + throw new Error("boom"); + }, + }, + }); + h.modules.set("good", { default: { activate: () => () => {} } }); + const driver = createFrontendDriver(h.deps); + await driver.sync([c("bad"), c("good")]); + expect(driver.states().get("bad")).toEqual({ kind: "failed", error: "boom" }); + expect(driver.states().get("good")).toEqual({ kind: "loaded" }); + // The failed plugin owns nothing — the sweep never ran for it. + expect(h.sweeps).toEqual([]); + }); + + it("a broken module (404 / missing activate) is a failed state, not a crash", async () => { + const h = harness(); + h.modules.set("no-activate", { default: {} }); + const driver = createFrontendDriver(h.deps); + await driver.sync([c("ghost"), c("no-activate")]); + expect(driver.states().get("ghost")?.kind).toBe("failed"); + expect(driver.states().get("no-activate")?.kind).toBe("failed"); + expect(h.logs.some((l) => l.includes("activation failed"))).toBe(true); + }); + + it("a throwing deactivate still gets the registry sweep (views never linger)", async () => { + const h = harness(); + h.modules.set("a", { + default: { + activate: () => () => { + throw new Error("disposer boom"); + }, + }, + }); + const driver = createFrontendDriver(h.deps); + await driver.sync([c("a")]); + await driver.sync([]); + expect(h.sweeps).toEqual(["a"]); + }); + + it("handshake-incompatible candidates are never imported (reported for the UI)", async () => { + const h = harness(); + const driver = createFrontendDriver(h.deps); + await driver.sync([{ ...c("bad"), apiVersion: "^2" }]); + expect(h.imports).toEqual([]); + expect(driver.states().get("bad")).toEqual({ kind: "incompatible", apiVersion: "^2" }); + }); + + it("overlapping syncs serialize (boot + extensions-changed race)", async () => { + const h = harness(); + let releaseImport: (() => void) | undefined; + h.modules.set("a", { + default: { + activate: () => { + return () => {}; + }, + }, + }); + const blockingImport = async (url: string) => { + h.imports.push(url); + await new Promise<void>((resolve) => { + releaseImport = resolve; + }); + return h.modules.get("a"); + }; + // Swap the import edge BEFORE the driver exists (the driver reads the + // dep object per call — the swap is the injection point for the race). + h.deps.importModule = blockingImport; + const driver = createFrontendDriver(h.deps); + const first = driver.sync([c("a")]); + const second = driver.sync([]); // queued behind the first + // While the first sync is blocked on import, the second must not have + // started (no unload of `a` mid-import). + await new Promise((r) => setTimeout(r, 10)); + releaseImport?.(); + await Promise.all([first, second]); + // The second pass sees `a` already imported (the queued sync ran after) + // and unloads it — exactly one sweep, no interleaving. + expect(h.sweeps).toEqual(["a"]); + }); +}); diff --git a/src/core/plugin-abi/driver.ts b/src/core/plugin-abi/driver.ts new file mode 100644 index 0000000..7ba326f --- /dev/null +++ b/src/core/plugin-abi/driver.ts @@ -0,0 +1,140 @@ +/** + * The frontend-module DRIVER (plan 03 §3.2) — the effectful lifecycle host. + * Owns the imported-module set and runs ONE sync pass per trigger (boot, + * `extensions-changed`, `workspace.set`): + * + * 1. diff (pure, `decideFrontendLoads`) the desired set against the + * imported set; + * 2. UNLOAD stragglers: run the module's deactivate disposer (if it + * returned one) then the host's unregister-by-pluginId sweep — the UI is + * immediately correct. HONEST LIMITATION: ESM cannot be un-fetched; the + * bytes linger in the module cache. Re-enable re-runs `activate` on the + * cached module (same URL ⇒ no re-fetch). + * 3. IMPORT newcomers: dynamic-import the URL (THE injected edge), call + * `activate(api)` with the plugin's own host API (forced pluginId + * attribution). A throwing/missing activate is LOUD: the plugin's FE + * half is marked failed (the backend half stays loaded) and the failure + * never crashes the host or blocks the other plugins. + * + * Syncs are serialized (a chain) — boot + extensions-changed + workspace + * switches can overlap, and import/unload ordering must stay coherent. + * No internal mocks: tests inject the import edge + host-API builder. + */ +import type { + FrontendCandidate, + FrontendLoadDecision, + ImportedFrontendModule, +} from "./load-diff.js"; +import { decideFrontendLoads } from "./load-diff.js"; +import type { FrontendHostAPI, FrontendPluginModule, FrontendPluginState } from "./types.js"; + +export interface FrontendDriverDeps { + /** + * THE edge: dynamic ESM import. Injected (the composition root binds + * `import(url)`; tests bind a fake) so nothing else in the driver touches + * the module system. + */ + readonly importModule: (url: string) => Promise<unknown>; + /** + * Build the plugin's host API — the composition root closes over the + * registry + surfaces ports + logger and FORCES the pluginId attribution. + */ + readonly buildHostApi: (pluginId: string) => FrontendHostAPI; + /** The unregister-by-pluginId sweep (the registry-side unload path). */ + readonly unregisterViewsByPlugin: (pluginId: string) => void; + readonly logger: { info: (msg: string) => void; warn: (msg: string) => void }; +} + +export interface FrontendDriver { + /** Run one sync pass. Serialized; never throws (failures are states). */ + sync(candidates: readonly FrontendCandidate[]): Promise<void>; + /** The per-plugin states the manager UI renders. */ + states(): ReadonlyMap<string, FrontendPluginState>; +} + +interface ImportedEntry extends ImportedFrontendModule { + readonly deactivate: (() => void) | undefined; +} + +export function createFrontendDriver(deps: FrontendDriverDeps): FrontendDriver { + /** pluginId → the imported module's entry. */ + const imported = new Map<string, ImportedEntry>(); + /** pluginId → its last-known FE state (for the manager UI). */ + const pluginStates = new Map<string, FrontendPluginState>(); + /** The sync chain (serialization across triggers). */ + let chain: Promise<void> = Promise.resolve(); + + /** Unload one plugin: module disposer FIRST, then the registry sweep. */ + function unload(id: string, entry: ImportedEntry): void { + try { + entry.deactivate?.(); + } catch (err) { + // A broken deactivate must not block the sweep — the registration + // ledger's force-clean philosophy (the views must not linger). + deps.logger.warn(`frontend-plugin "${id}": deactivate threw — ${String(err)}`); + } + try { + deps.unregisterViewsByPlugin(id); + } catch (err) { + deps.logger.warn(`frontend-plugin "${id}": unregister sweep threw — ${String(err)}`); + } + imported.delete(id); + pluginStates.delete(id); + deps.logger.info(`frontend-plugin "${id}": unloaded`); + } + + /** Import + activate one candidate. Failures are LOUD states, not throws. */ + async function load(candidate: FrontendCandidate): Promise<void> { + try { + const mod = (await deps.importModule(candidate.url)) as FrontendPluginModule | undefined; + const activate = mod?.default?.activate; + if (typeof activate !== "function") { + throw new Error("the module has no default.activate export"); + } + const disposer = activate(deps.buildHostApi(candidate.id)); + imported.set(candidate.id, { + url: candidate.url, + deactivate: typeof disposer === "function" ? disposer : undefined, + }); + pluginStates.set(candidate.id, { kind: "loaded" }); + deps.logger.info(`frontend-plugin "${candidate.id}": activated`); + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + pluginStates.set(candidate.id, { kind: "failed", error: message }); + deps.logger.warn(`frontend-plugin "${candidate.id}": activation failed — ${message}`); + } + } + + return { + async sync(candidates: readonly FrontendCandidate[]): Promise<void> { + const run = chain.then(() => runSync(candidates)); + // The NEXT sync waits for this one; a rejected run must not poison the + // chain (runSync itself never throws, but belt-and-suspenders). + chain = run.catch(() => {}); + await run; + }, + + states(): ReadonlyMap<string, FrontendPluginState> { + return new Map(pluginStates); + }, + }; + + async function runSync(candidates: readonly FrontendCandidate[]): Promise<void> { + const decision: FrontendLoadDecision = decideFrontendLoads(candidates, imported); + + // The handshake verdicts persist even when nothing else changes — the + // manager UI renders "incompatible frontend plugin" until the plugin's + // manifest (or the host) changes. + for (const bad of decision.incompatible) { + pluginStates.set(bad.id, { kind: "incompatible", apiVersion: bad.apiVersion }); + } + + for (const id of decision.toUnload) { + const entry = imported.get(id); + if (entry !== undefined) unload(id, entry); + } + for (const candidate of decision.toLoad) { + await load(candidate); + } + } +} diff --git a/src/core/plugin-abi/index.ts b/src/core/plugin-abi/index.ts new file mode 100644 index 0000000..3c97eca --- /dev/null +++ b/src/core/plugin-abi/index.ts @@ -0,0 +1,27 @@ +/** + * The frontend plugin ABI (plan 03) — the contract + lifecycle machinery for + * dynamically-imported frontend plugin modules. See `types.ts` (the surface), + * `version.ts` (the handshake), `load-diff.ts` (pure decisions), and + * `driver.ts` (the effectful host). The composition root (src/app) wires the + * driver to the registry, the surfaces ports, and the plugin HTTP edge. + */ +export { createFrontendDriver, type FrontendDriver, type FrontendDriverDeps } from "./driver.js"; +export { + decideFrontendLoads, + type FrontendCandidate, + type FrontendCandidatesInput, + type FrontendLoadDecision, + frontendCandidates, + type ImportedFrontendModule, + usablePluginIds, +} from "./load-diff.js"; +export type { + FrontendHostAPI, + FrontendPluginModule, + FrontendPluginState, + FrontendViewRegistration, +} from "./types.js"; +export { + FRONTEND_ABI_VERSION, + isFrontendCompatible, +} from "./version.js"; diff --git a/src/core/plugin-abi/load-diff.test.ts b/src/core/plugin-abi/load-diff.test.ts new file mode 100644 index 0000000..f8ebf6c --- /dev/null +++ b/src/core/plugin-abi/load-diff.test.ts @@ -0,0 +1,183 @@ +import type { PluginListResponse } from "@dispatch/transport-contract"; +import { describe, expect, it } from "vitest"; +import { + decideFrontendLoads, + frontendCandidates, + type ImportedFrontendModule, + usablePluginIds, +} from "./load-diff.js"; +import { FRONTEND_ABI_VERSION } from "./version.js"; + +const httpBase = "http://localhost:24203"; + +function listWith( + ...rows: readonly { + readonly id: string; + readonly apiVersion?: string; + readonly bust?: string; + }[] +): PluginListResponse { + return { + loaded: rows.map((r) => ({ + id: r.id, + name: r.id, + version: "1.0.0", + tier: "standard" as const, + trust: "local" as const, + locked: false, + dependsOn: [], + ...(r.apiVersion === undefined + ? {} + : { frontend: { apiVersion: r.apiVersion, entry: "frontend.js", bust: r.bust ?? "1" } }), + })), + disabledWithReason: [], + discovered: [], + missing: [], + }; +} + +describe("frontendCandidates", () => { + it("derives candidates from loaded ∧ frontend-declaring ∧ workspace-enabled rows", () => { + const candidates = frontendCandidates( + listWith( + { id: "a", apiVersion: "^1", bust: "111" }, + { id: "no-fe" }, + { id: "b", apiVersion: "^1", bust: "222" }, + ), + { workspaceId: "team-a", usableIds: new Set(["a", "b", "no-fe"]), httpBase }, + ); + expect(candidates).toEqual([ + { + id: "a", + apiVersion: "^1", + url: `${httpBase}/plugins/a/frontend?workspace=team-a&v=111`, + }, + { + id: "b", + apiVersion: "^1", + url: `${httpBase}/plugins/b/frontend?workspace=team-a&v=222`, + }, + ]); + }); + + it("excludes plugins disabled in the active workspace", () => { + const candidates = frontendCandidates(listWith({ id: "a", apiVersion: "^1" }), { + workspaceId: "team-a", + usableIds: new Set<string>(), + httpBase, + }); + expect(candidates).toEqual([]); + }); + + it("URL-encodes the workspace and bust", () => { + const candidates = frontendCandidates(listWith({ id: "a", apiVersion: "^1", bust: "1 2" }), { + workspaceId: "team a", + usableIds: new Set(["a"]), + httpBase, + }); + expect(candidates[0]?.url).toBe(`${httpBase}/plugins/a/frontend?workspace=team%20a&v=1%202`); + }); +}); + +describe("usablePluginIds", () => { + it("mirrors isPluginUsable: core usable unless disabled, others enabled-only", () => { + const usable = usablePluginIds([ + { pluginId: "core-one", tier: "core", state: "undefined", loaded: true, locked: true }, + { pluginId: "core-off", tier: "core", state: "disabled", loaded: true, locked: true }, + { pluginId: "std-on", tier: "standard", state: "enabled", loaded: true, locked: false }, + { pluginId: "std-blank", tier: "standard", state: "undefined", loaded: true, locked: false }, + { pluginId: "prov-on", tier: "providers", state: "enabled", loaded: true, locked: false }, + { pluginId: "prov-off", tier: "providers", state: "disabled", loaded: true, locked: false }, + ]); + expect([...usable].sort()).toEqual(["core-one", "prov-on", "std-on"]); + }); +}); + +describe("decideFrontendLoads — the load diff (boot / extensions-changed / workspace.set)", () => { + const c = (id: string, bust = "1"): { id: string; url: string; apiVersion: string } => ({ + id, + url: `${httpBase}/plugins/${id}/frontend?workspace=w&v=${bust}`, + apiVersion: "^1", + }); + + it("boot: every desired candidate imports (nothing imported yet)", () => { + const d = decideFrontendLoads([c("a"), c("b")], new Map()); + expect(d.toLoad.map((x) => x.id)).toEqual(["a", "b"]); + expect(d.toUnload).toEqual([]); + expect(d.incompatible).toEqual([]); + }); + + it("extensions-changed: a removed plugin unloads, a newcomer imports", () => { + const imported = new Map<string, ImportedFrontendModule>([ + ["a", { url: c("a").url }], + ["gone", { url: `${httpBase}/plugins/gone/frontend?workspace=w&v=1` }], + ]); + const d = decideFrontendLoads([c("a"), c("new")], imported); + expect(d.toLoad.map((x) => x.id)).toEqual(["new"]); + expect(d.toUnload).toEqual(["gone"]); + }); + + it("workspace.set: the enablement set change drives load AND unload", () => { + // The candidate list already reflects the NEW workspace (composition + // root re-derives per sync) — previously-imported ids absent from it unload. + const imported = new Map<string, ImportedFrontendModule>([ + ["a", { url: c("a").url }], + ["b", { url: c("b").url }], + ]); + const d = decideFrontendLoads([c("a")], imported); + expect(d.toLoad).toEqual([]); + expect(d.toUnload).toEqual(["b"]); + }); + + it("a stable desired set is a no-op (idempotent sync)", () => { + const imported = new Map<string, ImportedFrontendModule>([ + ["a", { url: c("a").url }], + ["b", { url: c("b").url }], + ]); + const d = decideFrontendLoads([c("a"), c("b")], imported); + expect(d.toLoad).toEqual([]); + expect(d.toUnload).toEqual([]); + }); + + it("a changed URL (backend reload → new bust) is unload + re-import", () => { + const imported = new Map<string, ImportedFrontendModule>([["a", { url: c("a", "old").url }]]); + const d = decideFrontendLoads([c("a", "new")], imported); + expect(d.toLoad.map((x) => x.id)).toEqual(["a"]); + expect(d.toUnload).toEqual(["a"]); + }); + + it("handshake failure → incompatible, never imported, still unloaded if it was", () => { + const bad = { ...c("bad"), apiVersion: "^2" }; + const imported = new Map<string, ImportedFrontendModule>([["bad", { url: bad.url }]]); + const d = decideFrontendLoads([bad], imported); + expect(d.toLoad).toEqual([]); + expect(d.incompatible).toEqual([{ id: "bad", apiVersion: "^2" }]); + // The previously-imported instance must be swept (the manifest changed + // under us — the UI must not keep running incompatible code). + expect(d.toUnload).toEqual(["bad"]); + }); + + it("the handshake checks against FRONTEND_ABI_VERSION", () => { + // A range that includes the host version passes — proving the version + // constant feeds the check. + expect(FRONTEND_ABI_VERSION).toBe("1.1"); + const d = decideFrontendLoads([c("ok")], new Map()); + expect(d.incompatible).toEqual([]); + }); + + it("the additive bump keeps v1 plugins loading (plan 04: ^1 spans 1.1)", () => { + // c() declares apiVersion "^1" — a bare-major range is a + // major-compatibility claim, so the 1.1 host (layout member added) still + // loads every v1 plugin. + const d = decideFrontendLoads([c("v1-plugin")], new Map()); + expect(d.toLoad.map((x) => x.id)).toEqual(["v1-plugin"]); + expect(d.incompatible).toEqual([]); + }); + + it("plugins requiring the 1.1 ABI (^1.1) load on the 1.1 host", () => { + const layout = { ...c("layout-plugin"), apiVersion: "^1.1" }; + const d = decideFrontendLoads([layout], new Map()); + expect(d.toLoad.map((x) => x.id)).toEqual(["layout-plugin"]); + expect(d.incompatible).toEqual([]); + }); +}); diff --git a/src/core/plugin-abi/load-diff.ts b/src/core/plugin-abi/load-diff.ts new file mode 100644 index 0000000..b846a40 --- /dev/null +++ b/src/core/plugin-abi/load-diff.ts @@ -0,0 +1,137 @@ +/** + * The frontend-module load DIFF (plan 03 §3.2) — pure decision logic. On + * boot, on `extensions-changed`, and on `workspace.set`, the composition + * root derives the DESIRED set of frontend plugins (loaded ∧ + * workspace-enabled ∧ frontend-declaring) from `GET /plugins` + the active + * workspace's enablement rows, and diffs it against the IMPORTED set: + * newcomers are `import()`-ed + activated; stragglers are deactivated + * (module disposer + unregister-by-pluginId sweep); a changed import URL + * (backend reload → new content bust) is an unload + re-import. + * + * The handshake runs HERE: a plugin whose `frontend.apiVersion` range does + * not include FRONTEND_ABI_VERSION is reported "incompatible" and is NEVER + * imported — it is also (re)unloaded if a previous version of it was + * imported. + */ +import type { PluginListResponse, WorkspacePluginRow } from "@dispatch/transport-contract"; +import { FRONTEND_ABI_VERSION, isFrontendCompatible } from "./version.js"; + +/** One desired frontend plugin, fully URL-resolved. */ +export interface FrontendCandidate { + readonly id: string; + /** The import URL: the backend route + workspace + the content bust. */ + readonly url: string; + /** The plugin's declared frontend ABI semver range (the handshake input). */ + readonly apiVersion: string; +} + +/** What the currently-imported modules look like to the diff. */ +export interface ImportedFrontendModule { + /** The URL the module was imported from (a change → unload + re-import). */ + readonly url: string; +} + +/** The diff verdict for one sync pass. */ +export interface FrontendLoadDecision { + /** Newcomers (or bust-changed plugins) to import + activate. */ + readonly toLoad: readonly FrontendCandidate[]; + /** Imported plugin ids no longer desired (or superseded by a new URL). */ + readonly toUnload: readonly string[]; + /** Handshake failures — "incompatible frontend plugin" in the manager UI. */ + readonly incompatible: readonly { readonly id: string; readonly apiVersion: string }[]; +} + +/** The composition-root inputs for candidate derivation. */ +export interface FrontendCandidatesInput { + /** The active workspace whose enablement set applies. */ + readonly workspaceId: string; + /** The plugin ids enabled in that workspace (see `usablePluginIds`). */ + readonly usableIds: ReadonlySet<string>; + /** The backend HTTP base (the import URL's origin). */ + readonly httpBase: string; +} + +/** + * The desired frontend-plugin set from a `GET /plugins` response: LOADED ∧ + * workspace-enabled ∧ frontend-declaring rows, one URL each. The URL is + * `/plugins/:id/frontend?workspace=<ws>&v=<bust>` — `v` is the content + * bust the additive list field carries (the entry's mtimeMs), so a + * backend-side reload (new mtime) produces a NEW URL → the diff re-imports. + */ +export function frontendCandidates( + list: PluginListResponse, + input: FrontendCandidatesInput, +): readonly FrontendCandidate[] { + const out: FrontendCandidate[] = []; + for (const row of list.loaded) { + if (row.frontend === undefined) continue; + if (!input.usableIds.has(row.id)) continue; + out.push({ + id: row.id, + apiVersion: row.frontend.apiVersion, + url: `${input.httpBase}/plugins/${encodeURIComponent(row.id)}/frontend?workspace=${encodeURIComponent(input.workspaceId)}&v=${encodeURIComponent(row.frontend.bust)}`, + }); + } + return out; +} + +/** + * Diff the desired set against the imported set. Handshake failures are + * NEVER in `toLoad` — and if such a plugin WAS imported (its manifest + * changed to an incompatible range), it lands in `toUnload`. + */ +export function decideFrontendLoads( + candidates: readonly FrontendCandidate[], + imported: ReadonlyMap<string, ImportedFrontendModule>, +): FrontendLoadDecision { + const toLoad: FrontendCandidate[] = []; + const incompatible: { id: string; apiVersion: string }[] = []; + const desired = new Set<string>(); + /** Imported ids whose URL changed (unload + re-import — fresh code). */ + const superseded: string[] = []; + + for (const candidate of candidates) { + if (!isFrontendCompatible(candidate.apiVersion)) { + incompatible.push({ id: candidate.id, apiVersion: candidate.apiVersion }); + continue; + } + desired.add(candidate.id); + const current = imported.get(candidate.id); + if (current === undefined) { + toLoad.push(candidate); + } else if (current.url !== candidate.url) { + // The backend reloaded the plugin (a new content bust → new URL): + // unload the OLD instance (its disposer + registry sweep run), then + // import the fresh code. + superseded.push(candidate.id); + toLoad.push(candidate); + } + } + + // Everything imported that is no longer a (handshake-passing) desired id, + // plus same-id candidates whose URL changed. + const toUnload = [...[...imported.keys()].filter((id) => !desired.has(id)), ...superseded]; + + return { toLoad, toUnload, incompatible }; +} + +/** + * The workspace-usable plugin ids from a `GET /workspaces/:id/plugins` + * response (plan-06 rows) — the FE mirror of the backend's + * `isPluginUsable`: tier "core" is usable unless explicitly `disabled`; + * every other tier is usable only when explicitly `enabled`. The response's + * `loaded: false` rows (records for unloaded plugins) cannot be desired + * anyway (candidates come from the LOADED set). + */ +export function usablePluginIds(rows: readonly WorkspacePluginRow[]): ReadonlySet<string> { + const usable = new Set<string>(); + for (const row of rows) { + if (row.tier === "core" ? row.state !== "disabled" : row.state === "enabled") { + usable.add(row.pluginId); + } + } + return usable; +} + +/** Re-export for the composition root's single-import convenience. */ +export { FRONTEND_ABI_VERSION, isFrontendCompatible }; diff --git a/src/core/plugin-abi/types.ts b/src/core/plugin-abi/types.ts new file mode 100644 index 0000000..bcb8139 --- /dev/null +++ b/src/core/plugin-abi/types.ts @@ -0,0 +1,113 @@ +/** + * The frontend plugin ABI surface (plan 03 §3.2) — the ONLY contract between + * the host app and a dynamically-imported frontend plugin module. + * + * A frontend plugin is a SINGLE prebuilt ESM file (svelte externalized — + * resolved through the host's import map to the app's own Svelte 5 runtime) + * whose default export is `{ activate(api) }`. `activate` registers views / + * stashes ports and MAY return a deactivate disposer (plan 03's deactivate + * contract). Trust model: plugin code runs with full page privileges — no + * sandbox (documented non-goal); the trust signal is the manifest's `trust` + * field rendered in the manager UI. + * + * Views get data via SURFACES (the ui-contract philosophy: backend-declared + * data rendered generically) — never via the app store. `registerView` + * FORCES the importing plugin's pluginId attribution (the host wraps the + * registry call), which is what makes unload-time sweeps exact. + */ +import type { ApplyOpsResult, LayoutOp, LayoutTree } from "@dispatch/layout-contract"; +import type { SurfaceSpec } from "@dispatch/ui-contract"; +import type { FRONTEND_ABI_VERSION } from "./version.js"; + +/** The registration input for one plugin view (as the ABI exposes it). */ +export interface FrontendViewRegistration { + /** Registry id — the same id a tile leaf's `view` field names. */ + readonly id: string; + /** Human label shown in panel dropdowns. */ + readonly label: string; + /** + * A Svelte 5 component (compiled by the plugin's build against the SHARED + * svelte runtime). Opaque here — the host casts it into the registry's + * heterogeneous component bag at the seam. + */ + readonly component: unknown; + /** Whether the view may appear in panel dropdowns (default true). */ + readonly panelEligible?: boolean; +} + +/** The per-plugin host API object handed to `activate`. */ +export interface FrontendHostAPI { + /** The host's ABI version (the other half of the semver handshake). */ + readonly abiVersion: typeof FRONTEND_ABI_VERSION; + /** + * Register a view into the SAME registry built-ins use. pluginId + * attribution is FORCED to the importing plugin (the host closes over it) + * — plugins cannot attribute views to another plugin. The returned + * disposer removes the view (call it from your deactivate). + */ + readonly registerView: (reg: FrontendViewRegistration) => () => void; + /** + * Talk to backend surfaces — the plugin-view data path. The app + * auto-subscribes every catalog entry, so `getSpec` typically returns + * data immediately for surfaces your plugin's backend half declares; + * subscribe/unsubscribe re-assert scopes explicitly. + */ + readonly surfaces: { + readonly subscribe: (surfaceId: string, conversationId?: string) => void; + readonly unsubscribe: (surfaceId: string, conversationId?: string) => void; + readonly invoke: ( + surfaceId: string, + actionId: string, + payload?: unknown, + conversationId?: string, + ) => void; + readonly getSpec: (surfaceId: string) => SurfaceSpec | null; + }; + readonly logger: { info: (msg: string) => void; warn: (msg: string) => void }; + /** + * Edit the ACTIVE workspace's layout through the provider (plan 04 §4.5 — + * the additive ABI bump; v1 plugins keep working, a plugin REQUIRING the + * layout member declares `frontend.apiVersion: "^1.1"`). A thin adapter + * over `POST /workspaces/:id/layout/ops` (the same kit `applyOps` the + * service exposes — one code path, two doors): + * - the ops are the `@dispatch/layout-contract` vocabulary (prefer the + * idempotent `ensure*` ops; address by node ID, never by numeric path); + * - the batch is ATOMIC — any failed op persists nothing and the result + * names it (`failedOp`); + * - no layout provider active (the route 404s) → the typed no-provider + * error `{ ok: false, error: "no layout provider active" }` — handle it; + * the provider may also load later (retry on workspace switches). + */ + readonly layout: { + readonly applyOps: (ops: readonly LayoutOp[]) => Promise<ApplyOpsResult>; + /** The ACTIVE workspace's tree; `null` in no-provider mode. */ + readonly getTree: () => LayoutTree | null; + }; +} + +/** + * What `activate` may return: nothing, or the deactivate disposer (the + * plan-03 deactivate contract). A named alias keeps the plan grammar + * readable; the suppression is needed because biome flags `void` inside a + * union type — here it is the plan-specified shape. + */ +// biome-ignore lint/suspicious/noConfusingVoidType: plan-specified activate grammar — "may return a disposer" (the deactivate contract) +export type FrontendActivateResult = void | (() => void); + +/** The module shape a frontend plugin's entry must default-export. */ +export interface FrontendPluginModule { + readonly default: { + /** + * Register the plugin's views/ports. MAY return a deactivate disposer — + * run on unload/disable BEFORE the host's unregister-by-pluginId sweep + * (the sweep is the safety net, not the primary path). + */ + activate(api: FrontendHostAPI): FrontendActivateResult; + }; +} + +/** The per-plugin FRONTEND lifecycle state the manager UI renders. */ +export type FrontendPluginState = + | { readonly kind: "loaded" } + | { readonly kind: "incompatible"; readonly apiVersion: string } + | { readonly kind: "failed"; readonly error: string }; diff --git a/src/core/plugin-abi/version.test.ts b/src/core/plugin-abi/version.test.ts new file mode 100644 index 0000000..ab4d34b --- /dev/null +++ b/src/core/plugin-abi/version.test.ts @@ -0,0 +1,65 @@ +import { describe, expect, it } from "vitest"; +import { FRONTEND_ABI_VERSION, isFrontendCompatible } from "./version.js"; + +describe("FRONTEND_ABI_VERSION", () => { + it("is 1.1 (plan 04 — the additive layout member bumped the minor)", () => { + expect(FRONTEND_ABI_VERSION).toBe("1.1"); + }); +}); + +describe("isFrontendCompatible — handshake accept/reject matrix (host 1.1.0)", () => { + it("accepts the canonical authoring range ^1 (v1 plugins keep working — the plan-04 promise)", () => { + expect(isFrontendCompatible("^1")).toBe(true); + expect(isFrontendCompatible("^1.0")).toBe(true); + expect(isFrontendCompatible("^1.0.0")).toBe(true); + }); + + it("accepts the bare major version (a major-compatibility claim: additive bumps never strand it)", () => { + expect(isFrontendCompatible("1")).toBe(true); + expect(isFrontendCompatible("~1")).toBe(true); + }); + + it("accepts plugins that REQUIRE the 1.1 ABI (^1.1 — the layout-declaring grammar)", () => { + expect(isFrontendCompatible("^1.1")).toBe(true); + expect(isFrontendCompatible("^1.1.0")).toBe(true); + }); + + it("accepts wildcard and >= ranges", () => { + expect(isFrontendCompatible("*")).toBe(true); + expect(isFrontendCompatible(">=1.0.0")).toBe(true); + }); + + it("rejects a next-major caret range", () => { + expect(isFrontendCompatible("^2")).toBe(false); + expect(isFrontendCompatible("^2.0.0")).toBe(false); + }); + + it("rejects an older pinned range", () => { + expect(isFrontendCompatible("0.9.0")).toBe(false); + expect(isFrontendCompatible("^0.1.0")).toBe(false); + }); + + it("rejects a tilde/pin at an OLDER explicit minor (explicit pins reject additive bumps)", () => { + expect(isFrontendCompatible("~0.9.0")).toBe(false); + expect(isFrontendCompatible("~1.0.0")).toBe(false); + expect(isFrontendCompatible("1.0.0")).toBe(false); + expect(isFrontendCompatible("^1.0")).not.toBe(false); // caret 1.x tracks the major + }); + + it("rejects malformed ranges (fail closed, never throws)", () => { + expect(isFrontendCompatible("")).toBe(false); + expect(isFrontendCompatible("banana")).toBe(false); + expect(isFrontendCompatible("^abc")).toBe(false); + expect(isFrontendCompatible("1.0.0.0")).toBe(false); + }); + + it("the host version pins the answer (injectable for tests)", () => { + // At a hypothetical 1.0.0 host, ^1.1 (needing the layout member) excludes; + // at 1.1.0 it includes — the gate is real, not decorative. + expect(isFrontendCompatible("^1.1", "1.0.0")).toBe(false); + expect(isFrontendCompatible("^1.1", "1.1.0")).toBe(true); + // "~1.0.0" pins minor 0: excludes a 1.1.0 host, includes a 1.0.x host. + expect(isFrontendCompatible("~1.0.0", "1.1.0")).toBe(false); + expect(isFrontendCompatible("~1.0.0", "1.0.0")).toBe(true); + }); +}); diff --git a/src/core/plugin-abi/version.ts b/src/core/plugin-abi/version.ts new file mode 100644 index 0000000..e5a0168 --- /dev/null +++ b/src/core/plugin-abi/version.ts @@ -0,0 +1,99 @@ +/** + * The frontend plugin ABI version + the semver handshake (plan 03 §3.2). + * + * FRONTEND_ABI_VERSION is bumped on BREAKING ABI changes only — an ADDITIVE + * member bumps the MINOR (plan 04 added `layout`: "1" → "1.1"; v1 plugins + * keep working, see the bare-major rule below). A plugin's + * `manifest.frontend.apiVersion` is a semver RANGE that must include the + * host's version; a mismatch lists the plugin as "incompatible frontend + * plugin" in the manager UI and its module is NEVER imported. + * + * `isFrontendCompatible` mirrors the backend kernel's + * `isApiVersionCompatible` semantics (host/src/host/version.ts) so both + * sides of the handshake behave identically. It additionally accepts + * major-only versions/ranges ("1", "^1") — the ABI's version grammar is + * single-number, normalized here to x.y.z with zeros — because a frontend + * ABI ("1") is not a kernel release ("0.1.0") and authors declare `"^1"`. + * A BARE-major range ("1", "^1", "~1" — no explicit minor) is a + * MAJOR-compatibility claim: it accepts every additive (minor) bump of that + * major, which is what makes "^1"/"1" plugins keep working on the 1.1 ABI + * (an author pinning minor/patch declares it explicitly: "1.0.0", "~1.0"). + */ + +export const FRONTEND_ABI_VERSION = "1.1"; + +/** + * The host version the handshake checks against (normalized to semver) — + * tracks FRONTEND_ABI_VERSION's major.minor at a 0 patch. + */ +const HOST_VERSION = "1.1.0"; + +interface SemVer { + readonly major: number; + readonly minor: number; + readonly patch: number; + /** The range string named the minor EXPLICITLY (a pin, not a bare major). */ + readonly explicitMinor: boolean; +} + +/** Parse a 1–3 component version ("1", "1.0", "1.0.0") — missing parts are 0. */ +function parseSemver(version: string): SemVer | null { + const match = version.match(/^(\d+)(?:\.(\d+))?(?:\.(\d+))?$/); + if (!match) return null; + return { + major: Number(match[1]), + minor: Number(match[2] ?? 0), + patch: Number(match[3] ?? 0), + explicitMinor: match[2] !== undefined, + }; +} + +function gte(a: SemVer, b: SemVer): boolean { + if (a.major !== b.major) return a.major > b.major; + if (a.minor !== b.minor) return a.minor > b.minor; + return a.patch >= b.patch; +} + +/** + * Whether the plugin's declared `frontend.apiVersion` range includes the + * host's FRONTEND_ABI_VERSION. A malformed range is NEVER compatible (a + * plugin-verify check catches it at authoring time; the handshake fails + * closed here). + */ +export function isFrontendCompatible(apiVersion: string, hostVersion = HOST_VERSION): boolean { + const range = apiVersion.trim(); + if (range === "*") return true; + const host = parseSemver(hostVersion); + if (host === null) return false; + + if (range.startsWith("^")) { + const min = parseSemver(range.slice(1)); + if (min === null) return false; + if (!gte(host, min)) return false; + if (min.major === 0) { + return host.major === 0 && (min.explicitMinor ? host.minor === min.minor : true); + } + return host.major === min.major; + } + if (range.startsWith("~")) { + const min = parseSemver(range.slice(1)); + if (min === null) return false; + if (!gte(host, min)) return false; + // A BARE "~1" is a major-compatibility claim (any 1.x.y); an explicit + // minor ("~1.0") pins the minor, as tilde always has. + if (!min.explicitMinor) return host.major === min.major; + return host.major === min.major && host.minor === min.minor; + } + if (range.startsWith(">=")) { + const min = parseSemver(range.slice(2)); + if (min === null) return false; + return gte(host, min); + } + const exact = parseSemver(range); + if (exact === null) return false; + // A BARE major ("1") names the major only — it accepts that major's + // additive bumps (the "v1 plugins keep working" rule). An explicit + // version ("1.0", "1.0.0") pins exactly, as before. + if (!exact.explicitMinor) return host.major === exact.major; + return host.major === exact.major && host.minor === exact.minor && host.patch === exact.patch; +} diff --git a/src/core/wire/conformance.test.ts b/src/core/wire/conformance.test.ts index 0d955d5..6bafda1 100644 --- a/src/core/wire/conformance.test.ts +++ b/src/core/wire/conformance.test.ts @@ -1,4 +1,8 @@ -import type { ChatSendMessage, ConversationHistoryResponse } from "@dispatch/transport-contract"; +import type { + ChatSendMessage, + ConversationHistoryResponse, + WsServerMessage, +} from "@dispatch/transport-contract"; import type { AgentEvent, StepId, StoredChunk } from "@dispatch/wire"; import { describe, expect, it } from "vitest"; import { @@ -75,6 +79,7 @@ describe("classifies every AgentEvent type", () => { { type: "error", conversationId: "c1", turnId: "t1", message: "oops" }, { type: "done", conversationId: "c1", turnId: "t1", reason: "complete" }, { type: "turn-sealed", conversationId: "c1", turnId: "t1" }, + { type: "turn-ended", conversationId: "c1", turnId: "t1", reason: "sealed" }, { type: "steering", conversationId: "c1", turnId: "t1", text: "steer mid-turn" }, { type: "provider-retry", @@ -103,13 +108,14 @@ describe("classifies every AgentEvent type", () => { "error", "done", "turn-sealed", + "turn-ended", "steering", "provider-retry", ]); }); - it("covers all 15 AgentEvent variants", () => { - expect(samples).toHaveLength(15); + it("covers all 16 AgentEvent variants", () => { + expect(samples).toHaveLength(16); }); }); @@ -172,6 +178,23 @@ describe("classifies every WsServerMessage type", () => { event: { type: "done" as const, conversationId: "c", turnId: "t", reason: "r" }, }, { type: "chat.error" as const, message: "e" }, + { type: "chat.accepted" as const, conversationId: "c1", requestId: "r1" }, + { + type: "chat.subscription-rejected" as const, + conversationId: "c1", + reason: "subscription-limit" as const, + }, + { + type: "chat.queued" as const, + conversationId: "conv-1", + reason: "compaction" as const, + messageId: "queued-1", + }, + { + type: "chat.resync-required" as const, + conversationId: "c1", + reason: "replay-limit" as const, + }, { type: "conversation.open" as const, conversationId: "c1", workspaceId: "w1" }, { type: "conversation.statusChanged" as const, @@ -184,9 +207,32 @@ describe("classifies every WsServerMessage type", () => { conversationId: "c1", newConversationId: "c2", messagesSummarized: 10, + historyEpoch: 0, messagesKept: 5, }, + { + type: "conversation.tabGroupChanged" as const, + conversationId: "c1", + workspaceId: "w1", + tabGroupParentId: null, + }, + { type: "extensions-changed" as const, action: "scan" as const }, + { + type: "layout" as const, + workspaceId: "w1", + tree: { version: 1 as const, root: { kind: "leaf" as const, id: "leaf1", view: "chat" } }, + }, + { + type: "tile-layout" as const, + mode: "wide" as const, + data: { version: 2 as const, tiles: [] }, + }, ]; + // Runtime samples stay exhaustive too, not just the switch guard. + const covered: Exclude<WsServerMessage["type"], (typeof msgs)[number]["type"]> extends never + ? true + : false = true; + expect(covered).toBe(true); const labels = msgs.map(assertWsServerMessageExhaustive); expect(labels).toEqual([ "catalog", @@ -195,9 +241,17 @@ describe("classifies every WsServerMessage type", () => { "error", "chat.delta", "chat.error", + "chat.accepted", + "chat.subscription-rejected", + "chat.queued", + "chat.resync-required", "conversation.open", "conversation.statusChanged", "conversation.compacted", + "conversation.tabGroupChanged", + "extensions-changed", + "layout", + "tile-layout", ]); }); }); @@ -212,6 +266,7 @@ describe("classifies every WsClientMessage type", () => { { type: "chat.subscribe" as const, conversationId: "c1" }, { type: "chat.unsubscribe" as const, conversationId: "c1" }, { type: "chat.queue" as const, conversationId: "c1", text: "steer" }, + { type: "chat.queue.cancel" as const, conversationId: "c1", messageId: "m1" }, ]; const labels = msgs.map(assertWsClientMessageExhaustive); expect(labels).toEqual([ @@ -222,6 +277,7 @@ describe("classifies every WsClientMessage type", () => { "chat.subscribe", "chat.unsubscribe", "chat.queue", + "chat.queue.cancel", ]); }); }); @@ -266,6 +322,7 @@ describe("ConversationHistoryResponse shape is constructible", () => { it("constructs a response with chunks", () => { const resp: ConversationHistoryResponse = { chunks: [{ seq: 1, role: "user", chunk: { type: "text", text: "hi" } }], + historyEpoch: 0, latestSeq: 1, }; expect(resp.chunks).toHaveLength(1); @@ -273,7 +330,7 @@ describe("ConversationHistoryResponse shape is constructible", () => { }); it("constructs an empty (caught-up) response", () => { - const resp: ConversationHistoryResponse = { chunks: [], latestSeq: 5 }; + const resp: ConversationHistoryResponse = { chunks: [], historyEpoch: 0, latestSeq: 5 }; expect(resp.chunks).toHaveLength(0); expect(resp.latestSeq).toBe(5); }); diff --git a/src/core/wire/conformance.ts b/src/core/wire/conformance.ts index bfa67dc..eff2330 100644 --- a/src/core/wire/conformance.ts +++ b/src/core/wire/conformance.ts @@ -18,6 +18,8 @@ export function assertAgentEventExhaustive(event: AgentEvent): string { return "text-delta"; case "reasoning-delta": return "reasoning-delta"; + case "reasoning-end": + return "reasoning-end"; case "tool-call": return "tool-call"; case "tool-result": @@ -32,6 +34,8 @@ export function assertAgentEventExhaustive(event: AgentEvent): string { return "done"; case "turn-sealed": return "turn-sealed"; + case "turn-ended": + return "turn-ended"; case "step-complete": return "step-complete"; case "steering": @@ -83,6 +87,14 @@ export function assertWsServerMessageExhaustive(msg: WsServerMessage): string { return "error"; case "chat.delta": return "chat.delta"; + case "chat.resync-required": + return "chat.resync-required"; + case "chat.accepted": + return "chat.accepted"; + case "chat.subscription-rejected": + return "chat.subscription-rejected"; + case "chat.queued": + return "chat.queued"; case "chat.error": return "chat.error"; case "conversation.open": @@ -91,6 +103,14 @@ export function assertWsServerMessageExhaustive(msg: WsServerMessage): string { return "conversation.statusChanged"; case "conversation.compacted": return "conversation.compacted"; + case "conversation.tabGroupChanged": + return "conversation.tabGroupChanged"; + case "extensions-changed": + return "extensions-changed"; + case "layout": + return "layout"; + case "tile-layout": + return "tile-layout"; default: return msg satisfies never; } @@ -116,6 +136,10 @@ export function assertWsClientMessageExhaustive(msg: WsClientMessage): string { return "chat.unsubscribe"; case "chat.queue": return "chat.queue"; + case "chat.queue.cancel": + return "chat.queue.cancel"; + case "workspace.set": + return "workspace.set"; default: return msg satisfies never; } |