feat(chunks): split live thinking at reasoning-end block boundaries

8 files changed, 223 insertions, 0 deletions

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":