summaryrefslogtreecommitdiffhomepage
path: root/src/core
diff options
context:
space:
mode:
Diffstat (limited to 'src/core')
-rw-r--r--src/core/chunks/groups.test.ts22
-rw-r--r--src/core/chunks/groups.ts14
-rw-r--r--src/core/chunks/index.ts2
-rw-r--r--src/core/chunks/local-echo.test.ts45
-rw-r--r--src/core/chunks/reconcile.test.ts143
-rw-r--r--src/core/chunks/reconcile.ts54
-rw-r--r--src/core/chunks/reducer.test.ts156
-rw-r--r--src/core/chunks/reducer.ts260
-rw-r--r--src/core/chunks/selectors.test.ts230
-rw-r--r--src/core/chunks/selectors.ts213
-rw-r--r--src/core/chunks/trim.test.ts34
-rw-r--r--src/core/chunks/trim.ts32
-rw-r--r--src/core/chunks/types.ts10
-rw-r--r--src/core/metrics/cache-timer.test.ts359
-rw-r--r--src/core/metrics/cache-timer.ts286
-rw-r--r--src/core/metrics/index.ts15
-rw-r--r--src/core/wire/conformance.test.ts63
-rw-r--r--src/core/wire/conformance.ts12
18 files changed, 1879 insertions, 71 deletions
diff --git a/src/core/chunks/groups.test.ts b/src/core/chunks/groups.test.ts
index cc2aa40..1196590 100644
--- a/src/core/chunks/groups.test.ts
+++ b/src/core/chunks/groups.test.ts
@@ -150,3 +150,25 @@ describe("groupRenderedChunks", () => {
expect(groups.map((g) => g.kind)).toEqual(["single", "tool"]);
});
});
+
+it("keeps windowed committed orphan calls in Show earlier even while another payload recovers", () => {
+ const groups = groupRenderedChunks(
+ [
+ {
+ seq: 50,
+ role: "tool",
+ provisional: false,
+ chunk: {
+ type: "tool-result",
+ toolCallId: "older",
+ toolName: "run_shell",
+ content: "output",
+ isError: false,
+ },
+ },
+ ],
+ true,
+ );
+ expect(groups[0]).toMatchObject({ kind: "tool", entry: { call: null } });
+ if (groups[0]?.kind === "tool") expect(groups[0].entry.recovering).toBeUndefined();
+});
diff --git a/src/core/chunks/groups.ts b/src/core/chunks/groups.ts
index c8493cc..6e0f8ea 100644
--- a/src/core/chunks/groups.ts
+++ b/src/core/chunks/groups.ts
@@ -3,6 +3,7 @@ import type { RenderedChunk } from "./types";
/** One tool call and its response. Null call = unloaded history; null result = pending. */
export interface ToolEntry {
+ readonly recovering?: boolean;
readonly toolCallId: string;
readonly toolName: string;
readonly call: ToolCallChunk | null;
@@ -26,7 +27,10 @@ export type RenderGroup =
* the stable t<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[] {
+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) {
@@ -55,7 +59,13 @@ export function groupRenderedChunks(rendered: readonly RenderedChunk[]): readonl
const turnId = callRow?.turnId ?? resultRow?.turnId;
groups.push({
kind: "tool",
- entry: { toolCallId: c.toolCallId, toolName: call?.toolName ?? c.toolName, call, result },
+ entry: {
+ toolCallId: c.toolCallId,
+ toolName: call?.toolName ?? c.toolName,
+ call,
+ result,
+ ...(recovering && call === null && resultRow?.provisional ? { recovering: true } : {}),
+ },
provisional: Boolean(callRow?.provisional || resultRow?.provisional),
...(stepId !== undefined ? { stepId } : {}),
...(turnId !== undefined ? { turnId } : {}),
diff --git a/src/core/chunks/index.ts b/src/core/chunks/index.ts
index 3822a79..03d687c 100644
--- a/src/core/chunks/index.ts
+++ b/src/core/chunks/index.ts
@@ -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 cba6fad..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,
@@ -278,6 +285,72 @@ describe("foldEvent — reasoning-delta", () => {
});
});
+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();
+ });
+});
+
describe("foldEvent — tool-call then tool-result", () => {
it("tool-call then tool-result render in order", () => {
let s = initialState();
@@ -1051,3 +1124,86 @@ describe("turnId provenance on live chunks", () => {
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 4eb08c1..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,13 @@ function flushAccumulating(
if (acc === null) return provisional;
const chunk: Chunk =
acc.kind === "text" ? { type: "text", text: acc.text } : { type: "thinking", text: acc.text };
- return [...provisional, withTurn({ role: "assistant", chunk }, acc.turnId)];
+ return [
+ ...provisional,
+ withTurn(
+ { role: "assistant", chunk, ...(acc.stepId !== undefined ? { stepId: acc.stepId } : {}) },
+ acc.turnId,
+ ),
+ ];
}
/** Attach a `turnId` to a provisional chunk (omitted when unknown). */
@@ -70,6 +77,26 @@ function tagTrailingUserEcho(
}
/**
+ * 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));
+}
+
+/**
* Content equality for two chunks of the SAME role (used to de-dup an
* optimistic echo against the authoritative committed version). Compares the
* discriminating payload only — `stepId` (generation provenance) is ignored
@@ -129,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
@@ -139,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);
@@ -149,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),
+ ),
+ }
+ : {}),
};
}
@@ -167,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");
@@ -179,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;
@@ -189,7 +263,17 @@ export function applyHistory(
}
}
- return { ...state, committed };
+ return {
+ ...state,
+ committed,
+ ...(state.historyRecoveryTurnIds !== undefined
+ ? {
+ historyRecoveryTurnIds: state.historyRecoveryTurnIds.filter(
+ (id) => !finishedTurnIds.includes(id),
+ ),
+ }
+ : {}),
+ };
}
/**
@@ -200,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
@@ -224,13 +310,26 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
case "tool-output":
return state;
- case "turn-start":
+ 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: tagTrailingUserEcho(state.provisional, event.turnId),
+ 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
@@ -246,11 +345,22 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
// text) emits no `user-message` and is not de-duped here.
if (event.text.length === 0) return state;
const run = trailingRun(state.provisional, "user");
- const alreadyEchoed = run.some((p) => p.chunk.type === "text" && p.chunk.text === event.text);
- if (alreadyEchoed) {
+ const echoed = run.find(
+ (p) =>
+ (p.turnId === undefined || p.turnId === event.turnId) &&
+ p.chunk.type === "text" &&
+ p.chunk.text === event.text,
+ );
+ if (echoed !== undefined) {
return {
...state,
- provisional: tagTrailingUserEcho(state.provisional, event.turnId),
+ // Preserve the live text independently of the local send. A delayed
+ // queued ack may remove its images, but cannot erase this server event.
+ provisional: tagTrailingUserEcho(state.provisional, event.turnId).map((p) => {
+ if (p.chunk !== echoed.chunk) return p;
+ const { localRequestId: _local, ...confirmed } = p;
+ return { ...confirmed, opening: true };
+ }),
generating: true,
};
}
@@ -259,7 +369,12 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
...state,
provisional: [
...provisional,
- { role: "user", chunk: { type: "text", text: event.text }, turnId: event.turnId },
+ {
+ role: "user",
+ chunk: { type: "text", text: event.text },
+ turnId: event.turnId,
+ opening: true,
+ },
],
accumulating: null,
generating: true,
@@ -268,7 +383,7 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
case "text-delta": {
const acc = state.accumulating;
- if (acc !== null && acc.kind === "text") {
+ if (acc !== null && acc.kind === "text" && acc.turnId === event.turnId) {
return {
...state,
accumulating: { ...acc, text: acc.text + event.delta },
@@ -286,7 +401,7 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
case "reasoning-delta": {
const acc = state.accumulating;
- if (acc !== null && acc.kind === "thinking") {
+ if (acc !== null && acc.kind === "thinking" && acc.turnId === event.turnId) {
return {
...state,
accumulating: { ...acc, text: acc.text + event.delta },
@@ -302,6 +417,30 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
};
}
+ 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,
+ };
+ }
+
case "tool-call": {
const provisional = flushAccumulating(state.provisional, state.accumulating);
const chunk: Chunk = {
@@ -355,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);
@@ -388,11 +543,21 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
// into the transcript by the sender).
if (event.text.length === 0) return state;
const provisional = flushAccumulating(state.provisional, state.accumulating);
+ const preceding = [...state.committed, ...provisional]
+ .reverse()
+ .find((p) => p.turnId === event.turnId && p.chunk.type === "tool-result");
+ const afterToolCallId =
+ preceding?.chunk.type === "tool-result" ? preceding.chunk.toolCallId : undefined;
return {
...state,
provisional: [
...provisional,
- { role: "user", chunk: { type: "text", text: event.text }, turnId: event.turnId },
+ {
+ role: "user",
+ chunk: { type: "text", text: event.text },
+ turnId: event.turnId,
+ ...(afterToolCallId !== undefined ? { afterToolCallId } : {}),
+ },
],
accumulating: null,
generating: true,
@@ -408,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.
}
}
@@ -426,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) {
@@ -460,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[] = [];
@@ -481,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
index 57c1415..90bb5cc 100644
--- a/src/core/chunks/selectors.test.ts
+++ b/src/core/chunks/selectors.test.ts
@@ -1,7 +1,8 @@
import type { AgentEvent, StepId, StoredChunk } from "@dispatch/wire";
import { describe, expect, it } from "vitest";
-import { applyHistory, foldEvent, initialState } from "./reducer";
+import { applyHistory, foldEvent, initialState, markHistoryRecovering } from "./reducer";
import { selectChunks } from "./selectors";
+import { trimTranscript } from "./trim";
import type { TranscriptState } from "./types";
const C = "conv";
@@ -19,6 +20,7 @@ const think = (delta: string): AgentEvent => ({
turnId: T,
delta,
});
+const thinkEnd: AgentEvent = { type: "reasoning-end", conversationId: C, turnId: T };
const text = (delta: string): AgentEvent => ({
type: "text-delta",
conversationId: C,
@@ -131,6 +133,34 @@ describe("selectChunks — de-dup of a late-join replay against committed histor
expect(rendered.filter((r) => r.provisional)).toHaveLength(3);
});
+ it("preserves an uncommitted accumulating tail even when its text repeats history", () => {
+ const committed: StoredChunk[] = [
+ ...committedStep0,
+ { seq: 5, role: "assistant", chunk: { type: "text", text: "final" }, turnId: T },
+ ];
+ let s = applyHistory(initialState(), committed);
+ s = markHistoryRecovering(s, T);
+ s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("a"), text("final")]);
+ const rendered = selectChunks(s);
+ expect(rendered).toHaveLength(6);
+ expect(rendered.slice(0, 5).every((r) => !r.provisional)).toBe(true);
+ expect(rendered[5]).toMatchObject({
+ provisional: true,
+ streaming: true,
+ chunk: { type: "text", text: "final" },
+ });
+ });
+
+ it("reconciles the same step identity even when provisional payload differs", () => {
+ let s = applyHistory(initialState(), committedStep0);
+ s = markHistoryRecovering(s, T);
+ s = fold(s, [userMsg, turnStart, think("DIFFERENT"), call("a"), result("a")]);
+ const rendered = selectChunks(s);
+ // Authoritative identity wins; payload equality cannot identify omitted/truncated content.
+ expect(rendered.filter((r) => r.provisional)).toHaveLength(0);
+ expect(rendered.map((r) => r.chunk)).toEqual(committedStep0.map((r) => r.chunk));
+ });
+
it("does nothing when the committed tail belongs to another turn", () => {
const other = committedStep0.map((c) => ({ ...c, turnId: "older" }));
let s = applyHistory(initialState(), other);
@@ -138,6 +168,204 @@ describe("selectChunks — de-dup of a late-join replay against committed histor
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 d32a628..8d32fcf 100644
--- a/src/core/chunks/selectors.ts
+++ b/src/core/chunks/selectors.ts
@@ -1,4 +1,5 @@
import type { ChatMessage, Chunk, StoredChunk, TurnProviderRetryEvent } from "@dispatch/wire";
+import { historyBackedProvisional } from "./reconcile";
import type { ProvisionalChunk, RenderedChunk, TranscriptState } from "./types";
/**
@@ -26,15 +27,163 @@ function sameChunk(p: ProvisionalChunk, c: StoredChunk): boolean {
}
/**
+ * 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 position-by-position with the committed run of turn T. Skip the
- * aligned prefix; stop at the first mismatch (fail-safe: a misalignment shows
- * a duplicate, never hides content).
+ * 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).
@@ -42,41 +191,59 @@ function sameChunk(p: ProvisionalChunk, c: StoredChunk): boolean {
function committedOverlap(state: TranscriptState): { skip: number; hideAccumulating: boolean } {
const first = state.provisional[0];
const turnId = first?.turnId ?? state.accumulating?.turnId;
- if (turnId === undefined) return { skip: 0, hideAccumulating: false };
+ if (turnId === undefined || state.historyRecoveryTurnIds?.includes(turnId))
+ return { skip: 0, hideAccumulating: false };
// The committed run for this turn (contiguous at the tail of `committed`).
let start = state.committed.length;
while (start > 0 && state.committed[start - 1]?.turnId === turnId) start--;
- const run = state.committed.length - start;
- if (run === 0) return { skip: 0, hideAccumulating: false };
-
- let skip = 0;
- while (skip < state.provisional.length && skip < run) {
- const p = state.provisional[skip];
- const c = state.committed[start + skip];
- if (p === undefined || c === undefined || p.turnId !== turnId || !sameChunk(p, c)) break;
- skip++;
- }
+ 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 && skip === state.provisional.length && skip < run) {
- const c = state.committed[start + skip];
+ 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;
+ (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) — minus provisional chunks already backed by
- * committed history (see `committedOverlap`).
+ * then provisional (seq: null), excluding history-backed identities. Unknown
+ * accumulating content stays live, even if its text equals an earlier step.
*/
export function selectChunks(state: TranscriptState): readonly RenderedChunk[] {
const result: RenderedChunk[] = [];
+ const backed = historyBackedProvisional(state);
const { skip, hideAccumulating } = committedOverlap(state);
for (const c of state.committed) {
result.push(
@@ -87,7 +254,7 @@ export function selectChunks(state: TranscriptState): readonly RenderedChunk[] {
}
for (let i = skip; i < state.provisional.length; i++) {
const p = state.provisional[i];
- if (p === undefined) continue;
+ if (p === undefined || backed.has(p)) continue;
result.push(
p.turnId !== undefined
? { seq: null, role: p.role, chunk: p.chunk, provisional: true, turnId: p.turnId }
diff --git a/src/core/chunks/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 949857a..2e2ee0c 100644
--- a/src/core/chunks/types.ts
+++ b/src/core/chunks/types.ts
@@ -2,6 +2,7 @@ import type { Chunk, Role, StoredChunk, TurnProviderRetryEvent, Usage } from "@d
/** A chunk being accumulated from streaming deltas (text or thinking). */
export interface AccumulatingChunk {
+ readonly stepId?: string;
readonly kind: "text" | "thinking";
readonly text: string;
/** The turn generating this chunk (from the live event's `turnId`). */
@@ -10,6 +11,13 @@ export interface AccumulatingChunk {
/** A provisional chunk that has no authoritative seq yet. */
export interface ProvisionalChunk {
+ /** Local provenance learned from tool-call/step-complete, not content. */
+ readonly stepId?: string;
+ readonly opening?: boolean;
+ /** Steering is persisted immediately after this completed tool batch. */
+ readonly afterToolCallId?: string;
+ /** Local send identity only; turn-start tagging does NOT confirm this content. */
+ readonly localRequestId?: string;
readonly role: Role;
readonly chunk: Chunk;
/**
@@ -22,6 +30,8 @@ export interface ProvisionalChunk {
/** The transcript reducer state. Holds committed history + live in-flight turn. */
export interface TranscriptState {
+ /** Turns with omitted delivery: never use payload equality as replay identity. */
+ readonly historyRecoveryTurnIds?: readonly string[];
readonly committed: readonly StoredChunk[];
readonly provisional: readonly ProvisionalChunk[];
readonly accumulating: AccumulatingChunk | null;
diff --git a/src/core/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/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/wire/conformance.test.ts b/src/core/wire/conformance.test.ts
index 2b98ca6..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",
]);
});
});
@@ -268,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);
@@ -275,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 855f7ea..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":
generated by cgit v1.2.3 (git 2.39.1) at 2026-10-26 03:33:48 +0000