From 53686ae6446c5c59928244f65c08150fb564b375 Mon Sep 17 00:00:00 2001 From: Adam Malczewski Date: Thu, 22 Oct 2026 15:48:50 +0900 Subject: fix(chunks): keep tool calls visible when opening an in-flight chat MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Opening (or switching back to) a conversation with a running turn showed a tail of "Thoughts" blocks with no tool cards. The late-join replay re-delivers the whole in-flight turn as provisional chunks; committedOverlap hides the ones already in committed history by aligning both runs chunk-by-chunk. When that alignment failed, the whole replay rendered as a duplicate, and because tool cards de-dup by toolCallId onto the first (committed) occurrence, the duplicate tail showed only thinking/text. The alignment broke in four ways, all reproduced against a live server: - Claude returns several thinking blocks per step (one signature each). History persists them separately, but reasoning-deltas carry no block boundary, so live/replay merges them. Now one provisional text/thinking chunk matches a run of committed chunks whose texts concatenate to it. - Images persisted mid-turn (tool attachments) have no live event. The walk now steps over them. - The chat-limit window/trim unloads the committed turn's head, so the committed run is a suffix while the replay starts at turn-start. Alignment now re-anchors (committed head, else the run's first tool chunk; text anchors must verify the whole overlap) and hides replayed chunks whose committed twins were unloaded. - Re-subscribing (tab switch / reconnect) replays the turn on top of the existing provisional copy. A turn-start for a turn that already has provisional content now replaces the stale copy. trimTranscript also counted those invisible duplicates toward the chat limit, unloading real history (incl. the turn head) to make room for them; it now counts rendered chunks (selectRenderedCount). svelte-check 0 errors · biome clean · 2488 tests pass · build OK --- src/core/chunks/reducer.ts | 39 ++++++++++++++++++++++++++++++++++----- 1 file changed, 34 insertions(+), 5 deletions(-) (limited to 'src/core/chunks/reducer.ts') 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 @@ -76,6 +76,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 @@ -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 -- cgit v1.2.3 From 75680eac7071c6968eaa5470686ad102a82103db Mon Sep 17 00:00:00 2001 From: Adam Malczewski Date: Thu, 22 Oct 2026 17:51:15 +0900 Subject: feat(chunks): split live thinking at reasoning-end block boundaries --- .dispatch/wire.reference.md | 26 +++++++++++++ backend-handoff.md | 12 ++++++ src/core/chunks/reducer.test.ts | 73 ++++++++++++++++++++++++++++++++++++ src/core/chunks/reducer.ts | 27 +++++++++++++ src/core/chunks/selectors.test.ts | 67 +++++++++++++++++++++++++++++++++ src/core/metrics/cache-timer.test.ts | 15 ++++++++ src/core/metrics/cache-timer.ts | 1 + src/core/wire/conformance.ts | 2 + 8 files changed, 223 insertions(+) (limited to 'src/core/chunks/reducer.ts') diff --git a/.dispatch/wire.reference.md b/.dispatch/wire.reference.md index 3743bd5..777a2ba 100644 --- a/.dispatch/wire.reference.md +++ b/.dispatch/wire.reference.md @@ -1,5 +1,15 @@ # `@dispatch/wire` — in-repo reference (read THIS, not node_modules) +> **2026-10-22 delta (reasoning block boundaries — ADDITIVE, no version bump):** `AgentEvent` +> gains `TurnReasoningEndEvent` `{ type: "reasoning-end", conversationId, turnId }`. The kernel emits +> one per provider reasoning block (Anthropic signs each thinking block separately), carrying NO +> opaque payload (signature/redacted data never leave the server). A live consumer splits thinking +> exactly as HTTP history persists it: the open thinking chunk closes; with none open (a redacted +> empty block) an empty thinking chunk is implied. Providers without block structure never emit it. +> The FE reducer mirrors the kernel's chunk builder; the old merged-run alignment stays as a fallback +> for older servers/replays. See `backend-handoff.md` and the backend's +> `notes/reasoning-end-live-event-plan.md`. +> > **2026-10-20 review 3.5:** history replacements preserve monotonic seqs (retained history need not start at 1). HTTP history and WS compaction carry required `historyEpoch`; unknown cached epochs must be reloaded. Consumed by review 3.6. > > MIRRORS the backend's `@dispatch/wire` package source so headless FE agents can read the wire @@ -414,6 +424,7 @@ export type AgentEvent = | TurnInputEvent | TurnTextDeltaEvent | TurnReasoningDeltaEvent + | TurnReasoningEndEvent | TurnToolCallEvent | TurnToolResultEvent | TurnToolOutputEvent @@ -474,6 +485,21 @@ export interface TurnReasoningDeltaEvent { readonly delta: string; } +/** + * End of one reasoning / thinking block. Emitted once per block by providers with + * block-structured reasoning (e.g. Anthropic, whose consecutive thinking blocks — each + * separately signed — persist as separate thinking chunks). Lets a live consumer split + * thinking exactly as history does: the open thinking chunk closes; if none is open (a + * redacted/empty block with no deltas) an EMPTY thinking chunk is implied. Carries no + * content: the opaque replay payloads (signature / redacted data) never leave the + * server. Providers without block structure never emit it (deltas merge as before). + */ +export interface TurnReasoningEndEvent { + readonly type: "reasoning-end"; + readonly conversationId: string; + readonly turnId: string; +} + /** The model has requested a tool to be run. */ export interface TurnToolCallEvent { readonly type: "tool-call"; diff --git a/backend-handoff.md b/backend-handoff.md index b4d53bc..0878b93 100644 --- a/backend-handoff.md +++ b/backend-handoff.md @@ -18,6 +18,18 @@ > ?archived=`) was made in the same branch and the FE consumes it via the `file:` deps (dists > rebuilt). Recorded below. +## Milestone: reasoning block boundaries on the live stream (2026-10-22) + +FE slice consuming a new ADDITIVE backend wire event. The kernel now emits one payload-free +`reasoning-end` `AgentEvent` per provider reasoning block (no version bump; opaque +signature/redacted data never leave the server). The pure chunk reducer mirrors the kernel's +builder — an open thinking chunk is flushed as its own provisional chunk; with none open it +appends an empty thinking chunk — so live thinking splits identically to HTTP history and the +replay/history de-dup lines up 1:1. The old merged-run alignment (`mergedRun`) stays as a +fallback for older servers/replays. The cache timer treats `reasoning-end` as resumed output; +`assertAgentEventExhaustive` covers it. Mirrored in `.dispatch/wire.reference.md` +(`TurnReasoningEndEvent`). No backend asks. + ## Milestone: Delivery warning display delay + Flush label (2026-10-21) Frontend-only: each keyed pending-input warning waits 1000 ms on mount (including diff --git a/src/core/chunks/reducer.test.ts b/src/core/chunks/reducer.test.ts index 4996e88..10d662e 100644 --- a/src/core/chunks/reducer.test.ts +++ b/src/core/chunks/reducer.test.ts @@ -7,6 +7,7 @@ import type { TurnInputEvent, TurnProviderRetryEvent, TurnReasoningDeltaEvent, + TurnReasoningEndEvent, TurnSealedEvent, TurnStartEvent, TurnSteeringEvent, @@ -45,6 +46,12 @@ const reasoningDelta = (turnId: string, delta: string): TurnReasoningDeltaEvent delta, }); +const reasoningEnd = (turnId: string): TurnReasoningEndEvent => ({ + type: "reasoning-end", + conversationId: "c1", + turnId, +}); + const toolCall = ( turnId: string, toolCallId: string, @@ -278,6 +285,72 @@ describe("foldEvent — reasoning-delta", () => { }); }); +describe("foldEvent — reasoning-end (thinking block boundary)", () => { + it("closes each block so consecutive deltas become separate chunks", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, reasoningDelta("t1", "a")); + s = foldEvent(s, reasoningEnd("t1")); + s = foldEvent(s, reasoningDelta("t1", "b")); + s = foldEvent(s, reasoningEnd("t1")); + expect(s.provisional.map((p) => p.chunk)).toEqual([ + { type: "thinking", text: "a" }, + { type: "thinking", text: "b" }, + ]); + expect(s.accumulating).toBeNull(); + }); + + it("still merges legacy deltas with no boundary (no reasoning-end)", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, reasoningDelta("t1", "a")); + s = foldEvent(s, reasoningDelta("t1", "b")); + expect(s.accumulating).toEqual({ kind: "thinking", text: "ab", turnId: "t1" }); + }); + + it("end with nothing open appends an empty thinking chunk for the turn", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, reasoningEnd("t1")); + expect(s.provisional).toHaveLength(1); + expect(s.provisional[0]?.chunk).toEqual({ type: "thinking", text: "" }); + expect(s.provisional[0]?.turnId).toBe("t1"); + expect(s.accumulating).toBeNull(); + }); + + it("text-delta then end flushes the text, then appends an empty thinking chunk", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "x")); + s = foldEvent(s, reasoningEnd("t1")); + expect(s.provisional.map((p) => p.chunk)).toEqual([ + { type: "text", text: "x" }, + { type: "thinking", text: "" }, + ]); + }); + + it("end for another turn flushes the open thinking and appends an empty one", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, reasoningDelta("t1", "open")); + s = foldEvent(s, reasoningEnd("t2")); + expect(s.provisional.map((p) => [p.turnId, p.chunk])).toEqual([ + ["t1", { type: "thinking", text: "open" }], + ["t2", { type: "thinking", text: "" }], + ]); + expect(s.accumulating).toBeNull(); + }); + + it("clears a pending provider-retry banner", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, providerRetry("t1", 0, 5000)); + expect(selectProviderRetry(s)).not.toBeNull(); + s = foldEvent(s, reasoningEnd("t1")); + expect(selectProviderRetry(s)).toBeNull(); + }); +}); + describe("foldEvent — tool-call then tool-result", () => { it("tool-call then tool-result render in order", () => { let s = initialState(); diff --git a/src/core/chunks/reducer.ts b/src/core/chunks/reducer.ts index 35bb1bb..71df4dd 100644 --- a/src/core/chunks/reducer.ts +++ b/src/core/chunks/reducer.ts @@ -284,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 @@ -415,6 +417,30 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState }; } + case "reasoning-end": { + const acc = state.accumulating; + // Close the open thinking block of this turn: flush it as its own provisional chunk. + if (acc !== null && acc.kind === "thinking" && acc.turnId === event.turnId) { + return { + ...state, + provisional: flushAccumulating(state.provisional, acc), + accumulating: null, + generating: true, + }; + } + // No open thinking (a redacted/empty block with no deltas): history persists an + // EMPTY thinking chunk for it, so mirror that 1:1. + return { + ...state, + provisional: [ + ...flushAccumulating(state.provisional, acc), + { role: "assistant", chunk: { type: "thinking", text: "" }, turnId: event.turnId }, + ], + accumulating: null, + generating: true, + }; + } + case "tool-call": { const provisional = flushAccumulating(state.provisional, state.accumulating); const chunk: Chunk = { @@ -568,6 +594,7 @@ const RETRY_CLEARING_EVENTS: ReadonlySet = new Set([ "turn-start", "text-delta", "reasoning-delta", + "reasoning-end", "tool-call", "tool-result", "error", diff --git a/src/core/chunks/selectors.test.ts b/src/core/chunks/selectors.test.ts index 3afc324..90bb5cc 100644 --- a/src/core/chunks/selectors.test.ts +++ b/src/core/chunks/selectors.test.ts @@ -20,6 +20,7 @@ const think = (delta: string): AgentEvent => ({ turnId: T, delta, }); +const thinkEnd: AgentEvent = { type: "reasoning-end", conversationId: C, turnId: T }; const text = (delta: string): AgentEvent => ({ type: "text-delta", conversationId: C, @@ -299,6 +300,72 @@ describe("selectChunks — de-dup of a late-join replay against committed histor 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/metrics/cache-timer.test.ts b/src/core/metrics/cache-timer.test.ts index e55eff4..d450135 100644 --- a/src/core/metrics/cache-timer.test.ts +++ b/src/core/metrics/cache-timer.test.ts @@ -32,6 +32,11 @@ const ev = { turnId, delta: "x", }), + reasoningEnd: (turnId: string): AgentEvent => ({ + type: "reasoning-end", + conversationId: C, + turnId, + }), call: (turnId: string, stepId: string, toolCallId: string): AgentEvent => ({ type: "tool-call", conversationId: C, @@ -151,6 +156,16 @@ describe("foldCacheTimerEvent — step gaps", () => { expect(s.stepStamps.get("t1#0")).toBe(8_000); }); + it("treats a reasoning-end boundary as resumed output too", () => { + const s = run([ + [0, ev.call("t1", "t1#0", "a")], + [1_000, ev.stepComplete("t1", "t1#0")], + [9_000, ev.reasoningEnd("t1")], + ]); + expect(s.running).toBeNull(); + expect(s.stepStamps.get("t1#0")).toBe(8_000); + }); + it("stops on a tool-call or usage from a DIFFERENT step, not its own", () => { const base = run([ [0, ev.call("t1", "t1#0", "a")], diff --git a/src/core/metrics/cache-timer.ts b/src/core/metrics/cache-timer.ts index 66d5fac..3d7270d 100644 --- a/src/core/metrics/cache-timer.ts +++ b/src/core/metrics/cache-timer.ts @@ -208,6 +208,7 @@ export function foldCacheTimerEvent( case "text-delta": case "reasoning-delta": + case "reasoning-end": case "provider-retry": // Output (or a retry) from the next request: the gap is over. return r?.kind === "step" ? stopStep(state, now) : state; diff --git a/src/core/wire/conformance.ts b/src/core/wire/conformance.ts index cd53b65..eff2330 100644 --- a/src/core/wire/conformance.ts +++ b/src/core/wire/conformance.ts @@ -18,6 +18,8 @@ export function assertAgentEventExhaustive(event: AgentEvent): string { return "text-delta"; case "reasoning-delta": return "reasoning-delta"; + case "reasoning-end": + return "reasoning-end"; case "tool-call": return "tool-call"; case "tool-result": -- cgit v1.2.3