diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/core/chunks/reducer.test.ts | 49 | ||||
| -rw-r--r-- | src/core/chunks/reducer.ts | 39 | ||||
| -rw-r--r-- | src/core/chunks/selectors.test.ts | 133 | ||||
| -rw-r--r-- | src/core/chunks/selectors.ts | 202 | ||||
| -rw-r--r-- | src/core/chunks/trim.test.ts | 13 | ||||
| -rw-r--r-- | src/core/chunks/trim.ts | 10 |
6 files changed, 420 insertions, 26 deletions
diff --git a/src/core/chunks/reducer.test.ts b/src/core/chunks/reducer.test.ts index 22b94d4..4996e88 100644 --- a/src/core/chunks/reducer.test.ts +++ b/src/core/chunks/reducer.test.ts @@ -1085,3 +1085,52 @@ describe("foldEvent — authoritative turn terminal", () => { 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 ee5858f..35bb1bb 100644 --- a/src/core/chunks/reducer.ts +++ b/src/core/chunks/reducer.ts @@ -77,6 +77,26 @@ function tagTrailingUserEcho( } /** + * Drop the previous copy of a turn that is being replayed from its start. + * Keeps the trailing user run — the opening this replay's `user-message` just + * appended (or the sender's echo that `turn-start` just tagged) — and every + * other turn's chunks. + */ +function dropReplayedTurn( + provisional: readonly ProvisionalChunk[], + turnId: string, +): readonly ProvisionalChunk[] { + let tail = provisional.length; + while (tail > 0 && provisional[tail - 1]?.role === "user") tail--; + // A stale steering bubble (replayed again as its own `steering` event) can + // end the old copy and so sit inside the trailing user run. + const stale = (p: ProvisionalChunk, i: number): boolean => + p.turnId === turnId && (i < tail || p.afterToolCallId !== undefined); + if (!provisional.some(stale)) return provisional; + return provisional.filter((p, i) => !stale(p, i)); +} + +/** * Content equality for two chunks of the SAME role (used to de-dup an * optimistic echo against the authoritative committed version). Compares the * discriminating payload only — `stepId` (generation provenance) is ignored @@ -288,17 +308,26 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState case "tool-output": return state; - case "turn-start": + case "turn-start": { + // A turn starts exactly once live, so a `turn-start` for a turn that + // already has provisional content is a late-join REPLAY restarting from + // the top (re-subscribe after a tab switch / WS reconnect): the server + // re-delivers the whole in-flight turn. Replace the stale copy instead of + // appending a second one (a duplicate copy rendered as a tail of thinking + // blocks with its tool cards collapsed onto the first copy's). + const stale = state.accumulating?.turnId === event.turnId; + const provisional = tagTrailingUserEcho( + flushAccumulating(state.provisional, stale ? null : state.accumulating), + event.turnId, + ); return { ...state, - provisional: tagTrailingUserEcho( - flushAccumulating(state.provisional, state.accumulating), - event.turnId, - ), + provisional: dropReplayedTurn(provisional, event.turnId), accumulating: null, currentTurnId: event.turnId, generating: true, }; + } case "user-message": { // The turn's USER prompt, surfaced on the event stream (backend CR-3) so a diff --git a/src/core/chunks/selectors.test.ts b/src/core/chunks/selectors.test.ts index 05dcc4e..3afc324 100644 --- a/src/core/chunks/selectors.test.ts +++ b/src/core/chunks/selectors.test.ts @@ -2,6 +2,7 @@ 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"; @@ -166,6 +167,138 @@ describe("selectChunks — de-dup of a late-join replay against committed histor expect(selectChunks(s).filter((r) => r.provisional)).toHaveLength(2); }); + it("de-dups the replay when the committed turn's head was unloaded (windowed/trimmed)", () => { + // A long in-flight turn: the chat-limit window/trim unloaded the turn's + // head (seq 1–2), so the committed run starts mid-turn while the late-join + // replay re-delivers the WHOLE turn from its opening. Positional alignment + // from the turn start fails at index 0; without re-anchoring, the whole + // replay rendered as a duplicate whose tool cards collapsed onto the + // committed ones — leaving a tail of thinking blocks with no tool calls. + let s = trimTranscript(applyHistory(initialState(), committedStep0), 2); + expect(s.committed.map((c) => c.seq)).toEqual([3, 4]); + s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("a"), think("next")]); + const rendered = selectChunks(s); + expect(rendered.map((r) => [r.seq, r.chunk.type])).toEqual([ + [3, "tool-call"], + [4, "tool-result"], + [null, "thinking"], + ]); + expect(rendered[2]).toMatchObject({ streaming: true, chunk: { text: "next" } }); + }); + + it("de-dups when the provisional replay starts after the committed run's head", () => { + // Provisional trimmed past the committed head (e.g. committed restored via + // "Show earlier" after an all-provisional overflow): re-anchor on the + // provisional head instead. + let s = applyHistory(initialState(), committedStep0); + s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("a"), call("b", 1)]); + s = { ...s, provisional: s.provisional.slice(2) }; // drop user + thinking + const rendered = selectChunks(s); + expect(rendered.map((r) => [r.seq, r.chunk.type])).toEqual([ + [1, "text"], + [2, "thinking"], + [3, "tool-call"], + [4, "tool-result"], + [null, "tool-call"], + ]); + }); + + it("re-anchors on a persisted tool call even if the replay diverges after it", () => { + // A toolCallId is a unique identity: everything the replay delivered before + // it was persisted earlier in the turn (append-only), so it is unloaded + // history — not divergent content. The divergence itself stays visible. + let s = applyHistory(initialState(), committedStep0.slice(2)); + s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("OTHER")]); + const provisional = selectChunks(s).filter((r) => r.provisional); + expect(provisional.map((r) => r.chunk)).toMatchObject([ + { type: "tool-result", toolCallId: "OTHER" }, + ]); + }); + + it("does not re-anchor on an unverified text match (never hides divergent content)", () => { + const committed: StoredChunk[] = [ + { seq: 7, role: "assistant", chunk: { type: "thinking", text: "plan" }, turnId: T }, + { seq: 8, role: "assistant", chunk: { type: "text", text: "persisted" }, turnId: T }, + ]; + let s = applyHistory(initialState(), committed); + s = fold(s, [userMsg, turnStart, think("plan"), text("diverged"), call("z")]); + // "plan" matches but the next chunk diverges: no anchor → nothing hidden. + expect(selectChunks(s).filter((r) => r.provisional)).toHaveLength(4); + }); + + it("steps over persisted images the replay can never carry", () => { + // A tool attached an image mid-turn: history persists it as a user image + // chunk, but no live event carries it, so the replay lacks it. + const image: StoredChunk = { + seq: 5, + role: "user", + chunk: { type: "image", url: "/images/conv/x.png" }, + turnId: T, + }; + const thinking: StoredChunk = { + seq: 6, + role: "assistant", + chunk: { type: "thinking", text: "look" }, + turnId: T, + }; + let s = applyHistory(initialState(), [...committedStep0, image, thinking]); + s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("a"), think("look")]); + s = fold(s, [call("b", 1)]); + const provisional = selectChunks(s).filter((r) => r.provisional); + expect(provisional.map((r) => r.chunk)).toMatchObject([{ type: "tool-call", toolCallId: "b" }]); + }); + + it("matches one merged thinking block against its split committed blocks (Claude)", () => { + // Claude may return several thinking blocks in one step; history persists + // each as its own chunk, but reasoning-deltas carry no block boundary, so + // the replay delivers them merged into one thinking chunk. + const split: StoredChunk[] = [ + committedStep0[0] as StoredChunk, + { seq: 2, role: "assistant", chunk: { type: "thinking", text: "first. " }, turnId: T }, + { seq: 3, role: "assistant", chunk: { type: "thinking", text: "second." }, turnId: T }, + { ...(committedStep0[2] as StoredChunk), seq: 4 }, + { ...(committedStep0[3] as StoredChunk), seq: 5 }, + ]; + let s = applyHistory(initialState(), split); + s = fold(s, [userMsg, turnStart, think("first. "), think("second."), call("a"), result("a")]); + s = fold(s, [think("next")]); + const rendered = selectChunks(s); + expect(rendered.map((r) => [r.seq, r.chunk.type])).toEqual([ + [1, "text"], + [2, "thinking"], + [3, "thinking"], + [4, "tool-call"], + [5, "tool-result"], + [null, "thinking"], + ]); + }); + + it("hides an accumulating merged block once its split blocks are committed", () => { + const split: StoredChunk[] = [ + ...committedStep0, + { seq: 5, role: "assistant", chunk: { type: "thinking", text: "a" }, turnId: T }, + { seq: 6, role: "assistant", chunk: { type: "thinking", text: "b" }, turnId: T }, + ]; + let s = applyHistory(initialState(), split); + s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("a"), think("ab")]); + expect(selectChunks(s).every((r) => !r.provisional)).toBe(true); + }); + + it("re-anchors on a tool chunk when the window cut between split blocks", () => { + // The committed head is the SECOND of two split blocks: it equals no + // provisional chunk, so alignment anchors on the run's first tool chunk. + const committed: StoredChunk[] = [ + { seq: 3, role: "assistant", chunk: { type: "thinking", text: "second." }, turnId: T }, + { ...(committedStep0[2] as StoredChunk), seq: 4 }, + { ...(committedStep0[3] as StoredChunk), seq: 5 }, + ]; + let s = applyHistory(initialState(), committed); + s = fold(s, [userMsg, turnStart, think("first. second."), call("a"), result("a")]); + s = fold(s, [call("b", 1)]); + const provisional = selectChunks(s).filter((r) => r.provisional); + expect(provisional.map((r) => r.chunk)).toMatchObject([{ type: "tool-call", toolCallId: "b" }]); + }); + it("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 b507cf7..8d32fcf 100644 --- a/src/core/chunks/selectors.ts +++ b/src/core/chunks/selectors.ts @@ -27,15 +27,163 @@ function sameChunk(p: ProvisionalChunk, c: StoredChunk): boolean { } /** + * A committed chunk with no live-event counterpart, so a late-join replay can + * never contain it: an image persisted into the turn's history (pasted-image + * openings, images a tool attaches mid-turn). Alignment steps over it. + */ +function replayless(c: StoredChunk): boolean { + return c.chunk.type === "image"; +} + +/** Where an alignment walk stopped, and whether it stopped on a mismatch. */ +interface Walk { + /** First provisional index NOT matched. */ + readonly p: number; + /** First committed index NOT matched (replayless chunks stepped over). */ + readonly c: number; + readonly matched: number; + /** Ended because a side ran out — no mismatch. */ + readonly complete: boolean; +} + +/** + * How many consecutive committed chunks starting at `c` one provisional text / + * thinking chunk stands for: their texts must concatenate EXACTLY to its text + * (≥ 2 chunks; 0 = no such run). A provider may return several blocks in one + * step (Claude emits multiple thinking blocks, each with its own signature) + * which history persists as separate chunks, while the delta stream carries + * no block boundary — so live (and in the replay) they arrive merged. + */ +function mergedRun( + state: TranscriptState, + text: string, + kind: string, + role: string, + c: number, +): number { + let joined = ""; + for (let k = c; k < state.committed.length; k++) { + const row = state.committed[k]; + if (row === undefined || row.role !== role || row.chunk.type !== kind) return 0; + if (row.chunk.type !== "text" && row.chunk.type !== "thinking") return 0; + joined += row.chunk.text; + if (joined === text) return k - c + 1 >= 2 ? k - c + 1 : 0; + if (!text.startsWith(joined)) return 0; + } + return 0; +} + +/** + * Walk provisional (from `p`) and committed (from `c`) of turn `turnId` in + * lockstep, stepping over committed chunks the replay cannot carry and + * matching one merged provisional block against its split committed run. + * Stops at the first mismatch or when either side runs out. + */ +function walk(state: TranscriptState, turnId: string, p: number, c: number): Walk { + let matched = 0; + for (;;) { + const prov = state.provisional[p]; + let row = state.committed[c]; + while (row !== undefined && replayless(row) && prov?.chunk.type !== "image") { + row = state.committed[++c]; + } + if (prov === undefined || row === undefined) return { p, c, matched, complete: true }; + if (prov.turnId !== turnId) return { p, c, matched, complete: false }; + let span = sameChunk(prov, row) ? 1 : 0; + if (span === 0 && (prov.chunk.type === "text" || prov.chunk.type === "thinking")) { + span = mergedRun(state, prov.chunk.text, prov.chunk.type, prov.role, c); + } + if (span === 0) return { p, c, matched, complete: false }; + p++; + c += span; + matched++; + } +} + +/** + * Accept a re-anchored alignment. A tool chunk is a unique identity + * (`toolCallId`), so a match on one is trusted outright; a text-bearing match + * must be confirmed by the WHOLE overlap aligning (no mismatch), so a repeated + * text can never hide divergent content. + */ +function anchored(w: Walk, anchor: StoredChunk): boolean { + if (w.matched === 0) return false; + return w.complete || anchor.chunk.type === "tool-call" || anchor.chunk.type === "tool-result"; +} + +/** + * Align the provisional run with the committed run of turn T (which begins at + * committed index `start`). Returns how many leading provisional chunks are + * hidden as already committed, and the walk's end. + * + * Normally both start at the turn's opening. But the chat-limit window (fresh + * load `?limit=` / `windowTranscript`) and `trimTranscript` unload the OLDEST + * committed chunks, so in a long in-flight turn the committed run is a SUFFIX + * of the turn while the late-join replay re-delivers the whole turn from its + * `turn-start`: re-anchor on the committed head inside the replay, and hide + * the replayed chunks before it too (persisted history, unloaded — pageable + * via "Show earlier"). Conversely, provisional overflow trimming can leave the + * replay starting after the committed head: re-anchor on the replay's head + * inside the committed run. Unverifiable → the plain prefix walk (fail-safe: + * a misalignment shows a duplicate, never hides content). + */ +function align(state: TranscriptState, turnId: string, start: number): Walk { + const direct = walk(state, turnId, 0, start); + // Aligned from the turn's head (or fully consistent): plain prefix semantics. + if (direct.complete || direct.matched > 0) return direct; + let headAt = start; + while ( + state.committed[headAt] !== undefined && + replayless(state.committed[headAt] as StoredChunk) + ) + headAt++; + // Anchor on the committed head — or, when the head is unmatchable (e.g. the + // window cut between two blocks a merged provisional chunk spans), on the + // run's first tool chunk, a unique identity. + const anchorsAt = [headAt]; + for (let i = headAt + 1; i < state.committed.length; i++) { + const t = state.committed[i]?.chunk.type; + if (t === "tool-call" || t === "tool-result") { + anchorsAt.push(i); + break; + } + } + for (const at of anchorsAt) { + const head = state.committed[at]; + if (head === undefined) continue; + for (let k = 1; k < state.provisional.length; k++) { + const p = state.provisional[k]; + if (p?.turnId !== turnId || !sameChunk(p, head)) continue; + const w = walk(state, turnId, k, at); + if (anchored(w, head)) return w; + } + } + const p0 = state.provisional[0]; + if (p0 !== undefined) { + for (let j = headAt + 1; j < state.committed.length; j++) { + const row = state.committed[j]; + if (row === undefined || !sameChunk(p0, row)) continue; + const w = walk(state, turnId, 0, j); + if (anchored(w, row)) return w; + } + } + return direct; +} + +/** * How many LEADING provisional chunks are already backed by committed chunks. * * The backend persists each step of an in-flight turn as it completes (CR-6), * so a client that opens a conversation mid-turn gets those steps from history * AND again from the late-join event replay (as provisional chunks). Both * carry the turn's chunks in the same order, so the provisional run of turn T - * is aligned position-by-position with the committed run of turn T. Skip the - * aligned prefix; stop at the first mismatch (fail-safe: a misalignment shows - * a duplicate, never hides content). + * is aligned chunk-by-chunk with the committed run of turn T (see `align`). + * Skip the aligned prefix; stop at the first mismatch. + * + * A failed alignment renders the ENTIRE replay as a duplicate — and because + * tool cards de-dup by `toolCallId` onto the first (committed) occurrence, + * that duplicate tail shows only thinking/text blocks ("the tool calls + * disappear"). Hence the re-anchoring and the replayless-image stepping. * * Also reports whether the ACCUMULATING chunk is already committed (the step * completed + history synced, but the next live event hasn't flushed it yet). @@ -48,31 +196,47 @@ function committedOverlap(state: TranscriptState): { skip: number; hideAccumulat // The committed run for this turn (contiguous at the tail of `committed`). let start = state.committed.length; while (start > 0 && state.committed[start - 1]?.turnId === turnId) start--; - const run = state.committed.length - start; - if (run === 0) return { skip: 0, hideAccumulating: false }; - - let skip = 0; - while (skip < state.provisional.length && skip < run) { - const p = state.provisional[skip]; - const c = state.committed[start + skip]; - if (p === undefined || c === undefined || p.turnId !== turnId || !sameChunk(p, c)) break; - skip++; - } + if (start === state.committed.length) return { skip: 0, hideAccumulating: false }; + + const end = align(state, turnId, start); + const skip = end.p; let hideAccumulating = false; const acc = state.accumulating; - if (acc !== null && acc.turnId === turnId && skip === state.provisional.length && skip < run) { - const c = state.committed[start + skip]; + if (acc !== null && acc.turnId === turnId && skip === state.provisional.length) { + let next = end.c; + while (state.committed[next] !== undefined && replayless(state.committed[next] as StoredChunk)) + next++; + const c = state.committed[next]; hideAccumulating = - c !== undefined && - c.role === "assistant" && - c.chunk.type === acc.kind && - c.chunk.text === acc.text; + (c !== undefined && + c.role === "assistant" && + c.chunk.type === acc.kind && + c.chunk.text === acc.text) || + mergedRun(state, acc.text, acc.kind, "assistant", next) > 0; } return { skip, hideAccumulating }; } /** + * How many chunks `selectChunks` would render, without materializing them. + * Provisional chunks hidden as history-backed / replay duplicates of committed + * chunks are NOT counted — the chat limit bounds what is on screen, and + * counting invisible duplicates would unload real history to make room for them. + */ +export function selectRenderedCount(state: TranscriptState): number { + const backed = historyBackedProvisional(state); + const { skip, hideAccumulating } = committedOverlap(state); + let count = state.committed.length; + for (let i = skip; i < state.provisional.length; i++) { + const p = state.provisional[i]; + if (p !== undefined && !backed.has(p)) count++; + } + if (state.accumulating !== null && !hideAccumulating) count++; + return count; +} + +/** * Select all chunks for rendering: committed first (seq order), * then provisional (seq: null), excluding history-backed identities. Unknown * accumulating content stays live, even if its text equals an earlier step. diff --git a/src/core/chunks/trim.test.ts b/src/core/chunks/trim.test.ts index 24c75fd..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) => ({ diff --git a/src/core/chunks/trim.ts b/src/core/chunks/trim.ts index 259c101..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 { |
