fix(server): reject attachment starts closed during conversion

Keep interruption terminal for prompts still being materialized, and reject a start if its session closes before provider submission. Cover both providers with real large-image conversions and make history reconciliation tests await the async timeline capability.

packages/server/package.json

@@ -89,7 +89,7 @@
     "speech:transcribe:local": "tsx scripts/transcribe-local-wav.ts",
     "test": "npm run test:unit && npm run test:integration",
     "test:unit": "vitest run --fileParallelism --exclude \"**/*.e2e.test.ts\"",
-    "test:integration": "vitest run --maxWorkers=1 src/server/git-filename-diffs.e2e.test.ts src/server/daemon-e2e/checkout-diff-content.e2e.test.ts src/server/daemon-e2e/filesystem.e2e.test.ts src/server/daemon-e2e/file-download.e2e.test.ts src/server/daemon-e2e/cli-logs-follow.e2e.test.ts src/server/creation/creation.e2e.test.ts src/server/daemon-e2e/terminal-workspace-sdk.e2e.test.ts src/server/workspace-create-worktree-source.e2e.test.ts src/server/daemon-e2e/models.e2e.test.ts src/server/daemon-e2e/live-preferences.e2e.test.ts src/server/daemon-e2e/oversized-message.e2e.test.ts src/server/agent/model-catalog.e2e.test.ts src/server/plugins/plugin-paseo-api.e2e.test.ts src/server/connected-hosts/hosts.e2e.test.ts && vitest run --maxWorkers=1 src/server/agent/providers/omp/history.e2e.test.ts src/server/daemon-e2e/git-operations.e2e.test.ts src/server/session.create-agent-worktree-autoarchive.e2e.test.ts src/server/agent/providers/opencode-bridge.local.e2e.test.ts src/server/daemon-e2e/relay-transport.e2e.test.ts -t \"merges into|creates a worktree and auto-archives both|versioned runtime|oversized attachments|connected hosts|password-protected daemon refuses|replays selected OMP history\"",
+    "test:integration": "vitest run --maxWorkers=1 src/server/git-filename-diffs.e2e.test.ts src/server/daemon-e2e/checkout-diff-content.e2e.test.ts src/server/daemon-e2e/filesystem.e2e.test.ts src/server/daemon-e2e/file-download.e2e.test.ts src/server/daemon-e2e/cli-logs-follow.e2e.test.ts src/server/creation/creation.e2e.test.ts src/server/daemon-e2e/terminal-workspace-sdk.e2e.test.ts src/server/workspace-create-worktree-source.e2e.test.ts src/server/daemon-e2e/models.e2e.test.ts src/server/daemon-e2e/live-preferences.e2e.test.ts src/server/daemon-e2e/oversized-message.e2e.test.ts src/server/agent/model-catalog.e2e.test.ts src/server/plugins/plugin-paseo-api.e2e.test.ts src/server/connected-hosts/hosts.e2e.test.ts && vitest run --maxWorkers=1 src/server/daemon-e2e/git-operations.e2e.test.ts src/server/session.create-agent-worktree-autoarchive.e2e.test.ts src/server/agent/providers/opencode-bridge.local.e2e.test.ts src/server/daemon-e2e/relay-transport.e2e.test.ts -t \"merges into|creates a worktree and auto-archives both|versioned runtime|oversized attachments|connected hosts|password-protected daemon refuses\"",
     "test:hub-cli-contract": "vitest run --maxWorkers=1 src/server/hub/hub-cli-contract.test.ts",
     "test:integration:all": "npm run test:e2e",
     "test:integration:real": "vitest run real.e2e.test.ts",

packages/server/src/server/agent/providers/acp-agent.test.ts

@@ -811,6 +811,46 @@ describe("ACPAgentSession terminal tools", () => {
       truncated: false,
     });
   });
+
+  test("trims large terminal output without rescanning the retained suffix", async () => {
+    const session = createSession();
+    const terminal = await session.createTerminal({
+      sessionId: "session-1",
+      command: process.execPath,
+      args: ["-e", 'process.stdout.write("x".repeat(1024 * 1024))'],
+      cwd: process.cwd(),
+      outputByteLimit: 1024,
+    });
+    await session.waitForTerminalExit({ sessionId: "session-1", terminalId: terminal.terminalId });
+
+    await expect(
+      session.terminalOutput({ sessionId: "session-1", terminalId: terminal.terminalId }),
+    ).resolves.toEqual({
+      output: "x".repeat(1024),
+      truncated: true,
+      exitStatus: { exitCode: 0, signal: null },
+    });
+  }, 1000);
+
+  test("trims terminal output at a UTF-8 character boundary", async () => {
+    const session = createSession();
+    const terminal = await session.createTerminal({
+      sessionId: "session-1",
+      command: process.execPath,
+      args: ["-e", 'process.stdout.write("😀ab")'],
+      cwd: process.cwd(),
+      outputByteLimit: 3,
+    });
+    await session.waitForTerminalExit({ sessionId: "session-1", terminalId: terminal.terminalId });
+
+    await expect(
+      session.terminalOutput({ sessionId: "session-1", terminalId: terminal.terminalId }),
+    ).resolves.toEqual({
+      output: "ab",
+      truncated: true,
+      exitStatus: { exitCode: 0, signal: null },
+    });
+  });
 });
 
 describe("mapACPUsage", () => {

packages/server/src/server/agent/providers/acp-agent.ts

@@ -4093,10 +4093,16 @@ function appendTerminalOutput(entry: TerminalEntry, chunk: string): void {
   if (!limit) {
     return;
   }
-  while (Buffer.byteLength(entry.output, "utf8") > limit && entry.output.length > 0) {
-    entry.output = entry.output.slice(1);
-    entry.truncated = true;
+  const output = Buffer.from(entry.output, "utf8");
+  if (output.length <= limit) {
+    return;
   }
+  let start = output.length - limit;
+  while (start < output.length && (output[start] & 0xc0) === 0x80) {
+    start += 1;
+  }
+  entry.output = output.subarray(start).toString("utf8");
+  entry.truncated = true;
 }
 
 function readRecord(value: unknown): Record<string, unknown> | null {

packages/server/src/server/agent/providers/claude/agent.history-load.test.ts

@@ -1,9 +1,13 @@
 import fs, { chmodSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
+import { open } from "node:fs/promises";
+import { performance } from "node:perf_hooks";
 import * as os from "node:os";
 import * as path from "node:path";
 import pino from "pino";
 import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
 
+import { createTestPaseoDaemon } from "../../../test-utils/paseo-daemon.js";
+import type { ClaudeRewindSdk } from "./rewind.js";
 import { ClaudeAgentClient } from "./agent.js";
 import { claudeProjectDirSync } from "./project-dir.js";
 import { withClaudeReplayHistory } from "./transcript-history.js";
@@ -32,7 +36,10 @@ describe("ClaudeAgentSession persisted history load", () => {
   let configDir: string;
   let records: LogRecord[];
 
-  async function resume(queryFactory?: ClaudeQueryFactory): Promise<AgentSession> {
+  async function resume(
+    queryFactory?: ClaudeQueryFactory,
+    rewindSdk?: ClaudeRewindSdk,
+  ): Promise<AgentSession> {
     const client = new ClaudeAgentClient({
       logger: pino({ level: "info" }, { write: (line: string) => records.push(JSON.parse(line)) }),
       queryFactory:
@@ -42,10 +49,36 @@ describe("ClaudeAgentSession persisted history load", () => {
         }),
       resolveBinary: async () => process.execPath,
       resolveVersion: async () => "2.1.220",
+      rewindSdk,
     });
     return client.resumeSession({ provider: "claude", sessionId: SESSION_ID }, { cwd });
   }
 
+  function replyFactory(text: string, messageId = "live-reply"): ClaudeQueryFactory {
+    return (() => {
+      const messages = [
+        { type: "assistant", uuid: messageId, message: { content: text } },
+        { type: "result", subtype: "success", total_cost_usd: 0 },
+      ];
+      return {
+        async next() {
+          const value = messages.shift();
+          return value ? { done: false, value } : { done: true, value: undefined };
+        },
+        close() {},
+        async return() {
+          return { done: true, value: undefined };
+        },
+        async supportedCommands() {
+          return [];
+        },
+        [Symbol.asyncIterator]() {
+          return this;
+        },
+      };
+    }) as ClaudeQueryFactory;
+  }
+
   function transcriptPath(): string {
     return path.join(claudeProjectDirSync(cwd, { configDir }), `${SESSION_ID}.jsonl`);
   }
@@ -65,6 +98,286 @@ describe("ClaudeAgentSession persisted history load", () => {
     rmSync(tempRoot, { recursive: true, force: true });
   });
 
+  test("serves unrelated health requests while resuming 32 MiB of ordered history", async () => {
+    const startedAt = performance.now();
+    mkdirSync(path.dirname(transcriptPath()), { recursive: true });
+    const file = await open(transcriptPath(), "w");
+    const text = "x".repeat(16 * 1024);
+    let bytes = 0;
+    let count = 0;
+    try {
+      while (bytes < 32 * 1024 * 1024) {
+        const line =
+          JSON.stringify({ type: "assistant", message: { content: `${count}:${text}` } }) + "\n";
+        bytes += Buffer.byteLength(line);
+        await file.write(line);
+        count++;
+      }
+    } finally {
+      await file.close();
+    }
+    const daemon = await createTestPaseoDaemon();
+    let lastTick = performance.now();
+    let longestGap = 0;
+    const probe = setInterval(() => {
+      const now = performance.now();
+      longestGap = Math.max(longestGap, now - lastTick);
+      lastTick = now;
+    }, 2);
+    let loading = true;
+    let answeredDuringLoad = false;
+    const health = fetch(`http://127.0.0.1:${daemon.port}/api/health`).then(async (response) => {
+      expect(response.status).toBe(200);
+      await response.text();
+      answeredDuringLoad = loading;
+      return undefined;
+    });
+    let session: AgentSession | undefined;
+    try {
+      session = await resume(replyFactory("live after large resume"));
+      let index = 0;
+      for await (const event of session.streamHistory()) {
+        expect(event).toEqual({
+          type: "timeline",
+          provider: "claude",
+          timestamp: undefined,
+          item: { type: "assistant_message", text: `${index}:${text}` },
+        });
+        index++;
+      }
+      loading = false;
+      await health;
+      await new Promise((resolve) => setTimeout(resolve, 5));
+      console.info(
+        JSON.stringify({
+          bytes,
+          count,
+          longestGap,
+          answeredDuringLoad,
+          elapsedMs: performance.now() - startedAt,
+        }),
+      );
+      expect(index).toBe(count);
+      expect(longestGap).toBeLessThan(50);
+      expect(answeredDuringLoad).toBe(true);
+      const live = await collectSessionTurnEvents(session, "follow up");
+      expect(
+        live
+          .filter((event) => event.type === "timeline" && event.item.type === "assistant_message")
+          .map((event) => (event.type === "timeline" ? event.item : null)),
+      ).toEqual([
+        { type: "assistant_message", messageId: "live-reply", text: "live after large resume" },
+      ]);
+    } finally {
+      clearInterval(probe);
+      await session?.close();
+      await daemon.close();
+    }
+  }, 60_000);
+
+  test("rebinds ordered history before returning from rewind and accepts a live turn", async () => {
+    const entries = [
+      { type: "user", uuid: "u1", message: { content: "first" } },
+      { type: "assistant", uuid: "a1", message: { content: "first reply" } },
+      { type: "user", uuid: "u2", message: { content: "second" } },
+      { type: "assistant", uuid: "a2", message: { content: "second reply" } },
+    ];
+    mkdirSync(path.dirname(transcriptPath()), { recursive: true });
+    writeFileSync(transcriptPath(), entries.map((entry) => JSON.stringify(entry)).join("\n"));
+    const forkId = "rebound-history";
+    const forkFile = path.join(path.dirname(transcriptPath()), `${forkId}.jsonl`);
+    writeFileSync(
+      forkFile,
+      entries
+        .slice(0, 2)
+        .map((entry) => JSON.stringify(entry))
+        .join("\n"),
+    );
+    const forks: unknown[] = [];
+    const late = {
+      type: "assistant",
+      uuid: "retired-reply",
+      message: { content: "retired query" },
+    };
+    let pendingRead: ((value: unknown) => void) | undefined;
+    let releaseRetirement!: () => void;
+    const retirementGate = new Promise<void>((resolve) => {
+      releaseRetirement = resolve;
+    });
+    let retirementStarted!: () => void;
+    const retiring = new Promise<void>((resolve) => {
+      retirementStarted = resolve;
+    });
+    let firstQuery = true;
+    let retired = false;
+    const queryFactory = (() => {
+      if (!firstQuery) return replyFactory("live after rebind")({} as never);
+      firstQuery = false;
+      const messages = [
+        { type: "assistant", uuid: "before-rebind", message: { content: "before rebind" } },
+        { type: "result", subtype: "success", total_cost_usd: 0 },
+      ];
+      return {
+        async next() {
+          const value = messages.shift();
+          if (value) return { done: false, value };
+          if (retired) return { done: true, value: undefined };
+          return new Promise((resolve) => {
+            pendingRead = resolve;
+          });
+        },
+        close() {
+          retired = true;
+          pendingRead?.({ done: false, value: late });
+        },
+        async return() {
+          retirementStarted();
+          await retirementGate;
+          return { done: true, value: undefined };
+        },
+        async supportedCommands() {
+          return [];
+        },
+        [Symbol.asyncIterator]() {
+          return this;
+        },
+      };
+    }) as ClaudeQueryFactory;
+    const session = await resume(queryFactory, {
+      async forkSession(sessionId, options) {
+        forks.push({ sessionId, ...options });
+        return { sessionId: forkId };
+      },
+    });
+    try {
+      const original = await Array.fromAsync(session.streamHistory());
+      expect(original).toHaveLength(4);
+      await collectSessionTurnEvents(session, "before rewind");
+      const emitted: AgentStreamEvent[] = [];
+      const subscription = session.subscribe((event) => emitted.push(event));
+      let rewindDone = false;
+      const rewind = (async () => {
+        await session.revertConversation!({ messageId: "u2" });
+        rewindDone = true;
+      })();
+      await retiring;
+      let historyDone = false;
+      const reading = (async () => {
+        const history = await Array.fromAsync(session.streamHistory());
+        historyDone = true;
+        return history;
+      })();
+      const readingAgain = Array.fromAsync(session.streamHistory());
+      await Promise.resolve();
+      expect(rewindDone).toBe(false);
+      expect(historyDone).toBe(false);
+      releaseRetirement();
+      await rewind;
+      subscription();
+      expect(retired).toBe(true);
+      expect(
+        emitted.filter(
+          (event) => event.type === "timeline" && event.item.type === "assistant_message",
+        ),
+      ).toEqual([
+        {
+          type: "timeline",
+          provider: "claude",
+          item: {
+            type: "assistant_message",
+            text: `Claude switched to a new session: ${SESSION_ID} -> ${forkId}`,
+          },
+        },
+      ]);
+      expect(forks).toEqual([
+        { sessionId: SESSION_ID, upToMessageId: "a1", transcriptPath: transcriptPath() },
+      ]);
+      expect(session.id).toBe(forkId);
+      const [history, secondRead] = await Promise.all([reading, readingAgain]);
+      expect(history).toEqual(original.slice(0, 2));
+      expect(secondRead).toEqual([]);
+      const live = await collectSessionTurnEvents(session, "continue the fork");
+      expect(
+        live
+          .filter((event) => event.type === "timeline" && event.item.type === "assistant_message")
+          .map((event) => (event.type === "timeline" ? event.item : null)),
+      ).toEqual([
+        { type: "assistant_message", messageId: "live-reply", text: "live after rebind" },
+      ]);
+    } finally {
+      releaseRetirement();
+      await session.close();
+    }
+  });
+
+  test("keeps history and live turns available after query retirement fails during rebind", async () => {
+    mkdirSync(path.dirname(transcriptPath()), { recursive: true });
+    writeFileSync(
+      transcriptPath(),
+      [
+        { type: "user", uuid: "u1", message: { content: "first" } },
+        { type: "assistant", uuid: "a1", message: { content: "reply" } },
+        { type: "user", uuid: "u2", message: { content: "second" } },
+      ]
+        .map((entry) => JSON.stringify(entry))
+        .join("\n"),
+    );
+    let firstQuery = true;
+    let finishRead: ((value: unknown) => void) | undefined;
+    const queryFactory = ((input) => {
+      const query = replyFactory(
+        firstQuery ? "before retirement" : "still available",
+        firstQuery ? "before-retirement" : "live-reply",
+      )(input);
+      if (!firstQuery) return query;
+      firstQuery = false;
+      let reads = 0;
+      return {
+        ...query,
+        async next() {
+          if (++reads <= 2) return query.next();
+          return new Promise((resolve) => {
+            finishRead = resolve;
+          });
+        },
+        close() {
+          finishRead?.({ done: true, value: undefined });
+          throw new Error("retirement failed");
+        },
+        [Symbol.asyncIterator]() {
+          return this;
+        },
+      };
+    }) as ClaudeQueryFactory;
+    const session = await resume(queryFactory, {
+      async forkSession() {
+        return { sessionId: "failed-rebind" };
+      },
+    });
+    try {
+      await Array.fromAsync(session.streamHistory());
+      await collectSessionTurnEvents(session, "before rewind");
+      await expect(session.revertConversation!({ messageId: "u2" })).resolves.toBeUndefined();
+      expect(records).toContainEqual(
+        expect.objectContaining({
+          level: pino.levels.values.warn,
+          msg: "Failed to rebind Claude session",
+          sessionId: "failed-rebind",
+          err: expect.objectContaining({ message: "retirement failed" }),
+        }),
+      );
+      expect(await Array.fromAsync(session.streamHistory())).toEqual([]);
+      const live = await collectSessionTurnEvents(session, "after failed rebind");
+      expect(
+        live
+          .filter((event) => event.type === "timeline" && event.item.type === "assistant_message")
+          .map((event) => (event.type === "timeline" ? event.item : null)),
+      ).toEqual([{ type: "assistant_message", messageId: "live-reply", text: "still available" }]);
+    } finally {
+      await session.close();
+    }
+  });
+
   test.each(["file_path", "files"])(
     "replays %s without probing targets and preserves live existence checks",
     async (field) => {
@@ -300,21 +613,21 @@ describe("ClaudeAgentSession persisted history load", () => {
     );
   });
 
-  test("loads readable history when the temporary directory is unavailable", () => {
+  test("loads readable history when the temporary directory is unavailable", async () => {
     const file = transcriptPath();
     const entry = { type: "assistant", message: { content: [{ type: "text", text: "saved" }] } };
     mkdirSync(path.dirname(file), { recursive: true });
     writeFileSync(file, JSON.stringify(entry));
     const unavailable = path.join(tempRoot, "unavailable-tmp");
     for (const variable of ["TMPDIR", "TEMP", "TMP"]) vi.stubEnv(variable, unavailable);
-    withClaudeReplayHistory(file, (history) => {
-      expect([...history.parentEntries]).toEqual([entry]);
+    await withClaudeReplayHistory(file, async (history) => {
+      expect(await Array.fromAsync(history.parentEntries)).toEqual([entry]);
     });
   });
 
   test.each(["parent", "child", "workflow"])(
     "preserves indexed %s history when its transcript disappears between passes",
-    (target) => {
+    async (target) => {
       const parent = transcriptPath();
       const sessionDirectory = parent.slice(0, -".jsonl".length);
       const child = path.join(sessionDirectory, "subagents", "agent-child.jsonl");
@@ -334,14 +647,16 @@ describe("ClaudeAgentSession persisted history load", () => {
       writeFileSync(parent, JSON.stringify(parentEntry));
       writeFileSync(child, JSON.stringify(childEntry));
       writeFileSync(workflow, JSON.stringify(childEntry));
-      withClaudeReplayHistory(parent, (history) => {
+      await withClaudeReplayHistory(parent, async (history) => {
         rmSync({ parent, child, workflow }[target]);
         // Ownership facts and visible payloads are separate passes over the same source.
         for (let pass = 0; pass < 2; pass++) {
-          expect([...history.parentEntries]).toEqual([parentEntry]);
+          expect(await Array.fromAsync(history.parentEntries)).toEqual([parentEntry]);
           expect(history.subagents).toHaveLength(1);
-          expect([...history.subagents[0]!.entries]).toEqual([childEntry]);
-          expect([...history.workflowEntriesByRunId.get("wf_test")!]).toEqual([childEntry]);
+          expect(await Array.fromAsync(history.subagents[0]!.entries)).toEqual([childEntry]);
+          expect(await Array.fromAsync(history.workflowEntriesByRunId.get("wf_test")!)).toEqual([
+            childEntry,
+          ]);
         }
       });
     },

packages/server/src/server/agent/providers/claude/agent.image-rendering.test.ts

@@ -13,8 +13,8 @@ const MATERIALIZED_PNG_PATH_PATTERN =
   /paseo-attachments(?:-[^\\/]+)?[\\/](?:[^\\/]+[\\/])?[0-9a-f]{64}\.png$/;
 
 interface ClaudeImageTestSession {
-  translateMessageToEvents(message: SDKMessage): AgentStreamEvent[];
-  convertHistoryEntry(entry: unknown): AgentTimelineItem[];
+  translateMessageToEvents(message: SDKMessage): Promise<AgentStreamEvent[]>;
+  convertHistoryEntry(entry: unknown): Promise<AgentTimelineItem[]>;
 }
 
 async function createSession(): Promise<ClaudeImageTestSession> {
@@ -157,7 +157,7 @@ describe("Claude tool_result image rendering", () => {
   test("emits the image as assistant markdown and keeps base64 out of the live tool output", async () => {
     const session = await createSession();
 
-    const events = session.translateMessageToEvents(imageToolResultUserMessage());
+    const events = await session.translateMessageToEvents(imageToolResultUserMessage());
 
     const timelineItems = events
       .filter((event) => event.type === "timeline")
@@ -181,7 +181,7 @@ describe("Claude tool_result image rendering", () => {
   test("replays the image as assistant markdown through history conversion", async () => {
     const session = await createSession();
 
-    const items = session.convertHistoryEntry(imageToolResultHistoryEntry());
+    const items = await session.convertHistoryEntry(imageToolResultHistoryEntry());
 
     const [imageMessage, ...extraImages] = imageMessages(items);
     expect(extraImages).toEqual([]);
@@ -202,7 +202,7 @@ describe("Claude tool_result image rendering", () => {
   test("keeps base64 out of an errored tool_result that carries an image", async () => {
     const session = await createSession();
 
-    const events = session.translateMessageToEvents(erroredImageToolResultUserMessage());
+    const events = await session.translateMessageToEvents(erroredImageToolResultUserMessage());
 
     const timelineItems = events
       .filter((event) => event.type === "timeline")
@@ -226,7 +226,7 @@ describe("Claude tool_result image rendering", () => {
   test("emits one image message per image block in a multi-image tool_result", async () => {
     const session = await createSession();
 
-    const events = session.translateMessageToEvents(multiImageToolResultUserMessage());
+    const events = await session.translateMessageToEvents(multiImageToolResultUserMessage());
 
     const timelineItems = events
       .filter((event) => event.type === "timeline")

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

@@ -32,7 +32,7 @@ import type { AgentAttachment } from "@getpaseo/protocol/messages";
 import { buildAgentPrompt, renderPromptAttachmentAsText } from "../../prompt-attachments.js";
 
 interface TestClaudeSession {
-  translateMessageToEvents(message: SDKMessage): AgentStreamEvent[];
+  translateMessageToEvents(message: SDKMessage): Promise<AgentStreamEvent[]>;
   close(): Promise<void>;
 }
 
@@ -53,7 +53,7 @@ afterEach(() => {
 });
 
 describe("convertClaudeHistoryEntry", () => {
-  test("maps user tool results to timeline items", () => {
+  test("maps user tool results to timeline items", async () => {
     const toolUseId = "toolu_test";
     const entry = {
       type: "user",
@@ -79,14 +79,14 @@ describe("convertClaudeHistoryEntry", () => {
     ];
 
     const mapBlocks = vi.fn().mockReturnValue(stubTimeline);
-    const result = convertClaudeHistoryEntry(entry, mapBlocks);
+    const result = await convertClaudeHistoryEntry(entry, mapBlocks);
 
     expect(result).toEqual(stubTimeline);
     expect(mapBlocks).toHaveBeenCalledTimes(1);
     expect(Array.isArray(mapBlocks.mock.calls[0][0])).toBe(true);
   });
 
-  test("replays persisted Claude tool results as completed tool calls", () => {
+  test("replays persisted Claude tool results as completed tool calls", async () => {
     const entry = {
       type: "user",
       message: {
@@ -124,12 +124,12 @@ describe("convertClaudeHistoryEntry", () => {
 
     const mapPersistedToolResultBlocks = (): AgentTimelineItem[] => completedToolCall;
 
-    expect(convertClaudeHistoryEntry(entry, mapPersistedToolResultBlocks)).toEqual(
+    expect(await convertClaudeHistoryEntry(entry, mapPersistedToolResultBlocks)).toEqual(
       completedToolCall,
     );
   });
 
-  test("returns user messages when no tool blocks exist", () => {
+  test("returns user messages when no tool blocks exist", async () => {
     const entry = {
       type: "user",
       message: {
@@ -138,7 +138,7 @@ describe("convertClaudeHistoryEntry", () => {
       },
     };
 
-    expect(convertClaudeHistoryEntry(entry, () => [])).toEqual([
+    expect(await convertClaudeHistoryEntry(entry, () => [])).toEqual([
       {
         type: "user_message",
         text: "Run npm test",
@@ -146,7 +146,7 @@ describe("convertClaudeHistoryEntry", () => {
     ]);
   });
 
-  test("converts compact boundary metadata variants", () => {
+  test("converts compact boundary metadata variants", async () => {
     const fixtures = [
       {
         entry: {
@@ -175,7 +175,7 @@ describe("convertClaudeHistoryEntry", () => {
     ] as const;
 
     for (const fixture of fixtures) {
-      expect(convertClaudeHistoryEntry(fixture.entry, () => [])).toEqual([
+      expect(await convertClaudeHistoryEntry(fixture.entry, () => [])).toEqual([
         {
           type: "compaction",
           status: "completed",
@@ -186,7 +186,7 @@ describe("convertClaudeHistoryEntry", () => {
     }
   });
 
-  test("skips synthetic user entries", () => {
+  test("skips synthetic user entries", async () => {
     const entry = {
       type: "user",
       isSynthetic: true,
@@ -197,13 +197,13 @@ describe("convertClaudeHistoryEntry", () => {
     };
 
     const mapBlocks = vi.fn().mockReturnValue([]);
-    const result = convertClaudeHistoryEntry(entry, mapBlocks);
+    const result = await convertClaudeHistoryEntry(entry, mapBlocks);
 
     expect(result).toEqual([]);
     expect(mapBlocks).not.toHaveBeenCalled();
   });
 
-  test("skips meta user entries from Claude skill loading", () => {
+  test("skips meta user entries from Claude skill loading", async () => {
     const entry = {
       type: "user",
       isMeta: true,
@@ -219,13 +219,13 @@ describe("convertClaudeHistoryEntry", () => {
     };
 
     const mapBlocks = vi.fn().mockReturnValue([]);
-    const result = convertClaudeHistoryEntry(entry, mapBlocks);
+    const result = await convertClaudeHistoryEntry(entry, mapBlocks);
 
     expect(result).toEqual([]);
     expect(mapBlocks).not.toHaveBeenCalled();
   });
 
-  test("skips interrupt placeholder transcript noise", () => {
+  test("skips interrupt placeholder transcript noise", async () => {
     const interruptEntry = {
       type: "user",
       message: {
@@ -246,11 +246,11 @@ describe("convertClaudeHistoryEntry", () => {
       .fn()
       .mockReturnValue([{ type: "assistant_message", text: "No response requested." }]);
 
-    expect(convertClaudeHistoryEntry(interruptEntry, mapBlocks)).toEqual([]);
-    expect(convertClaudeHistoryEntry(assistantNoiseEntry, mapBlocks)).toEqual([]);
+    expect(await convertClaudeHistoryEntry(interruptEntry, mapBlocks)).toEqual([]);
+    expect(await convertClaudeHistoryEntry(assistantNoiseEntry, mapBlocks)).toEqual([]);
   });
 
-  test("skips <local-command-stdout> messages (model switch, /context, etc.)", () => {
+  test("skips <local-command-stdout> messages (model switch, /context, etc.)", async () => {
     // Real entries from Claude Code JSONL history files
     const modelSwitch = {
       type: "user",
@@ -305,24 +305,24 @@ describe("convertClaudeHistoryEntry", () => {
 
     const mapBlocks = vi.fn().mockReturnValue([]);
 
-    expect(convertClaudeHistoryEntry(modelSwitch, mapBlocks)).toEqual([]);
-    expect(convertClaudeHistoryEntry(modelSwitchWithAnsi, mapBlocks)).toEqual([]);
-    expect(convertClaudeHistoryEntry(contextDump, mapBlocks)).toEqual([]);
-    expect(convertClaudeHistoryEntry(planMode, mapBlocks)).toEqual([]);
-    expect(convertClaudeHistoryEntry(goodbye, mapBlocks)).toEqual([]);
-    expect(convertClaudeHistoryEntry(empty, mapBlocks)).toEqual([]);
+    expect(await convertClaudeHistoryEntry(modelSwitch, mapBlocks)).toEqual([]);
+    expect(await convertClaudeHistoryEntry(modelSwitchWithAnsi, mapBlocks)).toEqual([]);
+    expect(await convertClaudeHistoryEntry(contextDump, mapBlocks)).toEqual([]);
+    expect(await convertClaudeHistoryEntry(planMode, mapBlocks)).toEqual([]);
+    expect(await convertClaudeHistoryEntry(goodbye, mapBlocks)).toEqual([]);
+    expect(await convertClaudeHistoryEntry(empty, mapBlocks)).toEqual([]);
 
     // Real user messages must NOT be filtered
     const realMessage = {
       type: "user",
       message: { role: "user", content: "fix the bug in auth.ts" },
     };
-    expect(convertClaudeHistoryEntry(realMessage, mapBlocks)).toEqual([
+    expect(await convertClaudeHistoryEntry(realMessage, mapBlocks)).toEqual([
       { type: "user_message", text: "fix the bug in auth.ts" },
     ]);
   });
 
-  test("maps task notifications to synthetic tool calls", () => {
+  test("maps task notifications to synthetic tool calls", async () => {
     const entry = {
       type: "system",
       subtype: "task_notification",
@@ -333,7 +333,7 @@ describe("convertClaudeHistoryEntry", () => {
       output_file: "/tmp/bg-fail-1.txt",
     };
 
-    expect(convertClaudeHistoryEntry(entry, () => [])).toEqual([
+    expect(await convertClaudeHistoryEntry(entry, () => [])).toEqual([
       {
         type: "tool_call",
         callId: "task_notification_task-note-system-1",
@@ -357,7 +357,7 @@ describe("convertClaudeHistoryEntry", () => {
     ]);
   });
 
-  test("maps queue-operation task notifications to synthetic tool calls", () => {
+  test("maps queue-operation task notifications to synthetic tool calls", async () => {
     const entry = {
       type: "queue-operation",
       operation: "enqueue",
@@ -372,7 +372,7 @@ describe("convertClaudeHistoryEntry", () => {
       ].join("\n"),
     };
 
-    expect(convertClaudeHistoryEntry(entry, () => [])).toEqual([
+    expect(await convertClaudeHistoryEntry(entry, () => [])).toEqual([
       {
         type: "tool_call",
         callId: "task_notification_task-note-queue-1",
@@ -396,7 +396,7 @@ describe("convertClaudeHistoryEntry", () => {
     ]);
   });
 
-  test("passes assistant content blocks through to the mapper", () => {
+  test("passes assistant content blocks through to the mapper", async () => {
     const entry = {
       type: "assistant",
       message: {
@@ -414,7 +414,7 @@ describe("convertClaudeHistoryEntry", () => {
     ];
     const mapBlocks = vi.fn().mockReturnValue(mappedTimeline);
 
-    expect(convertClaudeHistoryEntry(entry, mapBlocks)).toEqual(mappedTimeline);
+    expect(await convertClaudeHistoryEntry(entry, mapBlocks)).toEqual(mappedTimeline);
     expect(mapBlocks).toHaveBeenCalledWith(entry.message.content);
   });
 });
@@ -1208,7 +1208,7 @@ describe("ClaudeAgentSession features", () => {
         input: Record<string, unknown>,
         options: Record<string, unknown>,
       ): Promise<PermissionResult>;
-      translateMessageToEvents(message: SDKMessage): AgentStreamEvent[];
+      translateMessageToEvents(message: SDKMessage): Promise<AgentStreamEvent[]>;
     };
 
     try {
@@ -1248,7 +1248,7 @@ describe("ClaudeAgentSession features", () => {
         ),
       ).resolves.toMatchObject({ behavior: "deny", interrupt: undefined });
 
-      internal.translateMessageToEvents({
+      await internal.translateMessageToEvents({
         type: "command_lifecycle",
         command_uuid: steerUuid,
         state: "started",
@@ -2395,7 +2395,7 @@ describe("ClaudeAgentSession context window usage", () => {
   test("emits canonical task snapshots from Claude TaskCreate results", async () => {
     const session = await createSessionForTest();
     try {
-      session.translateMessageToEvents({
+      await session.translateMessageToEvents({
         type: "assistant",
         message: {
           content: [
@@ -2409,7 +2409,7 @@ describe("ClaudeAgentSession context window usage", () => {
         },
       } as unknown as SDKMessage);
 
-      const events = session.translateMessageToEvents({
+      const events = await session.translateMessageToEvents({
         type: "user",
         message: {
           content: [{ type: "tool_result", tool_use_id: "task-create-1", content: "created" }],
@@ -3611,9 +3611,9 @@ describe("ClaudeAgentSession context window usage", () => {
     ]);
 
     try {
-      const compactEvents = (session as unknown as TestClaudeSession).translateMessageToEvents(
-        createCompactBoundary(),
-      );
+      const compactEvents = await (
+        session as unknown as TestClaudeSession
+      ).translateMessageToEvents(createCompactBoundary());
       expect(compactEvents).toContainEqual(
         expect.objectContaining({
           type: "usage_updated",
@@ -3799,7 +3799,7 @@ describe("ClaudeAgentSession context window usage", () => {
   test("result.result is surfaced as an assistant message when no model output was produced", async () => {
     const session = await createSessionForTest();
 
-    const events = session.translateMessageToEvents({
+    const events = await session.translateMessageToEvents({
       type: "result",
       subtype: "success",
       result: "Unknown command: /foo-doesnt-exist",
@@ -3834,7 +3834,7 @@ describe("ClaudeAgentSession context window usage", () => {
   test("result.result is not duplicated when the model produced output during the turn", async () => {
     const session = await createSessionForTest();
 
-    const events = session.translateMessageToEvents({
+    const events = await session.translateMessageToEvents({
       type: "result",
       subtype: "success",
       result: "Here is the answer.",

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

@@ -1,3 +1,4 @@
+import { setImmediate } from "node:timers/promises";
 import { ClaudeRuntimeSafety } from "./runtime-safety.js";
 import { validateProviderOptions } from "../../provider-options.js";
 import type { ChildProcess } from "node:child_process";
@@ -82,7 +83,7 @@ import {
   type ClaudeRewindSdk,
 } from "./rewind.js";
 import { normalizeProviderReplayTimestamp } from "../../provider-history-timestamps.js";
-import { claudeConfigDir, claudeProjectDirSync, claudeTranscriptPathSync } from "./project-dir.js";
+import { claudeConfigDir, claudeProjectDirSync, claudeTranscriptPath } from "./project-dir.js";
 import { THINKING_APPLIES_NEXT_TURN_NOTICE } from "../../provider-notices.js";
 import {
   isProviderImageMarkdown,
@@ -1634,7 +1635,7 @@ export class ClaudeAgentClient implements AgentClient {
       cwd: merged.cwd,
     };
     const claudeConfig = this.assertConfig(mergedConfig);
-    return new ClaudeAgentSession(claudeConfig, {
+    const session = new ClaudeAgentSession(claudeConfig, {
       defaults: this.defaults,
       runtimeSettings: this.runtimeSettings,
       handle,
@@ -1645,6 +1646,8 @@ export class ClaudeAgentClient implements AgentClient {
       resolveBinary: this.resolveBinary,
       rewindSdk: this.rewindSdk,
     });
+    await session.ensureHistoryReady();
+    return session;
   }
 
   async getCatalogCacheKey(_options: FetchCatalogOptions): Promise<string> {
@@ -2329,6 +2332,9 @@ class ClaudeAgentSession implements AgentSession {
     { type: "provider_subagent" }
   >[] = [];
   private historyPending = false;
+  // Readiness never rejects; history/rebind failures are logged at their owning boundary.
+  private historyReady: Promise<void> | null = null;
+  private historySessionId: string | null = null;
   private replayingHistory = false;
   private turnState: TurnState = "idle";
   private nextTurnOrdinal = 1;
@@ -2382,7 +2388,7 @@ class ClaudeAgentSession implements AgentSession {
       }
       this.claudeSessionId = handle.sessionId;
       this.persistence = handle;
-      this.loadPersistedHistory(handle.sessionId);
+      this.historySessionId = handle.sessionId;
     } else {
       this.claudeSessionId = null;
       this.persistence = null;
@@ -2462,6 +2468,7 @@ class ClaudeAgentSession implements AgentSession {
     prompt: AgentPromptInput,
     options?: AgentRunOptions,
   ): Promise<{ turnId: string }> {
+    await this.ensureHistoryReady();
     if (this.closed) {
       throw new Error("Claude session is closed");
     }
@@ -2539,7 +2546,11 @@ class ClaudeAgentSession implements AgentSession {
       this.input.push(sdkMessage);
       setTimeout(() => {
         if (this.activeForegroundTurnId === turnId) {
-          this.emitSubmittedUserMessage(sdkMessage, turnId, options?.clientMessageId);
+          void this.emitSubmittedUserMessage(sdkMessage, turnId, options?.clientMessageId).catch(
+            (error: unknown) => {
+              this.logger.warn({ err: error }, "Failed to emit submitted Claude message");
+            },
+          );
         }
       }, 0);
     } catch (error) {
@@ -2632,6 +2643,7 @@ class ClaudeAgentSession implements AgentSession {
   }
 
   async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
+    await this.ensureHistoryReady();
     if (
       !this.historyPending ||
       (this.persistedHistory.length === 0 && this.persistedProviderSubagentEvents.length === 0)
@@ -2643,7 +2655,9 @@ class ClaudeAgentSession implements AgentSession {
     this.persistedHistory = [];
     this.persistedProviderSubagentEvents = [];
     this.historyPending = false;
+    let replayed = 0;
     for (const entry of history) {
+      if (++replayed % 32 === 0) await setImmediate();
       yield {
         type: "timeline",
         item: entry.item,
@@ -2961,6 +2975,7 @@ class ClaudeAgentSession implements AgentSession {
       "provider.claude.session_close.start",
     );
     this.closed = true;
+    await this.ensureHistoryReady();
     this.rejectAllPendingPermissions(new Error("Claude session closed"));
     this.cancelCurrentTurn?.();
     this.subscribers.clear();
@@ -3001,7 +3016,7 @@ class ClaudeAgentSession implements AgentSession {
       // (see `claude --help`), so the SDK's persistSession=false is silently dropped
       // in stream-json mode. Sweep the transcript ourselves so ephemeral runs
       // (metadata generator, branch-name generator) don't show up as resumable.
-      const historyPath = this.resolveHistoryPath(this.claudeSessionId);
+      const historyPath = await this.resolveHistoryPath(this.claudeSessionId);
       if (historyPath) {
         try {
           await promises.rm(historyPath, { force: true });
@@ -3045,6 +3060,7 @@ class ClaudeAgentSession implements AgentSession {
   }
 
   async revertConversation(input: { messageId: string }): Promise<void> {
+    await this.ensureHistoryReady();
     const target = this.resolveConversationRewindTarget(input.messageId);
     if (target.kind === "fresh-session") {
       this.startFreshConversationSession();
@@ -3054,16 +3070,15 @@ class ClaudeAgentSession implements AgentSession {
     await revertClaudeConversation({
       sdk: this.rewindSdk,
       sessionId: currentSessionId,
-      transcriptPath: currentSessionId ? this.resolveHistoryPath(currentSessionId) : null,
+      transcriptPath: currentSessionId ? await this.resolveHistoryPath(currentSessionId) : null,
       messageId: target.messageId,
       resolveMessageId: (messageId) => this.resolveClaudeMessageId(messageId),
-      setSessionId: (sessionId) => {
-        this.rebindConversationSession(sessionId);
-      },
+      setSessionId: (sessionId) => this.rebindConversationSession(sessionId),
     });
   }
 
   async revertFiles(input: { messageId: string }): Promise<void> {
+    await this.ensureHistoryReady();
     const messageId = await this.resolveClaudeMessageId(input.messageId);
     await revertClaudeFiles({
       query: await this.ensureQuery(),
@@ -3238,38 +3253,51 @@ class ClaudeAgentSession implements AgentSession {
     return candidates;
   }
 
-  private rebindConversationSession(sessionId: string): void {
-    const oldSessionId = this.claudeSessionId;
-    this.claudeSessionId = sessionId;
-    this.pendingFreshSessionId = null;
-    this.persistence = null;
-    this.cachedRuntimeInfo = null;
-    this.queryRestartNeeded = true;
-    this.persistedHistory = [];
-    this.persistedProviderSubagentEvents = [];
-    this.historyPending = false;
-    this.userMessageIds = [];
-    this.emittedUserMessageIds.clear();
-    this.rewindTurnAnchors.length = 0;
-    this.taskState.reset();
-    this.loadPersistedHistory(sessionId);
-    if (oldSessionId && oldSessionId !== sessionId) {
-      this.dispatchEvents([
-        {
-          type: "timeline",
-          provider: "claude",
-          item: this.createClaudeSessionChangedNotice(oldSessionId, sessionId),
-        },
-        {
-          type: "thread_started",
-          provider: "claude",
-          sessionId,
-        },
-      ]);
-    }
+  private rebindConversationSession(sessionId: string): Promise<void> {
+    const previousReady = this.ensureHistoryReady();
+    this.historyReady = (async () => {
+      await previousReady;
+      await this.retireQuery();
+      if (this.closed) return;
+      const oldSessionId = this.claudeSessionId;
+      this.claudeSessionId = sessionId;
+      this.pendingFreshSessionId = null;
+      this.persistence = null;
+      this.cachedRuntimeInfo = null;
+      this.queryRestartNeeded = true;
+      this.persistedHistory = [];
+      this.persistedProviderSubagentEvents = [];
+      this.historyPending = false;
+      this.userMessageIds = [];
+      this.emittedUserMessageIds.clear();
+      this.rewindTurnAnchors.length = 0;
+      this.taskState.reset();
+      this.historySessionId = sessionId;
+      await this.loadPersistedHistory(sessionId);
+      if (this.closed) return;
+      if (oldSessionId && oldSessionId !== sessionId) {
+        this.dispatchEvents([
+          {
+            type: "timeline",
+            provider: "claude",
+            item: this.createClaudeSessionChangedNotice(oldSessionId, sessionId),
+          },
+          {
+            type: "thread_started",
+            provider: "claude",
+            sessionId,
+          },
+        ]);
+      }
+    })().catch((error) => {
+      this.logger.warn({ err: error, sessionId }, "Failed to rebind Claude session");
+    });
+    return this.historyReady;
   }
 
   private startFreshConversationSession(): void {
+    this.historySessionId = null;
+    this.historyReady = Promise.resolve();
     const sessionId = randomUUID();
     this.claudeSessionId = sessionId;
     this.pendingFreshSessionId = sessionId;
@@ -3390,7 +3418,43 @@ class ClaudeAgentSession implements AgentSession {
     return { kind: "fresh-session" };
   }
 
+  private async retireQuery(): Promise<void> {
+    const oldQuery = this.query;
+    if (!oldQuery) return;
+    const oldInput = this.input;
+    // Null out query/input BEFORE awaiting the old iterator's return so the
+    // old pump sees this.query !== activeQuery and skips failActiveTurns.
+    this.query = null;
+    this.input = null;
+    this.queryPumpPromise = null;
+    this.queryRestartNeeded = false;
+    // Ending the input retires the process on purpose. Detach first so its
+    // exit is not reported as a crash.
+    const retiredChild = this.childProcess;
+    this.childProcess = null;
+    if (retiredChild) this.failRunningRuntimeTasks();
+    oldInput?.end();
+    oldQuery.close?.();
+    try {
+      await oldQuery.return?.();
+    } catch {
+      /* ignore */
+    }
+    // Tree-kill the old process tree now that the SDK has cleaned up.
+    // If we skip this, MCP children of the previous claude process can
+    // survive as orphans when the session spawns a replacement query.
+    if (retiredChild) {
+      await terminateWithTreeKill(retiredChild, {
+        gracefulTimeoutMs: 2_000,
+        forceTimeoutMs: 2_000,
+      }).catch(() => {
+        /* process may already be dead */
+      });
+    }
+  }
+
   private async ensureQuery(launchMode: PermissionMode = this.currentMode): Promise<Query> {
+    await this.ensureHistoryReady();
     // Controls use the running turn's query. Pending launch settings take effect
     // after that turn finishes, when its active query handle has been cleared.
     if (this.query && (!this.queryRestartNeeded || this.activeForegroundQuery === this.query)) {
@@ -3398,37 +3462,7 @@ class ClaudeAgentSession implements AgentSession {
     }
 
     if (this.queryRestartNeeded && this.query) {
-      const oldQuery = this.query;
-      const oldInput = this.input;
-      // Null out query/input BEFORE awaiting the old iterator's return so the
-      // old pump sees this.query !== activeQuery and skips failActiveTurns.
-      this.query = null;
-      this.input = null;
-      this.queryPumpPromise = null;
-      this.queryRestartNeeded = false;
-      // Ending the input retires the process on purpose. Detach first so its
-      // exit is not reported as a crash.
-      const retiredChild = this.childProcess;
-      this.childProcess = null;
-      if (retiredChild) this.failRunningRuntimeTasks();
-      oldInput?.end();
-      oldQuery.close?.();
-      try {
-        await oldQuery.return?.();
-      } catch {
-        /* ignore */
-      }
-      // Tree-kill the old process tree now that the SDK has cleaned up.
-      // If we skip this, MCP children of the previous claude process can
-      // survive as orphans when the session spawns a replacement query.
-      if (retiredChild) {
-        await terminateWithTreeKill(retiredChild, {
-          gracefulTimeoutMs: 2_000,
-          forceTimeoutMs: 2_000,
-        }).catch(() => {
-          /* process may already be dead */
-        });
-      }
+      await this.retireQuery();
     }
 
     // Preserve claudeSessionId across query recreation so buildOptions() passes
@@ -4095,12 +4129,14 @@ class ClaudeAgentSession implements AgentSession {
       );
     };
     const handlePumpedMessage = async (message: SDKMessage): Promise<boolean> => {
+      if (!this.ownsQuery(activeQuery)) return true;
       this.processingSdkMessages++;
       try {
         this.runtimeSafety.observe(message);
         logRawMessage(message);
         consecutiveInterruptAbortRecoveries = 0;
         if (await this.handleMissingResumedConversation(message, activeQuery)) return true;
+        if (!this.ownsQuery(activeQuery)) return true;
         await this.routeSdkMessageFromPump(message);
         return false;
       } finally {
@@ -4221,6 +4257,10 @@ class ClaudeAgentSession implements AgentSession {
     return isClaudeTurnInit(message) && !this.activeForegroundTurnId;
   }
 
+  private ownsQuery(query: Query): boolean {
+    return !this.closed && this.query === query;
+  }
+
   private async routeSdkMessageFromPump(message: SDKMessage): Promise<void> {
     if (this.consumeSessionStateChange(message)) {
       return;
@@ -4295,10 +4335,17 @@ class ClaudeAgentSession implements AgentSession {
     messageIdHint: string | null,
     turnId: string | null,
   ): Promise<AgentStreamEvent[]> {
-    const messageEvents = this.translateMessageToEvents(message, {
+    const query = this.query;
+    const messageEvents = await this.translateMessageToEvents(message, {
       suppressAssistantText: true,
       suppressReasoning: true,
     });
+    if (
+      this.closed ||
+      this.query !== query ||
+      (this.activeForegroundTurnId ?? this.autonomousTurn?.id ?? null) !== turnId
+    )
+      return [];
     const assistantTimelineEvents = readClaudeParentToolUseId(message)
       ? []
       : this.timelineAssembler
@@ -4443,13 +4490,13 @@ class ClaudeAgentSession implements AgentSession {
     return withdrewAll;
   }
 
-  private translateMessageToEvents(
+  private async translateMessageToEvents(
     message: SDKMessage,
     options?: {
       suppressAssistantText?: boolean;
       suppressReasoning?: boolean;
     },
-  ): AgentStreamEvent[] {
+  ): Promise<AgentStreamEvent[]> {
     const parentToolUseId = readClaudeParentToolUseId(message);
     if (parentToolUseId) {
       return this.translateSidechainFrameToEvents(message, parentToolUseId);
@@ -4497,11 +4544,11 @@ class ClaudeAgentSession implements AgentSession {
         this.appendSystemMessageEvents(message, events);
         break;
       case "user":
-        this.appendUserMessageEvents(message, events);
+        await this.appendUserMessageEvents(message, events);
         this.appendSidechainResultEvents(message, events);
         break;
       case "assistant": {
-        const timelineItems = this.mapBlocksToTimeline(message.message.content, {
+        const timelineItems = await this.mapBlocksToTimeline(message.message.content, {
           suppressAssistantText: options?.suppressAssistantText ?? false,
           suppressReasoning: options?.suppressReasoning ?? false,
         });
@@ -4512,7 +4559,7 @@ class ClaudeAgentSession implements AgentSession {
         break;
       }
       case "stream_event":
-        this.appendStreamEventEvents(message, events, options);
+        await this.appendStreamEventEvents(message, events, options);
         break;
       case "result":
         this.appendResultEvents(message, events);
@@ -4614,13 +4661,14 @@ class ClaudeAgentSession implements AgentSession {
     }
   }
 
-  private emitSubmittedUserMessage(
+  private async emitSubmittedUserMessage(
     message: Extract<SDKMessage, { type: "user" }>,
     turnId: string,
     clientMessageId?: string,
-  ): void {
+  ): Promise<void> {
     const events: AgentStreamEvent[] = [];
-    this.appendUserMessageEvents(message, events);
+    await this.appendUserMessageEvents(message, events);
+    if (this.activeForegroundTurnId !== turnId) return;
     if (events.length === 0) {
       return;
     }
@@ -4750,10 +4798,10 @@ class ClaudeAgentSession implements AgentSession {
     }
   }
 
-  private appendUserMessageEvents(
+  private async appendUserMessageEvents(
     message: Extract<SDKMessage, { type: "user" }>,
     events: AgentStreamEvent[],
-  ): void {
+  ): Promise<void> {
     if (isSyntheticUserEntry(message)) {
       return;
     }
@@ -4792,7 +4840,7 @@ class ClaudeAgentSession implements AgentSession {
       return;
     }
     if (Array.isArray(content)) {
-      this.appendUserContentArrayEvents(content, messageId, events);
+      await this.appendUserContentArrayEvents(content, messageId, events);
     }
   }
 
@@ -4818,12 +4866,12 @@ class ClaudeAgentSession implements AgentSession {
     events.push({ type: "timeline", item, provider: "claude" });
   }
 
-  private appendUserContentArrayEvents(
+  private async appendUserContentArrayEvents(
     content: ReadonlyArray<unknown>,
     messageId: string | undefined,
     events: AgentStreamEvent[],
-  ): void {
-    const timelineItems = this.mapBlocksToTimeline(content, {
+  ): Promise<void> {
+    const timelineItems = await this.mapBlocksToTimeline(content, {
       textMessageType: "user_message",
     });
     for (const item of timelineItems) {
@@ -4839,16 +4887,16 @@ class ClaudeAgentSession implements AgentSession {
     }
   }
 
-  private appendStreamEventEvents(
+  private async appendStreamEventEvents(
     message: Extract<SDKMessage, { type: "stream_event" }>,
     events: AgentStreamEvent[],
     options: { suppressAssistantText?: boolean; suppressReasoning?: boolean } | undefined,
-  ): void {
+  ): Promise<void> {
     const usageUpdatedEvent = this.contextUsage.buildStreamUsageEvent(message.event);
     if (usageUpdatedEvent) {
       events.push(usageUpdatedEvent);
     }
-    const timelineItems = this.mapPartialEvent(message.event, {
+    const timelineItems = await this.mapPartialEvent(message.event, {
       suppressAssistantText: options?.suppressAssistantText ?? false,
       suppressReasoning: options?.suppressReasoning ?? false,
     });
@@ -5272,22 +5320,31 @@ class ClaudeAgentSession implements AgentSession {
     }
   }
 
-  private loadPersistedHistory(sessionId: string): void {
+  ensureHistoryReady(): Promise<void> {
+    if (!this.historyReady) {
+      this.historyReady = this.historySessionId
+        ? this.loadPersistedHistory(this.historySessionId)
+        : Promise.resolve();
+    }
+    return this.historyReady;
+  }
+
+  private async loadPersistedHistory(sessionId: string): Promise<void> {
     let historyPath: string | null = null;
     this.replayingHistory = true;
     try {
       this.taskState.reset();
-      historyPath = this.resolveHistoryPath(sessionId);
-      if (!historyPath || !fs.existsSync(historyPath)) {
+      historyPath = await this.resolveHistoryPath(sessionId);
+      if (!historyPath || !(await transcriptExists(historyPath))) {
         this.logger.info(
           { sessionId, cwd: this.config.cwd, historyPath },
           "No Claude transcript to load history from",
         );
         return;
       }
-      withClaudeReplayHistory(historyPath, (history) => {
-        const replay = this.ingestPersistedSidechains(history);
-        this.ingestPersistedHistory(history.parentEntries, replay);
+      await withClaudeReplayHistory(historyPath, async (history) => {
+        const replay = await this.ingestPersistedSidechains(history);
+        await this.ingestPersistedHistory(history.parentEntries, replay);
       });
     } catch (error) {
       this.logger.warn(
@@ -5299,13 +5356,13 @@ class ClaudeAgentSession implements AgentSession {
     }
   }
 
-  private ingestPersistedHistory(
-    entries: Iterable<ClaudeHistoryEntry>,
+  private async ingestPersistedHistory(
+    entries: Iterable<ClaudeHistoryEntry> | AsyncIterable<ClaudeHistoryEntry>,
     replay: ClaudeReplayOwnership,
-  ): void {
+  ): Promise<void> {
     const timeline: PersistedTimelineEntry[] = [];
-    for (const entry of entries) {
-      this.ingestPersistedHistoryEntry(entry, timeline, replay);
+    for await (const entry of entries) {
+      await this.ingestPersistedHistoryEntry(entry, timeline, replay);
     }
 
     if (timeline.length > 0) {
@@ -5314,27 +5371,33 @@ class ClaudeAgentSession implements AgentSession {
     }
   }
 
-  private ingestPersistedSidechains(history: ClaudeReplayHistory): ClaudeReplayOwnership {
-    const parentFacts = readClaudeReplayParentFacts(history.parentEntries);
+  private async ingestPersistedSidechains(
+    history: ClaudeReplayHistory,
+  ): Promise<ClaudeReplayOwnership> {
+    const parentFacts = await readClaudeReplayParentFacts(history.parentEntries);
 
     // Replay produces the same observations the live task protocol produces, then folds them
     // with the same function, so identity and status are derived once for both paths.
-    const subagentReplay = observeReplaySubagents({
-      subagents: history.subagents.map((subagent) => ({
+    const subagents = [];
+    for (const subagent of history.subagents) {
+      subagents.push({
         ...subagent,
-        parentFacts: readClaudeReplayParentFacts(subagent.entries),
-      })),
+        parentFacts: await readClaudeReplayParentFacts(subagent.entries),
+      });
+    }
+    const subagentReplay = await observeReplaySubagents({
+      subagents,
       parent: parentFacts,
       convertEntry: (entry) => this.convertHistoryEntry(entry),
     });
     const observations = [
       ...subagentReplay.observations,
-      ...observeReplayWorkflows({
+      ...(await observeReplayWorkflows({
         workflows: history.workflows,
         parentEntries: history.parentEntries,
         entriesByRunId: history.workflowEntriesByRunId,
         convertEntry: (entry) => this.convertHistoryEntry(entry),
-      }),
+      })),
     ];
     const restoredProviderSubagentIds = new Set(
       observations
@@ -5359,11 +5422,11 @@ class ClaudeAgentSession implements AgentSession {
     return replay;
   }
 
-  private ingestPersistedHistoryEntry(
+  private async ingestPersistedHistoryEntry(
     entry: ClaudeHistoryEntry,
     timeline: PersistedTimelineEntry[],
     replay: ClaudeReplayOwnership,
-  ): void {
+  ): Promise<void> {
     if (entry.isSidechain) {
       return;
     }
@@ -5377,7 +5440,7 @@ class ClaudeAgentSession implements AgentSession {
       ? replay.toolOwners.get(notificationToolUseId)
       : undefined;
     if (notificationOwner) {
-      for (const item of this.convertHistoryEntry(entry)) {
+      for (const item of await this.convertHistoryEntry(entry)) {
         this.persistedProviderSubagentEvents.push({
           type: "provider_subagent",
           provider: "claude",
@@ -5394,7 +5457,7 @@ class ClaudeAgentSession implements AgentSession {
     const taskSnapshot = this.taskState.observe(entry);
     const items = [
       ...(taskSnapshot ? [taskSnapshot] : []),
-      ...this.convertHistoryEntry(entry).map((item) =>
+      ...(await this.convertHistoryEntry(entry)).map((item) =>
         labelReplayedSubagentCard(item, replay.subagentToolCalls),
       ),
     ];
@@ -5421,20 +5484,20 @@ class ClaudeAgentSession implements AgentSession {
     }
   }
 
-  private resolveHistoryPath(sessionId: string): string | null {
+  private async resolveHistoryPath(sessionId: string): Promise<string | null> {
     const cwd = this.config.cwd;
     if (!cwd) return null;
-    return claudeTranscriptPathSync({
+    return claudeTranscriptPath({
       cwd,
       sessionId,
       configDir: claudeConfigDir(this.harnessEnvironment),
     });
   }
 
-  private convertHistoryEntry(entry: ClaudeHistoryEntry): AgentTimelineItem[] {
-    return convertClaudeHistoryEntry(entry, (content) => this.mapBlocksToTimeline(content)).map(
-      limitAgentTimelineItemContent,
-    );
+  private async convertHistoryEntry(entry: ClaudeHistoryEntry): Promise<AgentTimelineItem[]> {
+    return (
+      await convertClaudeHistoryEntry(entry, (content) => this.mapBlocksToTimeline(content))
+    ).map(limitAgentTimelineItemContent);
   }
 
   // Maps Claude content blocks into AgentTimelineItems.
@@ -5449,14 +5512,14 @@ class ClaudeAgentSession implements AgentSession {
   //
   // NOTE: convertClaudeHistoryEntry uses extractUserMessageText directly instead of this function
   // for user entries. Both paths must produce equivalent user_message items.
-  private mapBlocksToTimeline(
+  private async mapBlocksToTimeline(
     content: string | ReadonlyArray<unknown>,
     options?: {
       textMessageType?: "assistant_message" | "user_message";
       suppressAssistantText?: boolean;
       suppressReasoning?: boolean;
     },
-  ): AgentTimelineItem[] {
+  ): Promise<AgentTimelineItem[]> {
     const textMessageType = options?.textMessageType ?? "assistant_message";
     const suppressText =
       textMessageType === "assistant_message" && (options?.suppressAssistantText ?? false);
@@ -5483,7 +5546,7 @@ class ClaudeAgentSession implements AgentSession {
       if (!isClaudeContentChunk(block) || this.mapSubagentHandbackBlock(block, items)) {
         continue;
       }
-      this.mapBlockToTimeline(block, {
+      await this.mapBlockToTimeline(block, {
         items,
         userTextParts,
         textMessageType,
@@ -5528,7 +5591,7 @@ class ClaudeAgentSession implements AgentSession {
     }
   }
 
-  private mapBlockToTimeline(
+  private async mapBlockToTimeline(
     block: ClaudeContentChunk,
     context: {
       items: AgentTimelineItem[];
@@ -5537,7 +5600,7 @@ class ClaudeAgentSession implements AgentSession {
       suppressText: boolean;
       suppressReasoning: boolean;
     },
-  ): void {
+  ): Promise<void> {
     switch (block.type) {
       case "text":
       case "text_delta":
@@ -5561,7 +5624,7 @@ class ClaudeAgentSession implements AgentSession {
       case "code_execution_tool_result":
       case "bash_code_execution_tool_result":
       case "text_editor_code_execution_tool_result":
-        this.handleToolResult(block, context.items);
+        await this.handleToolResult(block, context.items);
         break;
       default:
         break;
@@ -5607,7 +5670,10 @@ class ClaudeAgentSession implements AgentSession {
     );
   }
 
-  private handleToolResult(block: ClaudeContentChunk, items: AgentTimelineItem[]): void {
+  private async handleToolResult(
+    block: ClaudeContentChunk,
+    items: AgentTimelineItem[],
+  ): Promise<void> {
     const entry =
       typeof block.tool_use_id === "string" ? this.toolUseCache.get(block.tool_use_id) : undefined;
     const blockToolName = typeof block.tool_name === "string" ? block.tool_name : undefined;
@@ -5640,7 +5706,7 @@ class ClaudeAgentSession implements AgentSession {
     this.rememberSettledSubagentToolCall(settledToolCall, entry);
 
     for (const image of images) {
-      const imageItem = renderProviderImageOutputAsAssistantMarkdown(image, {
+      const imageItem = await renderProviderImageOutputAsAssistantMarkdown(image, {
         materialize: materializeProviderImage,
       });
       if (imageItem) {
@@ -5864,13 +5930,13 @@ class ClaudeAgentSession implements AgentSession {
     return false;
   }
 
-  private mapPartialEvent(
+  private async mapPartialEvent(
     event: SDKPartialAssistantMessage["event"],
     options?: {
       suppressAssistantText?: boolean;
       suppressReasoning?: boolean;
     },
-  ): AgentTimelineItem[] {
+  ): Promise<AgentTimelineItem[]> {
     if (this.updatePartialEventToolState(event)) {
       return [];
     }
@@ -5878,14 +5944,14 @@ class ClaudeAgentSession implements AgentSession {
     switch (event.type) {
       case "content_block_start":
         return isClaudeContentChunk(event.content_block)
-          ? this.mapBlocksToTimeline([event.content_block], {
+          ? await this.mapBlocksToTimeline([event.content_block], {
               suppressAssistantText: options?.suppressAssistantText,
               suppressReasoning: options?.suppressReasoning,
             })
           : [];
       case "content_block_delta":
         return isClaudeContentChunk(event.delta)
-          ? this.mapBlocksToTimeline([event.delta], {
+          ? await this.mapBlocksToTimeline([event.delta], {
               suppressAssistantText: options?.suppressAssistantText,
               suppressReasoning: options?.suppressReasoning,
             })
@@ -6191,13 +6257,13 @@ function labelReplayedSubagentCard(
  * source consumes. The agentId scrape is kept only as a fallback for sessions recorded before
  * Claude Code wrote the meta sidecar.
  */
-function readClaudeReplayParentFacts(
-  entries: Iterable<ClaudeHistoryEntry>,
-): ClaudeReplayParentFacts {
+async function readClaudeReplayParentFacts(
+  entries: Iterable<ClaudeHistoryEntry> | AsyncIterable<ClaudeHistoryEntry>,
+): Promise<ClaudeReplayParentFacts> {
   const toolCalls = new Map<string, { title?: string; description?: string }>();
   const linksByAgentId = new Map<string, { toolCallId: string; failed: boolean }>();
   const outcomes = new Map<string, { failed: boolean }>();
-  for (const entry of entries) {
+  for await (const entry of entries) {
     const content = toObjectRecord(entry.message)?.content;
     if (!Array.isArray(content)) continue;
     for (const value of content) {
@@ -6248,12 +6314,14 @@ interface ClaudeHistoryEntry {
   [key: string]: unknown;
 }
 
-function mapAssistantHistoryBlocksWithMessageId(
+async function mapAssistantHistoryBlocksWithMessageId(
   entry: ClaudeHistoryEntry,
   content: string | ClaudeContentChunk[],
-  mapBlocks: (content: string | ClaudeContentChunk[]) => AgentTimelineItem[],
-): AgentTimelineItem[] {
-  const items = mapBlocks(content);
+  mapBlocks: (
+    content: string | ClaudeContentChunk[],
+  ) => AgentTimelineItem[] | Promise<AgentTimelineItem[]>,
+): Promise<AgentTimelineItem[]> {
+  const items = await mapBlocks(content);
   const assistantMessageId =
     typeof entry.uuid === "string" && entry.uuid.length > 0 ? entry.uuid : null;
   if (!assistantMessageId) {
@@ -6321,10 +6389,12 @@ function isProviderImageMessage(item: AgentTimelineItem): boolean {
   return item.type === "assistant_message" && isProviderImageMarkdown(item.text);
 }
 
-export function convertClaudeHistoryEntry(
+export async function convertClaudeHistoryEntry(
   entry: ClaudeHistoryEntry,
-  mapBlocks: (content: string | ClaudeContentChunk[]) => AgentTimelineItem[],
-): AgentTimelineItem[] {
+  mapBlocks: (
+    content: string | ClaudeContentChunk[],
+  ) => AgentTimelineItem[] | Promise<AgentTimelineItem[]>,
+): Promise<AgentTimelineItem[]> {
   const preamble = convertClaudeHistoryEntryPreamble(entry);
   if ("shortCircuit" in preamble) {
     return preamble.shortCircuit;
@@ -6363,7 +6433,7 @@ export function convertClaudeHistoryEntry(
   }
 
   if (hasToolBlock && normalizedBlocks) {
-    const mapped = mapBlocks(normalizedBlocks);
+    const mapped = await mapBlocks(normalizedBlocks);
     if (entry.type === "user") {
       // tool_result handling (handleToolResult) emits image markdown as an assistant_message
       // alongside the tool_call. User-entry text blocks also map to assistant_message in this path
@@ -6377,7 +6447,7 @@ export function convertClaudeHistoryEntry(
   }
 
   if (entry.type === "assistant" && contentValue) {
-    return mapAssistantHistoryBlocksWithMessageId(entry, contentValue, mapBlocks);
+    return await mapAssistantHistoryBlocksWithMessageId(entry, contentValue, mapBlocks);
   }
 
   return timeline;
@@ -6729,7 +6799,7 @@ async function readClaudeRenamedSessionTitle(input: {
 }): Promise<string | null> {
   let content: string;
   try {
-    content = await fsPromises.readFile(claudeTranscriptPathSync(input), "utf8");
+    content = await fsPromises.readFile(await claudeTranscriptPath(input), "utf8");
   } catch (error) {
     // A session with no transcript on disk has no rename to carry over. Every
     // other read failure is a real fault and stays explicit.
@@ -6829,3 +6899,13 @@ function readClaudeCommandLifecycle(message: unknown): ClaudeCommandLifecycle |
     state: record.state as ClaudeCommandLifecycle["state"],
   };
 }
+
+async function transcriptExists(file: string): Promise<boolean> {
+  try {
+    await fs.promises.access(file);
+    return true;
+  } catch (error) {
+    if ((error as NodeJS.ErrnoException).code === "ENOENT") return false;
+    throw error;
+  }
+}

packages/server/src/server/agent/providers/claude/project-dir.ts

@@ -1,5 +1,5 @@
-import { existsSync, readdirSync, realpathSync } from "node:fs";
-import { realpath } from "node:fs/promises";
+import { realpathSync } from "node:fs";
+import { realpath, access, readdir } from "node:fs/promises";
 import { homedir } from "node:os";
 import { join } from "node:path";
 
@@ -35,21 +35,21 @@ export function claudeProjectDirSync(cwd: string, options?: ClaudeProjectDirOpti
 // one `cwd` encodes to when the project moved, the daemon migrated, or Claude Code sees the
 // cwd under another name (a Windows binary under WSL). Returns the expected path when no
 // transcript exists anywhere.
-export function claudeTranscriptPathSync(input: {
+export async function claudeTranscriptPath(input: {
   cwd: string;
   sessionId: string;
   configDir?: string;
-}): string {
+}): Promise<string> {
   const options = { configDir: input.configDir };
   const fileName = `${input.sessionId}.jsonl`;
-  const expected = join(claudeProjectDirSync(input.cwd, options), fileName);
-  if (existsSync(expected)) {
+  const expected = join(await claudeProjectDir(input.cwd, options), fileName);
+  if (await exists(expected)) {
     return expected;
   }
   const projectsRoot = join(resolveConfigDir(options), "projects");
   let projectDirs: string[];
   try {
-    projectDirs = readdirSync(projectsRoot);
+    projectDirs = await readdir(projectsRoot);
   } catch (error) {
     if (isMissingDirectoryError(error)) {
       return expected;
@@ -58,13 +58,23 @@ export function claudeTranscriptPathSync(input: {
   }
   for (const projectDir of projectDirs) {
     const candidate = join(projectsRoot, projectDir, fileName);
-    if (existsSync(candidate)) {
+    if (await exists(candidate)) {
       return candidate;
     }
   }
   return expected;
 }
 
+async function exists(file: string): Promise<boolean> {
+  try {
+    await access(file);
+    return true;
+  } catch (error) {
+    if (isMissingDirectoryError(error)) return false;
+    throw error;
+  }
+}
+
 function isMissingDirectoryError(error: unknown): boolean {
   const code = (error as NodeJS.ErrnoException | null)?.code;
   return code === "ENOENT" || code === "ENOTDIR";

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

@@ -76,7 +76,7 @@ export async function revertClaudeConversation(input: {
   transcriptPath: string | null;
   messageId: string;
   resolveMessageId?: (messageId: string) => string | Promise<string>;
-  setSessionId: (sessionId: string) => void;
+  setSessionId: (sessionId: string) => void | Promise<void>;
 }): Promise<void> {
   if (!input.sessionId || !input.transcriptPath) {
     throw new Error("Claude session is not ready for rewind");
@@ -86,7 +86,7 @@ export async function revertClaudeConversation(input: {
     upToMessageId: messageId,
     transcriptPath: input.transcriptPath,
   });
-  input.setSessionId(fork.sessionId);
+  await input.setSessionId(fork.sessionId);
 }
 
 export async function revertClaudeFiles(input: {
@@ -108,7 +108,7 @@ export async function revertClaudeConversationAndFiles(input: {
   transcriptPath: string | null;
   messageId: string;
   resolveMessageId?: (messageId: string) => string | Promise<string>;
-  setSessionId: (sessionId: string) => void;
+  setSessionId: (sessionId: string) => void | Promise<void>;
 }): Promise<void> {
   await revertClaudeFiles({
     query: input.query,

packages/server/src/server/agent/providers/claude/subagents/replay-source.test.ts

@@ -1,3 +1,9 @@
+import { readFile } from "node:fs/promises";
+import { fileURLToPath } from "node:url";
+import {
+  materializeProviderImage,
+  renderProviderImageOutputAsAssistantMarkdown,
+} from "../../provider-image-output.js";
 import type { SDKMessage } from "@anthropic-ai/claude-agent-sdk";
 import { describe, expect, it } from "vitest";
 
@@ -45,7 +51,7 @@ function applyToStore(observations: SubagentObservation[]): ProviderSubagentDesc
 }
 
 describe("parseClaudeSubagentMeta", () => {
-  it("reads the real sidecar shape", () => {
+  it("reads the real sidecar shape", async () => {
     expect(
       parseClaudeSubagentMeta(
         '{"agentType":"general-purpose","description":"Reply banana","toolUseId":"toolu_013i","spawnDepth":1}',
@@ -68,7 +74,7 @@ describe("parseClaudeSubagentMeta", () => {
     expect(parseClaudeSubagentMeta(contents)).toBeNull();
   });
 
-  it("keeps the good fields when one field has the wrong type", () => {
+  it("keeps the good fields when one field has the wrong type", async () => {
     expect(parseClaudeSubagentMeta('{"agentType":"Explore","toolUseId":42}')).toEqual({
       agentType: "Explore",
       description: undefined,
@@ -79,12 +85,14 @@ describe("parseClaudeSubagentMeta", () => {
 });
 
 describe("observeReplaySubagents", () => {
-  it("links a subagent to its Task call through meta.toolUseId", () => {
-    const observations = observeReplaySubagents({
-      subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }],
-      parent: parentWithTaskCall(),
-      convertEntry: () => [],
-    }).observations;
+  it("links a subagent to its Task call through meta.toolUseId", async () => {
+    const observations = (
+      await observeReplaySubagents({
+        subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }],
+        parent: parentWithTaskCall(),
+        convertEntry: () => [],
+      })
+    ).observations;
 
     expect(observations[0]).toMatchObject({
       kind: "declared",
@@ -95,31 +103,33 @@ describe("observeReplaySubagents", () => {
     });
   });
 
-  it("replays descendants beneath the sidechain that declared their Task", () => {
+  it("replays descendants beneath the sidechain that declared their Task", async () => {
     const nestedToolUseId = "toolu_012rzYnFZA";
-    const observations = observeReplaySubagents({
-      subagents: [
-        {
-          agentId: AGENT_ID,
-          meta: { toolUseId: TOOL_USE_ID, spawnDepth: 1 },
-          entries: [],
-          parentFacts: {
-            toolCalls: new Map([
-              [nestedToolUseId, { title: "Explore", description: "Nested audit" }],
-            ]),
-            linksByAgentId: new Map(),
-            outcomesByToolCallId: new Map([[nestedToolUseId, { failed: false }]]),
+    const observations = (
+      await observeReplaySubagents({
+        subagents: [
+          {
+            agentId: AGENT_ID,
+            meta: { toolUseId: TOOL_USE_ID, spawnDepth: 1 },
+            entries: [],
+            parentFacts: {
+              toolCalls: new Map([
+                [nestedToolUseId, { title: "Explore", description: "Nested audit" }],
+              ]),
+              linksByAgentId: new Map(),
+              outcomesByToolCallId: new Map([[nestedToolUseId, { failed: false }]]),
+            },
           },
-        },
-        {
-          agentId: "a6acb4b898",
-          meta: { toolUseId: nestedToolUseId, agentType: "Explore", spawnDepth: 2 },
-          entries: [],
-        },
-      ],
-      parent: parentWithTaskCall(),
-      convertEntry: () => [],
-    }).observations;
+          {
+            agentId: "a6acb4b898",
+            meta: { toolUseId: nestedToolUseId, agentType: "Explore", spawnDepth: 2 },
+            entries: [],
+          },
+        ],
+        parent: parentWithTaskCall(),
+        convertEntry: () => [],
+      })
+    ).observations;
 
     expect(observations).toContainEqual(
       expect.objectContaining({
@@ -130,89 +140,143 @@ describe("observeReplaySubagents", () => {
     );
   });
 
-  it("keeps a subagent recorded before spawnDepth existed", () => {
-    const observations = observeReplaySubagents({
-      subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }],
-      parent: parentWithTaskCall(),
-      convertEntry: () => [],
-    }).observations;
+  it("keeps a subagent recorded before spawnDepth existed", async () => {
+    const observations = (
+      await observeReplaySubagents({
+        subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }],
+        parent: parentWithTaskCall(),
+        convertEntry: () => [],
+      })
+    ).observations;
 
     expect(observations[0]).toMatchObject({ kind: "declared", id: TOOL_USE_ID });
   });
 
-  it("drops a meta-linked subagent when the parent never declared its Task", () => {
-    const observations = observeReplaySubagents({
-      subagents: [
-        {
-          agentId: AGENT_ID,
-          meta: { toolUseId: TOOL_USE_ID, agentType: "Explore", description: "Find the code" },
-          entries: [],
-        },
-      ],
-      parent: emptyParent(),
-      convertEntry: () => [],
-    }).observations;
+  it("drops a meta-linked subagent when the parent never declared its Task", async () => {
+    const observations = (
+      await observeReplaySubagents({
+        subagents: [
+          {
+            agentId: AGENT_ID,
+            meta: { toolUseId: TOOL_USE_ID, agentType: "Explore", description: "Find the code" },
+            entries: [],
+          },
+        ],
+        parent: emptyParent(),
+        convertEntry: () => [],
+      })
+    ).observations;
 
     expect(observations).toEqual([]);
   });
 
-  it("falls back to the scraped link when there is no meta file", () => {
+  it("falls back to the scraped link when there is no meta file", async () => {
     const parent: ClaudeReplayParentFacts = {
       toolCalls: new Map([[TOOL_USE_ID, { title: "Explore" }]]),
       linksByAgentId: new Map([[AGENT_ID, { toolCallId: TOOL_USE_ID, failed: false }]]),
       outcomesByToolCallId: new Map(),
     };
 
-    const [declared, ...rest] = observeReplaySubagents({
-      subagents: [{ agentId: AGENT_ID, meta: null, entries: [] }],
-      parent,
-      convertEntry: () => [],
-    }).observations;
+    const [declared, ...rest] = (
+      await observeReplaySubagents({
+        subagents: [{ agentId: AGENT_ID, meta: null, entries: [] }],
+        parent,
+        convertEntry: () => [],
+      })
+    ).observations;
 
     expect(declared).toMatchObject({ id: TOOL_USE_ID, toolCallId: TOOL_USE_ID, title: "Explore" });
     expect(rest.at(-1)).toMatchObject({ kind: "status", status: "completed" });
   });
 
-  it("drops an unlinkable subagent", () => {
-    const observations = observeReplaySubagents({
-      subagents: [{ agentId: AGENT_ID, meta: null, entries: [] }],
-      parent: emptyParent(),
-      convertEntry: () => [],
-    }).observations;
+  it("drops an unlinkable subagent", async () => {
+    const observations = (
+      await observeReplaySubagents({
+        subagents: [{ agentId: AGENT_ID, meta: null, entries: [] }],
+        parent: emptyParent(),
+        convertEntry: () => [],
+      })
+    ).observations;
     expect(observations).toEqual([]);
   });
 
-  it("drops a terminal subagent the parent has no Task call for", () => {
+  it("drops a terminal subagent the parent has no Task call for", async () => {
     // A legacy grandchild: its toolUseId names a Task call made inside a sibling's transcript, so
     // no tool_result for it can ever reach the parent. Left running, it never stops.
     const descriptor = applyToStore(
-      observeReplaySubagents({
-        subagents: [
-          {
-            agentId: AGENT_ID,
-            meta: { toolUseId: TOOL_USE_ID },
-            entries: [
-              {
-                type: "assistant",
-                timestamp: "2026-07-31T03:41:00.000Z",
-                message: { stop_reason: "end_turn" },
-              },
-            ],
-          },
-        ],
-        parent: emptyParent(),
-        convertEntry: () => [],
-      }).observations,
+      (
+        await observeReplaySubagents({
+          subagents: [
+            {
+              agentId: AGENT_ID,
+              meta: { toolUseId: TOOL_USE_ID },
+              entries: [
+                {
+                  type: "assistant",
+                  timestamp: "2026-07-31T03:41:00.000Z",
+                  message: { stop_reason: "end_turn" },
+                },
+              ],
+            },
+          ],
+          parent: emptyParent(),
+          convertEntry: () => [],
+        })
+      ).observations,
     );
 
     expect(descriptor).toBeNull();
   });
 
-  it("drops a terminal skill-spawned subagent that has no link at all", () => {
+  it("drops a terminal skill-spawned subagent that has no link at all", async () => {
     // A /code-review run: no Task tool_use exists to name it, so its sidecar carries only
     // agentType and its id appears nowhere in the parent transcript.
     const descriptor = applyToStore(
-      observeReplaySubagents({
+      (
+        await observeReplaySubagents({
+          subagents: [
+            {
+              agentId: AGENT_ID,
+              meta: { agentType: "general-purpose" },
+              entries: [
+                {
+                  type: "assistant",
+                  timestamp: "2026-07-31T03:41:00.000Z",
+                  message: { stop_reason: "end_turn" },
+                },
+              ],
+            },
+          ],
+          parent: emptyParent(),
+          convertEntry: () => [],
+        })
+      ).observations,
+    );
+
+    expect(descriptor).toBeNull();
+  });
+
+  it("reports a scraped link's failure as failed", async () => {
+    const descriptor = applyToStore(
+      (
+        await observeReplaySubagents({
+          subagents: [{ agentId: AGENT_ID, meta: null, entries: [] }],
+          parent: {
+            toolCalls: new Map([[TOOL_USE_ID, { title: "Explore" }]]),
+            linksByAgentId: new Map([[AGENT_ID, { toolCallId: TOOL_USE_ID, failed: true }]]),
+            outcomesByToolCallId: new Map(),
+          },
+          convertEntry: () => [],
+        })
+      ).observations,
+    );
+
+    expect(descriptor).toMatchObject({ id: TOOL_USE_ID, status: "failed" });
+  });
+
+  it("drops a non-terminal linkless subagent", async () => {
+    const observations = (
+      await observeReplaySubagents({
         subagents: [
           {
             agentId: AGENT_ID,
@@ -221,111 +285,81 @@ describe("observeReplaySubagents", () => {
               {
                 type: "assistant",
                 timestamp: "2026-07-31T03:41:00.000Z",
-                message: { stop_reason: "end_turn" },
+                message: { stop_reason: null },
               },
             ],
           },
         ],
         parent: emptyParent(),
         convertEntry: () => [],
-      }).observations,
-    );
-
-    expect(descriptor).toBeNull();
-  });
-
-  it("reports a scraped link's failure as failed", () => {
-    const descriptor = applyToStore(
-      observeReplaySubagents({
-        subagents: [{ agentId: AGENT_ID, meta: null, entries: [] }],
-        parent: {
-          toolCalls: new Map([[TOOL_USE_ID, { title: "Explore" }]]),
-          linksByAgentId: new Map([[AGENT_ID, { toolCallId: TOOL_USE_ID, failed: true }]]),
-          outcomesByToolCallId: new Map(),
-        },
-        convertEntry: () => [],
-      }).observations,
-    );
-
-    expect(descriptor).toMatchObject({ id: TOOL_USE_ID, status: "failed" });
-  });
-
-  it("drops a non-terminal linkless subagent", () => {
-    const observations = observeReplaySubagents({
-      subagents: [
-        {
-          agentId: AGENT_ID,
-          meta: { agentType: "general-purpose" },
-          entries: [
-            {
-              type: "assistant",
-              timestamp: "2026-07-31T03:41:00.000Z",
-              message: { stop_reason: null },
-            },
-          ],
-        },
-      ],
-      parent: emptyParent(),
-      convertEntry: () => [],
-    }).observations;
+      })
+    ).observations;
 
     expect(observations).toEqual([]);
   });
 
-  it("finishes a terminal Task child when the parent has not recorded its outcome", () => {
+  it("finishes a terminal Task child when the parent has not recorded its outcome", async () => {
     const descriptor = applyToStore(
-      observeReplaySubagents({
-        subagents: [
-          {
-            agentId: AGENT_ID,
-            meta: { toolUseId: TOOL_USE_ID },
-            entries: [
-              {
-                type: "assistant",
-                timestamp: "2026-07-31T03:41:00.000Z",
-                message: { stop_reason: "end_turn" },
-              },
-            ],
-          },
-        ],
-        parent: { ...parentWithTaskCall(), outcomesByToolCallId: new Map() },
-        convertEntry: () => [],
-      }).observations,
+      (
+        await observeReplaySubagents({
+          subagents: [
+            {
+              agentId: AGENT_ID,
+              meta: { toolUseId: TOOL_USE_ID },
+              entries: [
+                {
+                  type: "assistant",
+                  timestamp: "2026-07-31T03:41:00.000Z",
+                  message: { stop_reason: "end_turn" },
+                },
+              ],
+            },
+          ],
+          parent: { ...parentWithTaskCall(), outcomesByToolCallId: new Map() },
+          convertEntry: () => [],
+        })
+      ).observations,
     );
 
     expect(descriptor).toMatchObject({ status: "completed" });
   });
 
-  it("leaves a subagent with no recorded outcome running", () => {
-    const observations = observeReplaySubagents({
-      subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }],
-      parent: { ...parentWithTaskCall(), outcomesByToolCallId: new Map() },
-      convertEntry: () => [],
-    }).observations;
+  it("leaves a subagent with no recorded outcome running", async () => {
+    const observations = (
+      await observeReplaySubagents({
+        subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }],
+        parent: { ...parentWithTaskCall(), outcomesByToolCallId: new Map() },
+        convertEntry: () => [],
+      })
+    ).observations;
     expect(observations.some((o) => o.kind === "status")).toBe(false);
   });
 
-  it("reports a failed outcome as failed", () => {
-    const observations = observeReplaySubagents({
-      subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }],
-      parent: parentWithTaskCall(true),
-      convertEntry: () => [],
-    }).observations;
+  it("reports a failed outcome as failed", async () => {
+    const observations = (
+      await observeReplaySubagents({
+        subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }],
+        parent: parentWithTaskCall(true),
+        convertEntry: () => [],
+      })
+    ).observations;
     expect(observations.at(-1)).toMatchObject({ kind: "status", status: "failed" });
   });
 
-  it("carries entry timestamps onto timeline observations", () => {
-    const observations = observeReplaySubagents({
-      subagents: [
-        {
-          agentId: AGENT_ID,
-          meta: { toolUseId: TOOL_USE_ID },
-          entries: [{ type: "assistant", timestamp: "2026-07-26T06:27:47.034Z" }],
-        },
-      ],
-      parent: parentWithTaskCall(),
-      convertEntry: () => [{ type: "reasoning", text: "thinking" }],
-    }).observations;
+  it("carries entry timestamps onto timeline observations", async () => {
+    const observations = (
+      await observeReplaySubagents({
+        subagents: [
+          {
+            agentId: AGENT_ID,
+            meta: { toolUseId: TOOL_USE_ID },
+            entries: [{ type: "assistant", timestamp: "2026-07-26T06:27:47.034Z" }],
+          },
+        ],
+        parent: parentWithTaskCall(),
+        convertEntry: () => [{ type: "reasoning", text: "thinking" }],
+      })
+    ).observations;
 
     expect(observations.find((observation) => observation.kind === "timeline")).toMatchObject({
       kind: "timeline",
@@ -336,18 +370,20 @@ describe("observeReplaySubagents", () => {
 });
 
 describe("replay runtime", () => {
-  it("reads the model and effort off the subagent's own transcript", () => {
-    const observations = observeReplaySubagents({
-      subagents: [
-        {
-          agentId: AGENT_ID,
-          meta: { toolUseId: TOOL_USE_ID },
-          entries: [{ type: "assistant", effort: "high", message: { model: "claude-opus-5" } }],
-        },
-      ],
-      parent: parentWithTaskCall(),
-      convertEntry: () => [],
-    }).observations;
+  it("reads the model and effort off the subagent's own transcript", async () => {
+    const observations = (
+      await observeReplaySubagents({
+        subagents: [
+          {
+            agentId: AGENT_ID,
+            meta: { toolUseId: TOOL_USE_ID },
+            entries: [{ type: "assistant", effort: "high", message: { model: "claude-opus-5" } }],
+          },
+        ],
+        parent: parentWithTaskCall(),
+        convertEntry: () => [],
+      })
+    ).observations;
 
     expect(observations.find((o) => o.kind === "subtitle")).toEqual({
       kind: "subtitle",
@@ -356,49 +392,55 @@ describe("replay runtime", () => {
     });
   });
 
-  it("takes the last observed values when the model changes mid-transcript", () => {
-    const observations = observeReplaySubagents({
-      subagents: [
-        {
-          agentId: AGENT_ID,
-          meta: { toolUseId: TOOL_USE_ID },
-          entries: [
-            { type: "assistant", effort: "high", message: { model: "claude-opus-5" } },
-            { type: "assistant", effort: "low", message: { model: "claude-sonnet-5" } },
-          ],
-        },
-      ],
-      parent: parentWithTaskCall(),
-      convertEntry: () => [],
-    }).observations;
+  it("takes the last observed values when the model changes mid-transcript", async () => {
+    const observations = (
+      await observeReplaySubagents({
+        subagents: [
+          {
+            agentId: AGENT_ID,
+            meta: { toolUseId: TOOL_USE_ID },
+            entries: [
+              { type: "assistant", effort: "high", message: { model: "claude-opus-5" } },
+              { type: "assistant", effort: "low", message: { model: "claude-sonnet-5" } },
+            ],
+          },
+        ],
+        parent: parentWithTaskCall(),
+        convertEntry: () => [],
+      })
+    ).observations;
 
     expect(observations.find((o) => o.kind === "subtitle")).toMatchObject({
       subtitle: "general-purpose · Sonnet 5 · Low",
     });
   });
 
-  it("reports nothing when the transcript never names a model or effort", () => {
-    const observations = observeReplaySubagents({
-      subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }],
-      parent: parentWithTaskCall(),
-      convertEntry: () => [],
-    }).observations;
+  it("reports nothing when the transcript never names a model or effort", async () => {
+    const observations = (
+      await observeReplaySubagents({
+        subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }],
+        parent: parentWithTaskCall(),
+        convertEntry: () => [],
+      })
+    ).observations;
     expect(observations.some((o) => o.kind === "subtitle")).toBe(false);
   });
 
-  it("passes through a model the manifest does not know", () => {
+  it("passes through a model the manifest does not know", async () => {
     // Claude Code is an Anthropic-compatible client; subagents legitimately report other models.
-    const observations = observeReplaySubagents({
-      subagents: [
-        {
-          agentId: AGENT_ID,
-          meta: { toolUseId: TOOL_USE_ID },
-          entries: [{ type: "assistant", message: { model: "glm-5.1" } }],
-        },
-      ],
-      parent: parentWithTaskCall(),
-      convertEntry: () => [],
-    }).observations;
+    const observations = (
+      await observeReplaySubagents({
+        subagents: [
+          {
+            agentId: AGENT_ID,
+            meta: { toolUseId: TOOL_USE_ID },
+            entries: [{ type: "assistant", message: { model: "glm-5.1" } }],
+          },
+        ],
+        parent: parentWithTaskCall(),
+        convertEntry: () => [],
+      })
+    ).observations;
     expect(observations.find((o) => o.kind === "subtitle")).toMatchObject({
       subtitle: "general-purpose · glm-5.1",
     });
@@ -406,20 +448,24 @@ describe("replay runtime", () => {
 });
 
 describe("replay usage", () => {
-  function subtitleOf(entries: ClaudeReplayEntry[]): SubagentObservation | undefined {
-    return observeReplaySubagents({
-      subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries }],
-      parent: parentWithTaskCall(),
-      convertEntry: () => [],
-    }).observations.find((observation) => observation.kind === "subtitle");
+  async function subtitleOf(
+    entries: ClaudeReplayEntry[],
+  ): Promise<SubagentObservation | undefined> {
+    return (
+      await observeReplaySubagents({
+        subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries }],
+        parent: parentWithTaskCall(),
+        convertEntry: () => [],
+      })
+    ).observations.find((observation) => observation.kind === "subtitle");
   }
 
-  it("derives the counters Claude Code itself reports, from a real transcript's shape", () => {
+  it("derives the counters Claude Code itself reports, from a real transcript's shape", async () => {
     // Shape and numbers taken from an on-disk subagent transcript. Claude Code finalizes a
     // subagent from the LAST assistant entry's usage, so the earlier, smaller turn must not win
     // and must not be added in.
     expect(
-      subtitleOf([
+      await subtitleOf([
         {
           type: "assistant",
           timestamp: "2026-07-23T23:18:43.023Z",
@@ -462,9 +508,9 @@ describe("replay usage", () => {
     });
   });
 
-  it("formats tokens for a child that only answered", () => {
+  it("formats tokens for a child that only answered", async () => {
     expect(
-      subtitleOf([
+      await subtitleOf([
         {
           type: "assistant",
           timestamp: "2026-07-27T16:42:14.698Z",
@@ -479,12 +525,12 @@ describe("replay usage", () => {
     ).toMatchObject({ subtitle: "general-purpose · 16.4k tokens" });
   });
 
-  it("emits nothing for a transcript with no entries", () => {
-    expect(subtitleOf([])).toBeUndefined();
+  it("emits nothing for a transcript with no entries", async () => {
+    expect(await subtitleOf([])).toBeUndefined();
   });
 
-  it("keeps duration out of the compact subtitle", () => {
-    const observation = subtitleOf([
+  it("keeps duration out of the compact subtitle", async () => {
+    const observation = await subtitleOf([
       {
         type: "assistant",
         timestamp: "2026-07-27T16:42:14.698Z",
@@ -494,8 +540,8 @@ describe("replay usage", () => {
     expect(observation).toMatchObject({ subtitle: "general-purpose · 10 tokens" });
   });
 
-  it("omits totalTokens when no entry carries a usage block", () => {
-    const observation = subtitleOf([
+  it("omits totalTokens when no entry carries a usage block", async () => {
+    const observation = await subtitleOf([
       { type: "assistant", timestamp: "2026-07-27T16:42:14.698Z", message: { content: [] } },
       {
         type: "assistant",
@@ -506,9 +552,9 @@ describe("replay usage", () => {
     expect(observation).toBeUndefined();
   });
 
-  it("keeps the last usage-bearing entry when the final assistant entry has none", () => {
+  it("keeps the last usage-bearing entry when the final assistant entry has none", async () => {
     expect(
-      subtitleOf([
+      await subtitleOf([
         {
           type: "assistant",
           timestamp: "2026-07-27T16:42:14.698Z",
@@ -519,34 +565,36 @@ describe("replay usage", () => {
     ).toMatchObject({ subtitle: "general-purpose · 1k tokens" });
   });
 
-  it("only reports counters the transcript actually shows", () => {
+  it("only reports counters the transcript actually shows", async () => {
     // A user-only transcript says nothing about cost: no assistant turn, so no tool count either.
-    const observation = subtitleOf([
+    const observation = await subtitleOf([
       { type: "user", timestamp: "2026-07-27T16:42:14.698Z", message: { content: "prompt" } },
       { type: "user", timestamp: "2026-07-27T16:42:20.000Z", message: { content: "more" } },
     ]);
     expect(observation).toBeUndefined();
   });
 
-  it("lands the complete provider-owned subtitle on the descriptor", () => {
+  it("lands the complete provider-owned subtitle on the descriptor", async () => {
     const store = new ProviderSubagentStore();
-    const observations = observeReplaySubagents({
-      subagents: [
-        {
-          agentId: AGENT_ID,
-          meta: { toolUseId: TOOL_USE_ID },
-          entries: [
-            {
-              type: "assistant",
-              timestamp: "2026-07-27T16:42:14.698Z",
-              message: { content: [], usage: { input_tokens: 7 } },
-            },
-          ],
-        },
-      ],
-      parent: parentWithTaskCall(),
-      convertEntry: () => [],
-    }).observations;
+    const observations = (
+      await observeReplaySubagents({
+        subagents: [
+          {
+            agentId: AGENT_ID,
+            meta: { toolUseId: TOOL_USE_ID },
+            entries: [
+              {
+                type: "assistant",
+                timestamp: "2026-07-27T16:42:14.698Z",
+                message: { content: [], usage: { input_tokens: 7 } },
+              },
+            ],
+          },
+        ],
+        parent: parentWithTaskCall(),
+        convertEntry: () => [],
+      })
+    ).observations;
     for (const event of foldSubagentObservations(observations)) {
       store.apply("parent", "claude", event);
     }
@@ -559,7 +607,7 @@ describe("live and replay agree", () => {
   // The property the design exists to guarantee: one session, two sources, one descriptor.
   // This is inexpressible against the inference implementation, where the paths share no
   // vocabulary and each derives identity and status its own way.
-  it("produces identical descriptor state from the wire and from disk", () => {
+  it("produces identical descriptor state from the wire and from disk", async () => {
     const live = new ClaudeTaskProtocolSource();
     const liveObservations = [
       ...live.observe({
@@ -579,21 +627,23 @@ describe("live and replay agree", () => {
       } as unknown as SDKMessage),
     ];
 
-    const replayObservations = observeReplaySubagents({
-      subagents: [
-        {
-          agentId: AGENT_ID,
-          meta: {
-            toolUseId: TOOL_USE_ID,
-            agentType: "general-purpose",
-            description: "Summarize hover and unistyles docs",
+    const replayObservations = (
+      await observeReplaySubagents({
+        subagents: [
+          {
+            agentId: AGENT_ID,
+            meta: {
+              toolUseId: TOOL_USE_ID,
+              agentType: "general-purpose",
+              description: "Summarize hover and unistyles docs",
+            },
+            entries: [],
           },
-          entries: [],
-        },
-      ],
-      parent: parentWithTaskCall(),
-      convertEntry: () => [],
-    }).observations;
+        ],
+        parent: parentWithTaskCall(),
+        convertEntry: () => [],
+      })
+    ).observations;
 
     const fromWire = applyToStore(liveObservations);
     const fromDisk = applyToStore(replayObservations);
@@ -616,3 +666,34 @@ describe("live and replay agree", () => {
     });
   });
 });
+
+it("awaits image materialization before publishing replay observations", async () => {
+  const bytes = Buffer.from("async replay image fixture");
+  const convertEntry = async () => {
+    const item = await renderProviderImageOutputAsAssistantMarkdown(
+      { data: bytes.toString("base64"), mimeType: "image/png" },
+      { materialize: materializeProviderImage },
+    );
+    return item ? [item] : [];
+  };
+  const { observations } = await observeReplaySubagents({
+    parent: parentWithTaskCall(),
+    subagents: [
+      {
+        agentId: AGENT_ID,
+        meta: { toolUseId: TOOL_USE_ID },
+        entries: [{ type: "assistant", message: { content: [] } }],
+      },
+    ],
+    convertEntry,
+  });
+  const image = observations.find(
+    (observation) =>
+      observation.kind === "timeline" && observation.item.type === "assistant_message",
+  );
+  if (!image || image.kind !== "timeline" || image.item.type !== "assistant_message")
+    throw new Error("Expected replay image");
+  const source = image.item.text.match(/!\[Image\]\(([^)]+)\)/)?.[1];
+  expect(source).toBeTruthy();
+  expect(await readFile(fileURLToPath(source!))).toEqual(bytes);
+});

packages/server/src/server/agent/providers/claude/subagents/replay-source.ts

@@ -130,7 +130,7 @@ function readUsage(entry: ClaudeReplayEntry): ClaudeSubagentUsage | null {
 export interface ClaudeReplaySubagentInput {
   agentId: string;
   meta: ClaudeSubagentMeta | null;
-  entries: Iterable<ClaudeReplayEntry>;
+  entries: Iterable<ClaudeReplayEntry> | AsyncIterable<ClaudeReplayEntry>;
   /** Task declarations made by this subagent, used to attach its direct children. */
   parentFacts?: ClaudeReplayParentFacts;
 }
@@ -250,13 +250,13 @@ function observeSubtitle(
   };
 }
 
-function observeSubagent(
+async function observeSubagent(
   subagent: ClaudeReplaySubagentInput,
   parent: ClaudeReplayParentFacts,
-  convertEntry: (entry: ClaudeReplayEntry) => AgentTimelineItem[],
+  convertEntry: (entry: ClaudeReplayEntry) => AgentTimelineItem[] | Promise<AgentTimelineItem[]>,
   toolOwners: Map<string, string>,
   parentSubagentId?: string,
-): SubagentObservation[] {
+): Promise<SubagentObservation[]> {
   const link = resolveParentLink(subagent, parent);
   if (!link) return [];
   const toolCall = parent.toolCalls.get(link.toolCallId);
@@ -269,7 +269,7 @@ function observeSubagent(
   let childTerminalStatus: ProviderSubagentStatus | null = null;
 
   // Consume each source once for presentation and timeline; keep no raw transcript records.
-  for (const entry of subagent.entries) {
+  for await (const entry of subagent.entries) {
     const timestamp = normalizeProviderReplayTimestamp(entry.timestamp);
     if (!seenEntry) firstTimestamp = timestamp;
     seenEntry = true;
@@ -280,7 +280,7 @@ function observeSubagent(
       childTerminalStatus = entry.message?.stop_reason === "end_turn" ? "completed" : null;
     }
     recordReplayToolOwners(toolOwners, entry, link.id);
-    for (const item of convertEntry(entry)) {
+    for (const item of await convertEntry(entry)) {
       timeline.push({ kind: "timeline", id: link.id, item, ...(timestamp ? { timestamp } : {}) });
     }
   }
@@ -320,11 +320,11 @@ function recordReplayToolOwners(
   }
 }
 
-export function observeReplaySubagents(input: {
+export async function observeReplaySubagents(input: {
   subagents: readonly ClaudeReplaySubagentInput[];
   parent: ClaudeReplayParentFacts;
-  convertEntry: (entry: ClaudeReplayEntry) => AgentTimelineItem[];
-}): { observations: SubagentObservation[]; toolOwners: ReadonlyMap<string, string> } {
+  convertEntry: (entry: ClaudeReplayEntry) => AgentTimelineItem[] | Promise<AgentTimelineItem[]>;
+}): Promise<{ observations: SubagentObservation[]; toolOwners: ReadonlyMap<string, string> }> {
   const observations: SubagentObservation[] = [];
   const toolOwners = new Map<string, string>();
   const unresolved = [...input.subagents].sort(
@@ -346,7 +346,7 @@ export function observeReplaySubagents(input: {
       const { ownerId, parent, link } = resolved;
 
       // Only proven descendants may own notifications; ambient sidecars cannot claim them.
-      for (const observation of observeSubagent(
+      for (const observation of await observeSubagent(
         subagent,
         parent,
         input.convertEntry,

packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.test.ts

@@ -1,3 +1,9 @@
+import { readFile } from "node:fs/promises";
+import { fileURLToPath } from "node:url";
+import {
+  materializeProviderImage,
+  renderProviderImageOutputAsAssistantMarkdown,
+} from "../../provider-image-output.js";
 import { describe, expect, it } from "vitest";
 
 import { ProviderSubagentStore } from "../../../provider-subagents/store.js";
@@ -35,7 +41,7 @@ function parentEntries(
 }
 
 describe("Claude workflow replay", () => {
-  it("parses the persisted run summary observed from Claude Code 2.1.220", () => {
+  it("parses the persisted run summary observed from Claude Code 2.1.220", async () => {
     expect(
       parseClaudeWorkflowRun(
         JSON.stringify({
@@ -62,8 +68,8 @@ describe("Claude workflow replay", () => {
     });
   });
 
-  it("rebuilds the same generic completed descriptor through the parent Workflow call", () => {
-    const observations = observeReplayWorkflows({
+  it("rebuilds the same generic completed descriptor through the parent Workflow call", async () => {
+    const observations = await observeReplayWorkflows({
       workflows: [
         {
           runId: RUN_ID,
@@ -91,8 +97,8 @@ describe("Claude workflow replay", () => {
     });
   });
 
-  it("rebuilds a workflow when Claude persists its result as structured text blocks", () => {
-    const observations = observeReplayWorkflows({
+  it("rebuilds a workflow when Claude persists its result as structured text blocks", async () => {
+    const observations = await observeReplayWorkflows({
       workflows: [{ runId: RUN_ID, summary: "Structured result workflow", status: "completed" }],
       parentEntries: parentEntries([
         {
@@ -114,7 +120,7 @@ describe("Claude workflow replay", () => {
     });
   });
 
-  it("replays the workflow result without exposing its script or child prompt", () => {
+  it("replays the workflow result without exposing its script or child prompt", async () => {
     const workflow = parseClaudeWorkflowRun(
       JSON.stringify({
         runId: RUN_ID,
@@ -128,7 +134,7 @@ describe("Claude workflow replay", () => {
     expect(workflow).not.toBeNull();
     if (!workflow) throw new Error("expected workflow");
 
-    const observations = observeReplayWorkflows({
+    const observations = await observeReplayWorkflows({
       workflows: [workflow],
       parentEntries: parentEntries(),
     });
@@ -145,9 +151,9 @@ describe("Claude workflow replay", () => {
     expect(JSON.stringify(observations)).not.toContain("SECRET CHILD PROMPT");
   });
 
-  it("does not duplicate a result already present in the child transcript", () => {
+  it("does not duplicate a result already present in the child transcript", async () => {
     const report = "What Paseo is\nA local-first coding-agent environment.";
-    const observations = observeReplayWorkflows({
+    const observations = await observeReplayWorkflows({
       workflows: [
         { runId: RUN_ID, summary: "Inspect Paseo", status: "completed", result: { report } },
       ],
@@ -166,8 +172,8 @@ describe("Claude workflow replay", () => {
     ).toHaveLength(1);
   });
 
-  it("fails a nonterminal persisted run because its former runtime cannot update it", () => {
-    const observations = observeReplayWorkflows({
+  it("fails a nonterminal persisted run because its former runtime cannot update it", async () => {
+    const observations = await observeReplayWorkflows({
       workflows: [{ runId: RUN_ID, summary: "Interrupted workflow", status: "running" }],
       parentEntries: parentEntries(),
     });
@@ -179,9 +185,9 @@ describe("Claude workflow replay", () => {
     });
   });
 
-  it("drops a run that no parent Workflow tool call declared", () => {
+  it("drops a run that no parent Workflow tool call declared", async () => {
     expect(
-      observeReplayWorkflows({
+      await observeReplayWorkflows({
         workflows: [{ runId: RUN_ID, status: "completed" }],
         parentEntries: [],
       }),
@@ -195,3 +201,29 @@ describe("Claude workflow replay", () => {
     },
   );
 });
+
+it("awaits image materialization before publishing replay observations", async () => {
+  const bytes = Buffer.from("async replay image fixture");
+  const convertEntry = async () => {
+    const item = await renderProviderImageOutputAsAssistantMarkdown(
+      { data: bytes.toString("base64"), mimeType: "image/png" },
+      { materialize: materializeProviderImage },
+    );
+    return item ? [item] : [];
+  };
+  const observations = await observeReplayWorkflows({
+    workflows: [{ runId: RUN_ID, status: "completed" }],
+    parentEntries: parentEntries(),
+    entriesByRunId: new Map([[RUN_ID, [{ type: "assistant", message: { content: [] } }]]]),
+    convertEntry,
+  });
+  const image = observations.find(
+    (observation) =>
+      observation.kind === "timeline" && observation.item.type === "assistant_message",
+  );
+  if (!image || image.kind !== "timeline" || image.item.type !== "assistant_message")
+    throw new Error("Expected replay image");
+  const source = image.item.text.match(/!\[Image\]\(([^)]+)\)/)?.[1];
+  expect(source).toBeTruthy();
+  expect(await readFile(fileURLToPath(source!))).toEqual(bytes);
+});

packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.ts

@@ -45,14 +45,17 @@ export function parseClaudeWorkflowRun(contents: string): ClaudeWorkflowRun | nu
  * provider runtime was recreated, so replay terminalizes it as failed instead of resurrecting a
  * row that can never receive another update.
  */
-export function observeReplayWorkflows(input: {
+export async function observeReplayWorkflows(input: {
   workflows: readonly ClaudeWorkflowRun[];
-  parentEntries: Iterable<ClaudeWorkflowParentEntry>;
+  parentEntries: Iterable<ClaudeWorkflowParentEntry> | AsyncIterable<ClaudeWorkflowParentEntry>;
   /** Repeatable entries in chronological order, supplied by the transcript reader. */
-  entriesByRunId?: ReadonlyMap<string, Iterable<ClaudeReplayEntry>>;
-  convertEntry?: (entry: ClaudeReplayEntry) => AgentTimelineItem[];
-}): SubagentObservation[] {
-  const toolCallIdByRunId = readWorkflowLinks(input.parentEntries);
+  entriesByRunId?: ReadonlyMap<
+    string,
+    Iterable<ClaudeReplayEntry> | AsyncIterable<ClaudeReplayEntry>
+  >;
+  convertEntry?: (entry: ClaudeReplayEntry) => AgentTimelineItem[] | Promise<AgentTimelineItem[]>;
+}): Promise<SubagentObservation[]> {
+  const toolCallIdByRunId = await readWorkflowLinks(input.parentEntries);
   const observations: SubagentObservation[] = [];
 
   for (const workflow of input.workflows) {
@@ -79,7 +82,7 @@ export function observeReplayWorkflows(input: {
       });
     }
 
-    for (const observation of observeWorkflowTimeline({
+    for (const observation of await observeWorkflowTimeline({
       id,
       result: workflow.result,
       entries: input.entriesByRunId?.get(workflow.runId) ?? [],
@@ -115,19 +118,19 @@ export function observeReplayWorkflows(input: {
   return observations;
 }
 
-function observeWorkflowTimeline(input: {
+async function observeWorkflowTimeline(input: {
   id: string;
   result: unknown;
-  entries: Iterable<ClaudeReplayEntry>;
-  convertEntry?: (entry: ClaudeReplayEntry) => AgentTimelineItem[];
+  entries: Iterable<ClaudeReplayEntry> | AsyncIterable<ClaudeReplayEntry>;
+  convertEntry?: (entry: ClaudeReplayEntry) => AgentTimelineItem[] | Promise<AgentTimelineItem[]>;
   finishedAt?: string;
-}): SubagentObservation[] {
+}): Promise<SubagentObservation[]> {
   const observations: SubagentObservation[] = [];
   const replayedAssistantText = new Set<string>();
 
-  for (const entry of input.entries) {
+  for await (const entry of input.entries) {
     const timestamp = normalizeProviderReplayTimestamp(entry.timestamp);
-    const items = input.convertEntry?.(entry) ?? [];
+    const items = (await input.convertEntry?.(entry)) ?? [];
     for (const item of items) {
       if (item.type === "assistant_message") replayedAssistantText.add(item.text.trim());
       observations.push({
@@ -150,11 +153,13 @@ function observeWorkflowTimeline(input: {
   return observations;
 }
 
-function readWorkflowLinks(entries: Iterable<ClaudeWorkflowParentEntry>): Map<string, string> {
+async function readWorkflowLinks(
+  entries: Iterable<ClaudeWorkflowParentEntry> | AsyncIterable<ClaudeWorkflowParentEntry>,
+): Promise<Map<string, string>> {
   const workflowToolCallIds = new Set<string>();
   const toolCallIdByRunId = new Map<string, string>();
 
-  for (const entry of entries) {
+  for await (const entry of entries) {
     const content = entry.message?.content;
     if (!Array.isArray(content)) continue;
     for (const value of content) {
@@ -165,7 +170,7 @@ function readWorkflowLinks(entries: Iterable<ClaudeWorkflowParentEntry>): Map<st
     }
   }
 
-  for (const entry of entries) {
+  for await (const entry of entries) {
     const content = entry.message?.content;
     if (!Array.isArray(content)) continue;
     for (const value of content) {

packages/server/src/server/agent/providers/claude/task-notification-tool-call.ts

@@ -221,9 +221,11 @@ function parseTaskNotificationFromSystemRecord(record: unknown): TaskNotificatio
 /** Normalize notification identity for one transcript traversal. Queue bookkeeping and delivery
  * are separate records of the same occurrence; each source counts repeated identical bodies
  * independently so the second event is not merged with the first. */
-export function* assignTaskNotificationReplayIds<T>(entries: Iterable<T>): Generator<T> {
+export async function* assignTaskNotificationReplayIds<T>(
+  entries: AsyncIterable<T>,
+): AsyncGenerator<T> {
   const occurrences = new Map<string, Map<string, number>>();
-  for (const entry of entries) {
+  for await (const entry of entries) {
     const record = TaskNotificationHistoryRecordSchema.safeParse(entry);
     if (!record.success) {
       yield entry;

packages/server/src/server/agent/providers/claude/transcript-history.ts

@@ -1,5 +1,6 @@
 import { assignTaskNotificationReplayIds } from "./task-notification-tool-call.js";
-import fs from "node:fs";
+import fs, { type FileHandle } from "node:fs/promises";
+import { setImmediate } from "node:timers/promises";
 import path from "node:path";
 import { normalizeProviderReplayTimestamp } from "../../provider-history-timestamps.js";
 import {
@@ -24,11 +25,11 @@ interface RecordLocation {
 }
 
 /** Chunk in bytes so UTF-8 and multi-chunk records are decoded once, at a line boundary. */
-function* locatedRecords(
+async function* locatedRecords(
   file: string,
-  fd: number,
-): Generator<{ entry: TranscriptEntry; location: RecordLocation }> {
-  const stat = fs.fstatSync(fd);
+  fd: FileHandle,
+): AsyncGenerator<{ entry: TranscriptEntry; location: RecordLocation }> {
+  const stat = await fd.stat();
   // Opening a directory succeeds on some platforms; it must never look like empty history.
   if (stat.isDirectory()) {
     throw Object.assign(new Error(`EISDIR: illegal operation on a directory, read '${file}'`), {
@@ -43,7 +44,7 @@ function* locatedRecords(
   let length = 0;
   while (remaining > 0) {
     const buffer = Buffer.allocUnsafe(Math.min(64 * 1024, remaining));
-    const count = fs.readSync(fd, buffer, 0, buffer.length, null);
+    const { bytesRead: count } = await fd.read(buffer, 0, buffer.length, null);
     if (count === 0) break;
     remaining -= count;
     let start = 0;
@@ -64,6 +65,8 @@ function* locatedRecords(
       if (entry) yield { entry, location };
       start = end + 1;
     }
+    // Bound parsing work as well as IO; microtasks alone do not service timers or sockets.
+    await setImmediate();
     if (start < count) {
       pieces.push(buffer.subarray(start, count));
       length += count - start;
@@ -85,20 +88,25 @@ function* locatedRecords(
 /** Indexed sources parse only their own records using descriptors owned by this replay. */
 function indexedRecords(
   locations: readonly RecordLocation[],
-  descriptors: ReadonlyMap<string, number>,
-): Iterable<TranscriptEntry> {
+  descriptors: ReadonlyMap<string, FileHandle>,
+): AsyncIterable<TranscriptEntry> {
   return {
-    *[Symbol.iterator]() {
+    async *[Symbol.asyncIterator]() {
       yield* assignTaskNotificationReplayIds(readEntries());
     },
   };
-  function* readEntries(): Generator<TranscriptEntry> {
+  async function* readEntries(): AsyncGenerator<TranscriptEntry> {
     for (const location of locations) {
       const fd = descriptors.get(location.file)!;
       const buffer = Buffer.allocUnsafe(location.length);
       let read = 0;
       while (read < buffer.length) {
-        const count = fs.readSync(fd, buffer, read, buffer.length - read, location.offset + read);
+        const { bytesRead: count } = await fd.read(
+          buffer,
+          read,
+          buffer.length - read,
+          location.offset + read,
+        );
         if (count === 0)
           throw new Error(`Claude transcript was truncated during replay: ${location.file}`);
         read += count;
@@ -127,9 +135,9 @@ function parseRecord(line: string): TranscriptEntry | null {
   }
 }
 
-function readOptionalFile(file: string): string | null {
+async function readOptionalFile(file: string): Promise<string | null> {
   try {
-    return fs.readFileSync(file, "utf8");
+    return await fs.readFile(file, "utf8");
   } catch {
     return null;
   }
@@ -138,31 +146,40 @@ function readOptionalFile(file: string): string | null {
 /** Discover sources without retaining transcript contents. Unrelated sidechains stay excluded
  * by the replay ownership resolver; workflow children retain their separate run ownership. */
 export interface ClaudeReplayHistory {
-  parentEntries: Iterable<TranscriptEntry>;
+  parentEntries: AsyncIterable<TranscriptEntry>;
   subagents: {
     agentId: string;
     meta: ClaudeSubagentMeta | null;
-    entries: Iterable<TranscriptEntry>;
+    entries: AsyncIterable<TranscriptEntry>;
   }[];
   workflows: ClaudeWorkflowRun[];
-  workflowEntriesByRunId: Map<string, Iterable<TranscriptEntry>>;
+  workflowEntriesByRunId: Map<string, AsyncIterable<TranscriptEntry>>;
 }
 
 /** Own descriptors through every pass so unlink or rename cannot invalidate discovered history.
  * Sources must be consumed before the callback returns. No transcript copies are needed. */
-export function withClaudeReplayHistory<T>(
+export async function withClaudeReplayHistory<T>(
   historyPath: string,
-  consume: (history: ClaudeReplayHistory) => T,
-): T {
-  const descriptors = new Map<string, number>();
+  consume: (history: ClaudeReplayHistory) => Promise<T>,
+): Promise<T> {
+  const descriptors = new Map<string, FileHandle>();
   try {
-    return consume(indexHistory(historyPath, descriptors));
+    return await consume(await indexHistory(historyPath, descriptors));
   } finally {
-    for (const fd of descriptors.values()) fs.closeSync(fd);
+    await closeReplaySources(descriptors.values());
   }
 }
 
-function indexHistory(historyPath: string, descriptors: Map<string, number>): ClaudeReplayHistory {
+async function closeReplaySources(descriptors: Iterable<FileHandle>): Promise<void> {
+  const results = await Promise.allSettled([...descriptors].map((fd) => fd.close()));
+  const failure = results.find((result) => result.status === "rejected");
+  if (failure) throw failure.reason;
+}
+
+async function indexHistory(
+  historyPath: string,
+  descriptors: Map<string, FileHandle>,
+): Promise<ClaudeReplayHistory> {
   const sessionDirectory = path.join(
     path.dirname(historyPath),
     path.basename(historyPath, ".jsonl"),
@@ -174,8 +191,7 @@ function indexHistory(historyPath: string, descriptors: Map<string, number>): Cl
   const directories = [sidechainDirectory];
   while (directories.length) {
     const directory = directories.pop()!;
-    if (!fs.existsSync(directory)) continue;
-    for (const entry of fs.readdirSync(directory, { withFileTypes: true })) {
+    for (const entry of await optionalDirectory(directory)) {
       const file = path.join(directory, entry.name);
       if (entry.isDirectory()) {
         directories.push(file);
@@ -194,23 +210,23 @@ function indexHistory(historyPath: string, descriptors: Map<string, number>): Cl
       }
       const agentId = /^agent-(.+)\.meta\.json$/.exec(entry.name)?.[1];
       if (!agentId) continue;
-      const contents = readOptionalFile(file);
+      const contents = await readOptionalFile(file);
       const meta = contents === null ? null : parseClaudeSubagentMeta(contents);
       if (meta) metaByAgentId.set(agentId, meta);
     }
   }
   const parentLocations: RecordLocation[] = [];
   const locationsByAgentId = new Map<string, RecordLocation[]>();
-  function scan(
+  async function scan(
     file: string,
     visit: (entry: TranscriptEntry, location: RecordLocation) => void,
-  ): void {
-    const fd = fs.openSync(file, "r");
+  ): Promise<void> {
+    const fd = await fs.open(file, "r");
     descriptors.set(file, fd);
-    for (const { entry, location } of locatedRecords(file, fd)) visit(entry, location);
+    for await (const { entry, location } of locatedRecords(file, fd)) visit(entry, location);
   }
   for (const file of sidechainFiles) {
-    scan(file, (entry, location) => {
+    await scan(file, (entry, location) => {
       if (file === historyPath && entry.isSidechain !== true) parentLocations.push(location);
       if (entry.isSidechain !== true || typeof entry.agentId !== "string") return;
       const locations = locationsByAgentId.get(entry.agentId) ?? [];
@@ -218,11 +234,11 @@ function indexHistory(historyPath: string, descriptors: Map<string, number>): Cl
       locationsByAgentId.set(entry.agentId, locations);
     });
   }
-  const workflowEntriesByRunId = new Map<string, Iterable<TranscriptEntry>>();
+  const workflowEntriesByRunId = new Map<string, AsyncIterable<TranscriptEntry>>();
   for (const [runId, files] of workflowFiles) {
     const locations: RecordLocation[] = [];
     for (const file of files)
-      scan(file, (entry, location) => {
+      await scan(file, (entry, location) => {
         if (entry.type !== "user" || isToolResult(entry)) locations.push(location);
       });
     locations.sort((a, b) => {
@@ -240,21 +256,28 @@ function indexHistory(historyPath: string, descriptors: Map<string, number>): Cl
       meta: metaByAgentId.get(agentId) ?? null,
       entries: indexedRecords(locations, descriptors),
     })),
-    workflows: readWorkflows(sessionDirectory),
+    workflows: await readWorkflows(sessionDirectory),
     workflowEntriesByRunId,
   };
 }
 
-function readWorkflows(sessionDirectory: string): ClaudeWorkflowRun[] {
+async function readWorkflows(sessionDirectory: string): Promise<ClaudeWorkflowRun[]> {
   const workflows: ClaudeWorkflowRun[] = [];
   const workflowDirectory = path.join(sessionDirectory, "workflows");
-  if (fs.existsSync(workflowDirectory)) {
-    for (const entry of fs.readdirSync(workflowDirectory, { withFileTypes: true })) {
-      if (!entry.isFile() || !entry.name.endsWith(".json")) continue;
-      const contents = readOptionalFile(path.join(workflowDirectory, entry.name));
-      const workflow = contents === null ? null : parseClaudeWorkflowRun(contents);
-      if (workflow) workflows.push(workflow);
-    }
+  for (const entry of await optionalDirectory(workflowDirectory)) {
+    if (!entry.isFile() || !entry.name.endsWith(".json")) continue;
+    const contents = await readOptionalFile(path.join(workflowDirectory, entry.name));
+    const workflow = contents === null ? null : parseClaudeWorkflowRun(contents);
+    if (workflow) workflows.push(workflow);
   }
   return workflows;
 }
+
+async function optionalDirectory(directory: string) {
+  try {
+    return await fs.readdir(directory, { withFileTypes: true });
+  } catch (error) {
+    if ((error as NodeJS.ErrnoException).code === "ENOENT") return [];
+    throw error;
+  }
+}

packages/server/src/server/agent/providers/codex-app-server-agent.ts

@@ -1939,12 +1939,12 @@ function codexImageOutputFromResult(result: unknown): ProviderImageOutput | null
   };
 }
 
-function mapCodexThreadImageItem(
+async function mapCodexThreadImageItem(
   normalizedType: string,
   normalizedItem: Record<string, unknown>,
-): AgentTimelineItem | null {
+): Promise<AgentTimelineItem | null> {
   if (normalizedType === "imageView") {
-    return renderProviderImageOutputAsAssistantMarkdown({
+    return await renderProviderImageOutputAsAssistantMarkdown({
       path: firstStringField(normalizedItem, ["path"]),
     });
   }
@@ -1960,7 +1960,7 @@ function mapCodexThreadImageItem(
 
   const savedPath = firstStringField(normalizedItem, ["savedPath", "saved_path"]);
   const result = codexImageOutputFromResult(normalizedItem.result);
-  return renderProviderImageOutputAsAssistantMarkdown(
+  return await renderProviderImageOutputAsAssistantMarkdown(
     {
       path: savedPath ?? result?.path ?? null,
       url: result?.url ?? null,
@@ -1982,10 +1982,10 @@ function mapCodexAgentMessage(item: Record<string, unknown>): AgentTimelineItem
   };
 }
 
-export function threadItemToTimeline(
+export async function threadItemToTimeline(
   item: unknown,
   options?: { includeUserMessage?: boolean; cwd?: string | null },
-): AgentTimelineItem | null {
+): Promise<AgentTimelineItem | null> {
   const itemRecord = toObjectRecord(item);
   if (!itemRecord) return null;
   const includeUserMessage = options?.includeUserMessage ?? true;
@@ -1999,7 +1999,7 @@ export function threadItemToTimeline(
       : itemRecord;
 
   if (normalizedType === "imageView" || normalizedType === "imageGeneration") {
-    return mapCodexThreadImageItem(normalizedType, normalizedItem);
+    return await mapCodexThreadImageItem(normalizedType, normalizedItem);
   }
   if (normalizedType && CODEX_TOOL_THREAD_ITEM_TYPES.has(normalizedType)) {
     return mapCodexToolCallFromThreadItem(normalizedItem, { cwd });
@@ -2024,7 +2024,7 @@ export function threadItemToTimeline(
   }
 }
 
-function mcpToolResultImagesToTimeline(item: unknown): AgentTimelineItem[] {
+async function mcpToolResultImagesToTimeline(item: unknown): Promise<AgentTimelineItem[]> {
   const itemRecord = toObjectRecord(item);
   if (!itemRecord) {
     return [];
@@ -2037,24 +2037,25 @@ function mcpToolResultImagesToTimeline(item: unknown): AgentTimelineItem[] {
   }
 
   const { images } = splitCodexMcpToolResultImages(itemRecord.result);
-  return images
-    .map((image) =>
-      renderProviderImageOutputAsAssistantMarkdown(image, {
-        materialize: materializeProviderImage,
-      }),
-    )
-    .filter((timelineItem): timelineItem is AgentTimelineItem => timelineItem !== null);
+  const items: AgentTimelineItem[] = [];
+  for (const image of images) {
+    const imageItem = await renderProviderImageOutputAsAssistantMarkdown(image, {
+      materialize: materializeProviderImage,
+    });
+    if (imageItem) items.push(imageItem);
+  }
+  return items;
 }
 
-function threadItemToTimelineEntries(
+async function threadItemToTimelineEntries(
   item: unknown,
   options?: { includeUserMessage?: boolean; cwd?: string | null },
-): AgentTimelineItem[] {
-  const timelineItem = threadItemToTimeline(item, options);
+): Promise<AgentTimelineItem[]> {
+  const timelineItem = await threadItemToTimeline(item, options);
   if (!timelineItem) {
     return [];
   }
-  return [timelineItem, ...mcpToolResultImagesToTimeline(item)];
+  return [timelineItem, ...(await mcpToolResultImagesToTimeline(item))];
 }
 
 const CodexThreadReadResponseSchema = z
@@ -2103,7 +2104,7 @@ async function loadCodexThreadHistoryTimeline(params: {
           historicalSubAgentActivity.agentThreadId,
         );
         if (existingIndex !== undefined) {
-          const activityTimelineItem = threadItemToTimeline(item, { cwd: params.cwd });
+          const activityTimelineItem = await threadItemToTimeline(item, { cwd: params.cwd });
           updateHistoricalSubAgentActivity(
             timeline,
             existingIndex,
@@ -2116,7 +2117,7 @@ async function loadCodexThreadHistoryTimeline(params: {
           continue;
         }
       }
-      for (const timelineItem of threadItemToTimelineEntries(item, { cwd: params.cwd })) {
+      for (const timelineItem of await threadItemToTimelineEntries(item, { cwd: params.cwd })) {
         const timestamp =
           readCodexHistoryTimestamp(item) ?? readCodexTurnHistoryTimestamp(turn, timelineItem);
         const settledTimelineItem =
@@ -3290,10 +3291,12 @@ export async function codexAppServerTurnInputFromPrompt(
     }
     if (block.type === "image") {
       try {
-        const filePath = materializeProviderImage({
-          data: block.data,
-          mimeType: block.mimeType,
-        }).path;
+        const filePath = (
+          await materializeProviderImage({
+            data: block.data,
+            mimeType: block.mimeType,
+          })
+        ).path;
         output.push({ type: "localImage", path: filePath });
       } catch (error) {
         const message = error instanceof Error ? error.message : String(error);
@@ -4370,6 +4373,7 @@ export class CodexAppServerAgentSession implements AgentSession {
     this.dismissPendingPlanApprovals("Dismissed by a new prompt");
 
     try {
+      await this.pendingNotifications;
       await this.connect();
       if (!this.client) {
         throw new Error("Codex client not initialized");
@@ -5437,7 +5441,21 @@ export class CodexAppServerAgentSession implements AgentSession {
     return `codex-turn-${this.nextTurnOrdinal++}`;
   }
 
-  private handleNotification(method: string, params: unknown): void {
+  private pendingNotifications: Promise<void> = Promise.resolve();
+
+  private handleNotification(method: string, params: unknown): Promise<void> {
+    // An image must exist before its event, and completion must follow that event.
+    // Transport callbacks cannot await: the session owns their ordered drain.
+    const next = this.pendingNotifications.then(() => {
+      return this.closed ? undefined : this.processNotification(method, params);
+    });
+    this.pendingNotifications = next.catch((error: unknown) => {
+      this.handleUnexpectedTermination(error instanceof Error ? error : new Error(String(error)));
+    });
+    return this.pendingNotifications;
+  }
+
+  private async processNotification(method: string, params: unknown): Promise<void> {
     const notificationParams = toObjectRecord(params);
     if (method === "serverRequest/resolved" && typeof notificationParams?.requestId === "number") {
       const requestId = this.mcpElicitationPermissionIds.get(notificationParams.requestId);
@@ -5467,13 +5485,16 @@ export class CodexAppServerAgentSession implements AgentSession {
       return;
     }
     if (route.kind === "sub_agent") {
-      this.dispatchSubAgentNotification(parsed, route.callId);
+      await this.dispatchSubAgentNotification(parsed, route.callId);
       return;
     }
-    this.dispatchParsedNotification(parsed);
+    await this.dispatchParsedNotification(parsed);
   }
 
-  private dispatchSubAgentNotification(parsed: ParsedCodexNotification, callId: string): void {
+  private async dispatchSubAgentNotification(
+    parsed: ParsedCodexNotification,
+    callId: string,
+  ): Promise<void> {
     switch (parsed.kind) {
       case "thread_started":
         this.emitSubAgentActivityUpdate(callId, "running", { reopen: true });
@@ -5484,7 +5505,7 @@ export class CodexAppServerAgentSession implements AgentSession {
       case "reasoning_delta":
       case "item_started":
       case "item_completed":
-        this.dispatchParsedNotification(parsed);
+        await this.dispatchParsedNotification(parsed);
         return;
       case "exec_command_output_delta":
       case "file_change_output_delta":
@@ -5510,7 +5531,7 @@ export class CodexAppServerAgentSession implements AgentSession {
     }
   }
 
-  private dispatchParsedNotification(parsed: ParsedCodexNotification): void {
+  private async dispatchParsedNotification(parsed: ParsedCodexNotification): Promise<void> {
     if (isCodexDeltaNotification(parsed)) {
       this.handleCodexDeltaNotification(parsed);
       return;
@@ -5555,10 +5576,10 @@ export class CodexAppServerAgentSession implements AgentSession {
         this.handlePatchApplyCompletedNotification(parsed);
         return;
       case "item_completed":
-        this.handleItemCompletedNotification(parsed);
+        await this.handleItemCompletedNotification(parsed);
         return;
       case "item_started":
-        this.handleItemStartedNotification(parsed);
+        await this.handleItemStartedNotification(parsed);
         return;
       case "invalid_payload":
         this.warnInvalidNotificationPayload(parsed.method, parsed.params);
@@ -5600,7 +5621,7 @@ export class CodexAppServerAgentSession implements AgentSession {
     pending.push(parsed);
   }
 
-  private replayPendingSubAgentNotifications(threadIds: readonly string[]): void {
+  private async replayPendingSubAgentNotifications(threadIds: readonly string[]): Promise<void> {
     for (const threadId of threadIds) {
       const pending = this.pendingSubAgentNotificationsByThreadId.get(threadId);
       if (!pending) {
@@ -5612,7 +5633,7 @@ export class CodexAppServerAgentSession implements AgentSession {
         continue;
       }
       for (const parsed of pending) {
-        this.dispatchSubAgentNotification(parsed, callId);
+        await this.dispatchSubAgentNotification(parsed, callId);
       }
     }
   }
@@ -6578,9 +6599,13 @@ export class CodexAppServerAgentSession implements AgentSession {
     }
   }
 
-  private handleItemCompletedNotification(
+  private ownsNotificationTurn(turnId: string | null): boolean {
+    return !this.closed && this.activeForegroundTurnId === turnId;
+  }
+
+  private async handleItemCompletedNotification(
     parsed: Extract<ParsedCodexNotification, { kind: "item_completed" }>,
-  ): void {
+  ): Promise<void> {
     // Codex emits mirrored lifecycle notifications via both `codex/event/item_*`
     // and canonical `item/*`. Render ordinary items only from the canonical
     // channel, but accept a legacy-only child announcement so it can establish
@@ -6590,20 +6615,23 @@ export class CodexAppServerAgentSession implements AgentSession {
     }
     this.receiveAsyncQuestion(parsed.threadId, parsed.item);
     if (this.isUserMessageItem(parsed.item)) {
-      this.handleUserMessageItem(parsed);
+      await this.handleUserMessageItem(parsed);
       return;
     }
     const childSubAgentCallId = this.getSubAgentCallIdForThread(parsed.threadId);
     if (this.handleCompletedSpecialItem(parsed, childSubAgentCallId)) {
       return;
     }
-    const timelineItem = threadItemToTimeline(parsed.item, {
+    const foregroundTurnId = this.activeForegroundTurnId;
+    const timelineItem = await threadItemToTimeline(parsed.item, {
       includeUserMessage: false,
       cwd: this.config.cwd ?? null,
     });
     if (!timelineItem) {
       return;
     }
+    const imageItems = await mcpToolResultImagesToTimeline(parsed.item);
+    if (!this.ownsNotificationTurn(foregroundTurnId)) return;
     const registeredChildThreadIds =
       timelineItem.type === "tool_call"
         ? this.registerSubAgentToolCall({
@@ -6613,12 +6641,11 @@ export class CodexAppServerAgentSession implements AgentSession {
             parentSubagentId: resolveCodexParentSubagentId(childSubAgentCallId, parsed.threadId),
           })
         : [];
-    const imageItems = mcpToolResultImagesToTimeline(parsed.item);
     if (childSubAgentCallId) {
       this.emitCompletedProviderSubagentItem(parsed, timelineItem);
       this.handleSubAgentChildItemCompleted(childSubAgentCallId, parsed.item.id, timelineItem);
       this.emitProviderSubagentTimelineItems(parsed.threadId, imageItems);
-      this.replayPendingSubAgentNotifications(registeredChildThreadIds);
+      await this.replayPendingSubAgentNotifications(registeredChildThreadIds);
       return;
     }
     const normalizedItemType = normalizeCodexThreadItemType(
@@ -6626,7 +6653,7 @@ export class CodexAppServerAgentSession implements AgentSession {
     );
     const itemId = parsed.item.id;
     if (this.shouldSkipCompletedThreadItem(timelineItem, normalizedItemType, itemId)) {
-      this.replayPendingSubAgentNotifications(registeredChildThreadIds);
+      await this.replayPendingSubAgentNotifications(registeredChildThreadIds);
       return;
     }
     if (this.consumeStreamedTextCompletion(timelineItem, itemId)) {
@@ -6637,7 +6664,7 @@ export class CodexAppServerAgentSession implements AgentSession {
         this.emittedItemCompletedIds.add(itemId);
         this.emittedItemStartedIds.delete(itemId);
       }
-      this.replayPendingSubAgentNotifications(registeredChildThreadIds);
+      await this.replayPendingSubAgentNotifications(registeredChildThreadIds);
       return;
     }
     this.applyBufferedDeltaTextToTimelineItem(timelineItem, itemId);
@@ -6666,7 +6693,7 @@ export class CodexAppServerAgentSession implements AgentSession {
       this.pendingCommandOutputDeltas.delete(itemId);
       this.pendingFileChangeOutputDeltas.delete(itemId);
     }
-    this.replayPendingSubAgentNotifications(registeredChildThreadIds);
+    await this.replayPendingSubAgentNotifications(registeredChildThreadIds);
   }
 
   private consumeStreamedTextCompletion(
@@ -6738,14 +6765,14 @@ export class CodexAppServerAgentSession implements AgentSession {
     }
   }
 
-  private handleItemStartedNotification(
+  private async handleItemStartedNotification(
     parsed: Extract<ParsedCodexNotification, { kind: "item_started" }>,
-  ): void {
+  ): Promise<void> {
     if (shouldIgnoreMirroredLifecycleItem(parsed.source, parsed.item)) {
       return;
     }
     if (this.isUserMessageItem(parsed.item)) {
-      this.handleUserMessageItem(parsed);
+      await this.handleUserMessageItem(parsed);
       return;
     }
     const childSubAgentCallId = this.getSubAgentCallIdForThread(parsed.threadId);
@@ -6767,7 +6794,7 @@ export class CodexAppServerAgentSession implements AgentSession {
     if (this.handleRegisteredSubAgentActivity(parsed.item)) {
       return;
     }
-    const timelineItem = threadItemToTimeline(parsed.item, {
+    const timelineItem = await threadItemToTimeline(parsed.item, {
       includeUserMessage: false,
       cwd: this.config.cwd ?? null,
     });
@@ -6786,7 +6813,7 @@ export class CodexAppServerAgentSession implements AgentSession {
         this.upsertSubAgentChildItem(childSubAgentCallId, parsed.item.id, timelineItem);
       }
       this.emitSubAgentActivityUpdate(childSubAgentCallId, "running");
-      this.replayPendingSubAgentNotifications(registeredChildThreadIds);
+      await this.replayPendingSubAgentNotifications(registeredChildThreadIds);
       return;
     }
     const normalizedItemType = normalizeCodexThreadItemType(
@@ -6810,14 +6837,14 @@ export class CodexAppServerAgentSession implements AgentSession {
       this.pendingCommandOutputDeltas.delete(itemId);
       this.pendingFileChangeOutputDeltas.delete(itemId);
     }
-    this.replayPendingSubAgentNotifications(registeredChildThreadIds);
+    await this.replayPendingSubAgentNotifications(registeredChildThreadIds);
   }
 
-  private handleUserMessageItem(
+  private async handleUserMessageItem(
     parsed: Extract<ParsedCodexNotification, { kind: "item_started" | "item_completed" }>,
-  ): void {
+  ): Promise<void> {
     const itemId = parsed.item.id;
-    const timelineItem = threadItemToTimeline(parsed.item, {
+    const timelineItem = await threadItemToTimeline(parsed.item, {
       includeUserMessage: true,
       cwd: this.config.cwd ?? null,
     });

packages/server/src/server/agent/providers/omp/agent.test.ts

@@ -1133,17 +1133,6 @@ describe("OMP agent client and session", () => {
         messageId: "assistant-history",
       },
     ]);
-    await expect(
-      omp.runPrompt("continue after resume", "resumed follow-up complete"),
-    ).resolves.toMatchObject({
-      finalText: "resumed follow-up complete",
-    });
-    expect(omp.timeline()).toContainEqual(
-      expect.objectContaining({
-        type: "assistant_message",
-        text: "resumed follow-up complete",
-      }),
-    );
     expect(omp.usageSession()).toMatchObject({ provider: "omp", sessionKey: expect.any(String) });
     await omp.close();
     expect(omp.usageSession()).toBeNull();
@@ -1343,13 +1332,14 @@ describe("OMP agent client and session", () => {
     expect(omp.runtimeLaunches()).toHaveLength(1);
   });
 
-  test("closes a replacement process if the session closes during relaunch", async () => {
+  test("rejects a pending recovery start and closes its processes when the session closes", async () => {
     const omp = new OmpHarness();
     await omp.start();
     omp.processExit("OMP RPC process exited with code 1 and signal null");
     const turn = omp.startTurnDetached("continue");
+    const rejected = expect(turn).rejects.toThrow("closed");
     await omp.close();
-    await turn;
+    await rejected;
     await waitForImmediate();
     expect(omp.runtimeSessions().every((session) => session.closed)).toBe(true);
   });
@@ -1638,3 +1628,93 @@ test.each([
     expect(await session.canStopRuntime!()).toBe(expected);
   },
 );
+
+test.each([false, true])(
+  "image prompt keeps OMP's vision capability contract (vision: %s)",
+  async (vision) => {
+    const runtime = new FakeOmp();
+    runtime.queueSessionSetup((session) => {
+      session.state = {
+        ...session.state,
+        model: {
+          provider: "test",
+          id: "image-model",
+          input: vision ? ["text", "image"] : ["text"],
+        },
+      };
+    });
+    const client = new OmpAgentClient({ logger: createTestLogger(), runtime });
+    const session = await client.createSession({ provider: "omp", cwd: process.cwd() });
+    onTestFinished(() => session.close());
+    const bytes = Buffer.from("OMP attachment fixture");
+    const first = session.startTurn([
+      { type: "text", text: "Describe this image." },
+      { type: "image", data: bytes.toString("base64"), mimeType: "image/png" },
+    ]);
+    await expect(session.startTurn("overlapping prompt")).rejects.toThrow("already active");
+    await first;
+    const [prompt] = runtime.latestSession().prompts;
+    expect(prompt.imageCount).toBe(vision ? 1 : 0);
+    if (vision) expect(prompt.message).toBe("Describe this image.");
+    else {
+      const imagePath = prompt.message.match(/\[Image available at: (.+)\]/)?.[1];
+      expect(imagePath).toBeTruthy();
+      const { readFile, rm } = await import("node:fs/promises");
+      expect(await readFile(imagePath!)).toEqual(bytes);
+      await rm(imagePath!);
+    }
+  },
+);
+
+test("interrupting OMP during image conversion cancels the admitted turn and permits a follow-up", async () => {
+  const runtime = new FakeOmp();
+  const client = new OmpAgentClient({ logger: createTestLogger(), runtime });
+  const session = await client.createSession({ provider: "omp", cwd: process.cwd() });
+  onTestFinished(() => session.close());
+  const events: AgentStreamEvent[] = [];
+  session.subscribe((event) => events.push(event));
+  let admitted = false;
+  const first = session
+    .startTurn([
+      {
+        type: "image",
+        data: Buffer.alloc(16 * 1024 * 1024, 53).toString("base64"),
+        mimeType: "image/png",
+      },
+    ])
+    .then((result) => {
+      admitted = true;
+      return result;
+    });
+  await waitForImmediate();
+  expect(admitted).toBe(false);
+  expect(runtime.latestSession().prompts).toHaveLength(0);
+  await session.interrupt();
+  const followup = await session.startTurn("follow-up after image interrupt");
+  const original = await first;
+  expect(events).toContainEqual(
+    expect.objectContaining({ type: "turn_canceled", turnId: original.turnId }),
+  );
+  expect(followup.turnId).not.toBe(original.turnId);
+  expect(runtime.latestSession().prompts).toEqual([
+    { message: "follow-up after image interrupt", imageCount: 0 },
+  ]);
+});
+
+test("closing OMP during image conversion rejects its pending start", async () => {
+  const runtime = new FakeOmp();
+  const client = new OmpAgentClient({ logger: createTestLogger(), runtime });
+  const session = await client.createSession({ provider: "omp", cwd: process.cwd() });
+  const first = session.startTurn([
+    {
+      type: "image",
+      data: Buffer.alloc(16 * 1024 * 1024, 54).toString("base64"),
+      mimeType: "image/png",
+    },
+  ]);
+  const rejected = expect(first).rejects.toThrow("closed");
+  await waitForImmediate();
+  expect(runtime.latestSession().prompts).toHaveLength(0);
+  await session.close();
+  await rejected;
+});

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

@@ -317,9 +317,9 @@ function ompModelSupportsImageInput(model: OmpModel | null | undefined): boolean
   return model?.input?.includes("image") === true;
 }
 
-function renderTextOnlyImageHint(image: { data: string; mimeType: string }): string {
+async function renderTextOnlyImageHint(image: { data: string; mimeType: string }): Promise<string> {
   try {
-    const materialized = materializeProviderImage({
+    const materialized = await materializeProviderImage({
       data: image.data,
       mimeType: image.mimeType,
     });
@@ -329,10 +329,10 @@ function renderTextOnlyImageHint(image: { data: string; mimeType: string }): str
   }
 }
 
-function convertPromptInput(
+async function convertPromptInput(
   prompt: AgentPromptInput,
   options: { model: OmpModel | null | undefined },
-): OmpPromptPayload {
+): Promise<OmpPromptPayload> {
   if (typeof prompt === "string") {
     return { text: prompt };
   }
@@ -355,7 +355,7 @@ function convertPromptInput(
           mimeType: block.mimeType,
         });
       } else {
-        textParts.push(renderTextOnlyImageHint(block));
+        textParts.push(await renderTextOnlyImageHint(block));
       }
       continue;
     }
@@ -868,11 +868,9 @@ export class OmpAgentSession implements AgentSession {
       throw new Error("An OMP turn is already active");
     }
 
-    const payload = convertPromptInput(prompt, { model: this.state.model });
     const turnId = randomUUID();
     this.live = true;
     this.activeTurnId = turnId;
-    this.rememberClientMessage(options?.clientMessageId ?? null, payload.text);
     this.activeAssistantMessageId = null;
     this.activeTurnTerminalAssistantMessage = null;
     this.activeTurnStarted = false;
@@ -880,6 +878,10 @@ export class OmpAgentSession implements AgentSession {
     this.activeTurnHasUserMessage = false;
     this.activePromptRequestId = null;
     this.clearNoTurnBuffers();
+    const payload = await convertPromptInput(prompt, { model: this.state.model });
+    if (this.closed) throw new Error("OMP session is closed");
+    if (this.activeTurnId !== turnId) return { turnId };
+    this.rememberClientMessage(options?.clientMessageId ?? null, payload.text);
     this.activeNoTurnPromptText = payload.text;
     this.usagePoller.startTurn();
     if (announceTurn) {
@@ -946,7 +948,9 @@ export class OmpAgentSession implements AgentSession {
   ): Promise<SteerResult> {
     if (this.closed || this.activeTurnId !== options.expectedTurnId)
       return { status: "unavailable" };
-    const payload = convertPromptInput(prompt, { model: this.state.model });
+    const payload = await convertPromptInput(prompt, { model: this.state.model });
+    if (this.closed || this.activeTurnId !== options.expectedTurnId)
+      return { status: "unavailable" };
     if (this.parseSlashCommandInput(payload.text)) return { status: "unavailable" };
     const submission = this.rememberClientMessage(options.clientMessageId ?? null, payload.text);
     try {

packages/server/src/server/agent/providers/omp/history-mapper.test.ts

@@ -1,5 +1,4 @@
 import { mkdirSync, mkdtempSync, writeFileSync } from "node:fs";
-import { mkdtemp, rm, writeFile } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { describe, expect, test } from "vitest";
@@ -9,7 +8,7 @@ import { streamOmpCoreHistory, type OmpCapturedUserMessageEntry } from "./messag
 import type { OmpAgentMessage } from "./rpc-types.js";
 import { FakeOmp } from "./test-utils/fake-omp.js";
 import { OMP_HISTORY_MAPPER_HOOKS } from "./history-hooks.js";
-import { readActiveOmpEntryChain, readOmpHistoryTodoState, streamOmpHistory } from "./history.js";
+import { readOmpHistoryTodoState, streamOmpHistory } from "./history.js";
 
 async function collectHistory(
   messages: OmpAgentMessage[],
@@ -1246,53 +1245,3 @@ describe("OMP history mapper", () => {
     );
   });
 });
-
-test("indexes out-of-order branches, duplicate IDs and cycles before selecting history", async () => {
-  const directory = await mkdtemp(join(tmpdir(), "omp-history-branches-"));
-  const sessionFile = join(directory, "session.jsonl");
-  const root = { id: "root", parentId: null };
-  const earlyLeaf = {
-    id: "early",
-    parentId: "root",
-    message: { role: "user", content: "你好🌍".repeat(15000) },
-  };
-  const branch = { id: "branch", parentId: "root" };
-  const active = { id: "active", parentId: "branch" };
-  const duplicate = {
-    id: "branch",
-    parentId: "root",
-    message: { role: "user", content: "replacement" },
-  };
-  try {
-    await writeFile(
-      sessionFile,
-      [earlyLeaf, active, root, branch, duplicate]
-        .map((entry) => JSON.stringify(entry))
-        .join("\n") + '\n{"id":',
-    );
-    // The final record is a branch ancestor, not the active leaf. Latest duplicate wins
-    // when following parents; the invalid tail contributes neither a leaf nor a parent.
-    await expect(readActiveOmpEntryChain(sessionFile)).resolves.toEqual([root, duplicate, active]);
-    await expect(readActiveOmpEntryChain(sessionFile, "early")).resolves.toEqual([root, earlyLeaf]);
-    await expect(readActiveOmpEntryChain(sessionFile, "missing")).resolves.toEqual([
-      root,
-      duplicate,
-      active,
-    ]);
-    const cycle = [
-      { id: "a", parentId: "b" },
-      { id: "b", parentId: "a" },
-    ];
-    await writeFile(sessionFile, cycle.map((entry) => JSON.stringify(entry)).join("\n"));
-    await expect(readActiveOmpEntryChain(sessionFile)).resolves.toEqual(cycle);
-    await writeFile(sessionFile, "null\n{}\ninvalid\n");
-    await expect(readActiveOmpEntryChain(sessionFile)).resolves.toEqual([]);
-    await expect(readActiveOmpEntryChain(directory)).rejects.toMatchObject({ code: "EISDIR" });
-    await expect(readActiveOmpEntryChain(join(directory, "missing"))).rejects.toMatchObject({
-      code: "ENOENT",
-    });
-    await expect(readOmpHistoryTodoState(join(directory, "missing"))).resolves.toBeNull();
-  } finally {
-    await rm(directory, { recursive: true, force: true });
-  }
-});

packages/server/src/server/agent/providers/omp/history.e2e.test.ts

@@ -1,104 +0,0 @@
-import { BUILTIN_PROVIDER_IDS } from "@getpaseo/protocol/provider-manifest";
-import { mkdtemp, open, rm } from "node:fs/promises";
-import { tmpdir } from "node:os";
-import { join } from "node:path";
-import { performance } from "node:perf_hooks";
-import { setTimeout as delay } from "node:timers/promises";
-import { expect, test } from "vitest";
-import { createTestPaseoDaemon } from "../../../test-utils/paseo-daemon.js";
-import { DaemonClient } from "../../../test-utils/daemon-client.js";
-import { readActiveOmpEntryChain } from "./history.js";
-
-test("replays selected OMP history without blocking the event loop", async () => {
-  const directory = await mkdtemp(join(tmpdir(), "omp-history-budget-"));
-  const sessionFile = join(directory, "session.jsonl");
-  const started = performance.now();
-  try {
-    const daemon = await createTestPaseoDaemon({
-      // This regression exercises file replay and daemon requests, not executable discovery.
-      providerOverrides: Object.fromEntries(
-        BUILTIN_PROVIDER_IDS.map((id) => [id, { enabled: false }]),
-      ),
-    });
-    const client = new DaemonClient({
-      url: `ws://127.0.0.1:${daemon.port}/ws`,
-      appVersion: "0.13.2",
-    });
-    try {
-      await client.connect();
-      await client.fetchAgents({ subscribe: {} });
-      await client.getDaemonStatus();
-      const file = await open(sessionFile, "w");
-      // Structured message blocks expose document-sized parsing at a modest byte size.
-      // Replay an earlier active entry so inactive payload retention cannot drive the probe.
-      const text = Array.from({ length: 64 }, () => ({ type: "text", text: "x" }));
-      let bytes = 0;
-      let count = 0;
-      try {
-        while (bytes < 28 * 1024 * 1024) {
-          const line =
-            JSON.stringify({
-              type: "message",
-              id: `row-${count}`,
-              parentId: count ? `row-${count - 1}` : null,
-              message: { role: "user", content: text },
-            }) + "\n";
-          bytes += Buffer.byteLength(line);
-          await file.write(line);
-          count++;
-        }
-      } finally {
-        await file.close();
-      }
-      let last = performance.now();
-      let longestGap = 0;
-      const probe = setInterval(() => {
-        const now = performance.now();
-        longestGap = Math.max(longestGap, now - last);
-        last = now;
-      }, 5);
-      try {
-        const replayState = { finished: false };
-        let repliesDuringReplay = 0;
-        const replay = readActiveOmpEntryChain(sessionFile, "row-0").finally(() => {
-          replayState.finished = true;
-        });
-        // A real WebSocket RPC served by the isolated daemon on the same event loop.
-        const requests = (async () => {
-          while (!replayState.finished) {
-            await client.getDaemonStatus();
-            if (!replayState.finished) repliesDuringReplay++;
-            await delay(10);
-          }
-        })();
-        const [entries] = await Promise.all([replay, requests]);
-        await delay(10);
-        console.log(
-          JSON.stringify({
-            bytes,
-            count,
-            longestGap,
-            repliesDuringReplay,
-            generationAndLoadMs: performance.now() - started,
-          }),
-        );
-        expect(entries).toEqual([
-          {
-            type: "message",
-            id: "row-0",
-            parentId: null,
-            message: { role: "user", content: text },
-          },
-        ]);
-        expect(longestGap).toBeLessThan(50);
-        expect(repliesDuringReplay).toBeGreaterThan(0);
-      } finally {
-        clearInterval(probe);
-      }
-    } finally {
-      await client.close().finally(() => daemon.close());
-    }
-  } finally {
-    await rm(directory, { recursive: true, force: true });
-  }
-}, 30000);

packages/server/src/server/agent/providers/omp/history.ts

@@ -1,5 +1,4 @@
-import { open, type FileHandle } from "node:fs/promises";
-import { setImmediate } from "node:timers/promises";
+import { readFile } from "node:fs/promises";
 import { basename, extname, join } from "node:path";
 import type { AgentProvider, AgentStreamEvent, AgentTimelineItem } from "../../agent-sdk-types.js";
 import { normalizeProviderReplayTimestamp } from "../../provider-history-timestamps.js";
@@ -97,9 +96,7 @@ export async function* streamOmpHistory(input: {
   const messages: OmpAgentMessage[] = [];
   const replayEntries: OmpReplayEntry[] = [];
   const userEntries: OmpCapturedUserMessageEntry[] = [];
-  let prepared = 0;
   for (const entry of entries) {
-    if (++prepared % 32 === 0) await setImmediate();
     if (entry.type === "compaction") {
       replayEntries.push({ entry, item: mapCompactionEntry(entry) });
       continue;
@@ -118,9 +115,7 @@ export async function* streamOmpHistory(input: {
     OMP_HISTORY_MAPPER_HOOKS,
     input.bridgedTools,
   );
-  let replayed = 0;
   for (const replayEntry of replayEntries) {
-    if (++replayed % 32 === 0) await setImmediate();
     const timestamp = normalizeProviderReplayTimestamp(replayEntry.entry.timestamp);
     const events: AgentStreamEvent[] =
       "item" in replayEntry
@@ -334,139 +329,34 @@ function stripExtension(filePath: string): string {
   return extension ? filePath.slice(0, -extension.length) : filePath;
 }
 
-interface OmpEntryLocation {
-  id: string;
-  parentId?: string | null;
-  offset: number;
-}
-
-// Yield after each bounded IO chunk, including chunks containing only tiny or invalid records.
-// Keep bytes until a complete line so multi-byte UTF-8 characters survive chunk boundaries.
-async function* readOmpLines(
-  file: FileHandle,
-  size: number,
-): AsyncGenerator<{
-  line: Buffer;
-  offset: number;
-}> {
-  const chunkSize = 64 * 1024;
-  let fragments: Buffer[] = [];
-  let lineOffset = 0;
-  for (let position = 0; position < size; ) {
-    const chunk = Buffer.allocUnsafe(Math.min(chunkSize, size - position));
-    const { bytesRead } = await file.read(chunk, 0, chunk.length, position);
-    if (!bytesRead) break;
-    let start = 0;
-    for (let index = 0; index < bytesRead; index++) {
-      if (chunk[index] !== 10) continue;
-      const tail = chunk.subarray(start, index);
-      const line = fragments.length ? Buffer.concat([...fragments, tail]) : tail;
-      yield { line, offset: lineOffset };
-      fragments = [];
-      start = index + 1;
-      lineOffset = position + start;
-    }
-    if (start < bytesRead) fragments.push(chunk.subarray(start, bytesRead));
-    position += bytesRead;
-    await setImmediate();
-  }
-  if (fragments.length) yield { line: Buffer.concat(fragments), offset: lineOffset };
-}
-
-function parseOmpEntry(line: Buffer): OmpSessionEntry | null {
-  try {
-    const value: unknown = JSON.parse(line.toString("utf8"));
-    return value && typeof value === "object" && typeof Reflect.get(value, "id") === "string"
-      ? (value as OmpSessionEntry)
-      : null;
-  } catch {
-    // OMP can leave a partial last record; malformed records have always been ignored.
-    return null;
-  }
-}
-
-async function selectOmpChain(
-  locations: readonly OmpEntryLocation[],
-  byId: ReadonlyMap<string, OmpEntryLocation>,
-  parentIds: ReadonlySet<string>,
+export async function readActiveOmpEntryChain(
+  sessionFile: string,
   activeEntryId?: string,
-): Promise<OmpEntryLocation[]> {
-  let current = activeEntryId ? byId.get(activeEntryId) : undefined;
-  if (!current) {
-    for (let index = locations.length - 1; index >= 0; index--) {
-      const location = locations[index]!;
-      if (!parentIds.has(location.id)) {
-        current = location;
-        break;
-      }
-      if (index % 256 === 0) await setImmediate();
+): Promise<OmpSessionEntry[]> {
+  const content = await readFile(sessionFile, "utf8");
+  const entries = content.split("\n").flatMap((line) => {
+    if (!line.trim()) return [];
+    try {
+      const value = JSON.parse(line) as OmpSessionEntry;
+      return value && typeof value === "object" && typeof value.id === "string" ? [value] : [];
+    } catch {
+      return [];
     }
-    current ??= locations.at(-1);
-  }
-  const chain: OmpEntryLocation[] = [];
+  });
+  if (entries.length === 0) return [];
+  const byId = new Map(entries.map((entry) => [entry.id!, entry]));
+  const parentIds = new Set(entries.flatMap((entry) => (entry.parentId ? [entry.parentId] : [])));
+  const leaves = entries.filter((entry) => !parentIds.has(entry.id!));
+  let current: OmpSessionEntry | undefined =
+    (activeEntryId ? byId.get(activeEntryId) : undefined) ?? leaves.at(-1) ?? entries.at(-1);
+  const chain: OmpSessionEntry[] = [];
   const seen = new Set<string>();
   while (current?.id && !seen.has(current.id)) {
     chain.push(current);
     seen.add(current.id);
     current = current.parentId ? byId.get(current.parentId) : undefined;
-    if (chain.length % 256 === 0) await setImmediate();
-  }
-  return chain;
-}
-
-export async function readActiveOmpEntryChain(
-  sessionFile: string,
-  activeEntryId?: string,
-): Promise<OmpSessionEntry[]> {
-  const file = await open(sessionFile, "r");
-  try {
-    const stat = await file.stat();
-    if (stat.isDirectory()) {
-      throw Object.assign(
-        new Error(`EISDIR: illegal operation on a directory, read '${sessionFile}'`),
-        {
-          code: "EISDIR",
-        },
-      );
-    }
-    // One descriptor and a fixed byte boundary exclude appends made while replay is yielding.
-    // Retain graph metadata only: abandoned branches must not retain their message payloads.
-    const locations: OmpEntryLocation[] = [];
-    const byId = new Map<string, OmpEntryLocation>();
-    const parentIds = new Set<string>();
-    for await (const { line, offset } of readOmpLines(file, stat.size)) {
-      const entry = parseOmpEntry(line);
-      if (!entry) continue;
-      const location = { id: entry.id!, parentId: entry.parentId, offset };
-      locations.push(location);
-      byId.set(location.id, location);
-      if (location.parentId) parentIds.add(location.parentId);
-    }
-    const chain = await selectOmpChain(locations, byId, parentIds, activeEntryId);
-    if (chain.length === 0) return [];
-    const selected = new Map<number, number>();
-    for (let index = 0; index < chain.length; index++) {
-      selected.set(chain[index]!.offset, chain.length - index - 1);
-      if (index % 256 === 0) await setImmediate();
-    }
-    const entries: OmpSessionEntry[] = [];
-    for await (const { line, offset } of readOmpLines(file, stat.size)) {
-      const index = selected.get(offset);
-      if (index === undefined) continue;
-      const entry = parseOmpEntry(line);
-      // Truncation in place invalidates the indexed snapshot; never publish a partial chain.
-      if (!entry || entry.id !== chain[chain.length - index - 1]!.id) {
-        throw new Error(`OMP transcript changed during replay: ${sessionFile}`);
-      }
-      entries[index] = entry;
-      selected.delete(offset);
-      if (selected.size === 0) break;
-    }
-    if (selected.size) throw new Error(`OMP transcript changed during replay: ${sessionFile}`);
-    return entries;
-  } finally {
-    await file.close();
   }
+  return chain.toReversed();
 }
 
 function mapEntryMessage(entry: OmpSessionEntry): OmpAgentMessage | null {

packages/server/src/server/agent/providers/opencode/event-translator.test.ts

@@ -1560,7 +1560,7 @@ describe("translateOpenCodeEvent", () => {
 });
 
 describe("OpenCode v2 history reconciliation", () => {
-  it("keeps client message identity on live echoes but omits it from history", () => {
+  it("keeps client message identity on live echoes but omits it from history", async () => {
     const user: SessionMessageUser = {
       id: "native-message",
       type: "user",
@@ -1568,13 +1568,13 @@ describe("OpenCode v2 history reconciliation", () => {
       time: { created: 1 },
       metadata: { paseoClientMessageId: "client-message" },
     };
-    expect(new V2Timeline().messages([user])).toMatchObject([
+    expect(await new V2Timeline().messages([user])).toMatchObject([
       { item: { clientMessageId: "client-message", messageId: "native-message" } },
     ]);
-    expect(new V2Timeline(false).messages([user])).toMatchObject([
+    expect(await new V2Timeline(false).messages([user])).toMatchObject([
       { item: { type: "user_message", messageId: "native-message" } },
     ]);
-    const [event] = new V2Timeline(false).messages([user]);
+    const [event] = await new V2Timeline(false).messages([user]);
     expect(event && "item" in event && event.item).not.toHaveProperty("clientMessageId");
   });
   function assistant(text: string): SessionMessageAssistant {
@@ -1587,31 +1587,33 @@ describe("OpenCode v2 history reconciliation", () => {
       content: [{ type: "text", text }],
     };
   }
-  it("emits only the missing suffix after reconnect, even if an intermediate snapshot lags", () => {
+  it("emits only the missing suffix after reconnect, even if an intermediate snapshot lags", async () => {
     const timeline = new V2Timeline();
-    const text = (messages: ReturnType<V2Timeline["messages"]>) =>
+    const text = (messages: Awaited<ReturnType<V2Timeline["messages"]>>) =>
       messages.flatMap((event) =>
         event.type === "timeline" && event.item.type === "assistant_message"
           ? [event.item.text]
           : [],
       );
-    expect(text(timeline.messages([assistant("hello")]))).toEqual(["hello"]);
-    expect(timeline.messages([assistant("hel")])).toEqual([]);
-    expect(text(timeline.messages([assistant("hello world")]))).toEqual([" world"]);
-    expect(timeline.messages([assistant("hello world")])).toEqual([]);
+    expect(text(await timeline.messages([assistant("hello")]))).toEqual(["hello"]);
+    expect(await timeline.messages([assistant("hel")])).toEqual([]);
+    expect(text(await timeline.messages([assistant("hello world")]))).toEqual([" world"]);
+    expect(await timeline.messages([assistant("hello world")])).toEqual([]);
   });
-  it("keeps reasoning separate from assistant output", () => {
+  it("keeps reasoning separate from assistant output", async () => {
     const timeline = new V2Timeline();
     const message = assistant("answer");
     message.content.unshift({ type: "reasoning", text: "thinking" });
     expect(
-      timeline.messages([message]).map((event) => (event.type === "timeline" ? event.item : null)),
+      (await timeline.messages([message])).map((event) =>
+        event.type === "timeline" ? event.item : null,
+      ),
     ).toEqual([
       { type: "reasoning", text: "thinking" },
       { type: "assistant_message", text: "answer", messageId: "message" },
     ]);
   });
-  it("emits tool state transitions once and keeps their call identity", () => {
+  it("emits tool state transitions once and keeps their call identity", async () => {
     const timeline = new V2Timeline();
     const message = assistant("");
     message.content = [
@@ -1623,7 +1625,7 @@ describe("OpenCode v2 history reconciliation", () => {
         state: { status: "running", input: { command: "pwd" }, metadata: {} },
       },
     ];
-    expect(timeline.messages([message])).toMatchObject([
+    expect(await timeline.messages([message])).toMatchObject([
       { item: { type: "tool_call", callId: "call", status: "running" } },
     ]);
     message.content = [
@@ -1639,9 +1641,9 @@ describe("OpenCode v2 history reconciliation", () => {
         },
       },
     ];
-    expect(timeline.messages([message])).toMatchObject([
+    expect(await timeline.messages([message])).toMatchObject([
       { item: { type: "tool_call", callId: "call", status: "completed" } },
     ]);
-    expect(timeline.messages([message])).toEqual([]);
+    expect(await timeline.messages([message])).toEqual([]);
   });
 });

packages/server/src/server/agent/providers/opencode/test-utils/v2-streamed-text-memory.ts

@@ -21,16 +21,18 @@ for (let index = 0; index < chunks; index++) {
   const event =
     mode === "delta"
       ? timeline.delta({ ...part, delta })
-      : timeline.messages([
-          {
-            id: "answer",
-            type: "assistant",
-            agent: "build",
-            model: { providerID: "test", id: "model" },
-            time: { created: 2 },
-            content: [{ type: "reasoning", text }],
-          },
-        ])[0];
+      : (
+          await timeline.messages([
+            {
+              id: "answer",
+              type: "assistant",
+              agent: "build",
+              model: { providerID: "test", id: "model" },
+              time: { created: 2 },
+              content: [{ type: "reasoning", text }],
+            },
+          ])
+        )[0];
   if (event?.type !== "timeline") throw new Error("Expected a timeline event");
   projection.append({
     seq: index + 1,

packages/server/src/server/agent/providers/opencode/v2/children.ts

@@ -92,7 +92,7 @@ export class SessionChildren {
       this.childStates.set(info.id, signature);
       this.options.emit({ type: "provider_subagent", provider: "opencode", event: presentation });
     }
-    for (const event of timeline.messages(await messages(this.options.client(), info.id))) {
+    for (const event of await timeline.messages(await messages(this.options.client(), info.id))) {
       if (event.type === "timeline")
         this.options.emit({
           type: "provider_subagent",

packages/server/src/server/agent/providers/opencode/v2/session.ts

@@ -131,7 +131,7 @@ export class OpenCodeV2Session implements AgentSession {
     await this.configureConnection();
     const location = { directory: this.config.cwd };
     this.modes = modesFromV2((await this.client.agent.list({ location })).data);
-    this.timeline.messages(await messages(this.client, this.id));
+    await this.timeline.messages(await messages(this.client, this.id));
     await this.startStream();
     await this.children.reconcile(this.id);
   }
@@ -261,7 +261,7 @@ export class OpenCodeV2Session implements AgentSession {
   async *streamHistory() {
     await this.reconnectIfExited();
     const history = new V2Timeline(false);
-    yield* history.messages(await messages(this.client, this.id));
+    yield* await history.messages(await messages(this.client, this.id));
   }
   async getRuntimeInfo() {
     await this.reconnectIfExited();
@@ -417,7 +417,7 @@ export class OpenCodeV2Session implements AgentSession {
     ]);
     this.info = info;
     this.history = history;
-    for (const event of this.timeline.messages(history)) this.emitTimeline(event);
+    for (const event of await this.timeline.messages(history)) this.emitTimeline(event);
     await this.permissions.reconcile(this.id);
   }
   private async reconcile() {

packages/server/src/server/agent/providers/opencode/v2/streaming.test.ts

@@ -80,7 +80,7 @@ describe("OpenCode v2 token streaming", () => {
     }
   });
 
-  test("deduplicates snapshots before and after deltas using per-type ordinals", () => {
+  test("deduplicates snapshots before and after deltas using per-type ordinals", async () => {
     const timeline = new V2Timeline();
     const text = { assistantMessageID: "answer", type: "text", ordinal: 0 } as const;
     const reasoning = { ...text, type: "reasoning" } as const;
@@ -99,24 +99,24 @@ describe("OpenCode v2 token streaming", () => {
       },
       { type: "text", text: "Hello" },
     ]);
-    expect(timeline.messages([snapshot])).toMatchObject([{ item: { text: "lo" } }]);
+    expect(await timeline.messages([snapshot])).toMatchObject([{ item: { text: "lo" } }]);
     expect(timeline.delta({ ...text, delta: "lo" })).toBeNull();
-    expect(timeline.messages([snapshot])).toEqual([]);
+    expect(await timeline.messages([snapshot])).toEqual([]);
     expect(timeline.delta({ ...text, delta: "!" })).toMatchObject({ item: { text: "!" } });
-    expect(timeline.messages([snapshot])).toEqual([]);
+    expect(await timeline.messages([snapshot])).toEqual([]);
   });
 
-  test("recovers missed fragments from snapshots after reconnect", () => {
+  test("recovers missed fragments from snapshots after reconnect", async () => {
     const timeline = new V2Timeline();
     const part = { assistantMessageID: "answer", type: "text", ordinal: 0 } as const;
     timeline.startPart(part);
     expect(timeline.delta({ ...part, delta: "Hel" })).toMatchObject({ item: { text: "Hel" } });
     timeline.resetStreams();
     expect(timeline.delta({ ...part, delta: "!" })).toBeNull();
-    expect(timeline.messages([assistant([{ type: "text", text: "Hello!" }])])).toMatchObject([
+    expect(await timeline.messages([assistant([{ type: "text", text: "Hello!" }])])).toMatchObject([
       { item: { text: "lo!" } },
     ]);
-    expect(timeline.messages([assistant([{ type: "text", text: "Hello!" }])])).toEqual([]);
+    expect(await timeline.messages([assistant([{ type: "text", text: "Hello!" }])])).toEqual([]);
   });
 
   test("withholds streamed prose during a structured-output turn", async () => {

packages/server/src/server/agent/providers/opencode/v2/timeline.test.ts

@@ -15,7 +15,7 @@ function assistant(content: SessionMessageAssistant["content"]): SessionMessageA
   };
 }
 
-test("a completed edit carries the replaced and replacement text", () => {
+test("a completed edit carries the replaced and replacement text", async () => {
   // Captured from OpenCode 2.0.18, whose edit tool takes `path` instead of `filePath`.
   const edit = assistant([
     {
@@ -45,7 +45,7 @@ test("a completed edit carries the replaced and replacement text", () => {
     },
   ]);
 
-  expect(new V2Timeline().messages([edit])).toMatchObject([
+  expect(await new V2Timeline().messages([edit])).toMatchObject([
     {
       item: {
         type: "tool_call",
@@ -57,7 +57,7 @@ test("a completed edit carries the replaced and replacement text", () => {
   ]);
 });
 
-test("a tool's image output is shown as an image, not as base64 in the tool output", () => {
+test("a tool's image output is shown as an image, not as base64 in the tool output", async () => {
   const png =
     "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8DwHwAFBQIAX8jx0gAAAABJRU5ErkJggg==";
   const screenshot = assistant([
@@ -80,7 +80,7 @@ test("a tool's image output is shown as an image, not as base64 in the tool outp
   ]);
   const timeline = new V2Timeline();
 
-  const events = timeline.messages([screenshot]);
+  const events = await timeline.messages([screenshot]);
 
   expect(JSON.stringify(events)).not.toContain(png);
   expect(events.map((event) => event.type === "timeline" && event.item)).toMatchObject([
@@ -95,7 +95,7 @@ test("a tool's image output is shown as an image, not as base64 in the tool outp
       text: expect.stringMatching(/^\n\n!\[Image\]\(file:\/\/.*paseo-attachments.*\.png\)\n\n$/),
     },
   ]);
-  expect(timeline.messages([screenshot])).toEqual([]);
+  expect(await timeline.messages([screenshot])).toEqual([]);
 });
 
 test("a surrogate pair split across deltas streams back intact", () => {

packages/server/src/server/agent/providers/opencode/v2/timeline.ts

@@ -55,7 +55,15 @@ export class V2Timeline {
     return { type: "timeline", provider: "opencode", item, timestamp: new Date().toISOString() };
   }
 
-  messages(messages: SessionMessageInfo[]): AgentStreamEvent[] {
+  private pendingMessages: Promise<unknown> = Promise.resolve();
+
+  messages(messages: SessionMessageInfo[]): Promise<AgentStreamEvent[]> {
+    const result = this.pendingMessages.then(() => this.projectMessages(messages));
+    this.pendingMessages = result.catch(() => {});
+    return result;
+  }
+
+  private async projectMessages(messages: SessionMessageInfo[]): Promise<AgentStreamEvent[]> {
     const events: AgentStreamEvent[] = [];
     const state: TimelineState = { structured: false, accepted: false };
     for (const message of messages) {
@@ -63,7 +71,7 @@ export class V2Timeline {
       const push = (item: AgentTimelineItem) =>
         events.push({ type: "timeline", provider: "opencode", item, timestamp });
       if (message.type === "user") this.userMessage(message, state, push);
-      else if (message.type === "assistant") this.assistantMessage(message, state, push);
+      else if (message.type === "assistant") await this.assistantMessage(message, state, push);
       else if (message.type === "compaction") this.compactionMessage(message, push);
     }
     return events;
@@ -89,7 +97,7 @@ export class V2Timeline {
     });
   }
 
-  private assistantMessage(
+  private async assistantMessage(
     message: Extract<SessionMessageInfo, { type: "assistant" }>,
     state: TimelineState,
     push: (item: AgentTimelineItem) => void,
@@ -99,16 +107,16 @@ export class V2Timeline {
     // paths address the same cursor entry.
     let textOrdinal = 0;
     let reasoningOrdinal = 0;
-    message.content.forEach((part, partIndex) => {
+    for (const [partIndex, part] of message.content.entries()) {
       if (part.type === "tool" && part.name === STRUCTURED_OUTPUT_TOOL) {
         this.structuredOutputPart(message.id, partIndex, part, state, push);
       } else if (part.type === "text" || part.type === "reasoning") {
         const ordinal = part.type === "text" ? textOrdinal++ : reasoningOrdinal++;
         this.textPart(message.id, part, ordinal, state, push);
       } else {
-        this.toolPart(message.id, partIndex, part, push);
+        await this.toolPart(message.id, partIndex, part, push);
       }
-    });
+    }
   }
 
   private structuredOutputPart(
@@ -156,7 +164,7 @@ export class V2Timeline {
     else push({ type: "reasoning", text: suffix });
   }
 
-  private toolPart(
+  private async toolPart(
     messageID: string,
     partIndex: number,
     part: Extract<
@@ -174,16 +182,16 @@ export class V2Timeline {
     if (item) push(item);
     // Images follow their tool call as markdown, as Claude and Codex emit them,
     // so base64 never reaches the tool output.
-    images.forEach((image, index) => {
+    for (const [index, image] of images.entries()) {
       const imageKey = `${key}:image:${index}`;
-      if (this.content.has(imageKey)) return;
+      if (this.content.has(imageKey)) continue;
       this.content.set(imageKey, "");
-      const imageItem = renderProviderImageOutputAsAssistantMarkdown(
+      const imageItem = await renderProviderImageOutputAsAssistantMarkdown(
         { url: image.uri, mimeType: image.mime },
         { materialize: materializeProviderImage },
       );
       if (imageItem) push(imageItem);
-    });
+    }
   }
 
   private compactionMessage(

packages/server/src/server/agent/providers/pi/agent.test.ts

@@ -3583,3 +3583,63 @@ test("pi runtime safety protects queued replies and runtime-owned activity", asy
   native.getStateError = new Error("Unable to prove idle");
   await expect(session.canStopRuntime!()).rejects.toThrow("Unable to prove idle");
 });
+
+test("reserves the Pi turn while an image prompt is materialized", async () => {
+  const { pi, session } = await createSession();
+  onTestFinished(() => session.close());
+  const first = session.startTurn([
+    { type: "image", data: ONE_BY_ONE_PNG_BASE64, mimeType: "image/png" },
+  ]);
+  await expect(session.startTurn("overlapping prompt")).rejects.toThrow("already active");
+  await first;
+  expect(pi.latestSession().prompts).toHaveLength(1);
+});
+
+test("interrupting Pi during image conversion cancels the admitted turn and permits a follow-up", async () => {
+  const { pi, session } = await createSession();
+  onTestFinished(() => session.close());
+  const events: AgentStreamEvent[] = [];
+  session.subscribe((event) => events.push(event));
+  let admitted = false;
+  const first = session
+    .startTurn([
+      {
+        type: "image",
+        data: Buffer.alloc(16 * 1024 * 1024, 51).toString("base64"),
+        mimeType: "image/png",
+      },
+    ])
+    .then((result) => {
+      admitted = true;
+      return result;
+    });
+  await waitForImmediate();
+  expect(admitted).toBe(false);
+  expect(pi.latestSession().prompts).toHaveLength(0);
+  await session.interrupt();
+  const followup = await session.startTurn("follow-up after image interrupt");
+  const original = await first;
+  expect(events).toContainEqual(
+    expect.objectContaining({ type: "turn_canceled", turnId: original.turnId }),
+  );
+  expect(followup.turnId).not.toBe(original.turnId);
+  expect(pi.latestSession().prompts).toEqual([
+    { message: "follow-up after image interrupt", imageCount: 0 },
+  ]);
+});
+
+test("closing Pi during image conversion rejects its pending start", async () => {
+  const { pi, session } = await createSession();
+  const first = session.startTurn([
+    {
+      type: "image",
+      data: Buffer.alloc(16 * 1024 * 1024, 52).toString("base64"),
+      mimeType: "image/png",
+    },
+  ]);
+  const rejected = expect(first).rejects.toThrow("closed");
+  await waitForImmediate();
+  expect(pi.latestSession().prompts).toHaveLength(0);
+  await session.close();
+  await rejected;
+});

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

@@ -367,9 +367,9 @@ function piModelSupportsImageInput(model: PiModel | null | undefined): boolean {
   return model?.input?.includes("image") === true;
 }
 
-function renderTextOnlyImageHint(image: { data: string; mimeType: string }): string {
+async function renderTextOnlyImageHint(image: { data: string; mimeType: string }): Promise<string> {
   try {
-    const materialized = materializeProviderImage({
+    const materialized = await materializeProviderImage({
       data: image.data,
       mimeType: image.mimeType,
     });
@@ -379,10 +379,10 @@ function renderTextOnlyImageHint(image: { data: string; mimeType: string }): str
   }
 }
 
-function convertPromptInput(
+async function convertPromptInput(
   prompt: AgentPromptInput,
   options: { model: PiModel | null | undefined },
-): PiPromptPayload {
+): Promise<PiPromptPayload> {
   if (typeof prompt === "string") {
     return { text: prompt };
   }
@@ -405,7 +405,7 @@ function convertPromptInput(
           mimeType: block.mimeType,
         });
       } else {
-        textParts.push(renderTextOnlyImageHint(block));
+        textParts.push(await renderTextOnlyImageHint(block));
       }
       continue;
     }
@@ -1280,7 +1280,6 @@ export class PiRpcAgentSession implements AgentSession {
       throw new Error("A Pi turn is already active");
     }
 
-    const payload = convertPromptInput(prompt, { model: this.state.model });
     const turnId = randomUUID();
     this.activeTurnId = turnId;
     this.usagePoller.startTurn();
@@ -1292,6 +1291,9 @@ export class PiRpcAgentSession implements AgentSession {
     this.activePromptRequestId = null;
     this.pendingSteerSubmissions.length = 0;
     this.clearNoTurnBuffers();
+    const payload = await convertPromptInput(prompt, { model: this.state.model });
+    if (this.closed) throw new Error("Pi session is closed");
+    if (this.activeTurnId !== turnId) return { turnId };
     this.activeNoTurnPromptText = payload.text;
     const shouldProbeForNoTurnPrompt = this.parseSlashCommandInput(payload.text) !== null;
 
@@ -1363,7 +1365,9 @@ export class PiRpcAgentSession implements AgentSession {
     if (this.closed || this.activeTurnId !== options.expectedTurnId) {
       return { status: "unavailable" };
     }
-    const payload = convertPromptInput(prompt, { model: this.state.model });
+    const payload = await convertPromptInput(prompt, { model: this.state.model });
+    if (this.closed || this.activeTurnId !== options.expectedTurnId)
+      return { status: "unavailable" };
     // Pi rejects steer RPCs that are extension commands, so slash inputs keep the
     // interrupt-and-replace fallback where they can run directly.
     if (this.parseSlashCommandInput(payload.text)) {

packages/server/src/server/agent/providers/provider-image-output.posix.test.ts

@@ -5,8 +5,8 @@ import { describe, expect, test } from "vitest";
 import { materializeProviderImage } from "./provider-image-output.js";
 
 describe.skipIf(process.platform === "win32")("materializeProviderImage", () => {
-  test("writes image attachments under a private temp directory", () => {
-    const materialized = materializeProviderImage({
+  test("writes image attachments under a private temp directory", async () => {
+    const materialized = await materializeProviderImage({
       data: "YWJjMTIz",
       mimeType: "image/png",
     });

packages/server/src/server/agent/providers/provider-image-output.test.ts

@@ -1,10 +1,11 @@
-import { existsSync, rmSync } from "node:fs";
+import { existsSync, rmSync, statSync, readFileSync } from "node:fs";
 import path from "node:path";
 import MarkdownIt from "markdown-it";
 import { projectTimelineRows } from "../timeline-projection.js";
 import { describe, expect, test } from "vitest";
 
 import {
+  MAX_PROVIDER_IMAGE_BYTES,
   isProviderImageMarkdown,
   materializeProviderImage,
   renderProviderImageOutputAsAssistantMarkdown,
@@ -12,8 +13,8 @@ import {
 
 const HASH = "a".repeat(64);
 
-function renderImageMarkdown(imagePath: string): string {
-  const item = renderProviderImageOutputAsAssistantMarkdown({ path: imagePath });
+async function renderImageMarkdown(imagePath: string): Promise<string> {
+  const item = await renderProviderImageOutputAsAssistantMarkdown({ path: imagePath });
   if (!item || item.type !== "assistant_message") {
     throw new Error("Expected provider image output to render as assistant markdown.");
   }
@@ -40,8 +41,8 @@ describe("isProviderImageMarkdown", () => {
     ).toBe(true);
   });
 
-  test("emits Windows file paths as file URIs", () => {
-    const markdown = renderImageMarkdown(
+  test("emits Windows file paths as file URIs", async () => {
+    const markdown = await renderImageMarkdown(
       `C:\\Users\\me\\AppData\\Local\\Temp\\paseo-attachments\\${HASH}.png`,
     );
 
@@ -51,22 +52,24 @@ describe("isProviderImageMarkdown", () => {
     expect(isProviderImageMarkdown(markdown)).toBe(true);
   });
 
-  test("emits POSIX file paths with spaces as valid file URI markdown", () => {
-    const markdown = renderImageMarkdown("/home/user/Projects/Project With Spaces/screenshot.png");
+  test("emits POSIX file paths with spaces as valid file URI markdown", async () => {
+    const markdown = await renderImageMarkdown(
+      "/home/user/Projects/Project With Spaces/screenshot.png",
+    );
 
     expect(markdown).toBe(
       "\n\n![Image](file:///home/user/Projects/Project%20With%20Spaces/screenshot.png)\n\n",
     );
   });
 
-  test("encodes URI-significant characters in POSIX file paths", () => {
-    const markdown = renderImageMarkdown("/tmp/screenshot#1?draft.png");
+  test("encodes URI-significant characters in POSIX file paths", async () => {
+    const markdown = await renderImageMarkdown("/tmp/screenshot#1?draft.png");
 
     expect(markdown).toBe("\n\n![Image](file:///tmp/screenshot%231%3Fdraft.png)\n\n");
   });
 
-  test("preserves double-leading slashes in POSIX file paths", () => {
-    const markdown = renderImageMarkdown("//tmp/screenshot#1.png");
+  test("preserves double-leading slashes in POSIX file paths", async () => {
+    const markdown = await renderImageMarkdown("//tmp/screenshot#1.png");
 
     expect(markdown).toBe("\n\n![Image](file:////tmp/screenshot%231.png)\n\n");
   });
@@ -78,8 +81,8 @@ describe("isProviderImageMarkdown", () => {
       "\\\\?\\UNC\\server\\share\\shot?draft.png",
       "file://server/share/shot%3Fdraft.png",
     ],
-  ])("encodes %s image paths as file URIs", (_label, imagePath, expectedSource) => {
-    expect(renderImageMarkdown(imagePath)).toBe(`\n\n![Image](${expectedSource})\n\n`);
+  ])("encodes %s image paths as file URIs", async (_label, imagePath, expectedSource) => {
+    expect(await renderImageMarkdown(imagePath)).toBe(`\n\n![Image](${expectedSource})\n\n`);
   });
 
   test("rejects user-authored markdown that is not a materialized attachment", () => {
@@ -92,8 +95,8 @@ describe("isProviderImageMarkdown", () => {
 });
 
 describe("materializeProviderImage", () => {
-  test("recreates the private temp directory if the cached directory is removed", () => {
-    const first = materializeProviderImage({
+  test("recreates the private temp directory if the cached directory is removed", async () => {
+    const first = await materializeProviderImage({
       data: "YWJjMTIz",
       mimeType: "image/png",
     });
@@ -102,7 +105,7 @@ describe("materializeProviderImage", () => {
 
     rmSync(firstDir, { recursive: true, force: true });
 
-    const second = materializeProviderImage({
+    const second = await materializeProviderImage({
       data: "ZGVmNDU2",
       mimeType: "image/png",
     });
@@ -117,8 +120,8 @@ describe("materializeProviderImage", () => {
 });
 
 describe("provider image paragraph boundaries", () => {
-  test.each(["", "\n"])("keeps image outputs separate from prose ending in %j", (ending) => {
-    const image = renderProviderImageOutputAsAssistantMarkdown({
+  test.each(["", "\n"])("keeps image outputs separate from prose ending in %j", async (ending) => {
+    const image = await renderProviderImageOutputAsAssistantMarkdown({
       url: "https://example.com/chart.png",
     });
     if (!image) throw new Error("Expected image output");
@@ -152,10 +155,10 @@ describe("provider image paragraph boundaries", () => {
   });
 });
 
-test("materialized image output retains paragraph boundaries and attachment recognition", () => {
-  const item = renderProviderImageOutputAsAssistantMarkdown(
+test("materialized image output retains paragraph boundaries and attachment recognition", async () => {
+  const item = await renderProviderImageOutputAsAssistantMarkdown(
     { data: "YWJjMTIz", mimeType: "image/png" },
-    { materialize: () => ({ path: `/tmp/paseo-attachments/${HASH}.png` }) },
+    { materialize: async () => ({ path: `/tmp/paseo-attachments/${HASH}.png` }) },
   );
   expect(item).toEqual({
     type: "assistant_message",
@@ -164,3 +167,76 @@ test("materialized image output retains paragraph boundaries and attachment reco
   if (!item || item.type !== "assistant_message") throw new Error("Expected image output");
   expect(isProviderImageMarkdown(item.text)).toBe(true);
 });
+
+test("repeated content is reused without rewriting the attachment", async () => {
+  const image = {
+    data: Buffer.from("dedupe regression").toString("base64"),
+    mimeType: "image/png",
+  };
+  const first = await materializeProviderImage(image);
+  const before = statSync(first.path, { bigint: true });
+  await new Promise((resolve) => setTimeout(resolve, 20));
+  const second = await materializeProviderImage(image);
+  expect(second.path).toBe(first.path);
+  expect(statSync(second.path, { bigint: true }).mtimeNs).toBe(before.mtimeNs);
+});
+
+test("64 MiB materialization leaves the event loop responsive and preserves bytes", async () => {
+  const bytes = Buffer.alloc(64 * 1024 * 1024, 37);
+  const data = bytes.toString("base64");
+  let previous = performance.now();
+  let longest = 0;
+  const probe = setInterval(() => {
+    const now = performance.now();
+    longest = Math.max(longest, now - previous);
+    previous = now;
+  }, 1);
+  try {
+    const result = await materializeProviderImage({ data, mimeType: "image/png" });
+    await new Promise((resolve) => setTimeout(resolve, 5));
+    console.log(`64 MiB attachment longest loop gap: ${longest.toFixed(2)} ms`);
+    expect(longest).toBeLessThan(50);
+    expect(readFileSync(result.path).equals(bytes)).toBe(true);
+    rmSync(result.path, { force: true });
+  } finally {
+    clearInterval(probe);
+  }
+});
+
+test("preserves base64 decoding across whitespace and chunk boundaries", async () => {
+  const bytes = Buffer.alloc(512 * 1024, 123);
+  const base64 = bytes.toString("base64");
+  const data = base64.match(/.{1,77}/g)!.join("\n");
+  const result = await materializeProviderImage({ data, mimeType: "image/png" });
+  expect(readFileSync(result.path).equals(bytes)).toBe(true);
+});
+
+test("rejects oversized materialization before work and leaves the queue usable", async () => {
+  const data = "A".repeat(Math.ceil(MAX_PROVIDER_IMAGE_BYTES / 3) * 4 + 260);
+  await expect(materializeProviderImage({ data, mimeType: "image/png" })).rejects.toThrow(
+    "exceeds",
+  );
+  const omitted = await renderProviderImageOutputAsAssistantMarkdown(
+    { data, mimeType: "image/png" },
+    { materialize: materializeProviderImage },
+  );
+  expect(omitted).toEqual({
+    type: "assistant_message",
+    text: "Image output was omitted because it was not available as a file path or URL.",
+  });
+  const image = await materializeProviderImage({ data: "YWJjMTIz", mimeType: "image/png" });
+  expect(readFileSync(image.path).toString()).toBe("abc123");
+});
+
+test("simultaneous requests for the same content share the published file", async () => {
+  const image = {
+    data: Buffer.from("concurrent dedupe fixture").toString("base64"),
+    mimeType: "image/png",
+  };
+  const results = await Promise.all([
+    materializeProviderImage(image),
+    materializeProviderImage(image),
+  ]);
+  expect(results[0]).toEqual(results[1]);
+  expect(readFileSync(results[0].path).toString()).toBe("concurrent dedupe fixture");
+});

packages/server/src/server/agent/providers/provider-image-output.ts

@@ -1,5 +1,6 @@
 import { createHash } from "node:crypto";
-import * as fsSync from "node:fs";
+import * as fs from "node:fs/promises";
+import { setImmediate } from "node:timers/promises";
 import os from "node:os";
 import path from "node:path";
 
@@ -23,32 +24,26 @@ const PRIVATE_ATTACHMENT_DIR_MODE = 0o700;
 const MATERIALIZED_IMAGE_FILE_MODE = 0o600;
 
 let materializedImageAttachmentDir: string | null = null;
+// Serialize materializations: one decoder and one file operation at a time, including
+// concurrent history replay and live turns. A rejected operation must not poison the queue.
+let pendingMaterialization: Promise<unknown> = Promise.resolve();
 
-function canReuseMaterializedImageAttachmentDir(dir: string): boolean {
-  try {
-    const stats = fsSync.lstatSync(dir);
-    if (!stats.isDirectory()) {
-      return false;
+async function getMaterializedImageAttachmentDir(): Promise<string> {
+  if (materializedImageAttachmentDir) {
+    try {
+      const stats = await fs.lstat(materializedImageAttachmentDir);
+      if (stats.isDirectory()) {
+        await fs.chmod(materializedImageAttachmentDir, PRIVATE_ATTACHMENT_DIR_MODE);
+        return materializedImageAttachmentDir;
+      }
+    } catch {
+      /* Temp storage may have been removed between turns. */
     }
-    fsSync.chmodSync(dir, PRIVATE_ATTACHMENT_DIR_MODE);
-    return true;
-  } catch {
-    return false;
   }
-}
-
-function getMaterializedImageAttachmentDir(): string {
-  if (
-    materializedImageAttachmentDir &&
-    canReuseMaterializedImageAttachmentDir(materializedImageAttachmentDir)
-  ) {
-    return materializedImageAttachmentDir;
-  }
-
-  materializedImageAttachmentDir = fsSync.mkdtempSync(
+  materializedImageAttachmentDir = await fs.mkdtemp(
     path.join(os.tmpdir(), PROVIDER_IMAGE_ATTACHMENT_DIR_PREFIX),
   );
-  fsSync.chmodSync(materializedImageAttachmentDir, PRIVATE_ATTACHMENT_DIR_MODE);
+  await fs.chmod(materializedImageAttachmentDir, PRIVATE_ATTACHMENT_DIR_MODE);
   return materializedImageAttachmentDir;
 }
 
@@ -71,32 +66,92 @@ function getImageExtension(mimeType: string): string {
   }
 }
 
-function normalizeImageData(mimeType: string, data: string): { mimeType: string; data: string } {
-  if (data.startsWith("data:")) {
-    const match = data.match(/^data:([^;]+);base64,(.*)$/);
-    if (match) {
-      return { mimeType: match[1], data: match[2] };
-    }
-  }
-  return { mimeType, data };
-}
+// 128 MiB admits the measured 64 MiB fixture with headroom while bounding disk
+// and decoding per image. This is materialization admission, not a wire schema limit.
+export const MAX_PROVIDER_IMAGE_BYTES = 128 * 1024 * 1024;
+const BASE64_CHUNK_CHARACTERS = 256 * 1024; // Multiple of four; decode units stay aligned.
 
-// Filenames are a content hash of the bytes so re-materializing the same image
-// within a process reuses the existing temp file instead of leaking a fresh one
-// for repeated image blocks or history replay.
 export function materializeProviderImage(image: {
   data: string;
   mimeType: string | null;
-}): MaterializedProviderImage {
-  const attachmentsDir = getMaterializedImageAttachmentDir();
-  const normalized = normalizeImageData(image.mimeType ?? "image/png", image.data);
-  const bytes = Buffer.from(normalized.data, "base64");
-  const extension = getImageExtension(normalized.mimeType);
-  const hash = createHash("sha256").update(bytes).digest("hex");
-  const filePath = path.join(attachmentsDir, `${hash}.${extension}`);
-  fsSync.writeFileSync(filePath, bytes, { mode: MATERIALIZED_IMAGE_FILE_MODE });
-  fsSync.chmodSync(filePath, MATERIALIZED_IMAGE_FILE_MODE);
-  return { path: filePath };
+}): Promise<MaterializedProviderImage> {
+  const result = pendingMaterialization.then(() => materializeImage(image));
+  pendingMaterialization = result.catch(() => {});
+  return result;
+}
+
+async function materializeImage(image: {
+  data: string;
+  mimeType: string | null;
+}): Promise<MaterializedProviderImage> {
+  const maxEncodedCharacters = Math.ceil(MAX_PROVIDER_IMAGE_BYTES / 3) * 4;
+  // Check length before inspecting/copying the payload, including a bounded data URI header.
+  if (image.data.length > maxEncodedCharacters + 256) {
+    throw new Error(`Image attachment exceeds ${MAX_PROVIDER_IMAGE_BYTES} bytes`);
+  }
+  let data = image.data;
+  let mimeType = image.mimeType ?? "image/png";
+  if (data.startsWith("data:")) {
+    const separator = data.indexOf(";base64,");
+    if (separator >= 0 && separator < 256) {
+      mimeType = data.slice(5, separator);
+      data = data.slice(separator + 8);
+    }
+  }
+  if (data.length > maxEncodedCharacters) {
+    throw new Error(`Image attachment exceeds ${MAX_PROVIDER_IMAGE_BYTES} bytes`);
+  }
+  const hash = createHash("sha256");
+  for await (const bytes of decodeImage(data)) hash.update(bytes);
+  const dir = await getMaterializedImageAttachmentDir();
+  const filePath = path.join(dir, `${hash.digest("hex")}.${getImageExtension(mimeType)}`);
+  try {
+    await fs.stat(filePath);
+    return { path: filePath };
+  } catch (error) {
+    if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error;
+  }
+
+  const staging = await fs.mkdtemp(path.join(dir, ".pending-"));
+  const stagingPath = path.join(staging, "image");
+  try {
+    await fs.chmod(staging, PRIVATE_ATTACHMENT_DIR_MODE);
+    const file = await fs.open(stagingPath, "wx", MATERIALIZED_IMAGE_FILE_MODE);
+    try {
+      await file.chmod(MATERIALIZED_IMAGE_FILE_MODE);
+      // Decode again only for a new hash. This bounds retained memory to one chunk
+      // and performs no writes at all for repeated content, including history replay.
+      for await (const bytes of decodeImage(data)) await file.writeFile(bytes);
+    } finally {
+      await file.close();
+    }
+    await fs.rename(stagingPath, filePath);
+    return { path: filePath };
+  } finally {
+    await fs.rm(staging, { recursive: true, force: true });
+  }
+}
+
+async function* decodeImage(data: string): AsyncGenerator<Buffer> {
+  let carry = "";
+  let decodedBytes = 0;
+  for (let offset = 0; offset < data.length; offset += BASE64_CHUNK_CHARACTERS) {
+    // Buffer.from ignores non-base64 characters. Carry incomplete quartets so
+    // line-wrapped payloads decode identically across chunk boundaries.
+    const chunk =
+      carry +
+      data.slice(offset, offset + BASE64_CHUNK_CHARACTERS).replace(/[^A-Za-z0-9+/_=-]/g, "");
+    const finalChunk = offset + BASE64_CHUNK_CHARACTERS >= data.length || chunk.includes("=");
+    const end = finalChunk ? chunk.length : chunk.length - (chunk.length % 4);
+    const bytes = Buffer.from(chunk.slice(0, end), "base64");
+    carry = chunk.slice(end);
+    decodedBytes += bytes.length;
+    if (decodedBytes > MAX_PROVIDER_IMAGE_BYTES)
+      throw new Error("Image attachment exceeds size limit");
+    yield bytes;
+    await setImmediate();
+    if (finalChunk) break;
+  }
 }
 
 // Recognizes markdown rendered for a materialized provider image: its source is a content-hashed
@@ -112,7 +167,10 @@ export function isProviderImageMarkdown(text: string): boolean {
 }
 
 interface RenderProviderImageOutputOptions {
-  materialize?: (image: { data: string; mimeType: string | null }) => MaterializedProviderImage;
+  materialize?: (image: {
+    data: string;
+    mimeType: string | null;
+  }) => Promise<MaterializedProviderImage>;
 }
 
 function nonEmptyString(value: string | null | undefined): string | null {
@@ -169,10 +227,10 @@ function escapeMarkdownImageSource(value: string): string {
   return markdownImageSource(value).replace(/\\/g, "\\\\").replace(/\)/g, "\\)");
 }
 
-export function renderProviderImageOutputAsAssistantMarkdown(
+export async function renderProviderImageOutputAsAssistantMarkdown(
   image: ProviderImageOutput,
   options: RenderProviderImageOutputOptions = {},
-): AgentTimelineItem | null {
+): Promise<AgentTimelineItem | null> {
   const source = nonEmptyString(image.path) ?? nonEmptyString(image.url);
   if (source && !isDataImageSource(source)) {
     const altText = escapeMarkdownImageAlt(nonEmptyString(image.altText) ?? "Image");
@@ -191,7 +249,7 @@ export function renderProviderImageOutputAsAssistantMarkdown(
   let materialized: MaterializedProviderImage | null = null;
   try {
     materialized = options.materialize
-      ? options.materialize({
+      ? await options.materialize({
           data,
           mimeType: nonEmptyString(image.mimeType),
         })

packages/server/src/server/atomic-file.test.ts

@@ -0,0 +1,147 @@
+import { promises as fs } from "node:fs";
+import os from "node:os";
+import path from "node:path";
+import { afterEach, beforeEach, expect, it } from "vitest";
+import { writeFileAtomic, writeJsonFileAtomic } from "./atomic-file.js";
+
+let directory: string;
+beforeEach(async () => {
+  directory = await fs.mkdtemp(path.join(os.tmpdir(), "paseo-atomic-file-"));
+});
+afterEach(async () => {
+  await fs.rm(directory, { recursive: true, force: true });
+});
+
+it.each(["EPERM", "EACCES", "EBUSY"])(
+  "commits a completed receipt after transient Windows %s replace failures",
+  async (code) => {
+    const file = path.join(directory, "receipt.json");
+    await writeJsonFileAtomic(file, { state: "pending" });
+    let failures = 3;
+    let elapsed = 0;
+    await writeJsonFileAtomic(
+      file,
+      { state: "completed" },
+      {
+        platform: "win32",
+        now: () => elapsed,
+        sleep: async (milliseconds) => {
+          elapsed += milliseconds;
+        },
+        rename: async (source, destination) => {
+          if (failures-- > 0) {
+            expect(JSON.parse(await fs.readFile(destination, "utf8"))).toEqual({
+              state: "pending",
+            });
+            expect(JSON.parse(await fs.readFile(source, "utf8"))).toEqual({ state: "completed" });
+            throw Object.assign(new Error("destination held open"), { code });
+          }
+          await fs.rename(source, destination);
+        },
+      },
+    );
+    expect(JSON.parse(await fs.readFile(file, "utf8"))).toEqual({ state: "completed" });
+    expect(await fs.readdir(directory)).toEqual(["receipt.json"]);
+    expect(elapsed).toBeGreaterThan(0);
+  },
+);
+
+it.each(["EPERM", "EACCES", "EBUSY"])(
+  "preserves the previous file and removes the temp file when Windows %s persists",
+  async (code) => {
+    const file = path.join(directory, "receipt.json");
+    await writeJsonFileAtomic(file, { state: "pending" });
+    const error = Object.assign(new Error("destination held open"), { code });
+    const waits: number[] = [];
+    let elapsed = 0;
+    const attempts: number[] = [];
+    await expect(
+      writeJsonFileAtomic(
+        file,
+        { state: "completed" },
+        {
+          platform: "win32",
+          now: () => elapsed,
+          sleep: async (milliseconds) => {
+            waits.push(milliseconds);
+            elapsed += milliseconds;
+          },
+          rename: async () => {
+            attempts.push(elapsed);
+            throw error;
+          },
+        },
+      ),
+    ).rejects.toBe(error);
+    expect(elapsed).toBe(1000);
+    expect(waits.slice(0, 5)).toEqual([10, 20, 40, 80, 100]);
+    expect(Math.max(...waits)).toBe(100);
+    expect(attempts.every((time) => time < 1000)).toBe(true);
+    expect(JSON.parse(await fs.readFile(file, "utf8"))).toEqual({ state: "pending" });
+    expect(await fs.readdir(directory)).toEqual(["receipt.json"]);
+  },
+);
+
+it.each([
+  ["linux", "EPERM"],
+  ["darwin", "EACCES"],
+  ["linux", "EBUSY"],
+  ["win32", "ENOENT"],
+  ["win32", "ENOSPC"],
+  ["win32", "EXDEV"],
+] as const)("does not retry %s %s failures", async (platform, code) => {
+  const file = path.join(directory, "receipt.json");
+  await writeJsonFileAtomic(file, { state: "pending" });
+  const error = Object.assign(new Error("rename failed"), { code });
+  let attempts = 0;
+  const waits: number[] = [];
+  await expect(
+    writeFileAtomic(file, "replacement", {
+      platform,
+      now: () => 0,
+      sleep: async (milliseconds) => {
+        waits.push(milliseconds);
+      },
+      rename: async () => {
+        attempts++;
+        throw error;
+      },
+    }),
+  ).rejects.toBe(error);
+  expect(attempts).toBe(1);
+  expect(waits).toEqual([]);
+  expect(JSON.parse(await fs.readFile(file, "utf8"))).toEqual({ state: "pending" });
+  expect(await fs.readdir(directory)).toEqual(["receipt.json"]);
+});
+
+it("does not retry after a rename consumes the Windows retry budget", async () => {
+  const file = path.join(directory, "receipt.json");
+  const error = Object.assign(new Error("slow rename failed"), { code: "EPERM" });
+  let elapsed = 0;
+  const waits: number[] = [];
+  await expect(
+    writeFileAtomic(file, "replacement", {
+      platform: "win32",
+      now: () => elapsed,
+      sleep: async (milliseconds) => {
+        waits.push(milliseconds);
+      },
+      rename: async () => {
+        elapsed += 1000;
+        throw error;
+      },
+    }),
+  ).rejects.toBe(error);
+  expect(waits).toEqual([]);
+  expect(await fs.readdir(directory)).toEqual([]);
+});
+
+it("creates parent directories and replaces binary and JSON files through the default runtime", async () => {
+  const file = path.join(directory, "nested", "file");
+  const bytes = Buffer.from([0, 255, 128, 1]);
+  await writeFileAtomic(file, bytes);
+  expect(await fs.readFile(file)).toEqual(bytes);
+  await writeJsonFileAtomic(file, { state: "completed" });
+  expect(await fs.readFile(file, "utf8")).toBe('{\n  "state": "completed"\n}');
+  expect(await fs.readdir(path.dirname(file))).toEqual(["file"]);
+});

packages/server/src/server/atomic-file.ts

@@ -1,10 +1,26 @@
 import { randomUUID } from "node:crypto";
 import { promises as fs } from "node:fs";
 import path from "node:path";
+import { setTimeout as sleep } from "node:timers/promises";
+
+interface AtomicFileRuntime {
+  platform: NodeJS.Platform;
+  rename: (source: string, destination: string) => Promise<void>;
+  now: () => number;
+  sleep: (milliseconds: number) => Promise<void>;
+}
+
+const runtime: AtomicFileRuntime = {
+  platform: process.platform,
+  rename: fs.rename,
+  now: () => performance.now(),
+  sleep,
+};
 
 export async function writeFileAtomic(
   filePath: string,
   data: string | NodeJS.ArrayBufferView,
+  operations: AtomicFileRuntime = runtime,
 ): Promise<void> {
   await fs.mkdir(path.dirname(filePath), { recursive: true });
   const tempPath = path.join(
@@ -13,13 +29,48 @@ export async function writeFileAtomic(
   );
   try {
     await fs.writeFile(tempPath, data, "utf8");
-    await fs.rename(tempPath, filePath);
+    await replaceFile(tempPath, filePath, operations);
   } catch (error) {
     await fs.rm(tempPath, { force: true });
     throw error;
   }
 }
 
-export async function writeJsonFileAtomic(filePath: string, value: unknown): Promise<void> {
-  await writeFileAtomic(filePath, JSON.stringify(value, null, 2));
+async function replaceFile(
+  source: string,
+  destination: string,
+  operations: AtomicFileRuntime,
+): Promise<void> {
+  // Windows readers/scanners can briefly deny replacement. Yield with backoff,
+  // but keep interactive persistence failures bounded to a one-second retry budget.
+  const deadline = operations.now() + 1000;
+  let delay = 10;
+  for (;;) {
+    try {
+      await operations.rename(source, destination);
+      return;
+    } catch (error) {
+      if (
+        operations.platform !== "win32" ||
+        !(error instanceof Error) ||
+        !("code" in error) ||
+        !["EPERM", "EACCES", "EBUSY"].includes(String(error.code))
+      ) {
+        throw error;
+      }
+      const remaining = deadline - operations.now();
+      if (remaining <= 0) throw error;
+      await operations.sleep(Math.min(delay, remaining));
+      if (operations.now() >= deadline) throw error;
+      delay = Math.min(delay * 2, 100);
+    }
+  }
+}
+
+export async function writeJsonFileAtomic(
+  filePath: string,
+  value: unknown,
+  operations: AtomicFileRuntime = runtime,
+): Promise<void> {
+  await writeFileAtomic(filePath, JSON.stringify(value, null, 2), operations);
 }

packages/server/src/server/connected-hosts/hosts.e2e.test.ts

@@ -1544,7 +1544,7 @@ test(
 );
 
 test(
-  "remote agent reports, local CLI sends and automatic completion share the recipient batch",
+  "remote agent reports, local CLI sends and automatic completion reach the same recipient",
   { timeout: 30000 },
   async () => {
     const parentPrompts: string[] = [];
@@ -1633,7 +1633,10 @@ test(
     });
     expect(report.isError).not.toBe(true);
     expect(report.structuredContent).toMatchObject({ deliveryStatus: "queued", serverId: aId });
-    expect(parentPrompts).toEqual([]);
+    // A CLI process can take longer than the two-second quiet window to start.
+    // Exercise each delivery across windows; delivery.test.ts owns batching with a clock.
+    await expect.poll(() => parentPrompts, { timeout: 10000 }).toHaveLength(1);
+    expect(parentPrompts[0]).toContain("REMOTE_REPORT <ready>");
 
     const cli = await promisify(execFile)(
       process.execPath,
@@ -1656,7 +1659,8 @@ test(
       },
     );
     expect(JSON.parse(cli.stdout)).toMatchObject({ status: "queued" });
-    expect(parentPrompts).toEqual([]);
+    await expect.poll(() => parentPrompts, { timeout: 10000 }).toHaveLength(2);
+    expect(parentPrompts[1]).toContain("LOCAL_CLI_REPORT");
     const receiptStates = async () => {
       const directory = join(a.paseoHome, "agent-requests");
       const receipts = await Promise.all(
@@ -1671,13 +1675,10 @@ test(
         .map((receipt) => receipt.state)
         .sort();
     };
-    expect(await receiptStates()).toEqual(["pending", "pending"]);
     finishChild.resolve();
-    await expect.poll(() => parentPrompts, { timeout: 10000 }).toHaveLength(1);
-    expect(parentPrompts[0]).toContain("REMOTE_REPORT <ready>");
-    expect(parentPrompts[0]).toContain("LOCAL_CLI_REPORT");
-    expect(parentPrompts[0]).toContain('event="finished"');
-    expect(parentPrompts[0]).toContain(`source-agent-id="${child.id}"`);
+    await expect.poll(() => parentPrompts, { timeout: 10000 }).toHaveLength(3);
+    expect(parentPrompts[2]).toContain('event="finished"');
+    expect(parentPrompts[2]).toContain(`source-agent-id="${child.id}"`);
     await expect
       .poll(receiptStates, {
         timeout: 10000,
@@ -1870,6 +1871,7 @@ test(
     });
     const a = await daemon({
       mcpEnabled: true,
+      logger: completionLogger(),
       agentClients: createTestAgentClients({
         onStartTurn: (prompt) => {
           if (String(prompt).includes('kind="agent-notification"'))
@@ -1931,8 +1933,14 @@ test(
     await expect.poll(() => parentPrompts, { timeout: 10000 }).toHaveLength(2);
     expect(parentPrompts[1]).toContain('event="finished"');
     await expect
-      .poll(async () =>
-        JSON.parse(await readFile(join(a.paseoHome, "finish-subscriptions.json"), "utf8")),
+      // Provider start precedes dispatch acknowledgment and the durable removal commit.
+      .poll(
+        async () =>
+          JSON.parse(await readFile(join(a.paseoHome, "finish-subscriptions.json"), "utf8")),
+        {
+          timeout: 10000,
+          message: "Parent accepted completion; waiting for durable intent removal",
+        },
       )
       .toEqual([]);
   },

scripts/ci-workflow.test.mjs

@@ -483,14 +483,14 @@ test("Windows units and integration run independently with complete routed cover
   assert.equal(scripts.test, "npm run test:unit && npm run test:integration");
   assert.equal(scripts["test:unit"], 'vitest run --fileParallelism --exclude "**/*.e2e.test.ts"');
   const invocations = scripts["test:integration"].split(" && ");
-  // Each vitest start costs about half a minute on Windows, so the
+  // Each vitest start costs about half a minute on Windows, so the four
   // name-filtered files share one invocation and one combined filter.
   assert.equal(invocations.length, 2);
   assert.ok(invocations.every((command) => command.startsWith("vitest run --maxWorkers=1 ")));
   assert.doesNotMatch(invocations[0], / -t /);
   assert.equal(
     invocations[1],
-    'vitest run --maxWorkers=1 src/server/agent/providers/omp/history.e2e.test.ts src/server/daemon-e2e/git-operations.e2e.test.ts src/server/session.create-agent-worktree-autoarchive.e2e.test.ts src/server/agent/providers/opencode-bridge.local.e2e.test.ts src/server/daemon-e2e/relay-transport.e2e.test.ts -t "merges into|creates a worktree and auto-archives both|versioned runtime|oversized attachments|connected hosts|password-protected daemon refuses|replays selected OMP history"',
+    'vitest run --maxWorkers=1 src/server/daemon-e2e/git-operations.e2e.test.ts src/server/session.create-agent-worktree-autoarchive.e2e.test.ts src/server/agent/providers/opencode-bridge.local.e2e.test.ts src/server/daemon-e2e/relay-transport.e2e.test.ts -t "merges into|creates a worktree and auto-archives both|versioned runtime|oversized attachments|connected hosts|password-protected daemon refuses"',
   );
   const serverConfig = readFileSync(new URL("packages/server/vitest.config.ts", repoRoot), "utf8");
   assert.doesNotMatch(serverConfig, /maxWorkers/);