From 53686ae6446c5c59928244f65c08150fb564b375 Mon Sep 17 00:00:00 2001
From: Adam Malczewski <github@tradam.dev>
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.test.ts   |  49 +++++++++
 src/core/chunks/reducer.ts        |  39 +++++++-
 src/core/chunks/selectors.test.ts | 133 +++++++++++++++++++++++++
 src/core/chunks/selectors.ts      | 202 ++++++++++++++++++++++++++++++++++----
 src/core/chunks/trim.test.ts      |  13 +++
 src/core/chunks/trim.ts           |  10 +-
 6 files changed, 420 insertions(+), 26 deletions(-)

(limited to 'src/core/chunks')

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
@@ -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
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
@@ -26,6 +26,150 @@ 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.
  *
@@ -33,9 +177,13 @@ function sameChunk(p: ProvisionalChunk, c: StoredChunk): boolean {
  * 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,30 +196,46 @@ 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
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 {
-- 
cgit v1.2.3


From 75680eac7071c6968eaa5470686ad102a82103db Mon Sep 17 00:00:00 2001
From: Adam Malczewski <github@tradam.dev>
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')

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<AgentEvent["type"]> = 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