summaryrefslogtreecommitdiffhomepage
path: root/src/core
diff options
context:
space:
mode:
Diffstat (limited to 'src/core')
-rw-r--r--src/core/chunks/selectors.test.ts145
-rw-r--r--src/core/chunks/selectors.ts83
2 files changed, 223 insertions, 5 deletions
diff --git a/src/core/chunks/selectors.test.ts b/src/core/chunks/selectors.test.ts
new file mode 100644
index 0000000..57c1415
--- /dev/null
+++ b/src/core/chunks/selectors.test.ts
@@ -0,0 +1,145 @@
+import type { AgentEvent, StepId, StoredChunk } from "@dispatch/wire";
+import { describe, expect, it } from "vitest";
+import { applyHistory, foldEvent, initialState } from "./reducer";
+import { selectChunks } from "./selectors";
+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 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("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("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 8b265a5..d32a628 100644
--- a/src/core/chunks/selectors.ts
+++ b/src/core/chunks/selectors.ts
@@ -1,12 +1,83 @@
-import type { ChatMessage, Chunk, TurnProviderRetryEvent } from "@dispatch/wire";
-import type { RenderedChunk, TranscriptState } from "./types";
+import type { ChatMessage, Chunk, StoredChunk, TurnProviderRetryEvent } from "@dispatch/wire";
+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;
+ }
+}
+
+/**
+ * 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).
+ *
+ * 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) 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++;
+ }
+
+ let hideAccumulating = false;
+ const acc = state.accumulating;
+ if (acc !== null && skip === state.provisional.length && skip < run) {
+ const c = state.committed[start + skip];
+ hideAccumulating =
+ c !== undefined &&
+ c.role === "assistant" &&
+ c.chunk.type === acc.kind &&
+ c.chunk.text === acc.text;
+ }
+ return { skip, hideAccumulating };
+}
/**
* Select all chunks for rendering: committed first (seq order),
- * then provisional (seq: null).
+ * then provisional (seq: null) — minus provisional chunks already backed by
+ * committed history (see `committedOverlap`).
*/
export function selectChunks(state: TranscriptState): readonly RenderedChunk[] {
const result: RenderedChunk[] = [];
+ const { skip, hideAccumulating } = committedOverlap(state);
for (const c of state.committed) {
result.push(
c.turnId !== undefined
@@ -14,7 +85,9 @@ export function selectChunks(state: TranscriptState): readonly RenderedChunk[] {
: { seq: c.seq, role: c.role, chunk: c.chunk, provisional: false },
);
}
- for (const p of state.provisional) {
+ for (let i = skip; i < state.provisional.length; i++) {
+ const p = state.provisional[i];
+ if (p === undefined) continue;
result.push(
p.turnId !== undefined
? { seq: null, role: p.role, chunk: p.chunk, provisional: true, turnId: p.turnId }
@@ -22,7 +95,7 @@ export function selectChunks(state: TranscriptState): readonly RenderedChunk[] {
);
}
const acc = state.accumulating;
- if (acc !== null) {
+ if (acc !== null && !hideAccumulating) {
const chunk: Chunk =
acc.kind === "text" ? { type: "text", text: acc.text } : { type: "thinking", text: acc.text };
result.push({
generated by cgit v1.2.3 (git 2.39.1) at 2026-10-26 17:50:58 +0000