summaryrefslogtreecommitdiffhomepage
path: root/src/core/chunks
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/chunks')
-rw-r--r--src/core/chunks/groups.test.ts193
-rw-r--r--src/core/chunks/groups.ts122
-rw-r--r--src/core/chunks/index.ts4
-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.ts201
-rw-r--r--src/core/chunks/reducer.ts315
-rw-r--r--src/core/chunks/selectors.test.ts373
-rw-r--r--src/core/chunks/selectors.ts276
-rw-r--r--src/core/chunks/trim.test.ts34
-rw-r--r--src/core/chunks/trim.ts32
-rw-r--r--src/core/chunks/types.ts25
13 files changed, 1606 insertions, 211 deletions
diff --git a/src/core/chunks/groups.test.ts b/src/core/chunks/groups.test.ts
index c8b0fa2..1196590 100644
--- a/src/core/chunks/groups.test.ts
+++ b/src/core/chunks/groups.test.ts
@@ -3,16 +3,16 @@ import { describe, expect, it } from "vitest";
import { groupRenderedChunks } from "./groups";
import type { RenderedChunk } from "./types";
-const text = (seq: number, role: Role, t: string, provisional = false): RenderedChunk => ({
+const text = (seq: number, role: Role, t: string): RenderedChunk => ({
seq,
role,
chunk: { type: "text", text: t },
- provisional,
+ provisional: false,
});
-
const call = (seq: number, id: string, stepId?: string, provisional = false): RenderedChunk => ({
seq,
role: "assistant",
+ provisional,
chunk: {
type: "tool-call",
toolCallId: id,
@@ -20,12 +20,11 @@ const call = (seq: number, id: string, stepId?: string, provisional = false): Re
input: { id },
...(stepId !== undefined ? { stepId: stepId as StepId } : {}),
},
- provisional,
});
-
const result = (seq: number, id: string, stepId?: string, provisional = false): RenderedChunk => ({
seq,
role: "tool",
+ provisional,
chunk: {
type: "tool-result",
toolCallId: id,
@@ -34,92 +33,142 @@ const result = (seq: number, id: string, stepId?: string, provisional = false):
isError: false,
...(stepId !== undefined ? { stepId: stepId as StepId } : {}),
},
- provisional,
});
+function toolGroups(rows: readonly RenderedChunk[]) {
+ return groupRenderedChunks(rows).map((g) => {
+ if (g.kind !== "tool") throw new Error("expected a tool group");
+ return g;
+ });
+}
+
describe("groupRenderedChunks", () => {
it("returns no groups for an empty stream", () => {
expect(groupRenderedChunks([])).toEqual([]);
});
-
- it("passes non-tool chunks through as single groups, in order", () => {
- const groups = groupRenderedChunks([text(1, "user", "hi"), text(2, "assistant", "hello")]);
- expect(groups).toHaveLength(2);
- expect(groups.every((g) => g.kind === "single")).toBe(true);
- });
-
- it("does NOT batch a single tool call (one per step) — call+result stay separate singles", () => {
- const groups = groupRenderedChunks([call(1, "a", "s1"), result(2, "a", "s1")]);
- expect(groups).toHaveLength(2);
- expect(groups.map((g) => g.kind)).toEqual(["single", "single"]);
+ it("passes non-tool chunks through unchanged and in order", () => {
+ const rows = [text(1, "user", "hi"), text(2, "assistant", "hello")];
+ expect(groupRenderedChunks(rows)).toEqual(rows.map((chunk) => ({ kind: "single", chunk })));
});
-
- it("does NOT batch tool calls that have no stepId (pre-0.2.0 replay)", () => {
- const groups = groupRenderedChunks([
- call(1, "a"),
- call(2, "b"),
- result(3, "a"),
- result(4, "b"),
+ it("pairs a single call and result into one group", () => {
+ const c = call(1, "a", "s1");
+ const r = result(2, "a", "s1");
+ expect(toolGroups([c, r])).toEqual([
+ {
+ kind: "tool",
+ entry: { toolCallId: "a", toolName: "tool-a", call: c.chunk, result: r.chunk },
+ stepId: "s1",
+ provisional: false,
+ },
]);
- expect(groups).toHaveLength(4);
- expect(groups.every((g) => g.kind === "single")).toBe(true);
});
-
- it("batches 2+ calls sharing a stepId into one group, pairing each with its result", () => {
- const groups = groupRenderedChunks([
+ it("renders parallel calls independently in call order, with out-of-order results", () => {
+ const groups = toolGroups([
call(1, "a", "s1"),
call(2, "b", "s1"),
- result(3, "a", "s1"),
- result(4, "b", "s1"),
+ result(3, "b", "s1"),
+ result(4, "a", "s1"),
]);
- expect(groups).toHaveLength(1);
- const g = groups[0];
- if (g?.kind !== "tool-batch") throw new Error("expected a tool-batch group");
- expect(g.stepId).toBe("s1");
- expect(g.entries).toHaveLength(2);
- expect(g.entries[0]?.call.toolCallId).toBe("a");
- expect(g.entries[0]?.result?.content).toBe("result-a");
- expect(g.entries[1]?.call.toolCallId).toBe("b");
- expect(g.entries[1]?.result?.content).toBe("result-b");
+ expect(groups.map((g) => g.entry.toolCallId)).toEqual(["a", "b"]);
+ expect(groups.map((g) => g.entry.result?.content)).toEqual(["result-a", "result-b"]);
});
-
- it("positions the batch at the first call and keeps surrounding chunks in order", () => {
+ it("pairs legacy chunks without stepId and mismatching stepIds by toolCallId", () => {
+ const groups = toolGroups([
+ call(1, "a"),
+ result(2, "a"),
+ call(3, "b", "s1"),
+ result(4, "b", "s2"),
+ ]);
+ expect(groups).toHaveLength(2);
+ expect(groups[0]?.entry.result).not.toBeNull();
+ expect(groups[1]?.stepId).toBe("s1");
+ });
+ it("keeps surrounding text at its position and absorbs results", () => {
const groups = groupRenderedChunks([
text(1, "assistant", "before"),
call(2, "a", "s1"),
- call(3, "b", "s1"),
- result(4, "a", "s1"),
- result(5, "b", "s1"),
- text(6, "assistant", "after"),
+ text(3, "assistant", "between"),
+ call(4, "b", "s1"),
+ result(5, "a"),
+ result(6, "b"),
+ text(7, "assistant", "after"),
]);
- expect(groups.map((g) => g.kind)).toEqual(["single", "tool-batch", "single"]);
+ expect(groups.map((g) => g.kind)).toEqual(["single", "tool", "single", "tool", "single"]);
});
-
- it("marks the batch provisional when any of its calls/results is provisional", () => {
- const groups = groupRenderedChunks([call(1, "a", "s1"), call(2, "b", "s1", true)]);
- const g = groups[0];
- if (g?.kind !== "tool-batch") throw new Error("expected a tool-batch group");
- expect(g.provisional).toBe(true);
- expect(g.entries).toHaveLength(2);
- expect(g.entries[1]?.result).toBeNull(); // dangling call (no result yet)
+ it("keeps pending calls and orphan results", () => {
+ const groups = toolGroups([call(1, "pending"), result(2, "orphan", "s2", true)]);
+ expect(groups[0]?.entry.result).toBeNull();
+ expect(groups[1]?.entry.call).toBeNull();
+ expect(groups[1]?.entry.toolName).toBe("tool-orphan");
+ expect(groups[1]?.provisional).toBe(true);
+ expect(groups[1]?.stepId).toBe("s2");
});
-
- it("batches one step while leaving a different single-call step ungrouped", () => {
- const groups = groupRenderedChunks([
- call(1, "a", "s1"),
- call(2, "b", "s1"),
- call(3, "c", "s2"),
- result(4, "a", "s1"),
- result(5, "b", "s1"),
- result(6, "c", "s2"),
+ it.each([
+ [true, false],
+ [false, true],
+ [false, false],
+ ])("combines call/result provisional flags (%s, %s)", (cp, rp) => {
+ expect(toolGroups([call(1, "a", "s1", cp), result(2, "a", "s1", rp)])[0]?.provisional).toBe(
+ cp || rp,
+ );
+ });
+ it("prefers call provenance, falling back to result provenance", () => {
+ const groups = toolGroups([
+ { ...call(1, "a", "s1"), turnId: "t1" },
+ { ...result(2, "a", "s2"), turnId: "t2" },
+ call(3, "b"),
+ { ...result(4, "b", "s3"), turnId: "t3" },
+ ]);
+ expect(groups.map((g) => [g.stepId, g.turnId])).toEqual([
+ ["s1", "t1"],
+ ["s3", "t3"],
+ ]);
+ });
+ it("dedupes calls/results across steps, first occurrence wins", () => {
+ const first = call(1, "a", "s1", true);
+ const firstResult = result(4, "a", "s1");
+ const groups = toolGroups([
+ first,
+ call(2, "a", "s2"),
+ call(3, "b"),
+ firstResult,
+ result(5, "a", "s2"),
+ result(6, "b"),
+ result(7, "b"),
]);
- expect(groups.map((g) => g.kind)).toEqual(["tool-batch", "single", "single"]);
- const batch = groups[0];
- if (batch?.kind !== "tool-batch") throw new Error("expected a tool-batch group");
- expect(batch.entries).toHaveLength(2);
- // the s2 single call + its result remain as separate single groups
- const singles = groups.slice(1);
- expect(singles[0]?.kind === "single" && singles[0].chunk.chunk.type).toBe("tool-call");
- expect(singles[1]?.kind === "single" && singles[1].chunk.chunk.type).toBe("tool-result");
+ expect(groups).toHaveLength(2);
+ expect(groups[0]?.entry.call).toBe(first.chunk);
+ expect(groups[0]?.entry.result).toBe(firstResult.chunk);
+ expect(groups[0]?.provisional).toBe(true);
+ });
+ it("dedupes orphan results too", () => {
+ expect(toolGroups([result(1, "a"), result(2, "a")])).toHaveLength(1);
+ });
+ it("positions a card at its call even when its result came first", () => {
+ const c = call(3, "a");
+ const groups = groupRenderedChunks([result(1, "a"), text(2, "assistant", "before call"), c]);
+ expect(groups.map((g) => g.kind)).toEqual(["single", "tool"]);
});
});
+
+it("keeps windowed committed orphan calls in Show earlier even while another payload recovers", () => {
+ const groups = groupRenderedChunks(
+ [
+ {
+ seq: 50,
+ role: "tool",
+ provisional: false,
+ chunk: {
+ type: "tool-result",
+ toolCallId: "older",
+ toolName: "run_shell",
+ content: "output",
+ isError: false,
+ },
+ },
+ ],
+ true,
+ );
+ expect(groups[0]).toMatchObject({ kind: "tool", entry: { call: null } });
+ if (groups[0]?.kind === "tool") expect(groups[0].entry.recovering).toBeUndefined();
+});
diff --git a/src/core/chunks/groups.ts b/src/core/chunks/groups.ts
index 53a2873..6e0f8ea 100644
--- a/src/core/chunks/groups.ts
+++ b/src/core/chunks/groups.ts
@@ -1,95 +1,75 @@
import type { ToolCallChunk, ToolResultChunk } from "@dispatch/wire";
import type { RenderedChunk } from "./types";
-/**
- * One tool call within a batch, paired with its result (matched by `toolCallId`).
- * `result` is null while the call is still pending (no result chunk yet).
- */
-export interface ToolBatchEntry {
- readonly call: ToolCallChunk;
+/** One tool call and its response. Null call = unloaded history; null result = pending. */
+export interface ToolEntry {
+ readonly recovering?: boolean;
+ readonly toolCallId: string;
+ readonly toolName: string;
+ readonly call: ToolCallChunk | null;
readonly result: ToolResultChunk | null;
}
-/**
- * A render group: either a single rendered chunk (rendered as today) or a batch
- * of tool calls the model emitted together in one step (shared `stepId`), to be
- * rendered as one grouped unit.
- */
export type RenderGroup =
| { readonly kind: "single"; readonly chunk: RenderedChunk }
| {
- readonly kind: "tool-batch";
- readonly stepId: string;
- readonly entries: readonly ToolBatchEntry[];
+ readonly kind: "tool";
+ readonly entry: ToolEntry;
readonly provisional: boolean;
+ readonly stepId?: string;
+ readonly turnId?: string;
};
/**
- * Group a flat rendered-chunk stream for display. Tool calls sharing a `stepId`
- * (the backend's authoritative batch key) where the step has 2+ calls become one
- * `tool-batch` group, positioned at the first call and pairing each call with its
- * `tool-result` (by `toolCallId`); the absorbed result chunks are not emitted on
- * their own. Single tool calls (one per step, or no `stepId` — e.g. pre-0.2.0
- * replay rows) and every non-tool chunk render as `single` groups, in order.
- *
- * Pure: input → output, no DOM, no Svelte.
+ * One independent group per call, at the call's position, absorbing its result
+ * by toolCallId (including legacy chunks without stepId). Orphan results stay
+ * visible at their own position. First occurrence wins for duplicate ids, so
+ * the stable t<toolCallId> render keys are always unique, even on corrupt replay.
+ * Pure: input → output; no DOM, effects, or Svelte.
*/
-export function groupRenderedChunks(rendered: readonly RenderedChunk[]): readonly RenderGroup[] {
- // 1. Steps that batched 2+ tool calls.
- const callsPerStep = new Map<string, number>();
- for (const rc of rendered) {
- if (rc.chunk.type === "tool-call" && rc.chunk.stepId !== undefined) {
- callsPerStep.set(rc.chunk.stepId, (callsPerStep.get(rc.chunk.stepId) ?? 0) + 1);
- }
- }
- const batchSteps = new Set<string>();
- for (const [stepId, count] of callsPerStep) {
- if (count >= 2) batchSteps.add(stepId);
- }
-
- // 2. toolCallIds belonging to a batch (so their results are absorbed), and a
- // lookup of result chunks by toolCallId for pairing.
- const batchCallIds = new Set<string>();
- const resultByCallId = new Map<string, ToolResultChunk>();
+export function groupRenderedChunks(
+ rendered: readonly RenderedChunk[],
+ recovering = false,
+): readonly RenderGroup[] {
+ const calls = new Map<string, RenderedChunk>();
+ const results = new Map<string, RenderedChunk>();
for (const rc of rendered) {
- const chunk = rc.chunk;
- if (chunk.type === "tool-call" && chunk.stepId !== undefined && batchSteps.has(chunk.stepId)) {
- batchCallIds.add(chunk.toolCallId);
- } else if (chunk.type === "tool-result" && !resultByCallId.has(chunk.toolCallId)) {
- resultByCallId.set(chunk.toolCallId, chunk);
- }
+ const c = rc.chunk;
+ if (c.type === "tool-call" && !calls.has(c.toolCallId)) calls.set(c.toolCallId, rc);
+ if (c.type === "tool-result" && !results.has(c.toolCallId)) results.set(c.toolCallId, rc);
}
- // 3. Emit groups in stream order; each batch lands at its first call.
const groups: RenderGroup[] = [];
- const emittedSteps = new Set<string>();
+ const emitted = new Set<string>();
for (const rc of rendered) {
- const chunk = rc.chunk;
-
- if (chunk.type === "tool-call" && chunk.stepId !== undefined && batchSteps.has(chunk.stepId)) {
- const stepId = chunk.stepId;
- if (emittedSteps.has(stepId)) continue;
- emittedSteps.add(stepId);
-
- const entries: ToolBatchEntry[] = [];
- let provisional = false;
- for (const inner of rendered) {
- if (inner.chunk.type === "tool-call" && inner.chunk.stepId === stepId) {
- const result = resultByCallId.get(inner.chunk.toolCallId) ?? null;
- entries.push({ call: inner.chunk, result });
- if (inner.provisional) provisional = true;
- }
- }
- groups.push({ kind: "tool-batch", stepId, entries, provisional });
+ const c = rc.chunk;
+ if (c.type !== "tool-call" && c.type !== "tool-result") {
+ groups.push({ kind: "single", chunk: rc });
continue;
}
-
- if (chunk.type === "tool-result" && batchCallIds.has(chunk.toolCallId)) {
- continue; // absorbed into its batch
- }
-
- groups.push({ kind: "single", chunk: rc });
+ if (emitted.has(c.toolCallId)) continue;
+ // A matching call owns the card's position, even if the result came first.
+ if (c.type === "tool-result" && calls.has(c.toolCallId)) continue;
+ emitted.add(c.toolCallId);
+ const callRow = calls.get(c.toolCallId);
+ const resultRow = results.get(c.toolCallId);
+ const call = callRow?.chunk.type === "tool-call" ? callRow.chunk : null;
+ const result = resultRow?.chunk.type === "tool-result" ? resultRow.chunk : null;
+ const stepId = call?.stepId ?? result?.stepId;
+ const turnId = callRow?.turnId ?? resultRow?.turnId;
+ groups.push({
+ kind: "tool",
+ entry: {
+ toolCallId: c.toolCallId,
+ toolName: call?.toolName ?? c.toolName,
+ call,
+ result,
+ ...(recovering && call === null && resultRow?.provisional ? { recovering: true } : {}),
+ },
+ provisional: Boolean(callRow?.provisional || resultRow?.provisional),
+ ...(stepId !== undefined ? { stepId } : {}),
+ ...(turnId !== undefined ? { turnId } : {}),
+ });
}
-
return groups;
}
diff --git a/src/core/chunks/index.ts b/src/core/chunks/index.ts
index bdd6ce4..03d687c 100644
--- a/src/core/chunks/index.ts
+++ b/src/core/chunks/index.ts
@@ -1,4 +1,4 @@
-export type { RenderGroup, ToolBatchEntry } from "./groups";
+export type { RenderGroup, ToolEntry } from "./groups";
export { groupRenderedChunks } from "./groups";
export { resolveImageUrl } from "./image-url";
export {
@@ -7,6 +7,8 @@ export {
clearGenerating,
foldEvent,
initialState,
+ markHistoryRecovering,
+ removeLocalEcho,
} from "./reducer";
export type { ProviderRetryView } from "./retry-banner";
export { formatRetryDelay, viewProviderRetry } from "./retry-banner";
diff --git a/src/core/chunks/local-echo.test.ts b/src/core/chunks/local-echo.test.ts
new file mode 100644
index 0000000..565017c
--- /dev/null
+++ b/src/core/chunks/local-echo.test.ts
@@ -0,0 +1,45 @@
+import { describe, expect, it } from "vitest";
+import { appendUserMessage, applyHistory, foldEvent, initialState, removeLocalEcho } from "./index";
+
+describe("request-local optimistic echo", () => {
+ it("retains identity through turn-start tags without conflating identical sends", () => {
+ let state = appendUserMessage(initialState(), "same", [{ url: "data:first" }], "r1");
+ state = appendUserMessage(state, "same", [{ url: "data:second" }], "r2");
+ state = foldEvent(state, { type: "turn-start", conversationId: "c1", turnId: "t1" });
+ state = removeLocalEcho(state, "r1");
+ expect(state.provisional.map((p) => p.localRequestId)).toEqual(["r2", "r2"]);
+ expect(state.provisional.map((p) => p.chunk)).toEqual([
+ { type: "text", text: "same" },
+ { type: "image", url: "data:second" },
+ ]);
+ expect(state.generating).toBe(true);
+ expect(removeLocalEcho(state, "r1")).toBe(state);
+ });
+
+ it("a live user-message confirms only its deduplicated text, not every identical local send", () => {
+ let state = appendUserMessage(initialState(), "same", [{ url: "data:first" }], "r1");
+ state = appendUserMessage(state, "same", [{ url: "data:second" }], "r2");
+ state = foldEvent(state, { type: "turn-start", conversationId: "c1", turnId: "t1" });
+ state = foldEvent(state, {
+ type: "user-message",
+ conversationId: "c1",
+ turnId: "t1",
+ text: "same",
+ });
+ state = removeLocalEcho(removeLocalEcho(state, "r1"), "r2");
+ expect(state.provisional).toEqual([
+ { role: "user", turnId: "t1", opening: true, chunk: { type: "text", text: "same" } },
+ ]);
+ });
+
+ it("does not remove committed rows even if their content is identical", () => {
+ let state = appendUserMessage(initialState(), "same", undefined, "r1");
+ state = applyHistory(state, [{ seq: 1, role: "user", chunk: { type: "text", text: "same" } }]);
+ const committed = state.committed;
+ state = removeLocalEcho(state, "r1");
+ expect(state.committed).toBe(committed);
+ expect(state.provisional).toEqual([]);
+ expect(removeLocalEcho(state, "r1")).toBe(state);
+ expect(removeLocalEcho(initialState(), "r1")).toEqual(initialState());
+ });
+});
diff --git a/src/core/chunks/reconcile.test.ts b/src/core/chunks/reconcile.test.ts
new file mode 100644
index 0000000..3d90a51
--- /dev/null
+++ b/src/core/chunks/reconcile.test.ts
@@ -0,0 +1,143 @@
+import type { AgentEvent, StepId, StoredChunk } from "@dispatch/wire";
+import { describe, expect, it } from "vitest";
+import { applyHistory, foldEvent, initialState, markHistoryRecovering } from "./reducer";
+import { selectChunks } from "./selectors";
+import type { TranscriptState } from "./types";
+
+const stepId = "t1#1" as StepId;
+function fold(s: TranscriptState, e: AgentEvent) {
+ return foldEvent(s, e);
+}
+const call = (toolCallId = "a", turnId = "t1", step = stepId): AgentEvent => ({
+ type: "tool-call",
+ conversationId: "c1",
+ turnId,
+ stepId: step,
+ toolCallId,
+ toolName: "run_shell",
+ input: "partial",
+});
+const result = (toolCallId = "a"): AgentEvent => ({
+ type: "tool-result",
+ conversationId: "c1",
+ turnId: "t1",
+ stepId,
+ toolCallId,
+ toolName: "run_shell",
+ content: "partial",
+ isError: false,
+});
+const stored: StoredChunk[] = [
+ { seq: 50, role: "assistant", turnId: "t1", chunk: { type: "text", text: "full" } },
+ {
+ seq: 51,
+ role: "assistant",
+ turnId: "t1",
+ chunk: {
+ type: "tool-call",
+ stepId,
+ toolCallId: "a",
+ toolName: "run_shell",
+ input: { command: "full" },
+ },
+ },
+ {
+ seq: 52,
+ role: "tool",
+ turnId: "t1",
+ chunk: {
+ type: "tool-result",
+ stepId,
+ toolCallId: "a",
+ toolName: "run_shell",
+ content: "full result",
+ isError: false,
+ },
+ },
+];
+
+describe("history recovery identities", () => {
+ it.each([
+ "before-live",
+ "after-live",
+ ])("reconciles committed tool/text identity %s, not payload equality", (order) => {
+ let s = markHistoryRecovering(initialState(), "t1");
+ if (order === "before-live") s = applyHistory(s, stored);
+ s = fold(s, { type: "text-delta", conversationId: "c1", turnId: "t1", delta: "short" });
+ s = fold(s, call());
+ s = fold(s, result());
+ if (order === "after-live") s = applyHistory(s, stored);
+ expect(selectChunks(s).map((c) => c.chunk)).toEqual(stored.map((c) => c.chunk));
+ // Idempotent refresh; different step's identical text must remain live.
+ s = fold(s, { type: "text-delta", conversationId: "c1", turnId: "t1", delta: "full" });
+ s = applyHistory(s, stored);
+ expect(selectChunks(s).at(-1)).toMatchObject({
+ provisional: true,
+ streaming: true,
+ chunk: { type: "text", text: "full" },
+ });
+ });
+
+ it("does not conflate call/result, another turn, or another step's tool identity", () => {
+ let s = markHistoryRecovering(initialState(), "t1");
+ s = fold(s, call("a", "other"));
+ s = fold(s, call("a", "t1", "other-step" as StepId));
+ s = fold(s, result("a"));
+ s = applyHistory(s, stored);
+ expect(s.provisional).toHaveLength(2);
+ expect(s.provisional.map((p) => p.turnId)).toEqual(["other", "t1"]);
+ expect(s.provisional.every((p) => p.chunk.type === "tool-call")).toBe(true);
+ });
+
+ it("terminal sync removes only that turn and keeps optimistic/unrelated/carried content", () => {
+ let s = markHistoryRecovering(initialState(), "t1");
+ s = fold(s, call());
+ s = fold(s, { type: "turn-sealed", conversationId: "c1", turnId: "t1" });
+ s = fold(s, { type: "turn-start", conversationId: "c1", turnId: "t2" });
+ s = fold(s, { type: "text-delta", conversationId: "c1", turnId: "t2", delta: "queued tail" });
+ s = applyHistory(s, stored, ["t1"]);
+ expect(s.generating).toBe(true);
+ expect(s.currentTurnId).toBe("t2");
+ expect(s.accumulating?.text).toBe("queued tail");
+ expect(s.historyRecoveryTurnIds).toEqual([]);
+ });
+
+ it("does not resurrect watermark-hidden history; older content stays available to Show earlier", () => {
+ let s = markHistoryRecovering({ ...initialState(), hiddenBeforeSeq: 51 }, "t1");
+ s = fold(s, call());
+ s = applyHistory(s, stored);
+ expect(s.committed.map((c) => c.seq)).toEqual([51, 52]);
+ expect(s.provisional).toEqual([]);
+ expect(s.hiddenBeforeSeq).toBe(51);
+ });
+});
+
+it("keeps identical Stop→carry openings as distinct turn identities", () => {
+ let s = markHistoryRecovering(initialState(), "t1");
+ s = fold(s, { type: "user-message", conversationId: "c1", turnId: "t1", text: "same" });
+ s = fold(s, { type: "turn-ended", conversationId: "c1", turnId: "t1", reason: "aborted" });
+ s = fold(s, { type: "user-message", conversationId: "c1", turnId: "t2", text: "same" });
+ s = fold(s, { type: "turn-start", conversationId: "c1", turnId: "t2" });
+ s = applyHistory(
+ s,
+ [{ seq: 1, role: "user", turnId: "t1", chunk: { type: "text", text: "same" } }],
+ ["t1"],
+ );
+ expect(selectChunks(s).map((c) => [c.turnId, c.provisional])).toEqual([
+ ["t1", false],
+ ["t2", true],
+ ]);
+ expect(s.generating).toBe(true);
+});
+
+it("reconciles steering by its completed tool batch, not repeated user text", () => {
+ let s = markHistoryRecovering(initialState(), "t1");
+ s = fold(s, result());
+ s = fold(s, { type: "steering", conversationId: "c1", turnId: "t1", text: "short" });
+ s = applyHistory(s, [
+ ...stored,
+ { seq: 53, role: "user", turnId: "t1", chunk: { type: "text", text: "full steering" } },
+ ]);
+ expect(s.provisional).toEqual([]);
+ expect(selectChunks(s).at(-1)?.chunk).toEqual({ type: "text", text: "full steering" });
+});
diff --git a/src/core/chunks/reconcile.ts b/src/core/chunks/reconcile.ts
new file mode 100644
index 0000000..277d25b
--- /dev/null
+++ b/src/core/chunks/reconcile.ts
@@ -0,0 +1,54 @@
+import type { StoredChunk } from "@dispatch/wire";
+import type { ProvisionalChunk, TranscriptState } from "./types";
+
+/**
+ * A persisted tool call proves its entire assistant step committed atomically.
+ * Text/thinking wire chunks have no stepId: their provisional provenance is
+ * learned from tool-call/step-complete, never from text equality. This also
+ * handles a window starting midway through the step and an omitted prefix.
+ * Also usable when HTTP finished before replay/live flush.
+ */
+export function historyBackedProvisional(
+ state: TranscriptState,
+ committed: readonly StoredChunk[] = state.committed,
+): ReadonlySet<ProvisionalChunk> {
+ const backed = new Set<ProvisionalChunk>();
+ if (!state.historyRecoveryTurnIds?.length) return backed;
+ const steps = new Set<string>();
+ const steeringAnchors = new Set<string>();
+ let lastResult: StoredChunk | undefined;
+ const toolRows = new Map<string, Set<string | undefined>>();
+ for (const row of committed) {
+ const c = row.chunk;
+ if (row.role === "assistant" || row.turnId !== lastResult?.turnId) lastResult = undefined;
+ if (c.type === "tool-result") lastResult = row;
+ if (row.role === "user" && c.type === "text" && lastResult?.chunk.type === "tool-result") {
+ steeringAnchors.add(`${row.turnId}\0${lastResult.chunk.toolCallId}`);
+ }
+ if (c.type === "tool-call" || c.type === "tool-result") {
+ const key = `${row.turnId}\0${c.type}\0${c.toolCallId}`;
+ const ids = toolRows.get(key) ?? new Set<string | undefined>();
+ ids.add(c.stepId);
+ toolRows.set(key, ids);
+ if (c.type === "tool-call" && c.stepId !== undefined) steps.add(`${row.turnId}\0${c.stepId}`);
+ }
+ }
+ for (const p of state.provisional) {
+ if (p.turnId === undefined || !state.historyRecoveryTurnIds?.includes(p.turnId)) continue;
+ const c = p.chunk;
+ if (c.type === "tool-call" || c.type === "tool-result") {
+ const ids = toolRows.get(`${p.turnId}\0${c.type}\0${c.toolCallId}`);
+ if (ids !== undefined && (c.stepId === undefined || ids.has(undefined) || ids.has(c.stepId)))
+ backed.add(p);
+ } else if (p.afterToolCallId !== undefined && p.role === "user") {
+ if (steeringAnchors.has(`${p.turnId}\0${p.afterToolCallId}`)) backed.add(p);
+ } else if (p.opening && c.type === "text") {
+ // The opening is the first user row of this turn; steering is NOT opening.
+ const first = committed.find((row) => row.turnId === p.turnId);
+ if (first?.role === "user" && first.chunk.type === "text") backed.add(p);
+ } else if (p.stepId !== undefined && p.role === "assistant") {
+ if (steps.has(`${p.turnId}\0${p.stepId}`)) backed.add(p);
+ }
+ }
+ return backed;
+}
diff --git a/src/core/chunks/reducer.test.ts b/src/core/chunks/reducer.test.ts
index 058552e..10d662e 100644
--- a/src/core/chunks/reducer.test.ts
+++ b/src/core/chunks/reducer.test.ts
@@ -7,6 +7,7 @@ import type {
TurnInputEvent,
TurnProviderRetryEvent,
TurnReasoningDeltaEvent,
+ TurnReasoningEndEvent,
TurnSealedEvent,
TurnStartEvent,
TurnSteeringEvent,
@@ -45,6 +46,12 @@ const reasoningDelta = (turnId: string, delta: string): TurnReasoningDeltaEvent
delta,
});
+const reasoningEnd = (turnId: string): TurnReasoningEndEvent => ({
+ type: "reasoning-end",
+ conversationId: "c1",
+ turnId,
+});
+
const toolCall = (
turnId: string,
toolCallId: string,
@@ -226,7 +233,7 @@ describe("clearGenerating", () => {
s = foldEvent(s, textDelta("t1", "partial"));
const cleared = clearGenerating(s);
expect(cleared.generating).toBe(false);
- expect(cleared.accumulating).toEqual({ kind: "text", text: "partial" });
+ expect(cleared.accumulating).toEqual({ kind: "text", text: "partial", turnId: "t1" });
expect(cleared.currentTurnId).toBe("t1");
});
});
@@ -236,7 +243,7 @@ describe("foldEvent — text-delta", () => {
let s = initialState();
s = foldEvent(s, turnStart("t1"));
s = foldEvent(s, textDelta("t1", "hello"));
- expect(s.accumulating).toEqual({ kind: "text", text: "hello" });
+ expect(s.accumulating).toEqual({ kind: "text", text: "hello", turnId: "t1" });
expect(s.provisional).toEqual([]);
});
@@ -245,7 +252,7 @@ describe("foldEvent — text-delta", () => {
s = foldEvent(s, turnStart("t1"));
s = foldEvent(s, textDelta("t1", "hello "));
s = foldEvent(s, textDelta("t1", "world"));
- expect(s.accumulating).toEqual({ kind: "text", text: "hello world" });
+ expect(s.accumulating).toEqual({ kind: "text", text: "hello world", turnId: "t1" });
expect(s.provisional).toEqual([]);
});
@@ -254,7 +261,7 @@ describe("foldEvent — text-delta", () => {
s = foldEvent(s, turnStart("t1"));
s = foldEvent(s, reasoningDelta("t1", "thinking..."));
s = foldEvent(s, textDelta("t1", "answer"));
- expect(s.accumulating).toEqual({ kind: "text", text: "answer" });
+ expect(s.accumulating).toEqual({ kind: "text", text: "answer", turnId: "t1" });
expect(s.provisional).toHaveLength(1);
expect(s.provisional[0]?.chunk).toEqual({ type: "thinking", text: "thinking..." });
expect(s.provisional[0]?.role).toBe("assistant");
@@ -266,7 +273,7 @@ describe("foldEvent — reasoning-delta", () => {
let s = initialState();
s = foldEvent(s, turnStart("t1"));
s = foldEvent(s, reasoningDelta("t1", "hmm"));
- expect(s.accumulating).toEqual({ kind: "thinking", text: "hmm" });
+ expect(s.accumulating).toEqual({ kind: "thinking", text: "hmm", turnId: "t1" });
});
it("successive reasoning-deltas extend the same chunk", () => {
@@ -274,7 +281,73 @@ describe("foldEvent — reasoning-delta", () => {
s = foldEvent(s, turnStart("t1"));
s = foldEvent(s, reasoningDelta("t1", "hmm "));
s = foldEvent(s, reasoningDelta("t1", "ok"));
- expect(s.accumulating).toEqual({ kind: "thinking", text: "hmm ok" });
+ expect(s.accumulating).toEqual({ kind: "thinking", text: "hmm ok", turnId: "t1" });
+ });
+});
+
+describe("foldEvent — reasoning-end (thinking block boundary)", () => {
+ it("closes each block so consecutive deltas become separate chunks", () => {
+ let s = initialState();
+ s = foldEvent(s, turnStart("t1"));
+ s = foldEvent(s, reasoningDelta("t1", "a"));
+ s = foldEvent(s, reasoningEnd("t1"));
+ s = foldEvent(s, reasoningDelta("t1", "b"));
+ s = foldEvent(s, reasoningEnd("t1"));
+ expect(s.provisional.map((p) => p.chunk)).toEqual([
+ { type: "thinking", text: "a" },
+ { type: "thinking", text: "b" },
+ ]);
+ expect(s.accumulating).toBeNull();
+ });
+
+ it("still merges legacy deltas with no boundary (no reasoning-end)", () => {
+ let s = initialState();
+ s = foldEvent(s, turnStart("t1"));
+ s = foldEvent(s, reasoningDelta("t1", "a"));
+ s = foldEvent(s, reasoningDelta("t1", "b"));
+ expect(s.accumulating).toEqual({ kind: "thinking", text: "ab", turnId: "t1" });
+ });
+
+ it("end with nothing open appends an empty thinking chunk for the turn", () => {
+ let s = initialState();
+ s = foldEvent(s, turnStart("t1"));
+ s = foldEvent(s, reasoningEnd("t1"));
+ expect(s.provisional).toHaveLength(1);
+ expect(s.provisional[0]?.chunk).toEqual({ type: "thinking", text: "" });
+ expect(s.provisional[0]?.turnId).toBe("t1");
+ expect(s.accumulating).toBeNull();
+ });
+
+ it("text-delta then end flushes the text, then appends an empty thinking chunk", () => {
+ let s = initialState();
+ s = foldEvent(s, turnStart("t1"));
+ s = foldEvent(s, textDelta("t1", "x"));
+ s = foldEvent(s, reasoningEnd("t1"));
+ expect(s.provisional.map((p) => p.chunk)).toEqual([
+ { type: "text", text: "x" },
+ { type: "thinking", text: "" },
+ ]);
+ });
+
+ it("end for another turn flushes the open thinking and appends an empty one", () => {
+ let s = initialState();
+ s = foldEvent(s, turnStart("t1"));
+ s = foldEvent(s, reasoningDelta("t1", "open"));
+ s = foldEvent(s, reasoningEnd("t2"));
+ expect(s.provisional.map((p) => [p.turnId, p.chunk])).toEqual([
+ ["t1", { type: "thinking", text: "open" }],
+ ["t2", { type: "thinking", text: "" }],
+ ]);
+ expect(s.accumulating).toBeNull();
+ });
+
+ it("clears a pending provider-retry banner", () => {
+ let s = initialState();
+ s = foldEvent(s, turnStart("t1"));
+ s = foldEvent(s, providerRetry("t1", 0, 5000));
+ expect(selectProviderRetry(s)).not.toBeNull();
+ s = foldEvent(s, reasoningEnd("t1"));
+ expect(selectProviderRetry(s)).toBeNull();
});
});
@@ -531,7 +604,7 @@ describe("clearGenerating also clears a stale provider-retry banner (reconnect)"
s = foldEvent(s, textDelta("t1", "partial"));
s = foldEvent(s, providerRetry("t1", 0, 5000));
const cleared = clearGenerating(s);
- expect(cleared.accumulating).toEqual({ kind: "text", text: "partial" });
+ expect(cleared.accumulating).toEqual({ kind: "text", text: "partial", turnId: "t1" });
expect(selectProviderRetry(cleared)).toBeNull();
});
});
@@ -859,7 +932,7 @@ describe("appendUserMessage", () => {
let s = initialState();
s = foldEvent(s, turnStart("t1"));
s = foldEvent(s, textDelta("t1", "partial"));
- expect(s.accumulating).toEqual({ kind: "text", text: "partial" });
+ expect(s.accumulating).toEqual({ kind: "text", text: "partial", turnId: "t1" });
s = appendUserMessage(s, "user msg");
expect(s.accumulating).toBeNull();
@@ -1022,3 +1095,115 @@ describe("applyHistory — multi-chunk image echo is superseded by committed", (
expect(msgs[0]?.chunks[1]).toEqual({ type: "image", url: PNG, mimeType: "image/png" });
});
});
+
+describe("turnId provenance on live chunks", () => {
+ it("tags every live chunk with its event's turnId, and the optimistic echo on turn-start", () => {
+ let s = initialState();
+ s = appendUserMessage(s, "q", [{ url: "data:image/png;base64,AA", mimeType: "image/png" }]);
+ expect(selectChunks(s).map((c) => c.turnId)).toEqual([undefined, undefined]);
+ s = foldEvent(s, turnStart("t1"));
+ s = foldEvent(s, textDelta("t1", "a"));
+ const chunks = selectChunks(s);
+ expect(chunks.map((c) => c.turnId)).toEqual(["t1", "t1", "t1"]);
+ });
+
+ it("turn-start does not re-tag an echo already owned by an earlier turn", () => {
+ let s = initialState();
+ s = appendUserMessage(s, "q1");
+ s = foldEvent(s, turnStart("t1"));
+ s = foldEvent(s, textDelta("t1", "a1"));
+ s = foldEvent(s, turnStart("t2"));
+ expect(selectChunks(s).map((c) => c.turnId)).toEqual(["t1", "t1"]);
+ });
+
+ it("committed chunks expose StoredChunk.turnId", () => {
+ const s = applyHistory(initialState(), [
+ { seq: 1, role: "user", chunk: { type: "text", text: "q" }, turnId: "t9" },
+ { seq: 2, role: "assistant", chunk: { type: "text", text: "a" } },
+ ]);
+ expect(selectChunks(s).map((c) => c.turnId)).toEqual(["t9", undefined]);
+ });
+});
+
+describe("foldEvent — additive wire compatibility", () => {
+ it.each(["future-event"])("preserves transcript identity for %s", (type) => {
+ const state = foldEvent(foldEvent(initialState(), turnStart("t1")), textDelta("t1", "partial"));
+ const before = structuredClone(state);
+ const event = { type, conversationId: "c1", turnId: "t1", reason: "sealed" };
+ expect(foldEvent(state, event as Parameters<typeof foldEvent>[1])).toBe(state);
+ expect(state).toEqual(before);
+ });
+});
+
+describe("foldEvent — authoritative turn terminal", () => {
+ it.each([
+ "sealed",
+ "aborted",
+ "error",
+ "rejected",
+ ] as const)("ends generation on %s", (reason) => {
+ const before = foldEvent(
+ foldEvent(initialState(), turnStart("t1")),
+ textDelta("t1", "partial"),
+ );
+ const next = foldEvent(before, {
+ type: "turn-ended",
+ conversationId: "c1",
+ turnId: "t1",
+ reason,
+ });
+ expect(next.generating).toBe(false);
+ expect(next.accumulating).toBeNull();
+ expect(selectChunks(next).map((c) => c.chunk)).toEqual([{ type: "text", text: "partial" }]);
+ expect(before.accumulating?.text).toBe("partial");
+ });
+});
+
+describe("foldEvent — re-replay of an in-flight turn (re-subscribe after tab switch / reconnect)", () => {
+ const userMessage = (turnId: string, text: string): TurnInputEvent => ({
+ type: "user-message",
+ conversationId: "c1",
+ turnId,
+ text,
+ });
+ const replay = (upTo: number) =>
+ [
+ userMessage("t1", "go"),
+ turnStart("t1"),
+ reasoningDelta("t1", "plan"),
+ toolCall("t1", "tc1", "read", {}),
+ toolResult("t1", "tc1", "read", "out"),
+ { type: "steering", conversationId: "c1", turnId: "t1", text: "also this" } as const,
+ reasoningDelta("t1", "more"),
+ toolCall("t1", "tc2", "read", {}),
+ ].slice(0, upTo);
+
+ it("replaces the stale copy instead of appending a duplicate turn", () => {
+ let s = replay(7).reduce(foldEvent, initialState()); // first view, mid-thinking
+ s = replay(8).reduce(foldEvent, s); // came back: the whole turn replays again
+ const once = replay(8).reduce(foldEvent, initialState());
+ expect(selectChunks(s)).toEqual(selectChunks(once));
+ expect(s.accumulating).toBeNull();
+ });
+
+ it("drops a stale steering bubble that ended the old copy", () => {
+ let s = replay(6).reduce(foldEvent, initialState()); // old copy ends in steering
+ s = replay(8).reduce(foldEvent, s);
+ expect(selectChunks(s).filter((c) => c.role === "user")).toHaveLength(2); // opening + steer
+ });
+
+ it("keeps other turns' provisional chunks", () => {
+ let s = [turnStart("t0"), textDelta("t0", "earlier"), doneEvent("t0")].reduce(
+ foldEvent,
+ initialState(),
+ );
+ s = replay(4).reduce(foldEvent, s);
+ s = replay(4).reduce(foldEvent, s);
+ expect(selectChunks(s).map((c) => c.chunk.type)).toEqual([
+ "text",
+ "text",
+ "thinking",
+ "tool-call",
+ ]);
+ });
+});
diff --git a/src/core/chunks/reducer.ts b/src/core/chunks/reducer.ts
index 64e41b9..71df4dd 100644
--- a/src/core/chunks/reducer.ts
+++ b/src/core/chunks/reducer.ts
@@ -1,5 +1,6 @@
import type { AgentEvent, Chunk, ImageInput, Role, StoredChunk } from "@dispatch/wire";
import { assertChunkExhaustive } from "../wire/conformance";
+import { historyBackedProvisional } from "./reconcile";
import type { AccumulatingChunk, ProvisionalChunk, TranscriptState } from "./types";
/** The initial empty transcript state. */
@@ -40,7 +41,59 @@ function flushAccumulating(
if (acc === null) return provisional;
const chunk: Chunk =
acc.kind === "text" ? { type: "text", text: acc.text } : { type: "thinking", text: acc.text };
- return [...provisional, { role: "assistant", chunk }];
+ return [
+ ...provisional,
+ withTurn(
+ { role: "assistant", chunk, ...(acc.stepId !== undefined ? { stepId: acc.stepId } : {}) },
+ acc.turnId,
+ ),
+ ];
+}
+
+/** Attach a `turnId` to a provisional chunk (omitted when unknown). */
+function withTurn(p: ProvisionalChunk, turnId: string | undefined): ProvisionalChunk {
+ return turnId !== undefined ? { ...p, turnId } : p;
+}
+
+/**
+ * Tag the trailing UNTAGGED user run (the sender's optimistic echo, appended
+ * before the server assigned a turn id) with the turn that just started.
+ */
+function tagTrailingUserEcho(
+ provisional: readonly ProvisionalChunk[],
+ turnId: string,
+): readonly ProvisionalChunk[] {
+ let start = provisional.length;
+ while (start > 0) {
+ const p = provisional[start - 1];
+ if (p === undefined || p.role !== "user" || p.turnId !== undefined) break;
+ start--;
+ }
+ if (start === provisional.length) return provisional;
+ return [
+ ...provisional.slice(0, start),
+ ...provisional.slice(start).map((p) => ({ ...p, turnId })),
+ ];
+}
+
+/**
+ * Drop the previous copy of a turn that is being replayed from its start.
+ * Keeps the trailing user run — the opening this replay's `user-message` just
+ * appended (or the sender's echo that `turn-start` just tagged) — and every
+ * other turn's chunks.
+ */
+function dropReplayedTurn(
+ provisional: readonly ProvisionalChunk[],
+ turnId: string,
+): readonly ProvisionalChunk[] {
+ let tail = provisional.length;
+ while (tail > 0 && provisional[tail - 1]?.role === "user") tail--;
+ // A stale steering bubble (replayed again as its own `steering` event) can
+ // end the old copy and so sit inside the trailing user run.
+ const stale = (p: ProvisionalChunk, i: number): boolean =>
+ p.turnId === turnId && (i < tail || p.afterToolCallId !== undefined);
+ if (!provisional.some(stale)) return provisional;
+ return provisional.filter((p, i) => !stale(p, i));
}
/**
@@ -103,8 +156,8 @@ function trailingRun(
/**
* Merge authoritative seq-keyed chunks into the committed history.
* Dedupes by seq (new wins), keeps seq-monotonic order, idempotent.
- * When sealedTurnId is set, drops all provisional chunks (now superseded)
- * and clears sealedTurnId.
+ * A sealed/finally-synced turn supersedes only its own provisional chunks;
+ * unrelated turns and an already-running queued carry remain live.
*
* Chunks below the chat-limit unload watermark (`hiddenBeforeSeq`) are
* REJECTED: a full-cache or tail merge must not resurrect what the trim
@@ -113,6 +166,7 @@ function trailingRun(
export function applyHistory(
state: TranscriptState,
chunks: readonly StoredChunk[],
+ finishedTurnIds: readonly string[] = [],
): TranscriptState {
const seqMap = new Map<number, StoredChunk>();
for (const c of state.committed) seqMap.set(c.seq, c);
@@ -123,14 +177,49 @@ export function applyHistory(
seqMap.set(c.seq, c);
}
const committed = Array.from(seqMap.values()).sort((a, b) => a.seq - b.seq);
+ const identityRows = new Map(seqMap);
+ for (const c of chunks) identityRows.set(c.seq, c);
+ const backed = historyBackedProvisional(
+ state,
+ [...identityRows.values()].sort((a, b) => a.seq - b.seq),
+ );
+ const finished = new Set(finishedTurnIds);
+ if (state.sealedTurnId !== null) finished.add(state.sealedTurnId);
+ state = {
+ ...state,
+ provisional: state.provisional.filter((p) => !backed.has(p) && !finished.has(p.turnId ?? "")),
+ accumulating:
+ state.accumulating?.turnId !== undefined && finished.has(state.accumulating.turnId)
+ ? null
+ : state.accumulating,
+ };
if (state.sealedTurnId !== null) {
+ // Legacy (non-recovery) histories may carry no turn provenance at all.
+ // Retain their seal cleanup, but never erase an already-live queued carry.
+ const legacySeal =
+ (state.historyRecoveryTurnIds?.length ?? 0) === 0 && finishedTurnIds.length === 0;
return {
...state,
committed,
- provisional: [],
- accumulating: null,
+ provisional: legacySeal
+ ? state.provisional.filter(
+ (p) =>
+ state.generating &&
+ p.turnId === state.currentTurnId &&
+ p.turnId !== state.sealedTurnId,
+ )
+ : state.provisional,
+ accumulating:
+ legacySeal && state.accumulating?.turnId === state.sealedTurnId ? null : state.accumulating,
sealedTurnId: null,
+ ...(state.historyRecoveryTurnIds !== undefined
+ ? {
+ historyRecoveryTurnIds: state.historyRecoveryTurnIds.filter(
+ (id) => !finishedTurnIds.includes(id),
+ ),
+ }
+ : {}),
};
}
@@ -141,7 +230,14 @@ export function applyHistory(
// match the trailing provisional user-run against the trailing committed
// user-run by content equality and drop the whole echo when it is fully
// backed. Leaves the accumulating (streaming) chunk untouched.
- if (addedNew && state.generating && state.provisional.length > 0) {
+ if (
+ addedNew &&
+ state.generating &&
+ state.provisional.length > 0 &&
+ !state.provisional.some(
+ (p) => p.turnId !== undefined && state.historyRecoveryTurnIds?.includes(p.turnId),
+ )
+ ) {
const provRun = trailingRun(state.provisional, "user");
if (provRun.length > 0) {
const commRun = trailingRun(committed, "user");
@@ -153,7 +249,11 @@ export function applyHistory(
commRun.length >= provRun.length &&
provRun.every((p, i) => {
const c = commRun[i];
- return c !== undefined && chunkContentEquals(p.chunk, c.chunk);
+ return (
+ c !== undefined &&
+ (c.turnId === undefined || p.turnId === undefined || c.turnId === p.turnId) &&
+ chunkContentEquals(p.chunk, c.chunk)
+ );
});
if (fullyBacked) {
const dropStart = state.provisional.length - provRun.length;
@@ -163,7 +263,17 @@ export function applyHistory(
}
}
- return { ...state, committed };
+ return {
+ ...state,
+ committed,
+ ...(state.historyRecoveryTurnIds !== undefined
+ ? {
+ historyRecoveryTurnIds: state.historyRecoveryTurnIds.filter(
+ (id) => !finishedTurnIds.includes(id),
+ ),
+ }
+ : {}),
+ };
}
/**
@@ -174,6 +284,8 @@ export function applyHistory(
* optimistic echo) so a watcher renders it mid-turn.
* - `text-delta` extends the current accumulating TextChunk (or starts one).
* - `reasoning-delta` extends the current accumulating ThinkingChunk (or starts one).
+ * - `reasoning-end` closes the open ThinkingChunk as its own chunk; with none open
+ * (a redacted/empty block) it appends an empty thinking chunk, mirroring history.
* - `tool-call` / `tool-result` / `error` finalize any accumulating chunk and
* add a new provisional chunk.
* - `steering` appends a user bubble mid-turn (drained from the message queue
@@ -198,8 +310,26 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
case "tool-output":
return state;
- case "turn-start":
- return { ...state, currentTurnId: event.turnId, generating: true };
+ case "turn-start": {
+ // A turn starts exactly once live, so a `turn-start` for a turn that
+ // already has provisional content is a late-join REPLAY restarting from
+ // the top (re-subscribe after a tab switch / WS reconnect): the server
+ // re-delivers the whole in-flight turn. Replace the stale copy instead of
+ // appending a second one (a duplicate copy rendered as a tail of thinking
+ // blocks with its tool cards collapsed onto the first copy's).
+ const stale = state.accumulating?.turnId === event.turnId;
+ const provisional = tagTrailingUserEcho(
+ flushAccumulating(state.provisional, stale ? null : state.accumulating),
+ event.turnId,
+ );
+ return {
+ ...state,
+ provisional: dropReplayedTurn(provisional, event.turnId),
+ accumulating: null,
+ currentTurnId: event.turnId,
+ generating: true,
+ };
+ }
case "user-message": {
// The turn's USER prompt, surfaced on the event stream (backend CR-3) so a
@@ -215,14 +345,37 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
// text) emits no `user-message` and is not de-duped here.
if (event.text.length === 0) return state;
const run = trailingRun(state.provisional, "user");
- const alreadyEchoed = run.some((p) => p.chunk.type === "text" && p.chunk.text === event.text);
- if (alreadyEchoed) {
- return { ...state, generating: true };
+ const echoed = run.find(
+ (p) =>
+ (p.turnId === undefined || p.turnId === event.turnId) &&
+ p.chunk.type === "text" &&
+ p.chunk.text === event.text,
+ );
+ if (echoed !== undefined) {
+ return {
+ ...state,
+ // Preserve the live text independently of the local send. A delayed
+ // queued ack may remove its images, but cannot erase this server event.
+ provisional: tagTrailingUserEcho(state.provisional, event.turnId).map((p) => {
+ if (p.chunk !== echoed.chunk) return p;
+ const { localRequestId: _local, ...confirmed } = p;
+ return { ...confirmed, opening: true };
+ }),
+ generating: true,
+ };
}
const provisional = flushAccumulating(state.provisional, state.accumulating);
return {
...state,
- provisional: [...provisional, { role: "user", chunk: { type: "text", text: event.text } }],
+ provisional: [
+ ...provisional,
+ {
+ role: "user",
+ chunk: { type: "text", text: event.text },
+ turnId: event.turnId,
+ opening: true,
+ },
+ ],
accumulating: null,
generating: true,
};
@@ -230,10 +383,10 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
case "text-delta": {
const acc = state.accumulating;
- if (acc !== null && acc.kind === "text") {
+ if (acc !== null && acc.kind === "text" && acc.turnId === event.turnId) {
return {
...state,
- accumulating: { kind: "text", text: acc.text + event.delta },
+ accumulating: { ...acc, text: acc.text + event.delta },
generating: true,
};
}
@@ -241,17 +394,17 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
return {
...state,
provisional,
- accumulating: { kind: "text", text: event.delta },
+ accumulating: { kind: "text", text: event.delta, turnId: event.turnId },
generating: true,
};
}
case "reasoning-delta": {
const acc = state.accumulating;
- if (acc !== null && acc.kind === "thinking") {
+ if (acc !== null && acc.kind === "thinking" && acc.turnId === event.turnId) {
return {
...state,
- accumulating: { kind: "thinking", text: acc.text + event.delta },
+ accumulating: { ...acc, text: acc.text + event.delta },
generating: true,
};
}
@@ -259,7 +412,31 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
return {
...state,
provisional,
- accumulating: { kind: "thinking", text: event.delta },
+ accumulating: { kind: "thinking", text: event.delta, turnId: event.turnId },
+ generating: true,
+ };
+ }
+
+ case "reasoning-end": {
+ const acc = state.accumulating;
+ // Close the open thinking block of this turn: flush it as its own provisional chunk.
+ if (acc !== null && acc.kind === "thinking" && acc.turnId === event.turnId) {
+ return {
+ ...state,
+ provisional: flushAccumulating(state.provisional, acc),
+ accumulating: null,
+ generating: true,
+ };
+ }
+ // No open thinking (a redacted/empty block with no deltas): history persists an
+ // EMPTY thinking chunk for it, so mirror that 1:1.
+ return {
+ ...state,
+ provisional: [
+ ...flushAccumulating(state.provisional, acc),
+ { role: "assistant", chunk: { type: "thinking", text: "" }, turnId: event.turnId },
+ ],
+ accumulating: null,
generating: true,
};
}
@@ -275,7 +452,7 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
};
return {
...state,
- provisional: [...provisional, { role: "assistant", chunk }],
+ provisional: [...provisional, { role: "assistant", chunk, turnId: event.turnId }],
accumulating: null,
generating: true,
};
@@ -293,7 +470,7 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
};
return {
...state,
- provisional: [...provisional, { role: "tool", chunk }],
+ provisional: [...provisional, { role: "tool", chunk, turnId: event.turnId }],
accumulating: null,
generating: true,
};
@@ -307,7 +484,7 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
: { type: "error", message: event.message };
return {
...state,
- provisional: [...provisional, { role: "assistant", chunk }],
+ provisional: [...provisional, { role: "assistant", chunk, turnId: event.turnId }],
accumulating: null,
generating: false,
};
@@ -317,8 +494,24 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
return { ...state, latestUsage: event.usage };
case "step-complete":
- // Timing metadata — no content chunk; handled by the telemetry reducer.
- return state;
+ // A persistence boundary may lag this notification. Stamp provenance and
+ // finish the generation chunk, but keep it until history really backs it.
+ return {
+ ...state,
+ provisional: flushAccumulating(state.provisional, state.accumulating),
+ accumulating: null,
+ };
+
+ case "turn-ended":
+ return {
+ ...state,
+ provisional:
+ state.accumulating?.turnId === event.turnId
+ ? flushAccumulating(state.provisional, state.accumulating)
+ : state.provisional,
+ accumulating: state.accumulating?.turnId === event.turnId ? null : state.accumulating,
+ generating: state.currentTurnId === event.turnId ? false : state.generating,
+ };
case "done": {
const provisional = flushAccumulating(state.provisional, state.accumulating);
@@ -350,9 +543,22 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
// into the transcript by the sender).
if (event.text.length === 0) return state;
const provisional = flushAccumulating(state.provisional, state.accumulating);
+ const preceding = [...state.committed, ...provisional]
+ .reverse()
+ .find((p) => p.turnId === event.turnId && p.chunk.type === "tool-result");
+ const afterToolCallId =
+ preceding?.chunk.type === "tool-result" ? preceding.chunk.toolCallId : undefined;
return {
...state,
- provisional: [...provisional, { role: "user", chunk: { type: "text", text: event.text } }],
+ provisional: [
+ ...provisional,
+ {
+ role: "user",
+ chunk: { type: "text", text: event.text },
+ turnId: event.turnId,
+ ...(afterToolCallId !== undefined ? { afterToolCallId } : {}),
+ },
+ ],
accumulating: null,
generating: true,
};
@@ -367,6 +573,9 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
// (already true from `turn-start`) is left untouched.
return { ...state, providerRetry: event };
}
+
+ default:
+ return state; // Forward-compatible with additive wire events.
}
}
@@ -385,14 +594,37 @@ const RETRY_CLEARING_EVENTS: ReadonlySet<AgentEvent["type"]> = new Set([
"turn-start",
"text-delta",
"reasoning-delta",
+ "reasoning-end",
"tool-call",
"tool-result",
"error",
"done",
"turn-sealed",
+ "turn-ended",
]);
export function foldEvent(state: TranscriptState, event: AgentEvent): TranscriptState {
+ if (
+ event.type === "tool-call" ||
+ event.type === "tool-result" ||
+ event.type === "step-complete"
+ ) {
+ state = {
+ ...state,
+ provisional: state.provisional.map((p) =>
+ p.turnId === event.turnId &&
+ p.role === "assistant" &&
+ p.stepId === undefined &&
+ (p.chunk.type === "text" || p.chunk.type === "thinking")
+ ? { ...p, stepId: event.stepId }
+ : p,
+ ),
+ accumulating:
+ state.accumulating?.turnId === event.turnId
+ ? { ...state.accumulating, stepId: event.stepId }
+ : state.accumulating,
+ };
+ }
const next = reduceEvent(state, event);
if (event.type === "provider-retry") return next; // set by reduceEvent; not a clearing event
if (RETRY_CLEARING_EVENTS.has(event.type) && state.providerRetry !== null) {
@@ -419,6 +651,7 @@ export function appendUserMessage(
state: TranscriptState,
text: string,
images?: readonly ImageInput[],
+ localRequestId?: string,
): TranscriptState {
const provisional = flushAccumulating(state.provisional, state.accumulating);
const userChunks: Chunk[] = [];
@@ -440,7 +673,35 @@ export function appendUserMessage(
}
return {
...state,
- provisional: [...provisional, ...userChunks.map((chunk) => ({ role: "user", chunk }) as const)],
+ provisional: [
+ ...provisional,
+ ...userChunks.map(
+ (chunk) =>
+ ({
+ role: "user",
+ chunk,
+ ...(localRequestId !== undefined ? { localRequestId } : {}),
+ }) as const,
+ ),
+ ],
accumulating: null,
};
}
+
+/** Remove only still-local artifacts of one accepted queued send; never history/live text. */
+export function removeLocalEcho(state: TranscriptState, requestId: string): TranscriptState {
+ const provisional = state.provisional.filter((p) => p.localRequestId !== requestId);
+ return provisional.length === state.provisional.length ? state : { ...state, provisional };
+}
+
+/** A resync notice is lifecycle evidence, never fabricated transcript content. */
+export function markHistoryRecovering(state: TranscriptState, turnId: string): TranscriptState {
+ return {
+ ...state,
+ currentTurnId: turnId,
+ generating: true,
+ historyRecoveryTurnIds: state.historyRecoveryTurnIds?.includes(turnId)
+ ? state.historyRecoveryTurnIds
+ : [...(state.historyRecoveryTurnIds ?? []), turnId],
+ };
+}
diff --git a/src/core/chunks/selectors.test.ts b/src/core/chunks/selectors.test.ts
new file mode 100644
index 0000000..90bb5cc
--- /dev/null
+++ b/src/core/chunks/selectors.test.ts
@@ -0,0 +1,373 @@
+import type { AgentEvent, StepId, StoredChunk } from "@dispatch/wire";
+import { describe, expect, it } from "vitest";
+import { applyHistory, foldEvent, initialState, markHistoryRecovering } from "./reducer";
+import { selectChunks } from "./selectors";
+import { trimTranscript } from "./trim";
+import type { TranscriptState } from "./types";
+
+const C = "conv";
+const T = "turn-1";
+
+function fold(state: TranscriptState, events: readonly AgentEvent[]): TranscriptState {
+ return events.reduce(foldEvent, state);
+}
+
+const turnStart: AgentEvent = { type: "turn-start", conversationId: C, turnId: T };
+const userMsg: AgentEvent = { type: "user-message", conversationId: C, turnId: T, text: "hi" };
+const think = (delta: string): AgentEvent => ({
+ type: "reasoning-delta",
+ conversationId: C,
+ turnId: T,
+ delta,
+});
+const thinkEnd: AgentEvent = { type: "reasoning-end", conversationId: C, turnId: T };
+const text = (delta: string): AgentEvent => ({
+ type: "text-delta",
+ conversationId: C,
+ turnId: T,
+ delta,
+});
+const call = (id: string, step = 0): AgentEvent => ({
+ type: "tool-call",
+ conversationId: C,
+ turnId: T,
+ stepId: `${T}#${step}` as StepId,
+ toolCallId: id,
+ toolName: "run_shell",
+ input: {},
+});
+const result = (id: string, step = 0): AgentEvent => ({
+ type: "tool-result",
+ conversationId: C,
+ turnId: T,
+ stepId: `${T}#${step}` as StepId,
+ toolCallId: id,
+ toolName: "run_shell",
+ content: "ok",
+ isError: false,
+});
+
+/** The persisted chunks for steps 0 (user + think + call/result). */
+const committedStep0: StoredChunk[] = [
+ { seq: 1, role: "user", chunk: { type: "text", text: "hi" }, turnId: T },
+ { seq: 2, role: "assistant", chunk: { type: "thinking", text: "plan" }, turnId: T },
+ {
+ seq: 3,
+ role: "assistant",
+ chunk: {
+ type: "tool-call",
+ toolCallId: "a",
+ toolName: "run_shell",
+ input: {},
+ stepId: `${T}#0` as StepId,
+ },
+ turnId: T,
+ },
+ {
+ seq: 4,
+ role: "tool",
+ chunk: {
+ type: "tool-result",
+ toolCallId: "a",
+ toolName: "run_shell",
+ content: "ok",
+ isError: false,
+ stepId: `${T}#0` as StepId,
+ },
+ turnId: T,
+ },
+];
+
+describe("selectChunks — de-dup of a late-join replay against committed history", () => {
+ it("hides replayed provisional chunks already persisted for the in-flight turn", () => {
+ // Fresh load mid-turn: history has step 0; the replay re-delivers step 0
+ // and then step 1 (still streaming).
+ let s = applyHistory(initialState(), committedStep0);
+ s = fold(s, [
+ userMsg,
+ turnStart,
+ think("pl"),
+ think("an"),
+ call("a"),
+ result("a"),
+ text("now"),
+ ]);
+
+ const rendered = selectChunks(s);
+ expect(rendered.map((r) => [r.seq, r.chunk.type])).toEqual([
+ [1, "text"],
+ [2, "thinking"],
+ [3, "tool-call"],
+ [4, "tool-result"],
+ [null, "text"],
+ ]);
+ expect(rendered[4]?.streaming).toBe(true);
+ });
+
+ it("keeps provisional chunks beyond the committed run", () => {
+ let s = applyHistory(initialState(), committedStep0);
+ s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("a"), call("b", 1)]);
+ const rendered = selectChunks(s);
+ expect(rendered).toHaveLength(5);
+ expect(rendered[4]?.chunk).toMatchObject({ type: "tool-call", toolCallId: "b" });
+ expect(rendered[4]?.provisional).toBe(true);
+ });
+
+ it("hides the accumulating chunk once its content is committed", () => {
+ const committed: StoredChunk[] = [
+ ...committedStep0,
+ { seq: 5, role: "assistant", chunk: { type: "text", text: "final" }, turnId: T },
+ ];
+ let s = applyHistory(initialState(), committed);
+ s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("a"), text("final")]);
+ const rendered = selectChunks(s);
+ expect(rendered).toHaveLength(5);
+ expect(rendered.every((r) => !r.provisional)).toBe(true);
+ });
+
+ it("stops de-duping at the first mismatch (never hides divergent content)", () => {
+ let s = applyHistory(initialState(), committedStep0);
+ s = fold(s, [userMsg, turnStart, think("DIFFERENT"), call("a"), result("a")]);
+ const rendered = selectChunks(s);
+ // user echo matched (1 hidden); everything from the mismatch on is shown.
+ expect(rendered.filter((r) => r.provisional)).toHaveLength(3);
+ });
+
+ it("preserves an uncommitted accumulating tail even when its text repeats history", () => {
+ const committed: StoredChunk[] = [
+ ...committedStep0,
+ { seq: 5, role: "assistant", chunk: { type: "text", text: "final" }, turnId: T },
+ ];
+ let s = applyHistory(initialState(), committed);
+ s = markHistoryRecovering(s, T);
+ s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("a"), text("final")]);
+ const rendered = selectChunks(s);
+ expect(rendered).toHaveLength(6);
+ expect(rendered.slice(0, 5).every((r) => !r.provisional)).toBe(true);
+ expect(rendered[5]).toMatchObject({
+ provisional: true,
+ streaming: true,
+ chunk: { type: "text", text: "final" },
+ });
+ });
+
+ it("reconciles the same step identity even when provisional payload differs", () => {
+ let s = applyHistory(initialState(), committedStep0);
+ s = markHistoryRecovering(s, T);
+ s = fold(s, [userMsg, turnStart, think("DIFFERENT"), call("a"), result("a")]);
+ const rendered = selectChunks(s);
+ // Authoritative identity wins; payload equality cannot identify omitted/truncated content.
+ expect(rendered.filter((r) => r.provisional)).toHaveLength(0);
+ expect(rendered.map((r) => r.chunk)).toEqual(committedStep0.map((r) => r.chunk));
+ });
+
+ it("does nothing when the committed tail belongs to another turn", () => {
+ const other = committedStep0.map((c) => ({ ...c, turnId: "older" }));
+ let s = applyHistory(initialState(), other);
+ s = fold(s, [userMsg, turnStart, think("plan")]);
+ expect(selectChunks(s).filter((r) => r.provisional)).toHaveLength(2);
+ });
+
+ it("de-dups the replay when the committed turn's head was unloaded (windowed/trimmed)", () => {
+ // A long in-flight turn: the chat-limit window/trim unloaded the turn's
+ // head (seq 1–2), so the committed run starts mid-turn while the late-join
+ // replay re-delivers the WHOLE turn from its opening. Positional alignment
+ // from the turn start fails at index 0; without re-anchoring, the whole
+ // replay rendered as a duplicate whose tool cards collapsed onto the
+ // committed ones — leaving a tail of thinking blocks with no tool calls.
+ let s = trimTranscript(applyHistory(initialState(), committedStep0), 2);
+ expect(s.committed.map((c) => c.seq)).toEqual([3, 4]);
+ s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("a"), think("next")]);
+ const rendered = selectChunks(s);
+ expect(rendered.map((r) => [r.seq, r.chunk.type])).toEqual([
+ [3, "tool-call"],
+ [4, "tool-result"],
+ [null, "thinking"],
+ ]);
+ expect(rendered[2]).toMatchObject({ streaming: true, chunk: { text: "next" } });
+ });
+
+ it("de-dups when the provisional replay starts after the committed run's head", () => {
+ // Provisional trimmed past the committed head (e.g. committed restored via
+ // "Show earlier" after an all-provisional overflow): re-anchor on the
+ // provisional head instead.
+ let s = applyHistory(initialState(), committedStep0);
+ s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("a"), call("b", 1)]);
+ s = { ...s, provisional: s.provisional.slice(2) }; // drop user + thinking
+ const rendered = selectChunks(s);
+ expect(rendered.map((r) => [r.seq, r.chunk.type])).toEqual([
+ [1, "text"],
+ [2, "thinking"],
+ [3, "tool-call"],
+ [4, "tool-result"],
+ [null, "tool-call"],
+ ]);
+ });
+
+ it("re-anchors on a persisted tool call even if the replay diverges after it", () => {
+ // A toolCallId is a unique identity: everything the replay delivered before
+ // it was persisted earlier in the turn (append-only), so it is unloaded
+ // history — not divergent content. The divergence itself stays visible.
+ let s = applyHistory(initialState(), committedStep0.slice(2));
+ s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("OTHER")]);
+ const provisional = selectChunks(s).filter((r) => r.provisional);
+ expect(provisional.map((r) => r.chunk)).toMatchObject([
+ { type: "tool-result", toolCallId: "OTHER" },
+ ]);
+ });
+
+ it("does not re-anchor on an unverified text match (never hides divergent content)", () => {
+ const committed: StoredChunk[] = [
+ { seq: 7, role: "assistant", chunk: { type: "thinking", text: "plan" }, turnId: T },
+ { seq: 8, role: "assistant", chunk: { type: "text", text: "persisted" }, turnId: T },
+ ];
+ let s = applyHistory(initialState(), committed);
+ s = fold(s, [userMsg, turnStart, think("plan"), text("diverged"), call("z")]);
+ // "plan" matches but the next chunk diverges: no anchor → nothing hidden.
+ expect(selectChunks(s).filter((r) => r.provisional)).toHaveLength(4);
+ });
+
+ it("steps over persisted images the replay can never carry", () => {
+ // A tool attached an image mid-turn: history persists it as a user image
+ // chunk, but no live event carries it, so the replay lacks it.
+ const image: StoredChunk = {
+ seq: 5,
+ role: "user",
+ chunk: { type: "image", url: "/images/conv/x.png" },
+ turnId: T,
+ };
+ const thinking: StoredChunk = {
+ seq: 6,
+ role: "assistant",
+ chunk: { type: "thinking", text: "look" },
+ turnId: T,
+ };
+ let s = applyHistory(initialState(), [...committedStep0, image, thinking]);
+ s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("a"), think("look")]);
+ s = fold(s, [call("b", 1)]);
+ const provisional = selectChunks(s).filter((r) => r.provisional);
+ expect(provisional.map((r) => r.chunk)).toMatchObject([{ type: "tool-call", toolCallId: "b" }]);
+ });
+
+ it("matches one merged thinking block against its split committed blocks (Claude)", () => {
+ // Claude may return several thinking blocks in one step; history persists
+ // each as its own chunk, but reasoning-deltas carry no block boundary, so
+ // the replay delivers them merged into one thinking chunk.
+ const split: StoredChunk[] = [
+ committedStep0[0] as StoredChunk,
+ { seq: 2, role: "assistant", chunk: { type: "thinking", text: "first. " }, turnId: T },
+ { seq: 3, role: "assistant", chunk: { type: "thinking", text: "second." }, turnId: T },
+ { ...(committedStep0[2] as StoredChunk), seq: 4 },
+ { ...(committedStep0[3] as StoredChunk), seq: 5 },
+ ];
+ let s = applyHistory(initialState(), split);
+ s = fold(s, [userMsg, turnStart, think("first. "), think("second."), call("a"), result("a")]);
+ s = fold(s, [think("next")]);
+ const rendered = selectChunks(s);
+ expect(rendered.map((r) => [r.seq, r.chunk.type])).toEqual([
+ [1, "text"],
+ [2, "thinking"],
+ [3, "thinking"],
+ [4, "tool-call"],
+ [5, "tool-result"],
+ [null, "thinking"],
+ ]);
+ });
+
+ it("hides an accumulating merged block once its split blocks are committed", () => {
+ const split: StoredChunk[] = [
+ ...committedStep0,
+ { seq: 5, role: "assistant", chunk: { type: "thinking", text: "a" }, turnId: T },
+ { seq: 6, role: "assistant", chunk: { type: "thinking", text: "b" }, turnId: T },
+ ];
+ let s = applyHistory(initialState(), split);
+ s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("a"), think("ab")]);
+ expect(selectChunks(s).every((r) => !r.provisional)).toBe(true);
+ });
+
+ it("re-anchors on a tool chunk when the window cut between split blocks", () => {
+ // The committed head is the SECOND of two split blocks: it equals no
+ // provisional chunk, so alignment anchors on the run's first tool chunk.
+ const committed: StoredChunk[] = [
+ { seq: 3, role: "assistant", chunk: { type: "thinking", text: "second." }, turnId: T },
+ { ...(committedStep0[2] as StoredChunk), seq: 4 },
+ { ...(committedStep0[3] as StoredChunk), seq: 5 },
+ ];
+ let s = applyHistory(initialState(), committed);
+ s = fold(s, [userMsg, turnStart, think("first. second."), call("a"), result("a")]);
+ s = fold(s, [call("b", 1)]);
+ const provisional = selectChunks(s).filter((r) => r.provisional);
+ expect(provisional.map((r) => r.chunk)).toMatchObject([{ type: "tool-call", toolCallId: "b" }]);
+ });
+
+ it("aligns a boundary-carrying replay 1:1 against split committed blocks", () => {
+ const split: StoredChunk[] = [
+ committedStep0[0] as StoredChunk,
+ { seq: 2, role: "assistant", chunk: { type: "thinking", text: "first. " }, turnId: T },
+ { seq: 3, role: "assistant", chunk: { type: "thinking", text: "second." }, turnId: T },
+ { ...(committedStep0[2] as StoredChunk), seq: 4 },
+ { ...(committedStep0[3] as StoredChunk), seq: 5 },
+ ];
+ let s = applyHistory(initialState(), split);
+ s = fold(s, [
+ userMsg,
+ turnStart,
+ think("first. "),
+ thinkEnd,
+ think("second."),
+ thinkEnd,
+ call("a"),
+ result("a"),
+ text("next"),
+ ]);
+ const rendered = selectChunks(s);
+ expect(rendered.map((r) => [r.seq, r.chunk.type])).toEqual([
+ [1, "text"],
+ [2, "thinking"],
+ [3, "thinking"],
+ [4, "tool-call"],
+ [5, "tool-result"],
+ [null, "text"],
+ ]);
+ expect(rendered[5]?.streaming).toBe(true);
+ });
+
+ it("aligns a redacted empty block with a bare reasoning-end", () => {
+ const split: StoredChunk[] = [
+ committedStep0[0] as StoredChunk,
+ { seq: 2, role: "assistant", chunk: { type: "thinking", text: "first. " }, turnId: T },
+ { seq: 3, role: "assistant", chunk: { type: "thinking", text: "" }, turnId: T },
+ { seq: 4, role: "assistant", chunk: { type: "thinking", text: "second." }, turnId: T },
+ { ...(committedStep0[2] as StoredChunk), seq: 5 },
+ { ...(committedStep0[3] as StoredChunk), seq: 6 },
+ ];
+ let s = applyHistory(initialState(), split);
+ s = fold(s, [
+ userMsg,
+ turnStart,
+ think("first. "),
+ thinkEnd,
+ thinkEnd,
+ think("second."),
+ thinkEnd,
+ call("a"),
+ result("a"),
+ text("next"),
+ ]);
+ const rendered = selectChunks(s);
+ expect(rendered.map((r) => [r.seq, r.chunk.type])).toEqual([
+ [1, "text"],
+ [2, "thinking"],
+ [3, "thinking"],
+ [4, "thinking"],
+ [5, "tool-call"],
+ [6, "tool-result"],
+ [null, "text"],
+ ]);
+ });
+
+ it("is the identity when there is no committed history", () => {
+ const s = fold(initialState(), [userMsg, turnStart, text("hello")]);
+ expect(selectChunks(s)).toHaveLength(2);
+ });
+});
diff --git a/src/core/chunks/selectors.ts b/src/core/chunks/selectors.ts
index e46d5f5..8d32fcf 100644
--- a/src/core/chunks/selectors.ts
+++ b/src/core/chunks/selectors.ts
@@ -1,24 +1,278 @@
-import type { ChatMessage, Chunk, TurnProviderRetryEvent } from "@dispatch/wire";
-import type { RenderedChunk, TranscriptState } from "./types";
+import type { ChatMessage, Chunk, StoredChunk, TurnProviderRetryEvent } from "@dispatch/wire";
+import { historyBackedProvisional } from "./reconcile";
+import type { ProvisionalChunk, RenderedChunk, TranscriptState } from "./types";
+
+/**
+ * Whether a provisional chunk is the same content as a committed chunk at the
+ * same position of the same turn (identity for de-dup, not deep equality).
+ * Tool chunks match by `toolCallId`; text-bearing chunks by exact text.
+ */
+function sameChunk(p: ProvisionalChunk, c: StoredChunk): boolean {
+ if (p.role !== c.role || p.chunk.type !== c.chunk.type) return false;
+ const a = p.chunk;
+ const b = c.chunk;
+ switch (a.type) {
+ case "tool-call":
+ case "tool-result":
+ return a.toolCallId === (b as typeof a).toolCallId;
+ case "text":
+ case "thinking":
+ case "system":
+ return a.text === (b as typeof a).text;
+ case "error":
+ return a.message === (b as typeof a).message;
+ case "image":
+ return true;
+ }
+}
+
+/**
+ * A committed chunk with no live-event counterpart, so a late-join replay can
+ * never contain it: an image persisted into the turn's history (pasted-image
+ * openings, images a tool attaches mid-turn). Alignment steps over it.
+ */
+function replayless(c: StoredChunk): boolean {
+ return c.chunk.type === "image";
+}
+
+/** Where an alignment walk stopped, and whether it stopped on a mismatch. */
+interface Walk {
+ /** First provisional index NOT matched. */
+ readonly p: number;
+ /** First committed index NOT matched (replayless chunks stepped over). */
+ readonly c: number;
+ readonly matched: number;
+ /** Ended because a side ran out — no mismatch. */
+ readonly complete: boolean;
+}
+
+/**
+ * How many consecutive committed chunks starting at `c` one provisional text /
+ * thinking chunk stands for: their texts must concatenate EXACTLY to its text
+ * (≥ 2 chunks; 0 = no such run). A provider may return several blocks in one
+ * step (Claude emits multiple thinking blocks, each with its own signature)
+ * which history persists as separate chunks, while the delta stream carries
+ * no block boundary — so live (and in the replay) they arrive merged.
+ */
+function mergedRun(
+ state: TranscriptState,
+ text: string,
+ kind: string,
+ role: string,
+ c: number,
+): number {
+ let joined = "";
+ for (let k = c; k < state.committed.length; k++) {
+ const row = state.committed[k];
+ if (row === undefined || row.role !== role || row.chunk.type !== kind) return 0;
+ if (row.chunk.type !== "text" && row.chunk.type !== "thinking") return 0;
+ joined += row.chunk.text;
+ if (joined === text) return k - c + 1 >= 2 ? k - c + 1 : 0;
+ if (!text.startsWith(joined)) return 0;
+ }
+ return 0;
+}
+
+/**
+ * Walk provisional (from `p`) and committed (from `c`) of turn `turnId` in
+ * lockstep, stepping over committed chunks the replay cannot carry and
+ * matching one merged provisional block against its split committed run.
+ * Stops at the first mismatch or when either side runs out.
+ */
+function walk(state: TranscriptState, turnId: string, p: number, c: number): Walk {
+ let matched = 0;
+ for (;;) {
+ const prov = state.provisional[p];
+ let row = state.committed[c];
+ while (row !== undefined && replayless(row) && prov?.chunk.type !== "image") {
+ row = state.committed[++c];
+ }
+ if (prov === undefined || row === undefined) return { p, c, matched, complete: true };
+ if (prov.turnId !== turnId) return { p, c, matched, complete: false };
+ let span = sameChunk(prov, row) ? 1 : 0;
+ if (span === 0 && (prov.chunk.type === "text" || prov.chunk.type === "thinking")) {
+ span = mergedRun(state, prov.chunk.text, prov.chunk.type, prov.role, c);
+ }
+ if (span === 0) return { p, c, matched, complete: false };
+ p++;
+ c += span;
+ matched++;
+ }
+}
+
+/**
+ * Accept a re-anchored alignment. A tool chunk is a unique identity
+ * (`toolCallId`), so a match on one is trusted outright; a text-bearing match
+ * must be confirmed by the WHOLE overlap aligning (no mismatch), so a repeated
+ * text can never hide divergent content.
+ */
+function anchored(w: Walk, anchor: StoredChunk): boolean {
+ if (w.matched === 0) return false;
+ return w.complete || anchor.chunk.type === "tool-call" || anchor.chunk.type === "tool-result";
+}
+
+/**
+ * Align the provisional run with the committed run of turn T (which begins at
+ * committed index `start`). Returns how many leading provisional chunks are
+ * hidden as already committed, and the walk's end.
+ *
+ * Normally both start at the turn's opening. But the chat-limit window (fresh
+ * load `?limit=` / `windowTranscript`) and `trimTranscript` unload the OLDEST
+ * committed chunks, so in a long in-flight turn the committed run is a SUFFIX
+ * of the turn while the late-join replay re-delivers the whole turn from its
+ * `turn-start`: re-anchor on the committed head inside the replay, and hide
+ * the replayed chunks before it too (persisted history, unloaded — pageable
+ * via "Show earlier"). Conversely, provisional overflow trimming can leave the
+ * replay starting after the committed head: re-anchor on the replay's head
+ * inside the committed run. Unverifiable → the plain prefix walk (fail-safe:
+ * a misalignment shows a duplicate, never hides content).
+ */
+function align(state: TranscriptState, turnId: string, start: number): Walk {
+ const direct = walk(state, turnId, 0, start);
+ // Aligned from the turn's head (or fully consistent): plain prefix semantics.
+ if (direct.complete || direct.matched > 0) return direct;
+ let headAt = start;
+ while (
+ state.committed[headAt] !== undefined &&
+ replayless(state.committed[headAt] as StoredChunk)
+ )
+ headAt++;
+ // Anchor on the committed head — or, when the head is unmatchable (e.g. the
+ // window cut between two blocks a merged provisional chunk spans), on the
+ // run's first tool chunk, a unique identity.
+ const anchorsAt = [headAt];
+ for (let i = headAt + 1; i < state.committed.length; i++) {
+ const t = state.committed[i]?.chunk.type;
+ if (t === "tool-call" || t === "tool-result") {
+ anchorsAt.push(i);
+ break;
+ }
+ }
+ for (const at of anchorsAt) {
+ const head = state.committed[at];
+ if (head === undefined) continue;
+ for (let k = 1; k < state.provisional.length; k++) {
+ const p = state.provisional[k];
+ if (p?.turnId !== turnId || !sameChunk(p, head)) continue;
+ const w = walk(state, turnId, k, at);
+ if (anchored(w, head)) return w;
+ }
+ }
+ const p0 = state.provisional[0];
+ if (p0 !== undefined) {
+ for (let j = headAt + 1; j < state.committed.length; j++) {
+ const row = state.committed[j];
+ if (row === undefined || !sameChunk(p0, row)) continue;
+ const w = walk(state, turnId, 0, j);
+ if (anchored(w, row)) return w;
+ }
+ }
+ return direct;
+}
+
+/**
+ * How many LEADING provisional chunks are already backed by committed chunks.
+ *
+ * The backend persists each step of an in-flight turn as it completes (CR-6),
+ * so a client that opens a conversation mid-turn gets those steps from history
+ * AND again from the late-join event replay (as provisional chunks). Both
+ * carry the turn's chunks in the same order, so the provisional run of turn T
+ * is aligned chunk-by-chunk with the committed run of turn T (see `align`).
+ * Skip the aligned prefix; stop at the first mismatch.
+ *
+ * A failed alignment renders the ENTIRE replay as a duplicate — and because
+ * tool cards de-dup by `toolCallId` onto the first (committed) occurrence,
+ * that duplicate tail shows only thinking/text blocks ("the tool calls
+ * disappear"). Hence the re-anchoring and the replayless-image stepping.
+ *
+ * Also reports whether the ACCUMULATING chunk is already committed (the step
+ * completed + history synced, but the next live event hasn't flushed it yet).
+ */
+function committedOverlap(state: TranscriptState): { skip: number; hideAccumulating: boolean } {
+ const first = state.provisional[0];
+ const turnId = first?.turnId ?? state.accumulating?.turnId;
+ if (turnId === undefined || state.historyRecoveryTurnIds?.includes(turnId))
+ return { skip: 0, hideAccumulating: false };
+ // The committed run for this turn (contiguous at the tail of `committed`).
+ let start = state.committed.length;
+ while (start > 0 && state.committed[start - 1]?.turnId === turnId) start--;
+ if (start === state.committed.length) return { skip: 0, hideAccumulating: false };
+
+ const end = align(state, turnId, start);
+ const skip = end.p;
+
+ let hideAccumulating = false;
+ const acc = state.accumulating;
+ if (acc !== null && acc.turnId === turnId && skip === state.provisional.length) {
+ let next = end.c;
+ while (state.committed[next] !== undefined && replayless(state.committed[next] as StoredChunk))
+ next++;
+ const c = state.committed[next];
+ hideAccumulating =
+ (c !== undefined &&
+ c.role === "assistant" &&
+ c.chunk.type === acc.kind &&
+ c.chunk.text === acc.text) ||
+ mergedRun(state, acc.text, acc.kind, "assistant", next) > 0;
+ }
+ return { skip, hideAccumulating };
+}
+
+/**
+ * How many chunks `selectChunks` would render, without materializing them.
+ * Provisional chunks hidden as history-backed / replay duplicates of committed
+ * chunks are NOT counted — the chat limit bounds what is on screen, and
+ * counting invisible duplicates would unload real history to make room for them.
+ */
+export function selectRenderedCount(state: TranscriptState): number {
+ const backed = historyBackedProvisional(state);
+ const { skip, hideAccumulating } = committedOverlap(state);
+ let count = state.committed.length;
+ for (let i = skip; i < state.provisional.length; i++) {
+ const p = state.provisional[i];
+ if (p !== undefined && !backed.has(p)) count++;
+ }
+ if (state.accumulating !== null && !hideAccumulating) count++;
+ return count;
+}
/**
* Select all chunks for rendering: committed first (seq order),
- * then provisional (seq: null).
+ * then provisional (seq: null), excluding history-backed identities. Unknown
+ * accumulating content stays live, even if its text equals an earlier step.
*/
export function selectChunks(state: TranscriptState): readonly RenderedChunk[] {
const result: RenderedChunk[] = [];
+ const backed = historyBackedProvisional(state);
+ const { skip, hideAccumulating } = committedOverlap(state);
for (const c of state.committed) {
- result.push({ seq: c.seq, role: c.role, chunk: c.chunk, provisional: false });
+ result.push(
+ c.turnId !== undefined
+ ? { seq: c.seq, role: c.role, chunk: c.chunk, provisional: false, turnId: c.turnId }
+ : { seq: c.seq, role: c.role, chunk: c.chunk, provisional: false },
+ );
}
- for (const p of state.provisional) {
- result.push({ seq: null, role: p.role, chunk: p.chunk, provisional: true });
+ for (let i = skip; i < state.provisional.length; i++) {
+ const p = state.provisional[i];
+ if (p === undefined || backed.has(p)) continue;
+ result.push(
+ p.turnId !== undefined
+ ? { seq: null, role: p.role, chunk: p.chunk, provisional: true, turnId: p.turnId }
+ : { seq: null, role: p.role, chunk: p.chunk, provisional: true },
+ );
}
- if (state.accumulating !== null) {
+ const acc = state.accumulating;
+ if (acc !== null && !hideAccumulating) {
const chunk: Chunk =
- state.accumulating.kind === "text"
- ? { type: "text", text: state.accumulating.text }
- : { type: "thinking", text: state.accumulating.text };
- result.push({ seq: null, role: "assistant", chunk, provisional: true, streaming: true });
+ acc.kind === "text" ? { type: "text", text: acc.text } : { type: "thinking", text: acc.text };
+ result.push({
+ seq: null,
+ role: "assistant",
+ chunk,
+ provisional: true,
+ streaming: true,
+ ...(acc.turnId !== undefined ? { turnId: acc.turnId } : {}),
+ });
}
return result;
}
diff --git a/src/core/chunks/trim.test.ts b/src/core/chunks/trim.test.ts
index 7c4bbef..dcc1755 100644
--- a/src/core/chunks/trim.test.ts
+++ b/src/core/chunks/trim.test.ts
@@ -103,6 +103,19 @@ describe("trimTranscript", () => {
expect(next.accumulating).not.toBeNull();
});
+ it("does not count a late-join replay's duplicates of committed chunks", () => {
+ // An in-flight turn of 8 persisted chunks + its late-join replay (8 more
+ // provisional copies). Only 8 are rendered, so a limit of 10 keeps them all
+ // instead of unloading real history to make room for invisible duplicates.
+ const T = "turn";
+ const committed = chunks(1, 8).map((c) => ({ ...c, turnId: T }));
+ const state: TranscriptState = {
+ ...stateWith(committed),
+ provisional: committed.map((c) => ({ role: c.role, chunk: c.chunk, turnId: T })),
+ };
+ expect(trimTranscript(state, 10)).toBe(state);
+ });
+
it("drops oldest provisional when committed is exhausted", () => {
const base = stateWith(chunks(1, 2));
const provisional = Array.from({ length: 20 }, (_, i) => ({
@@ -146,7 +159,7 @@ describe("windowTranscript", () => {
it("is the identity within the window", () => {
const state = stateWith(chunks(1, 50));
expect(windowTranscript(state, 75)).toBe(state);
- expect(selectHasEarlier(state)).toBe(false);
+ expect(selectHasEarlier(state, 1)).toBe(false);
});
});
@@ -178,21 +191,21 @@ describe("restoreEarlier", () => {
expect(selectHasEarlier(restored)).toBe(true);
});
- it("restoring down to seq 1 reaches the contractual origin (hasEarlier clears)", () => {
+ it("restoring down to a known boundary clears hasEarlier", () => {
const windowed = windowTranscript(stateWith(chunks(1, 100)), 75); // hidden: 1..25
const restored = restoreEarlier(windowed, chunks(1, 100), 64);
expect(restored.committed).toHaveLength(100);
expect(restored.committed[0]?.seq).toBe(1);
expect(restored.hiddenBeforeSeq).toBe(1); // floor at the origin — inert
expect(restored.hiddenThinkingCount).toBe(0);
- expect(selectHasEarlier(restored)).toBe(false);
+ expect(selectHasEarlier(restored, 1)).toBe(false);
});
it("is the identity when nothing older is known locally (server may still hold more)", () => {
const windowed = windowTranscript(stateWith(chunks(50, 200)), 75);
const restored = restoreEarlier(windowed, [], 64);
expect(restored).toBe(windowed);
- // seqs are 1-based gap-free: window starts at 126 ⇒ older chunks DO exist.
+ // The server boundary is unknown, so another page may exist.
expect(selectHasEarlier(restored)).toBe(true);
});
@@ -203,7 +216,7 @@ describe("restoreEarlier", () => {
it("works on a server-windowed transcript (no local watermark)", () => {
// A cold-cache fresh load with `?limit=` commits a suffix (seq 809..1000)
- // with hiddenBeforeSeq still 0 — hasEarlier derives from seq > 1, and a
+ // with hiddenBeforeSeq still 0 — the server boundary is unknown, and a
// backfilled run merges below it.
const state = stateWith(chunks(809, 1000));
expect(state.hiddenBeforeSeq).toBe(0);
@@ -230,6 +243,15 @@ describe("restoreEarlier", () => {
const restored = restoreEarlier(trimmed, original, 1000);
expect(restored.committed).toEqual(original);
expect(restored.hiddenBeforeSeq).toBe(1);
- expect(selectHasEarlier(restored)).toBe(false);
+ expect(selectHasEarlier(restored, 1)).toBe(false);
});
});
+
+it("uses a non-1 history boundary rather than the seq origin", () => {
+ const state = stateWith(chunks(101, 110));
+ expect(selectHasEarlier(state, 101)).toBe(false);
+ expect(selectHasEarlier(state, null)).toBe(true);
+ const windowed = windowTranscript(state, 5);
+ expect(selectHasEarlier(windowed, 101)).toBe(true);
+ expect(selectHasEarlier(restoreEarlier(windowed, state.committed, 5), 101)).toBe(false);
+});
diff --git a/src/core/chunks/trim.ts b/src/core/chunks/trim.ts
index 9846357..6612bfe 100644
--- a/src/core/chunks/trim.ts
+++ b/src/core/chunks/trim.ts
@@ -14,6 +14,7 @@
// "Show earlier messages" affordance knows where to page back in from.
import type { StoredChunk } from "@dispatch/wire";
+import { selectRenderedCount } from "./selectors";
import type { TranscriptState } from "./types";
/** Default chat limit (max loaded chunks per conversation). */
@@ -46,9 +47,14 @@ export function initialWindowSize(limit: number): number {
return Math.max(1, Math.floor(limit * 0.75));
}
-/** Total loaded (rendered) chunk count: committed + provisional + accumulating. */
+/**
+ * Total loaded (rendered) chunk count: committed + provisional + accumulating,
+ * EXCLUDING provisional replay duplicates of committed chunks (a late-join
+ * replay re-delivers an in-flight turn's already-persisted steps). Counting
+ * those invisible duplicates unloaded real history to make room for them.
+ */
function totalCount(state: TranscriptState): number {
- return state.committed.length + state.provisional.length + (state.accumulating !== null ? 1 : 0);
+ return selectRenderedCount(state);
}
function countThinking(chunks: readonly StoredChunk[]): number {
@@ -150,8 +156,7 @@ function oldestLoadedSeq(state: TranscriptState): number {
* `?beforeSeq=` backfill; chunks at/inside the window are ignored). The newest
* `count` of them are merged back in front of `committed`, and the watermark
* follows the new window start so history merges still can't resurrect what
- * remains unloaded. Identity when the window already starts at seq 1 (the
- * contractual origin) or nothing older is known locally.
+ * remains unloaded. Identity when nothing older is known locally.
*/
export function restoreEarlier(
state: TranscriptState,
@@ -159,7 +164,7 @@ export function restoreEarlier(
count: number,
): TranscriptState {
const oldest = oldestLoadedSeq(state);
- if (oldest <= 1) return state;
+ if (oldest === 0) return state;
const below = earlier.filter((c) => c.seq < oldest).sort((a, b) => a.seq - b.seq);
if (below.length === 0) return state;
const keep = below.slice(-Math.max(1, count));
@@ -172,14 +177,11 @@ export function restoreEarlier(
};
}
-/**
- * Whether earlier history exists below the loaded window — drives the
- * "Show earlier messages" affordance. Derived from the [email protected] CONTRACT
- * that per-conversation seqs are 1-based and gap-free: a loaded window that
- * starts above seq 1 means older chunks exist (locally cached or server-side),
- * whether the window came from a local trim or a server-windowed (`?limit=`)
- * fresh load.
- */
-export function selectHasEarlier(state: TranscriptState): boolean {
- return oldestLoadedSeq(state) > 1;
+/** Earlier availability uses the known server boundary, never a seq origin. */
+export function selectHasEarlier(
+ state: TranscriptState,
+ earliestSeq: number | null = null,
+): boolean {
+ const oldest = oldestLoadedSeq(state);
+ return oldest > 0 && (earliestSeq === null || oldest > earliestSeq);
}
diff --git a/src/core/chunks/types.ts b/src/core/chunks/types.ts
index 2ced736..2e2ee0c 100644
--- a/src/core/chunks/types.ts
+++ b/src/core/chunks/types.ts
@@ -2,18 +2,36 @@ import type { Chunk, Role, StoredChunk, TurnProviderRetryEvent, Usage } from "@d
/** A chunk being accumulated from streaming deltas (text or thinking). */
export interface AccumulatingChunk {
+ readonly stepId?: string;
readonly kind: "text" | "thinking";
readonly text: string;
+ /** The turn generating this chunk (from the live event's `turnId`). */
+ readonly turnId?: string;
}
/** A provisional chunk that has no authoritative seq yet. */
export interface ProvisionalChunk {
+ /** Local provenance learned from tool-call/step-complete, not content. */
+ readonly stepId?: string;
+ readonly opening?: boolean;
+ /** Steering is persisted immediately after this completed tool batch. */
+ readonly afterToolCallId?: string;
+ /** Local send identity only; turn-start tagging does NOT confirm this content. */
+ readonly localRequestId?: string;
readonly role: Role;
readonly chunk: Chunk;
+ /**
+ * The turn this chunk belongs to (from the live event's `turnId`). Absent
+ * only on the sender's optimistic user echo until its `turn-start` arrives
+ * (which tags it).
+ */
+ readonly turnId?: string;
}
/** The transcript reducer state. Holds committed history + live in-flight turn. */
export interface TranscriptState {
+ /** Turns with omitted delivery: never use payload equality as replay identity. */
+ readonly historyRecoveryTurnIds?: readonly string[];
readonly committed: readonly StoredChunk[];
readonly provisional: readonly ProvisionalChunk[];
readonly accumulating: AccumulatingChunk | null;
@@ -74,4 +92,11 @@ export interface RenderedChunk {
* dots on streaming thinking) and drop it the moment generation moves on.
*/
readonly streaming?: boolean;
+ /**
+ * The turn that produced this chunk — `StoredChunk.turnId` for committed
+ * chunks, the live event's `turnId` for provisional ones. The exact join key
+ * for placing per-turn metrics. Absent on legacy rows (persisted before the
+ * backend stamped it) and on the optimistic echo before `turn-start`.
+ */
+ readonly turnId?: string;
}
generated by cgit v1.2.3 (git 2.39.1) at 2026-10-25 17:31:33 +0000