Merge remote-tracking branch 'origin/main' into fix/6199-live-narration-764ea71e-1

The discard assertions pass but synchronous temporary-directory removal races a Windows lock after daemon cleanup. Use the existing asynchronous rm retry pattern without weakening the behavior assertions.

packages/server/src/server/agent/providers/claude/agent.progress-updates.real.e2e.test.ts

@@ -0,0 +1,120 @@
+import { afterEach, expect, test } from "vitest";
+import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import pino from "pino";
+
+import type { AgentSession, AgentStreamEvent, AgentTimelineItem } from "../../agent-sdk-types.js";
+import { ClaudeAgentClient } from "./agent.js";
+
+// Reproduces #6199 through the real Claude provider. Claude Opus 5.5 writes the
+// text it means for the user between tool calls as a "progress update": a
+// thinking block that Claude Code marks with `narration_block_indexes`.
+const PROMPT =
+  "There are three plan files in this directory: plan-a.txt, plan-b.txt and plan-c.txt. Read each one, tell me how they compare on cost, time and risk, and then ask me which plan to pick using your AskUserQuestion tool. Keep me posted as you go.";
+
+const handles: Array<{ cwd: string; session: AgentSession }> = [];
+
+afterEach(async () => {
+  for (const handle of handles.splice(0)) {
+    await handle.session.close().catch(() => undefined);
+    rmSync(handle.cwd, { recursive: true, force: true });
+  }
+});
+
+function writePlans(cwd: string): void {
+  writeFileSync(
+    path.join(cwd, "plan-a.txt"),
+    "Plan A\nCost: 3 credits\nTime: 2 days\nRisk: needs a manual review step\n",
+  );
+  writeFileSync(
+    path.join(cwd, "plan-b.txt"),
+    "Plan B\nCost: 5 credits\nTime: 1 day\nRisk: none known\n",
+  );
+  writeFileSync(
+    path.join(cwd, "plan-c.txt"),
+    "Plan C\nCost: 2 credits\nTime: 4 days\nRisk: depends on an external vendor\n",
+  );
+}
+
+async function runPlanComparison(): Promise<{
+  items: AgentTimelineItem[];
+  narration: string[];
+}> {
+  const cwd = mkdtempSync(path.join(tmpdir(), "claude-progress-updates-"));
+  writePlans(cwd);
+  const narration: string[] = [];
+  const logger = pino(
+    { level: "trace" },
+    {
+      write(line) {
+        const record = JSON.parse(line) as {
+          rawEvent?: {
+            type?: string;
+            narration_block_indexes?: number[];
+            message?: { content?: Array<{ type: string; thinking?: string }> };
+          };
+        };
+        const frame = record.rawEvent;
+        if (frame?.type !== "assistant") return;
+        for (const index of frame.narration_block_indexes ?? []) {
+          const block = frame.message?.content?.[index];
+          if (block?.type === "thinking" && block.thinking) narration.push(block.thinking);
+        }
+      },
+    },
+  );
+  const client = new ClaudeAgentClient({ logger });
+  const session = await client.createSession({
+    provider: "claude",
+    cwd,
+    model: "claude-opus-5-5",
+    thinkingOptionId: "xhigh",
+    modeId: "bypassPermissions",
+  });
+  handles.push({ cwd, session });
+
+  const items: AgentTimelineItem[] = [];
+  await new Promise<void>((resolve, reject) => {
+    session.subscribe((event: AgentStreamEvent) => {
+      if (event.type === "timeline") items.push(event.item);
+      if (event.type === "turn_failed") reject(new Error(event.error));
+      if (
+        event.type === "permission_requested" ||
+        event.type === "turn_completed" ||
+        event.type === "turn_failed"
+      ) {
+        resolve();
+      }
+    });
+    void session.startTurn(PROMPT).catch(reject);
+  });
+
+  return { items, narration };
+}
+
+test("the comparison written before AskUserQuestion reaches the timeline as assistant text", async () => {
+  const { items, narration } = await runPlanComparison();
+  const lastReadIndex = items.findLastIndex(
+    (item) => item.type === "tool_call" && item.name !== "AskUserQuestion",
+  );
+  const askIndex = items.findIndex(
+    (item) => item.type === "tool_call" && item.name === "AskUserQuestion",
+  );
+  expect(askIndex).toBeGreaterThan(lastReadIndex);
+  const between = items.slice(lastReadIndex + 1, askIndex);
+  const assistantText = between
+    .flatMap((item) => (item.type === "assistant_message" ? [item.text] : []))
+    .join("");
+  expect(assistantText).toMatch(/plan/i);
+  // A plain-text model response would not exercise narration classification.
+  expect(narration.length).toBeGreaterThan(0);
+  for (const text of narration) {
+    expect(
+      items.flatMap((item) => (item.type === "assistant_message" ? [item.text] : [])).join(""),
+    ).toContain(text);
+    expect(
+      items.flatMap((item) => (item.type === "reasoning" ? [item.text] : [])).join(""),
+    ).not.toContain(text);
+  }
+}, 280_000);

packages/server/src/server/agent/providers/claude/agent.redesign.test.ts

@@ -1,5 +1,6 @@
 import { afterEach, beforeEach, expect, test, vi } from "vitest";
 import type { Logger } from "pino";
+import type { Query, SDKMessage } from "@anthropic-ai/claude-agent-sdk";
 
 import { createTestLogger } from "../../../../test-utils/test-logger.js";
 import { asInternals } from "../../../test-utils/class-mocks.js";
@@ -2060,3 +2061,340 @@ test("does not use stream_event uuid as assistant message identity when message_
 
   await session.close();
 });
+
+const MULTI_BLOCK_SESSION_ID = "multi-block-session";
+
+function streamFrame(event: Record<string, unknown>) {
+  return { type: "stream_event", parent_tool_use_id: null, event };
+}
+
+// A block's start and deltas. Its stop is separate because the SDK sends the block's snapshot
+// before content_block_stop.
+function streamedBlock(input: { index: number; kind: "thinking" | "text"; deltas: string[] }) {
+  const { index, kind, deltas } = input;
+  const deltaType = kind === "thinking" ? "thinking_delta" : "text_delta";
+  return [
+    streamFrame({ type: "content_block_start", index, content_block: { type: kind, [kind]: "" } }),
+    ...deltas.map((text) =>
+      streamFrame({ type: "content_block_delta", index, delta: { type: deltaType, [kind]: text } }),
+    ),
+  ];
+}
+
+function blockStop(index: number) {
+  return streamFrame({ type: "content_block_stop", index });
+}
+
+// The SDK's per-block assistant message: same message id as its siblings, one block of content.
+function blockSnapshot(input: { uuid: string; kind: "thinking" | "text"; text: string }) {
+  const { uuid, kind, text } = input;
+  return {
+    type: "assistant",
+    uuid,
+    parent_tool_use_id: null,
+    session_id: MULTI_BLOCK_SESSION_ID,
+    message: { id: "msg-multi", role: "assistant", content: [{ type: kind, [kind]: text }] },
+  };
+}
+
+async function runScriptedTurn(
+  frames: Array<Record<string, unknown> | Error | "interrupt">,
+): Promise<AgentTimelineItem[]> {
+  let session: Awaited<ReturnType<ClaudeAgentClient["createSession"]>>;
+  // An injected SDK transport exercises the public session, without module mocks.
+  const client = new ClaudeAgentClient({
+    logger: createTestLogger(),
+    resolveBinary: async () => "/test/claude/bin",
+    queryFactory: ({ prompt }) => {
+      const readPromptUuid = createPromptUuidReader(prompt as AsyncIterable<unknown>);
+      const iterator = (async function* (): AsyncGenerator<SDKMessage, void> {
+        // These fixtures deliberately include legacy/incomplete wire frames. Their cast stays
+        // at the input boundary; the injected transport itself implements the full SDK interface.
+        yield { ...claudeTurnInit(), session_id: MULTI_BLOCK_SESSION_ID } as SDKMessage;
+        yield {
+          type: "user",
+          message: { role: "user", content: "multi-block prompt" },
+          parent_tool_use_id: null,
+          uuid: (await readPromptUuid()) ?? "missing-prompt-uuid",
+          session_id: MULTI_BLOCK_SESSION_ID,
+        };
+        for (const frame of frames) {
+          if (frame instanceof Error) throw frame;
+          if (frame === "interrupt") {
+            await session.interrupt();
+            return;
+          }
+          yield frame as SDKMessage;
+        }
+        yield successResult() as SDKMessage;
+      })();
+      const unsupported = async (): Promise<never> => {
+        throw new Error("SDK operation not supported by scripted transport");
+      };
+      const transport: Query = Object.assign(iterator, {
+        interrupt: async () => undefined,
+        close: () => undefined,
+        setPermissionMode: async () => undefined,
+        setModel: async () => undefined,
+        supportedModels: async () => [
+          { value: "opus", displayName: "Opus", description: "Test model" },
+        ],
+        supportedCommands: async () => [],
+        setMcpPermissionModeOverride: unsupported,
+        setMaxThinkingTokens: unsupported,
+        applyFlagSettings: unsupported,
+        initializationResult: unsupported,
+        reinitialize: unsupported,
+        supportedAgents: unsupported,
+        mcpServerStatus: unsupported,
+        getContextUsage: unsupported,
+        usage_EXPERIMENTAL_MAY_CHANGE_DO_NOT_RELY_ON_THIS_API_YET: unsupported,
+        readFile: unsupported,
+        reloadPlugins: unsupported,
+        reloadSkills: unsupported,
+        accountInfo: unsupported,
+        rewindFiles: unsupported,
+        seedReadState: unsupported,
+        reconnectMcpServer: unsupported,
+        toggleMcpServer: unsupported,
+        setMcpServers: unsupported,
+        streamInput: unsupported,
+        stopTask: unsupported,
+        backgroundTasks: unsupported,
+      });
+      return transport;
+    },
+  });
+  session = await client.createSession({ provider: "claude", cwd: process.cwd() });
+  try {
+    const events = await collectUntilTerminal(streamSession(session, "multi-block prompt"));
+    return events.flatMap((event) => (event.type === "timeline" ? [event.item] : []));
+  } finally {
+    await session.close();
+  }
+}
+
+// Each chunk as emitted, so a test sees duplicates and lost chunks, not just the final text.
+function emittedChunks(input: {
+  items: AgentTimelineItem[];
+  type: "reasoning" | "assistant_message";
+}): string[] {
+  const { items, type } = input;
+  return items.flatMap((item) => (item.type === type ? [item.text] : []));
+}
+
+const MESSAGE_START = streamFrame({
+  type: "message_start",
+  message: { id: "msg-multi", role: "assistant" },
+});
+const MESSAGE_STOP = streamFrame({ type: "message_stop" });
+
+test("shows each thinking block of a multi-block message once", async () => {
+  const items = await runScriptedTurn([
+    MESSAGE_START,
+    ...streamedBlock({
+      index: 0,
+      kind: "thinking",
+      deltas: ["First paragraph.\n\n", "Second paragraph.\n\n"],
+    }),
+    blockSnapshot({
+      uuid: "uuid-block-0",
+      kind: "thinking",
+      text: "First paragraph.\n\nSecond paragraph.\n\n",
+    }),
+    blockStop(0),
+    ...streamedBlock({ index: 1, kind: "thinking", deltas: ["Last paragraph.\n\n"] }),
+    blockSnapshot({ uuid: "uuid-block-1", kind: "thinking", text: "Last paragraph.\n\n" }),
+    blockStop(1),
+    MESSAGE_STOP,
+  ]);
+
+  expect(emittedChunks({ items, type: "reasoning" })).toEqual([
+    "First paragraph.\n\nSecond paragraph.\n\n",
+    "Last paragraph.\n\n",
+  ]);
+});
+
+test("shows each text block of a multi-block message once", async () => {
+  const items = await runScriptedTurn([
+    MESSAGE_START,
+    ...streamedBlock({ index: 0, kind: "text", deltas: ["Checking ", "the tests."] }),
+    blockSnapshot({ uuid: "uuid-block-0", kind: "text", text: "Checking the tests." }),
+    blockStop(0),
+    ...streamedBlock({ index: 1, kind: "text", deltas: [" All green."] }),
+    blockSnapshot({ uuid: "uuid-block-1", kind: "text", text: " All green." }),
+    blockStop(1),
+    MESSAGE_STOP,
+  ]);
+
+  expect(emittedChunks({ items, type: "assistant_message" })).toEqual([
+    "Checking ",
+    "the tests.",
+    " All green.",
+  ]);
+});
+
+test("fills in text a block snapshot has beyond what streamed", async () => {
+  const items = await runScriptedTurn([
+    MESSAGE_START,
+    ...streamedBlock({ index: 0, kind: "thinking", deltas: ["Partial"] }),
+    blockSnapshot({ uuid: "uuid-block-0", kind: "thinking", text: "Partial thought." }),
+    blockStop(0),
+    ...streamedBlock({ index: 1, kind: "thinking", deltas: ["Next."] }),
+    blockSnapshot({ uuid: "uuid-block-1", kind: "thinking", text: "Next." }),
+    blockStop(1),
+    MESSAGE_STOP,
+  ]);
+
+  expect(emittedChunks({ items, type: "reasoning" })).toEqual(["Partial thought.", "Next."]);
+});
+
+test("keeps identical blocks that arrive only as snapshots", async () => {
+  const items = await runScriptedTurn([
+    blockSnapshot({ uuid: "uuid-block-0", kind: "text", text: "Same line." }),
+    blockSnapshot({ uuid: "uuid-block-1", kind: "text", text: "Same line." }),
+  ]);
+
+  expect(emittedChunks({ items, type: "assistant_message" })).toEqual(["Same line.", "Same line."]);
+});
+
+test("ignores a redelivered block snapshot", async () => {
+  const items = await runScriptedTurn([
+    blockSnapshot({ uuid: "uuid-block-0", kind: "text", text: "Once." }),
+    blockSnapshot({ uuid: "uuid-block-0", kind: "text", text: "Once." }),
+  ]);
+
+  expect(emittedChunks({ items, type: "assistant_message" })).toEqual(["Once."]);
+});
+
+// Older fixtures send deltas without an index. This also covers a snapshot arriving after its
+// block's stop, the reverse of the documented order.
+test("continues the open block with deltas that carry no index", async () => {
+  const items = await runScriptedTurn([
+    MESSAGE_START,
+    ...streamedBlock({ index: 0, kind: "text", deltas: ["A"] }),
+    streamFrame({ type: "content_block_delta", delta: { type: "text_delta", text: "B" } }),
+    blockStop(0),
+    blockSnapshot({ uuid: "uuid-block-0", kind: "text", text: "AB" }),
+    streamFrame({ type: "content_block_delta", delta: { type: "text_delta", text: "C" } }),
+    blockStop(1),
+    blockSnapshot({ uuid: "uuid-block-1", kind: "text", text: "C" }),
+    MESSAGE_STOP,
+  ]);
+
+  expect(emittedChunks({ items, type: "assistant_message" })).toEqual(["A", "B", "C"]);
+});
+
+// With thinking display "omitted", a thinking block streams an empty delta and a signature, and
+// the SDK sends no snapshot for it because it only snapshots non-empty blocks.
+test("does not hold a snapshot slot for an empty thinking block", async () => {
+  const items = await runScriptedTurn([
+    MESSAGE_START,
+    ...streamedBlock({ index: 0, kind: "thinking", deltas: [""] }),
+    streamFrame({
+      type: "content_block_delta",
+      index: 0,
+      delta: { type: "signature_delta", signature: "sig" },
+    }),
+    blockStop(0),
+    ...streamedBlock({ index: 1, kind: "thinking", deltas: ["Visible."] }),
+    blockSnapshot({ uuid: "uuid-block-1", kind: "thinking", text: "Visible." }),
+    blockStop(1),
+    MESSAGE_STOP,
+  ]);
+
+  expect(emittedChunks({ items, type: "reasoning" })).toEqual(["Visible."]);
+});
+
+test("shows a streamed narration block as assistant text before its question", async () => {
+  const comparison = "Plan B is fastest. Plan C costs least.";
+  const items = await runScriptedTurn([
+    MESSAGE_START,
+    ...streamedBlock({ index: 0, kind: "thinking", deltas: ["Compare cost", " and time."] }),
+    blockSnapshot({ uuid: "reason", kind: "thinking", text: "Compare cost and time." }),
+    blockStop(0),
+    ...streamedBlock({ index: 1, kind: "thinking", deltas: [comparison] }),
+    {
+      ...blockSnapshot({ uuid: "narration", kind: "thinking", text: comparison }),
+      narration_block_indexes: [0],
+    },
+    blockStop(1),
+    {
+      type: "assistant",
+      parent_tool_use_id: null,
+      message: {
+        id: "msg-multi",
+        content: [{ type: "tool_use", id: "ask-plan", name: "AskUserQuestion", input: {} }],
+      },
+    },
+    MESSAGE_STOP,
+  ]);
+
+  expect(emittedChunks({ items, type: "reasoning" }).join("")).toBe("Compare cost and time.");
+  expect(emittedChunks({ items, type: "assistant_message" })).toEqual([comparison]);
+  const comparisonIndex = items.findIndex((item) => item.type === "assistant_message");
+  const questionIndex = items.findIndex((item) => item.type === "tool_call");
+  expect(questionIndex).toBeGreaterThan(comparisonIndex);
+});
+
+test("classifies only the marked thinking block of a snapshot-only message", async () => {
+  const items = await runScriptedTurn([
+    {
+      type: "assistant",
+      uuid: "snapshot-only",
+      parent_tool_use_id: null,
+      narration_block_indexes: [2],
+      message: {
+        id: "msg-multi",
+        content: [
+          { type: "text", text: "Reading plans." },
+          { type: "thinking", thinking: "Compare the options." },
+          { type: "thinking", thinking: "Plan B is fastest." },
+        ],
+      },
+    },
+  ]);
+  expect(emittedChunks({ items, type: "assistant_message" })).toEqual([
+    "Reading plans.",
+    "Plan B is fastest.",
+  ]);
+  expect(emittedChunks({ items, type: "reasoning" })).toEqual(["Compare the options."]);
+});
+
+test("ignores malformed narration markers and retains unmarked reasoning", async () => {
+  for (const marker of [undefined, null, "0", ["0"], [-1], [1.5], [3]]) {
+    const items = await runScriptedTurn([
+      {
+        ...blockSnapshot({ uuid: "unmarked", kind: "thinking", text: "A thought." }),
+        narration_block_indexes: marker,
+      },
+    ]);
+    expect(emittedChunks({ items, type: "reasoning" })).toEqual(["A thought."]);
+    expect(emittedChunks({ items, type: "assistant_message" })).toEqual([]);
+  }
+});
+
+test("retains thinking at message stop when no completed block snapshot arrives", async () => {
+  const items = await runScriptedTurn([
+    MESSAGE_START,
+    ...streamedBlock({ index: 0, kind: "thinking", deltas: ["Partial", " thought."] }),
+    blockStop(0),
+    MESSAGE_STOP,
+  ]);
+  expect(emittedChunks({ items, type: "reasoning" }).join("")).toBe("Partial thought.");
+});
+
+test.each([
+  ["fails", new Error("Provider disconnected")],
+  ["is interrupted", "interrupt" as const],
+])(
+  "retains unfinished thinking when the turn %s without a message stop",
+  async (_name, terminal) => {
+    const items = await runScriptedTurn([
+      MESSAGE_START,
+      ...streamedBlock({ index: 0, kind: "thinking", deltas: ["Partial", " thought."] }),
+      terminal,
+    ]);
+    expect(emittedChunks({ items, type: "reasoning" })).toEqual(["Partial thought."]);
+  },
+);

packages/server/src/server/agent/providers/claude/agent.ts

@@ -1144,14 +1144,29 @@ function buildClaudePlanPermissionActions(
 interface TimelineFragment {
   kind: "assistant" | "reasoning";
   text: string;
+  narration: boolean;
+}
+
+// While streaming, the SDK emits one assistant message per completed non-empty content block,
+// carrying only that block and sent before the block's content_block_stop (see
+// SDKAssistantMessage). Snapshots therefore reconcile against their own block, never against the
+// whole message. Empty blocks get no snapshot, so they are never registered as blocks.
+interface TimelineBlockState {
+  kind: TimelineFragment["kind"];
+  text: string;
+  emittedLength: number;
+  snapshotApplied: boolean;
 }
 
 interface TimelineMessageState {
   id: string;
-  assistantText: string;
-  reasoningText: string;
-  emittedAssistantLength: number;
-  emittedReasoningLength: number;
+  // Insertion order is content order. Streamed blocks are keyed by stream block index.
+  blocks: Map<string, TimelineBlockState>;
+  // The block that deltas without an index continue, until content_block_stop or its snapshot.
+  // The Messages API always sends an index; index-less deltas only come from older fixtures.
+  openStreamBlockKey: string | null;
+  appliedSnapshotUuids: Set<string>;
+  syntheticBlockCount: number;
   stopped: boolean;
 }
 
@@ -1175,6 +1190,11 @@ class TimelineAssembler {
     return [];
   }
 
+  finishRun(runId: string): AgentTimelineItem[] {
+    const messageId = this.activeMessageByRun.get(runId);
+    return messageId ? this.finalizeMessage(messageId, runId) : [];
+  }
+
   private consumeAssistantMessage(
     message: SDKMessage & { type: "assistant" },
     runId: string | null,
@@ -1191,8 +1211,24 @@ class TimelineAssembler {
       return [];
     }
     const state = this.ensureMessageState(messageId, runId);
-    const fragments = this.extractFragments(message.message?.content);
-    return this.applyAbsoluteFragments(state, fragments);
+    const snapshotUuid = readTrimmedString(message.uuid);
+    if (snapshotUuid) {
+      if (state.appliedSnapshotUuids.has(snapshotUuid)) {
+        return [];
+      }
+      state.appliedSnapshotUuids.add(snapshotUuid);
+    }
+    const marker = toObjectRecord(message)?.narration_block_indexes;
+    const narrationIndexes = new Set<number>(
+      Array.isArray(marker)
+        ? marker.filter((index): index is number => Number.isInteger(index) && index >= 0)
+        : [],
+    );
+    const fragments = this.extractFragments({
+      content: message.message?.content,
+      narrationIndexes,
+    });
+    return this.applyBlockSnapshots({ state, fragments });
   }
 
   private consumeStreamEvent(
@@ -1230,22 +1266,47 @@ class TimelineAssembler {
     }
 
     if (eventType === "content_block_start") {
-      return this.consumeDeltaContent(event.content_block, runId, streamEventMessageId);
+      return this.consumeDeltaContent({
+        content: event.content_block,
+        blockIndex: event.index,
+        runId,
+        messageIdHint: streamEventMessageId,
+      });
     }
 
     if (eventType === "content_block_delta") {
-      return this.consumeDeltaContent(event.delta, runId, streamEventMessageId);
+      return this.consumeDeltaContent({
+        content: event.delta,
+        blockIndex: event.index,
+        runId,
+        messageIdHint: streamEventMessageId,
+      });
+    }
+
+    if (eventType === "content_block_stop") {
+      const messageId = this.resolveMessageId({
+        runId,
+        createIfMissing: false,
+        messageId: streamEventMessageId,
+      });
+      const state = messageId ? this.messages.get(messageId) : undefined;
+      if (state) {
+        state.openStreamBlockKey = null;
+      }
+      return [];
     }
 
     return [];
   }
 
-  private consumeDeltaContent(
-    content: unknown,
-    runId: string | null,
-    messageIdHint: string | null,
-  ): AgentTimelineItem[] {
-    const fragments = this.extractFragments(content);
+  private consumeDeltaContent(input: {
+    content: unknown;
+    blockIndex: unknown;
+    runId: string | null;
+    messageIdHint: string | null;
+  }): AgentTimelineItem[] {
+    const { content, blockIndex, runId, messageIdHint } = input;
+    const fragments = this.extractFragments({ content });
     if (fragments.length === 0) {
       return [];
     }
@@ -1258,49 +1319,77 @@ class TimelineAssembler {
       return [];
     }
     const state = this.ensureMessageState(messageId, runId);
-    return this.appendFragments(state, fragments);
-  }
-
-  private appendFragments(
-    state: TimelineMessageState,
-    fragments: TimelineFragment[],
-  ): AgentTimelineItem[] {
     for (const fragment of fragments) {
-      if (fragment.kind === "assistant") {
-        state.assistantText += fragment.text;
-      } else {
-        state.reasoningText += fragment.text;
-      }
+      const key = this.resolveStreamBlockKey({ state, blockIndex, kind: fragment.kind });
+      state.openStreamBlockKey = key;
+      const block = state.blocks.get(key) ?? this.addBlock({ state, key, kind: fragment.kind });
+      block.text += fragment.text;
     }
-    return this.emitNewContent(state);
+    return this.emitNewContent({ state });
   }
 
-  private applyAbsoluteFragments(
-    state: TimelineMessageState,
-    fragments: TimelineFragment[],
-  ): AgentTimelineItem[] {
-    const assistantText = fragments
-      .filter((fragment) => fragment.kind === "assistant")
-      .map((fragment) => fragment.text)
-      .join("");
-    const reasoningText = fragments
-      .filter((fragment) => fragment.kind === "reasoning")
-      .map((fragment) => fragment.text)
-      .join("");
+  private resolveStreamBlockKey(input: {
+    state: TimelineMessageState;
+    blockIndex: unknown;
+    kind: TimelineFragment["kind"];
+  }): string {
+    const { state, blockIndex, kind } = input;
+    if (typeof blockIndex === "number") {
+      return `stream:${blockIndex}`;
+    }
+    const openKey = state.openStreamBlockKey;
+    if (openKey && state.blocks.get(openKey)?.kind === kind) {
+      return openKey;
+    }
+    return `stream:unindexed:${++state.syntheticBlockCount}`;
+  }
 
-    if (assistantText.length > 0) {
-      if (!assistantText.startsWith(state.assistantText)) {
-        state.emittedAssistantLength = 0;
+  private applyBlockSnapshots(input: {
+    state: TimelineMessageState;
+    fragments: TimelineFragment[];
+  }): AgentTimelineItem[] {
+    const { state, fragments } = input;
+    for (const fragment of fragments) {
+      // Snapshots arrive in content order, so each belongs to the earliest block of its kind
+      // that has not had one yet.
+      const match = [...state.blocks.entries()].find(
+        ([, candidate]) => candidate.kind === fragment.kind && !candidate.snapshotApplied,
+      );
+      if (!match) {
+        const key = `snapshot:${++state.syntheticBlockCount}`;
+        const snapshotOnly = this.addBlock({ state, key, kind: fragment.kind });
+        snapshotOnly.kind = fragment.narration ? "assistant" : fragment.kind;
+        snapshotOnly.text = fragment.text;
+        snapshotOnly.snapshotApplied = true;
+        continue;
       }
-      state.assistantText = assistantText;
-    }
-    if (reasoningText.length > 0) {
-      if (!reasoningText.startsWith(state.reasoningText)) {
-        state.emittedReasoningLength = 0;
+      const [key, block] = match;
+      block.snapshotApplied = true;
+      if (fragment.narration) {
+        block.kind = "assistant";
+      }
+      // A snapshot completes its block, so later deltas without an index start a new one.
+      if (state.openStreamBlockKey === key) {
+        state.openStreamBlockKey = null;
+      }
+      // Buffered thinking has not been emitted, so its completed text is authoritative.
+      // Plain text has streamed already and can only be extended.
+      if (block.emittedLength === 0 || fragment.text.startsWith(block.text)) {
+        block.text = fragment.text;
       }
-      state.reasoningText = reasoningText;
     }
-    return this.emitNewContent(state);
+    return this.emitNewContent({ state });
+  }
+
+  private addBlock(input: {
+    state: TimelineMessageState;
+    key: string;
+    kind: TimelineFragment["kind"];
+  }): TimelineBlockState {
+    const { state, key, kind } = input;
+    const block: TimelineBlockState = { kind, text: "", emittedLength: 0, snapshotApplied: false };
+    state.blocks.set(key, block);
+    return block;
   }
 
   private finalizeMessage(messageId: string, runId: string | null): AgentTimelineItem[] {
@@ -1309,7 +1398,7 @@ class TimelineAssembler {
       return [];
     }
     state.stopped = true;
-    const items = this.emitNewContent(state);
+    const items = this.emitNewContent({ state, finalizing: true });
     if (runId && this.activeMessageByRun.get(runId) === messageId) {
       this.activeMessageByRun.delete(runId);
     }
@@ -1318,22 +1407,32 @@ class TimelineAssembler {
     return items;
   }
 
-  private emitNewContent(state: TimelineMessageState): AgentTimelineItem[] {
+  private emitNewContent(input: {
+    state: TimelineMessageState;
+    finalizing?: boolean;
+  }): AgentTimelineItem[] {
+    const { state, finalizing = false } = input;
     const items: AgentTimelineItem[] = [];
-    const nextAssistantText = state.assistantText.slice(state.emittedAssistantLength);
-    if (
-      nextAssistantText.length > 0 &&
-      nextAssistantText !== INTERRUPT_TOOL_USE_PLACEHOLDER &&
-      !isClaudeTranscriptNoiseText(nextAssistantText)
-    ) {
-      state.emittedAssistantLength = state.assistantText.length;
-      items.push({ type: "assistant_message", text: nextAssistantText, messageId: state.id });
-    }
-
-    const nextReasoningText = state.reasoningText.slice(state.emittedReasoningLength);
-    if (nextReasoningText.length > 0) {
-      state.emittedReasoningLength = state.reasoningText.length;
-      items.push({ type: "reasoning", text: nextReasoningText });
+    for (const block of state.blocks.values()) {
+      // Narration is identified only by the completed snapshot. Hold thinking until then,
+      // so a progress update never reaches clients as reasoning. Plain text still streams.
+      if (block.kind === "reasoning" && !block.snapshotApplied && !finalizing) {
+        continue;
+      }
+      const nextText = block.text.slice(block.emittedLength);
+      if (nextText.length === 0) {
+        continue;
+      }
+      if (block.kind === "reasoning") {
+        block.emittedLength = block.text.length;
+        items.push({ type: "reasoning", text: nextText });
+        continue;
+      }
+      if (nextText === INTERRUPT_TOOL_USE_PLACEHOLDER || isClaudeTranscriptNoiseText(nextText)) {
+        continue;
+      }
+      block.emittedLength = block.text.length;
+      items.push({ type: "assistant_message", text: nextText, messageId: state.id });
     }
     return items;
   }
@@ -1349,10 +1448,10 @@ class TimelineAssembler {
     }
     const created: TimelineMessageState = {
       id: messageId,
-      assistantText: "",
-      reasoningText: "",
-      emittedAssistantLength: 0,
-      emittedReasoningLength: 0,
+      blocks: new Map(),
+      appliedSnapshotUuids: new Set(),
+      openStreamBlockKey: null,
+      syntheticBlockCount: 0,
       stopped: false,
     };
     this.messages.set(messageId, created);
@@ -1386,16 +1485,20 @@ class TimelineAssembler {
     return synthetic;
   }
 
-  private extractFragments(content: unknown): TimelineFragment[] {
+  private extractFragments(input: {
+    content: unknown;
+    narrationIndexes?: ReadonlySet<number>;
+  }): TimelineFragment[] {
+    const { content, narrationIndexes = new Set<number>() } = input;
     if (typeof content === "string") {
       if (content.length === 0) {
         return [];
       }
-      return [{ kind: "assistant", text: content }];
+      return [{ kind: "assistant", text: content, narration: false }];
     }
     const blocks = Array.isArray(content) ? content : [content];
     const fragments: TimelineFragment[] = [];
-    for (const rawBlock of blocks) {
+    for (const [index, rawBlock] of blocks.entries()) {
       if (!isClaudeContentChunk(rawBlock)) {
         continue;
       }
@@ -1404,14 +1507,18 @@ class TimelineAssembler {
         typeof rawBlock.text === "string" &&
         rawBlock.text.length > 0
       ) {
-        fragments.push({ kind: "assistant", text: rawBlock.text });
+        fragments.push({ kind: "assistant", text: rawBlock.text, narration: false });
       }
       if (
         (rawBlock.type === "thinking" || rawBlock.type === "thinking_delta") &&
         typeof rawBlock.thinking === "string" &&
         rawBlock.thinking.length > 0
       ) {
-        fragments.push({ kind: "reasoning", text: rawBlock.thinking });
+        fragments.push({
+          kind: "reasoning",
+          text: rawBlock.thinking,
+          narration: narrationIndexes.has(index),
+        });
       }
     }
     return fragments;
@@ -5057,6 +5164,11 @@ class ClaudeAgentSession implements AgentSession {
 
   private notifySubscribers(event: AgentStreamEvent): void {
     const turnId = this.activeForegroundTurnId ?? this.autonomousTurn?.id;
+    if (turnId && this.isTerminalTurnEvent(event)) {
+      for (const item of this.timelineAssembler.finishRun(turnId)) {
+        this.notifySubscribers({ type: "timeline", provider: "claude", item });
+      }
+    }
     const tagged = turnId ? { ...event, turnId } : event;
     this.logger.trace(
       {