Merge local dev into feature/close-tab-cascade-subagents

13 files changed, 1148 insertions, 67 deletions

diff --git a/src/core/chunks/groups.test.ts b/src/core/chunks/groups.test.ts
index cc2aa40..1196590 100644
--- a/src/core/chunks/groups.test.ts
+++ b/src/core/chunks/groups.test.ts
@@ -150,3 +150,25 @@ describe("groupRenderedChunks", () => {
     expect(groups.map((g) => g.kind)).toEqual(["single", "tool"]);
   });
 });
+
+it("keeps windowed committed orphan calls in Show earlier even while another payload recovers", () => {
+  const groups = groupRenderedChunks(
+    [
+      {
+        seq: 50,
+        role: "tool",
+        provisional: false,
+        chunk: {
+          type: "tool-result",
+          toolCallId: "older",
+          toolName: "run_shell",
+          content: "output",
+          isError: false,
+        },
+      },
+    ],
+    true,
+  );
+  expect(groups[0]).toMatchObject({ kind: "tool", entry: { call: null } });
+  if (groups[0]?.kind === "tool") expect(groups[0].entry.recovering).toBeUndefined();
+});
diff --git a/src/core/chunks/groups.ts b/src/core/chunks/groups.ts
index c8493cc..6e0f8ea 100644
--- a/src/core/chunks/groups.ts
+++ b/src/core/chunks/groups.ts
@@ -3,6 +3,7 @@ import type { RenderedChunk } from "./types";
 
 /** One tool call and its response. Null call = unloaded history; null result = pending. */
 export interface ToolEntry {
+  readonly recovering?: boolean;
   readonly toolCallId: string;
   readonly toolName: string;
   readonly call: ToolCallChunk | null;
@@ -26,7 +27,10 @@ export type RenderGroup =
  * the stable t<toolCallId> render keys are always unique, even on corrupt replay.
  * Pure: input → output; no DOM, effects, or Svelte.
  */
-export function groupRenderedChunks(rendered: readonly RenderedChunk[]): readonly RenderGroup[] {
+export function groupRenderedChunks(
+  rendered: readonly RenderedChunk[],
+  recovering = false,
+): readonly RenderGroup[] {
   const calls = new Map<string, RenderedChunk>();
   const results = new Map<string, RenderedChunk>();
   for (const rc of rendered) {
@@ -55,7 +59,13 @@ export function groupRenderedChunks(rendered: readonly RenderedChunk[]): readonl
     const turnId = callRow?.turnId ?? resultRow?.turnId;
     groups.push({
       kind: "tool",
-      entry: { toolCallId: c.toolCallId, toolName: call?.toolName ?? c.toolName, call, result },
+      entry: {
+        toolCallId: c.toolCallId,
+        toolName: call?.toolName ?? c.toolName,
+        call,
+        result,
+        ...(recovering && call === null && resultRow?.provisional ? { recovering: true } : {}),
+      },
       provisional: Boolean(callRow?.provisional || resultRow?.provisional),
       ...(stepId !== undefined ? { stepId } : {}),
       ...(turnId !== undefined ? { turnId } : {}),
diff --git a/src/core/chunks/index.ts b/src/core/chunks/index.ts
index 3822a79..03d687c 100644
--- a/src/core/chunks/index.ts
+++ b/src/core/chunks/index.ts
@@ -7,6 +7,8 @@ export {
   clearGenerating,
   foldEvent,
   initialState,
+  markHistoryRecovering,
+  removeLocalEcho,
 } from "./reducer";
 export type { ProviderRetryView } from "./retry-banner";
 export { formatRetryDelay, viewProviderRetry } from "./retry-banner";
diff --git a/src/core/chunks/local-echo.test.ts b/src/core/chunks/local-echo.test.ts
new file mode 100644
index 0000000..565017c
--- /dev/null
+++ b/src/core/chunks/local-echo.test.ts
@@ -0,0 +1,45 @@
+import { describe, expect, it } from "vitest";
+import { appendUserMessage, applyHistory, foldEvent, initialState, removeLocalEcho } from "./index";
+
+describe("request-local optimistic echo", () => {
+  it("retains identity through turn-start tags without conflating identical sends", () => {
+    let state = appendUserMessage(initialState(), "same", [{ url: "data:first" }], "r1");
+    state = appendUserMessage(state, "same", [{ url: "data:second" }], "r2");
+    state = foldEvent(state, { type: "turn-start", conversationId: "c1", turnId: "t1" });
+    state = removeLocalEcho(state, "r1");
+    expect(state.provisional.map((p) => p.localRequestId)).toEqual(["r2", "r2"]);
+    expect(state.provisional.map((p) => p.chunk)).toEqual([
+      { type: "text", text: "same" },
+      { type: "image", url: "data:second" },
+    ]);
+    expect(state.generating).toBe(true);
+    expect(removeLocalEcho(state, "r1")).toBe(state);
+  });
+
+  it("a live user-message confirms only its deduplicated text, not every identical local send", () => {
+    let state = appendUserMessage(initialState(), "same", [{ url: "data:first" }], "r1");
+    state = appendUserMessage(state, "same", [{ url: "data:second" }], "r2");
+    state = foldEvent(state, { type: "turn-start", conversationId: "c1", turnId: "t1" });
+    state = foldEvent(state, {
+      type: "user-message",
+      conversationId: "c1",
+      turnId: "t1",
+      text: "same",
+    });
+    state = removeLocalEcho(removeLocalEcho(state, "r1"), "r2");
+    expect(state.provisional).toEqual([
+      { role: "user", turnId: "t1", opening: true, chunk: { type: "text", text: "same" } },
+    ]);
+  });
+
+  it("does not remove committed rows even if their content is identical", () => {
+    let state = appendUserMessage(initialState(), "same", undefined, "r1");
+    state = applyHistory(state, [{ seq: 1, role: "user", chunk: { type: "text", text: "same" } }]);
+    const committed = state.committed;
+    state = removeLocalEcho(state, "r1");
+    expect(state.committed).toBe(committed);
+    expect(state.provisional).toEqual([]);
+    expect(removeLocalEcho(state, "r1")).toBe(state);
+    expect(removeLocalEcho(initialState(), "r1")).toEqual(initialState());
+  });
+});
diff --git a/src/core/chunks/reconcile.test.ts b/src/core/chunks/reconcile.test.ts
new file mode 100644
index 0000000..3d90a51
--- /dev/null
+++ b/src/core/chunks/reconcile.test.ts
@@ -0,0 +1,143 @@
+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 type { TranscriptState } from "./types";
+
+const stepId = "t1#1" as StepId;
+function fold(s: TranscriptState, e: AgentEvent) {
+  return foldEvent(s, e);
+}
+const call = (toolCallId = "a", turnId = "t1", step = stepId): AgentEvent => ({
+  type: "tool-call",
+  conversationId: "c1",
+  turnId,
+  stepId: step,
+  toolCallId,
+  toolName: "run_shell",
+  input: "partial",
+});
+const result = (toolCallId = "a"): AgentEvent => ({
+  type: "tool-result",
+  conversationId: "c1",
+  turnId: "t1",
+  stepId,
+  toolCallId,
+  toolName: "run_shell",
+  content: "partial",
+  isError: false,
+});
+const stored: StoredChunk[] = [
+  { seq: 50, role: "assistant", turnId: "t1", chunk: { type: "text", text: "full" } },
+  {
+    seq: 51,
+    role: "assistant",
+    turnId: "t1",
+    chunk: {
+      type: "tool-call",
+      stepId,
+      toolCallId: "a",
+      toolName: "run_shell",
+      input: { command: "full" },
+    },
+  },
+  {
+    seq: 52,
+    role: "tool",
+    turnId: "t1",
+    chunk: {
+      type: "tool-result",
+      stepId,
+      toolCallId: "a",
+      toolName: "run_shell",
+      content: "full result",
+      isError: false,
+    },
+  },
+];
+
+describe("history recovery identities", () => {
+  it.each([
+    "before-live",
+    "after-live",
+  ])("reconciles committed tool/text identity %s, not payload equality", (order) => {
+    let s = markHistoryRecovering(initialState(), "t1");
+    if (order === "before-live") s = applyHistory(s, stored);
+    s = fold(s, { type: "text-delta", conversationId: "c1", turnId: "t1", delta: "short" });
+    s = fold(s, call());
+    s = fold(s, result());
+    if (order === "after-live") s = applyHistory(s, stored);
+    expect(selectChunks(s).map((c) => c.chunk)).toEqual(stored.map((c) => c.chunk));
+    // Idempotent refresh; different step's identical text must remain live.
+    s = fold(s, { type: "text-delta", conversationId: "c1", turnId: "t1", delta: "full" });
+    s = applyHistory(s, stored);
+    expect(selectChunks(s).at(-1)).toMatchObject({
+      provisional: true,
+      streaming: true,
+      chunk: { type: "text", text: "full" },
+    });
+  });
+
+  it("does not conflate call/result, another turn, or another step's tool identity", () => {
+    let s = markHistoryRecovering(initialState(), "t1");
+    s = fold(s, call("a", "other"));
+    s = fold(s, call("a", "t1", "other-step" as StepId));
+    s = fold(s, result("a"));
+    s = applyHistory(s, stored);
+    expect(s.provisional).toHaveLength(2);
+    expect(s.provisional.map((p) => p.turnId)).toEqual(["other", "t1"]);
+    expect(s.provisional.every((p) => p.chunk.type === "tool-call")).toBe(true);
+  });
+
+  it("terminal sync removes only that turn and keeps optimistic/unrelated/carried content", () => {
+    let s = markHistoryRecovering(initialState(), "t1");
+    s = fold(s, call());
+    s = fold(s, { type: "turn-sealed", conversationId: "c1", turnId: "t1" });
+    s = fold(s, { type: "turn-start", conversationId: "c1", turnId: "t2" });
+    s = fold(s, { type: "text-delta", conversationId: "c1", turnId: "t2", delta: "queued tail" });
+    s = applyHistory(s, stored, ["t1"]);
+    expect(s.generating).toBe(true);
+    expect(s.currentTurnId).toBe("t2");
+    expect(s.accumulating?.text).toBe("queued tail");
+    expect(s.historyRecoveryTurnIds).toEqual([]);
+  });
+
+  it("does not resurrect watermark-hidden history; older content stays available to Show earlier", () => {
+    let s = markHistoryRecovering({ ...initialState(), hiddenBeforeSeq: 51 }, "t1");
+    s = fold(s, call());
+    s = applyHistory(s, stored);
+    expect(s.committed.map((c) => c.seq)).toEqual([51, 52]);
+    expect(s.provisional).toEqual([]);
+    expect(s.hiddenBeforeSeq).toBe(51);
+  });
+});
+
+it("keeps identical Stop→carry openings as distinct turn identities", () => {
+  let s = markHistoryRecovering(initialState(), "t1");
+  s = fold(s, { type: "user-message", conversationId: "c1", turnId: "t1", text: "same" });
+  s = fold(s, { type: "turn-ended", conversationId: "c1", turnId: "t1", reason: "aborted" });
+  s = fold(s, { type: "user-message", conversationId: "c1", turnId: "t2", text: "same" });
+  s = fold(s, { type: "turn-start", conversationId: "c1", turnId: "t2" });
+  s = applyHistory(
+    s,
+    [{ seq: 1, role: "user", turnId: "t1", chunk: { type: "text", text: "same" } }],
+    ["t1"],
+  );
+  expect(selectChunks(s).map((c) => [c.turnId, c.provisional])).toEqual([
+    ["t1", false],
+    ["t2", true],
+  ]);
+  expect(s.generating).toBe(true);
+});
+
+it("reconciles steering by its completed tool batch, not repeated user text", () => {
+  let s = markHistoryRecovering(initialState(), "t1");
+  s = fold(s, result());
+  s = fold(s, { type: "steering", conversationId: "c1", turnId: "t1", text: "short" });
+  s = applyHistory(s, [
+    ...stored,
+    { seq: 53, role: "user", turnId: "t1", chunk: { type: "text", text: "full steering" } },
+  ]);
+  expect(s.provisional).toEqual([]);
+  expect(selectChunks(s).at(-1)?.chunk).toEqual({ type: "text", text: "full steering" });
+});
diff --git a/src/core/chunks/reconcile.ts b/src/core/chunks/reconcile.ts
new file mode 100644
index 0000000..277d25b
--- /dev/null
+++ b/src/core/chunks/reconcile.ts
@@ -0,0 +1,54 @@
+import type { StoredChunk } from "@dispatch/wire";
+import type { ProvisionalChunk, TranscriptState } from "./types";
+
+/**
+ * A persisted tool call proves its entire assistant step committed atomically.
+ * Text/thinking wire chunks have no stepId: their provisional provenance is
+ * learned from tool-call/step-complete, never from text equality. This also
+ * handles a window starting midway through the step and an omitted prefix.
+ * Also usable when HTTP finished before replay/live flush.
+ */
+export function historyBackedProvisional(
+  state: TranscriptState,
+  committed: readonly StoredChunk[] = state.committed,
+): ReadonlySet<ProvisionalChunk> {
+  const backed = new Set<ProvisionalChunk>();
+  if (!state.historyRecoveryTurnIds?.length) return backed;
+  const steps = new Set<string>();
+  const steeringAnchors = new Set<string>();
+  let lastResult: StoredChunk | undefined;
+  const toolRows = new Map<string, Set<string | undefined>>();
+  for (const row of committed) {
+    const c = row.chunk;
+    if (row.role === "assistant" || row.turnId !== lastResult?.turnId) lastResult = undefined;
+    if (c.type === "tool-result") lastResult = row;
+    if (row.role === "user" && c.type === "text" && lastResult?.chunk.type === "tool-result") {
+      steeringAnchors.add(`${row.turnId}\0${lastResult.chunk.toolCallId}`);
+    }
+    if (c.type === "tool-call" || c.type === "tool-result") {
+      const key = `${row.turnId}\0${c.type}\0${c.toolCallId}`;
+      const ids = toolRows.get(key) ?? new Set<string | undefined>();
+      ids.add(c.stepId);
+      toolRows.set(key, ids);
+      if (c.type === "tool-call" && c.stepId !== undefined) steps.add(`${row.turnId}\0${c.stepId}`);
+    }
+  }
+  for (const p of state.provisional) {
+    if (p.turnId === undefined || !state.historyRecoveryTurnIds?.includes(p.turnId)) continue;
+    const c = p.chunk;
+    if (c.type === "tool-call" || c.type === "tool-result") {
+      const ids = toolRows.get(`${p.turnId}\0${c.type}\0${c.toolCallId}`);
+      if (ids !== undefined && (c.stepId === undefined || ids.has(undefined) || ids.has(c.stepId)))
+        backed.add(p);
+    } else if (p.afterToolCallId !== undefined && p.role === "user") {
+      if (steeringAnchors.has(`${p.turnId}\0${p.afterToolCallId}`)) backed.add(p);
+    } else if (p.opening && c.type === "text") {
+      // The opening is the first user row of this turn; steering is NOT opening.
+      const first = committed.find((row) => row.turnId === p.turnId);
+      if (first?.role === "user" && first.chunk.type === "text") backed.add(p);
+    } else if (p.stepId !== undefined && p.role === "assistant") {
+      if (steps.has(`${p.turnId}\0${p.stepId}`)) backed.add(p);
+    }
+  }
+  return backed;
+}
diff --git a/src/core/chunks/reducer.test.ts b/src/core/chunks/reducer.test.ts
index cba6fad..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();
@@ -1051,3 +1124,86 @@ describe("turnId provenance on live chunks", () => {
     expect(selectChunks(s).map((c) => c.turnId)).toEqual(["t9", undefined]);
   });
 });
+
+describe("foldEvent — additive wire compatibility", () => {
+  it.each(["future-event"])("preserves transcript identity for %s", (type) => {
+    const state = foldEvent(foldEvent(initialState(), turnStart("t1")), textDelta("t1", "partial"));
+    const before = structuredClone(state);
+    const event = { type, conversationId: "c1", turnId: "t1", reason: "sealed" };
+    expect(foldEvent(state, event as Parameters<typeof foldEvent>[1])).toBe(state);
+    expect(state).toEqual(before);
+  });
+});
+
+describe("foldEvent — authoritative turn terminal", () => {
+  it.each([
+    "sealed",
+    "aborted",
+    "error",
+    "rejected",
+  ] as const)("ends generation on %s", (reason) => {
+    const before = foldEvent(
+      foldEvent(initialState(), turnStart("t1")),
+      textDelta("t1", "partial"),
+    );
+    const next = foldEvent(before, {
+      type: "turn-ended",
+      conversationId: "c1",
+      turnId: "t1",
+      reason,
+    });
+    expect(next.generating).toBe(false);
+    expect(next.accumulating).toBeNull();
+    expect(selectChunks(next).map((c) => c.chunk)).toEqual([{ type: "text", text: "partial" }]);
+    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 4eb08c1..71df4dd 100644
--- a/src/core/chunks/reducer.ts
+++ b/src/core/chunks/reducer.ts
@@ -1,5 +1,6 @@
 import type { AgentEvent, Chunk, ImageInput, Role, StoredChunk } from "@dispatch/wire";
 import { assertChunkExhaustive } from "../wire/conformance";
+import { historyBackedProvisional } from "./reconcile";
 import type { AccumulatingChunk, ProvisionalChunk, TranscriptState } from "./types";
 
 /** The initial empty transcript state. */
@@ -40,7 +41,13 @@ function flushAccumulating(
   if (acc === null) return provisional;
   const chunk: Chunk =
     acc.kind === "text" ? { type: "text", text: acc.text } : { type: "thinking", text: acc.text };
-  return [...provisional, withTurn({ role: "assistant", chunk }, acc.turnId)];
+  return [
+    ...provisional,
+    withTurn(
+      { role: "assistant", chunk, ...(acc.stepId !== undefined ? { stepId: acc.stepId } : {}) },
+      acc.turnId,
+    ),
+  ];
 }
 
 /** Attach a `turnId` to a provisional chunk (omitted when unknown). */
@@ -70,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
@@ -129,8 +156,8 @@ function trailingRun(
 /**
  * Merge authoritative seq-keyed chunks into the committed history.
  * Dedupes by seq (new wins), keeps seq-monotonic order, idempotent.
- * When sealedTurnId is set, drops all provisional chunks (now superseded)
- * and clears sealedTurnId.
+ * A sealed/finally-synced turn supersedes only its own provisional chunks;
+ * unrelated turns and an already-running queued carry remain live.
  *
  * Chunks below the chat-limit unload watermark (`hiddenBeforeSeq`) are
  * REJECTED: a full-cache or tail merge must not resurrect what the trim
@@ -139,6 +166,7 @@ function trailingRun(
 export function applyHistory(
   state: TranscriptState,
   chunks: readonly StoredChunk[],
+  finishedTurnIds: readonly string[] = [],
 ): TranscriptState {
   const seqMap = new Map<number, StoredChunk>();
   for (const c of state.committed) seqMap.set(c.seq, c);
@@ -149,14 +177,49 @@ export function applyHistory(
     seqMap.set(c.seq, c);
   }
   const committed = Array.from(seqMap.values()).sort((a, b) => a.seq - b.seq);
+  const identityRows = new Map(seqMap);
+  for (const c of chunks) identityRows.set(c.seq, c);
+  const backed = historyBackedProvisional(
+    state,
+    [...identityRows.values()].sort((a, b) => a.seq - b.seq),
+  );
+  const finished = new Set(finishedTurnIds);
+  if (state.sealedTurnId !== null) finished.add(state.sealedTurnId);
+  state = {
+    ...state,
+    provisional: state.provisional.filter((p) => !backed.has(p) && !finished.has(p.turnId ?? "")),
+    accumulating:
+      state.accumulating?.turnId !== undefined && finished.has(state.accumulating.turnId)
+        ? null
+        : state.accumulating,
+  };
 
   if (state.sealedTurnId !== null) {
+    // Legacy (non-recovery) histories may carry no turn provenance at all.
+    // Retain their seal cleanup, but never erase an already-live queued carry.
+    const legacySeal =
+      (state.historyRecoveryTurnIds?.length ?? 0) === 0 && finishedTurnIds.length === 0;
     return {
       ...state,
       committed,
-      provisional: [],
-      accumulating: null,
+      provisional: legacySeal
+        ? state.provisional.filter(
+            (p) =>
+              state.generating &&
+              p.turnId === state.currentTurnId &&
+              p.turnId !== state.sealedTurnId,
+          )
+        : state.provisional,
+      accumulating:
+        legacySeal && state.accumulating?.turnId === state.sealedTurnId ? null : state.accumulating,
       sealedTurnId: null,
+      ...(state.historyRecoveryTurnIds !== undefined
+        ? {
+            historyRecoveryTurnIds: state.historyRecoveryTurnIds.filter(
+              (id) => !finishedTurnIds.includes(id),
+            ),
+          }
+        : {}),
     };
   }
 
@@ -167,7 +230,14 @@ export function applyHistory(
   // match the trailing provisional user-run against the trailing committed
   // user-run by content equality and drop the whole echo when it is fully
   // backed. Leaves the accumulating (streaming) chunk untouched.
-  if (addedNew && state.generating && state.provisional.length > 0) {
+  if (
+    addedNew &&
+    state.generating &&
+    state.provisional.length > 0 &&
+    !state.provisional.some(
+      (p) => p.turnId !== undefined && state.historyRecoveryTurnIds?.includes(p.turnId),
+    )
+  ) {
     const provRun = trailingRun(state.provisional, "user");
     if (provRun.length > 0) {
       const commRun = trailingRun(committed, "user");
@@ -179,7 +249,11 @@ export function applyHistory(
         commRun.length >= provRun.length &&
         provRun.every((p, i) => {
           const c = commRun[i];
-          return c !== undefined && chunkContentEquals(p.chunk, c.chunk);
+          return (
+            c !== undefined &&
+            (c.turnId === undefined || p.turnId === undefined || c.turnId === p.turnId) &&
+            chunkContentEquals(p.chunk, c.chunk)
+          );
         });
       if (fullyBacked) {
         const dropStart = state.provisional.length - provRun.length;
@@ -189,7 +263,17 @@ export function applyHistory(
     }
   }
 
-  return { ...state, committed };
+  return {
+    ...state,
+    committed,
+    ...(state.historyRecoveryTurnIds !== undefined
+      ? {
+          historyRecoveryTurnIds: state.historyRecoveryTurnIds.filter(
+            (id) => !finishedTurnIds.includes(id),
+          ),
+        }
+      : {}),
+  };
 }
 
 /**
@@ -200,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
@@ -224,13 +310,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(state.provisional, 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
@@ -246,11 +345,22 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
       // text) emits no `user-message` and is not de-duped here.
       if (event.text.length === 0) return state;
       const run = trailingRun(state.provisional, "user");
-      const alreadyEchoed = run.some((p) => p.chunk.type === "text" && p.chunk.text === event.text);
-      if (alreadyEchoed) {
+      const echoed = run.find(
+        (p) =>
+          (p.turnId === undefined || p.turnId === event.turnId) &&
+          p.chunk.type === "text" &&
+          p.chunk.text === event.text,
+      );
+      if (echoed !== undefined) {
         return {
           ...state,
-          provisional: tagTrailingUserEcho(state.provisional, event.turnId),
+          // Preserve the live text independently of the local send. A delayed
+          // queued ack may remove its images, but cannot erase this server event.
+          provisional: tagTrailingUserEcho(state.provisional, event.turnId).map((p) => {
+            if (p.chunk !== echoed.chunk) return p;
+            const { localRequestId: _local, ...confirmed } = p;
+            return { ...confirmed, opening: true };
+          }),
           generating: true,
         };
       }
@@ -259,7 +369,12 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
         ...state,
         provisional: [
           ...provisional,
-          { role: "user", chunk: { type: "text", text: event.text }, turnId: event.turnId },
+          {
+            role: "user",
+            chunk: { type: "text", text: event.text },
+            turnId: event.turnId,
+            opening: true,
+          },
         ],
         accumulating: null,
         generating: true,
@@ -268,7 +383,7 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
 
     case "text-delta": {
       const acc = state.accumulating;
-      if (acc !== null && acc.kind === "text") {
+      if (acc !== null && acc.kind === "text" && acc.turnId === event.turnId) {
         return {
           ...state,
           accumulating: { ...acc, text: acc.text + event.delta },
@@ -286,7 +401,7 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
 
     case "reasoning-delta": {
       const acc = state.accumulating;
-      if (acc !== null && acc.kind === "thinking") {
+      if (acc !== null && acc.kind === "thinking" && acc.turnId === event.turnId) {
         return {
           ...state,
           accumulating: { ...acc, text: acc.text + event.delta },
@@ -302,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 = {
@@ -355,8 +494,24 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
       return { ...state, latestUsage: event.usage };
 
     case "step-complete":
-      // Timing metadata — no content chunk; handled by the telemetry reducer.
-      return state;
+      // A persistence boundary may lag this notification. Stamp provenance and
+      // finish the generation chunk, but keep it until history really backs it.
+      return {
+        ...state,
+        provisional: flushAccumulating(state.provisional, state.accumulating),
+        accumulating: null,
+      };
+
+    case "turn-ended":
+      return {
+        ...state,
+        provisional:
+          state.accumulating?.turnId === event.turnId
+            ? flushAccumulating(state.provisional, state.accumulating)
+            : state.provisional,
+        accumulating: state.accumulating?.turnId === event.turnId ? null : state.accumulating,
+        generating: state.currentTurnId === event.turnId ? false : state.generating,
+      };
 
     case "done": {
       const provisional = flushAccumulating(state.provisional, state.accumulating);
@@ -388,11 +543,21 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
       // into the transcript by the sender).
       if (event.text.length === 0) return state;
       const provisional = flushAccumulating(state.provisional, state.accumulating);
+      const preceding = [...state.committed, ...provisional]
+        .reverse()
+        .find((p) => p.turnId === event.turnId && p.chunk.type === "tool-result");
+      const afterToolCallId =
+        preceding?.chunk.type === "tool-result" ? preceding.chunk.toolCallId : undefined;
       return {
         ...state,
         provisional: [
           ...provisional,
-          { role: "user", chunk: { type: "text", text: event.text }, turnId: event.turnId },
+          {
+            role: "user",
+            chunk: { type: "text", text: event.text },
+            turnId: event.turnId,
+            ...(afterToolCallId !== undefined ? { afterToolCallId } : {}),
+          },
         ],
         accumulating: null,
         generating: true,
@@ -408,6 +573,9 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState
       // (already true from `turn-start`) is left untouched.
       return { ...state, providerRetry: event };
     }
+
+    default:
+      return state; // Forward-compatible with additive wire events.
   }
 }
 
@@ -426,14 +594,37 @@ const RETRY_CLEARING_EVENTS: ReadonlySet<AgentEvent["type"]> = new Set([
   "turn-start",
   "text-delta",
   "reasoning-delta",
+  "reasoning-end",
   "tool-call",
   "tool-result",
   "error",
   "done",
   "turn-sealed",
+  "turn-ended",
 ]);
 
 export function foldEvent(state: TranscriptState, event: AgentEvent): TranscriptState {
+  if (
+    event.type === "tool-call" ||
+    event.type === "tool-result" ||
+    event.type === "step-complete"
+  ) {
+    state = {
+      ...state,
+      provisional: state.provisional.map((p) =>
+        p.turnId === event.turnId &&
+        p.role === "assistant" &&
+        p.stepId === undefined &&
+        (p.chunk.type === "text" || p.chunk.type === "thinking")
+          ? { ...p, stepId: event.stepId }
+          : p,
+      ),
+      accumulating:
+        state.accumulating?.turnId === event.turnId
+          ? { ...state.accumulating, stepId: event.stepId }
+          : state.accumulating,
+    };
+  }
   const next = reduceEvent(state, event);
   if (event.type === "provider-retry") return next; // set by reduceEvent; not a clearing event
   if (RETRY_CLEARING_EVENTS.has(event.type) && state.providerRetry !== null) {
@@ -460,6 +651,7 @@ export function appendUserMessage(
   state: TranscriptState,
   text: string,
   images?: readonly ImageInput[],
+  localRequestId?: string,
 ): TranscriptState {
   const provisional = flushAccumulating(state.provisional, state.accumulating);
   const userChunks: Chunk[] = [];
@@ -481,7 +673,35 @@ export function appendUserMessage(
   }
   return {
     ...state,
-    provisional: [...provisional, ...userChunks.map((chunk) => ({ role: "user", chunk }) as const)],
+    provisional: [
+      ...provisional,
+      ...userChunks.map(
+        (chunk) =>
+          ({
+            role: "user",
+            chunk,
+            ...(localRequestId !== undefined ? { localRequestId } : {}),
+          }) as const,
+      ),
+    ],
     accumulating: null,
   };
 }
+
+/** Remove only still-local artifacts of one accepted queued send; never history/live text. */
+export function removeLocalEcho(state: TranscriptState, requestId: string): TranscriptState {
+  const provisional = state.provisional.filter((p) => p.localRequestId !== requestId);
+  return provisional.length === state.provisional.length ? state : { ...state, provisional };
+}
+
+/** A resync notice is lifecycle evidence, never fabricated transcript content. */
+export function markHistoryRecovering(state: TranscriptState, turnId: string): TranscriptState {
+  return {
+    ...state,
+    currentTurnId: turnId,
+    generating: true,
+    historyRecoveryTurnIds: state.historyRecoveryTurnIds?.includes(turnId)
+      ? state.historyRecoveryTurnIds
+      : [...(state.historyRecoveryTurnIds ?? []), turnId],
+  };
+}
diff --git a/src/core/chunks/selectors.test.ts b/src/core/chunks/selectors.test.ts
index 57c1415..90bb5cc 100644
--- a/src/core/chunks/selectors.test.ts
+++ b/src/core/chunks/selectors.test.ts
@@ -1,7 +1,8 @@
 import type { AgentEvent, StepId, StoredChunk } from "@dispatch/wire";
 import { describe, expect, it } from "vitest";
-import { applyHistory, foldEvent, initialState } from "./reducer";
+import { applyHistory, foldEvent, initialState, markHistoryRecovering } from "./reducer";
 import { selectChunks } from "./selectors";
+import { trimTranscript } from "./trim";
 import type { TranscriptState } from "./types";
 
 const C = "conv";
@@ -19,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,
@@ -131,6 +133,34 @@ describe("selectChunks — de-dup of a late-join replay against committed histor
     expect(rendered.filter((r) => r.provisional)).toHaveLength(3);
   });
 
+  it("preserves an uncommitted accumulating tail even when its text repeats history", () => {
+    const committed: StoredChunk[] = [
+      ...committedStep0,
+      { seq: 5, role: "assistant", chunk: { type: "text", text: "final" }, turnId: T },
+    ];
+    let s = applyHistory(initialState(), committed);
+    s = markHistoryRecovering(s, T);
+    s = fold(s, [userMsg, turnStart, think("plan"), call("a"), result("a"), text("final")]);
+    const rendered = selectChunks(s);
+    expect(rendered).toHaveLength(6);
+    expect(rendered.slice(0, 5).every((r) => !r.provisional)).toBe(true);
+    expect(rendered[5]).toMatchObject({
+      provisional: true,
+      streaming: true,
+      chunk: { type: "text", text: "final" },
+    });
+  });
+
+  it("reconciles the same step identity even when provisional payload differs", () => {
+    let s = applyHistory(initialState(), committedStep0);
+    s = markHistoryRecovering(s, T);
+    s = fold(s, [userMsg, turnStart, think("DIFFERENT"), call("a"), result("a")]);
+    const rendered = selectChunks(s);
+    // Authoritative identity wins; payload equality cannot identify omitted/truncated content.
+    expect(rendered.filter((r) => r.provisional)).toHaveLength(0);
+    expect(rendered.map((r) => r.chunk)).toEqual(committedStep0.map((r) => r.chunk));
+  });
+
   it("does nothing when the committed tail belongs to another turn", () => {
     const other = committedStep0.map((c) => ({ ...c, turnId: "older" }));
     let s = applyHistory(initialState(), other);
@@ -138,6 +168,204 @@ 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("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/chunks/selectors.ts b/src/core/chunks/selectors.ts
index d32a628..8d32fcf 100644
--- a/src/core/chunks/selectors.ts
+++ b/src/core/chunks/selectors.ts
@@ -1,4 +1,5 @@
 import type { ChatMessage, Chunk, StoredChunk, TurnProviderRetryEvent } from "@dispatch/wire";
+import { historyBackedProvisional } from "./reconcile";
 import type { ProvisionalChunk, RenderedChunk, TranscriptState } from "./types";
 
 /**
@@ -26,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).
@@ -42,41 +191,59 @@ function sameChunk(p: ProvisionalChunk, c: StoredChunk): boolean {
 function committedOverlap(state: TranscriptState): { skip: number; hideAccumulating: boolean } {
   const first = state.provisional[0];
   const turnId = first?.turnId ?? state.accumulating?.turnId;
-  if (turnId === undefined) return { skip: 0, hideAccumulating: false };
+  if (turnId === undefined || state.historyRecoveryTurnIds?.includes(turnId))
+    return { skip: 0, hideAccumulating: false };
   // The committed run for this turn (contiguous at the tail of `committed`).
   let start = state.committed.length;
   while (start > 0 && state.committed[start - 1]?.turnId === turnId) start--;
-  const run = state.committed.length - start;
-  if (run === 0) return { skip: 0, hideAccumulating: false };
-
-  let skip = 0;
-  while (skip < state.provisional.length && skip < run) {
-    const p = state.provisional[skip];
-    const c = state.committed[start + skip];
-    if (p === undefined || c === undefined || p.turnId !== turnId || !sameChunk(p, c)) break;
-    skip++;
-  }
+  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 && 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) — minus provisional chunks already backed by
- * committed history (see `committedOverlap`).
+ * then provisional (seq: null), excluding history-backed identities. Unknown
+ * accumulating content stays live, even if its text equals an earlier step.
  */
 export function selectChunks(state: TranscriptState): readonly RenderedChunk[] {
   const result: RenderedChunk[] = [];
+  const backed = historyBackedProvisional(state);
   const { skip, hideAccumulating } = committedOverlap(state);
   for (const c of state.committed) {
     result.push(
@@ -87,7 +254,7 @@ export function selectChunks(state: TranscriptState): readonly RenderedChunk[] {
   }
   for (let i = skip; i < state.provisional.length; i++) {
     const p = state.provisional[i];
-    if (p === undefined) continue;
+    if (p === undefined || backed.has(p)) continue;
     result.push(
       p.turnId !== undefined
         ? { seq: null, role: p.role, chunk: p.chunk, provisional: true, turnId: p.turnId }
diff --git a/src/core/chunks/trim.test.ts b/src/core/chunks/trim.test.ts
index 7c4bbef..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) => ({
@@ -146,7 +159,7 @@ describe("windowTranscript", () => {
   it("is the identity within the window", () => {
     const state = stateWith(chunks(1, 50));
     expect(windowTranscript(state, 75)).toBe(state);
-    expect(selectHasEarlier(state)).toBe(false);
+    expect(selectHasEarlier(state, 1)).toBe(false);
   });
 });
 
@@ -178,21 +191,21 @@ describe("restoreEarlier", () => {
     expect(selectHasEarlier(restored)).toBe(true);
   });
 
-  it("restoring down to seq 1 reaches the contractual origin (hasEarlier clears)", () => {
+  it("restoring down to a known boundary clears hasEarlier", () => {
     const windowed = windowTranscript(stateWith(chunks(1, 100)), 75); // hidden: 1..25
     const restored = restoreEarlier(windowed, chunks(1, 100), 64);
     expect(restored.committed).toHaveLength(100);
     expect(restored.committed[0]?.seq).toBe(1);
     expect(restored.hiddenBeforeSeq).toBe(1); // floor at the origin — inert
     expect(restored.hiddenThinkingCount).toBe(0);
-    expect(selectHasEarlier(restored)).toBe(false);
+    expect(selectHasEarlier(restored, 1)).toBe(false);
   });
 
   it("is the identity when nothing older is known locally (server may still hold more)", () => {
     const windowed = windowTranscript(stateWith(chunks(50, 200)), 75);
     const restored = restoreEarlier(windowed, [], 64);
     expect(restored).toBe(windowed);
-    // seqs are 1-based gap-free: window starts at 126 ⇒ older chunks DO exist.
+    // The server boundary is unknown, so another page may exist.
     expect(selectHasEarlier(restored)).toBe(true);
   });
 
@@ -203,7 +216,7 @@ describe("restoreEarlier", () => {
 
   it("works on a server-windowed transcript (no local watermark)", () => {
     // A cold-cache fresh load with `?limit=` commits a suffix (seq 809..1000)
-    // with hiddenBeforeSeq still 0 — hasEarlier derives from seq > 1, and a
+    // with hiddenBeforeSeq still 0 — the server boundary is unknown, and a
     // backfilled run merges below it.
     const state = stateWith(chunks(809, 1000));
     expect(state.hiddenBeforeSeq).toBe(0);
@@ -230,6 +243,15 @@ describe("restoreEarlier", () => {
     const restored = restoreEarlier(trimmed, original, 1000);
     expect(restored.committed).toEqual(original);
     expect(restored.hiddenBeforeSeq).toBe(1);
-    expect(selectHasEarlier(restored)).toBe(false);
+    expect(selectHasEarlier(restored, 1)).toBe(false);
   });
 });
+
+it("uses a non-1 history boundary rather than the seq origin", () => {
+  const state = stateWith(chunks(101, 110));
+  expect(selectHasEarlier(state, 101)).toBe(false);
+  expect(selectHasEarlier(state, null)).toBe(true);
+  const windowed = windowTranscript(state, 5);
+  expect(selectHasEarlier(windowed, 101)).toBe(true);
+  expect(selectHasEarlier(restoreEarlier(windowed, state.committed, 5), 101)).toBe(false);
+});
diff --git a/src/core/chunks/trim.ts b/src/core/chunks/trim.ts
index 9846357..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 {
@@ -150,8 +156,7 @@ function oldestLoadedSeq(state: TranscriptState): number {
  * `?beforeSeq=` backfill; chunks at/inside the window are ignored). The newest
  * `count` of them are merged back in front of `committed`, and the watermark
  * follows the new window start so history merges still can't resurrect what
- * remains unloaded. Identity when the window already starts at seq 1 (the
- * contractual origin) or nothing older is known locally.
+ * remains unloaded. Identity when nothing older is known locally.
  */
 export function restoreEarlier(
   state: TranscriptState,
@@ -159,7 +164,7 @@ export function restoreEarlier(
   count: number,
 ): TranscriptState {
   const oldest = oldestLoadedSeq(state);
-  if (oldest <= 1) return state;
+  if (oldest === 0) return state;
   const below = earlier.filter((c) => c.seq < oldest).sort((a, b) => a.seq - b.seq);
   if (below.length === 0) return state;
   const keep = below.slice(-Math.max(1, count));
@@ -172,14 +177,11 @@ export function restoreEarlier(
   };
 }
 
-/**
- * Whether earlier history exists below the loaded window — drives the
- * "Show earlier messages" affordance. Derived from the [email protected] CONTRACT
- * that per-conversation seqs are 1-based and gap-free: a loaded window that
- * starts above seq 1 means older chunks exist (locally cached or server-side),
- * whether the window came from a local trim or a server-windowed (`?limit=`)
- * fresh load.
- */
-export function selectHasEarlier(state: TranscriptState): boolean {
-  return oldestLoadedSeq(state) > 1;
+/** Earlier availability uses the known server boundary, never a seq origin. */
+export function selectHasEarlier(
+  state: TranscriptState,
+  earliestSeq: number | null = null,
+): boolean {
+  const oldest = oldestLoadedSeq(state);
+  return oldest > 0 && (earliestSeq === null || oldest > earliestSeq);
 }
diff --git a/src/core/chunks/types.ts b/src/core/chunks/types.ts
index 949857a..2e2ee0c 100644
--- a/src/core/chunks/types.ts
+++ b/src/core/chunks/types.ts
@@ -2,6 +2,7 @@ import type { Chunk, Role, StoredChunk, TurnProviderRetryEvent, Usage } from "@d
 
 /** A chunk being accumulated from streaming deltas (text or thinking). */
 export interface AccumulatingChunk {
+  readonly stepId?: string;
   readonly kind: "text" | "thinking";
   readonly text: string;
   /** The turn generating this chunk (from the live event's `turnId`). */
@@ -10,6 +11,13 @@ export interface AccumulatingChunk {
 
 /** A provisional chunk that has no authoritative seq yet. */
 export interface ProvisionalChunk {
+  /** Local provenance learned from tool-call/step-complete, not content. */
+  readonly stepId?: string;
+  readonly opening?: boolean;
+  /** Steering is persisted immediately after this completed tool batch. */
+  readonly afterToolCallId?: string;
+  /** Local send identity only; turn-start tagging does NOT confirm this content. */
+  readonly localRequestId?: string;
   readonly role: Role;
   readonly chunk: Chunk;
   /**
@@ -22,6 +30,8 @@ export interface ProvisionalChunk {
 
 /** The transcript reducer state. Holds committed history + live in-flight turn. */
 export interface TranscriptState {
+  /** Turns with omitted delivery: never use payload equality as replay identity. */
+  readonly historyRecoveryTurnIds?: readonly string[];
   readonly committed: readonly StoredChunk[];
   readonly provisional: readonly ProvisionalChunk[];
   readonly accumulating: AccumulatingChunk | null;