test(server): calibrate Claude replay fixture for CI

Load persisted history through session readiness rather than the constructor. Async descriptor-owned replay preserves ordering while serving other requests, and rebind retires the old query before rebuilding history and rewind state.

.github/workflows/desktop-release.yml

@@ -169,6 +169,34 @@ jobs:
           NODE_OPTIONS: --max-old-space-size=${{ matrix.node_heap_mb }}
         run: |
           set -euo pipefail
+          export PASEO_REAL_XCRUN="$(command -v xcrun)"
+          stapler_bin="$(mktemp -d)"
+          trap 'rm -rf "$stapler_bin"' EXIT
+          cat > "$stapler_bin/xcrun" <<'STAPLER'
+          #!/bin/bash
+          if [[ "${1:-}" != stapler || "${2:-}" != staple ]]; then
+            exec "$PASEO_REAL_XCRUN" "$@"
+          fi
+          output="$(mktemp)"
+          trap 'rm -f "$output"' EXIT
+          for attempt in 1 2 3 4; do
+            if "$PASEO_REAL_XCRUN" "$@" >"$output" 2>&1; then
+              cat "$output"
+              exit 0
+            else
+              status=$?
+            fi
+            cat "$output"
+            if [[ "$status" != 68 || "$attempt" == 4 ]] ||
+              ! grep -Fq 'NSURLErrorDomain Code=-1005' "$output"; then
+              exit "$status"
+            fi
+            echo "::warning::Apple staple connection lost; retrying the same signed app ($attempt/4)" >&2
+            sleep 30
+          done
+          STAPLER
+          chmod +x "$stapler_bin/xcrun"
+          export PATH="$stapler_bin:$PATH"
           build_args=(-- --publish never --mac --${{ matrix.electron_arch }})
           build_args+=("-c.publish.releaseType=$RELEASE_TYPE")
           build_args+=("-c.publish.channel=$RELEASE_CHANNEL")

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.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");
     }
@@ -2632,6 +2639,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 +2651,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 +2971,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 +3012,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 +3056,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 +3066,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 +3249,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 +3414,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 +3458,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,13 +4125,15 @@ 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;
-        await this.routeSdkMessageFromPump(message);
+        if (!this.ownsQuery(activeQuery)) return true;
+        this.routeSdkMessageFromPump(message);
         return false;
       } finally {
         this.processingSdkMessages--;
@@ -4221,7 +4253,11 @@ class ClaudeAgentSession implements AgentSession {
     return isClaudeTurnInit(message) && !this.activeForegroundTurnId;
   }
 
-  private async routeSdkMessageFromPump(message: SDKMessage): Promise<void> {
+  private ownsQuery(query: Query): boolean {
+    return !this.closed && this.query === query;
+  }
+
+  private routeSdkMessageFromPump(message: SDKMessage): void {
     if (this.consumeSessionStateChange(message)) {
       return;
     }
@@ -4255,7 +4291,7 @@ class ClaudeAgentSession implements AgentSession {
       "provider.claude.parsed_event",
     );
 
-    const events = await this.buildPumpedMessageEvents(message, identifiers.messageId, turnId);
+    const events = this.buildPumpedMessageEvents(message, identifiers.messageId, turnId);
 
     if (events.length === 0) {
       return;
@@ -4290,11 +4326,11 @@ class ClaudeAgentSession implements AgentSession {
     this.dispatchEvents(events);
   }
 
-  private async buildPumpedMessageEvents(
+  private buildPumpedMessageEvents(
     message: SDKMessage,
     messageIdHint: string | null,
     turnId: string | null,
-  ): Promise<AgentStreamEvent[]> {
+  ): AgentStreamEvent[] {
     const messageEvents = this.translateMessageToEvents(message, {
       suppressAssistantText: true,
       suppressReasoning: true,
@@ -5272,22 +5308,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,12 +5344,12 @@ 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) {
+    for await (const entry of entries) {
       this.ingestPersistedHistoryEntry(entry, timeline, replay);
     }
 
@@ -5314,27 +5359,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
@@ -5421,10 +5472,10 @@ 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),
@@ -6191,13 +6242,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) {
@@ -6729,7 +6780,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 +6880,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

@@ -45,7 +45,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 +68,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 +79,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 +97,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 +134,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 +279,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 +364,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 +386,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 +442,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 +502,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 +519,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 +534,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 +546,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 +559,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 +601,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 +621,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);

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[],
   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;
@@ -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> } {
+}): 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

@@ -35,7 +35,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 +62,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 +91,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 +114,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 +128,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 +145,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 +166,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 +179,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: [],
       }),

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>>;
+  entriesByRunId?: ReadonlyMap<
+    string,
+    Iterable<ClaudeReplayEntry> | AsyncIterable<ClaudeReplayEntry>
+  >;
   convertEntry?: (entry: ClaudeReplayEntry) => AgentTimelineItem[];
-}): SubagentObservation[] {
-  const toolCallIdByRunId = readWorkflowLinks(input.parentEntries);
+}): 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,17 +118,17 @@ export function observeReplayWorkflows(input: {
   return observations;
 }
 
-function observeWorkflowTimeline(input: {
+async function observeWorkflowTimeline(input: {
   id: string;
   result: unknown;
-  entries: Iterable<ClaudeReplayEntry>;
+  entries: Iterable<ClaudeReplayEntry> | AsyncIterable<ClaudeReplayEntry>;
   convertEntry?: (entry: ClaudeReplayEntry) => 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) ?? [];
     for (const item of items) {
@@ -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/pi/agent.test.ts

@@ -3063,6 +3063,81 @@ describe("PiRpcAgentClient", () => {
     ]);
   });
 
+  test.each([
+    { outcome: "successful", error: null },
+    { outcome: "failed", error: new Error("summarizer failed") },
+  ])("waits for $outcome manual compaction before sending a new prompt", async ({ error }) => {
+    const { pi, session, events } = await createSession();
+    onTestFinished(() => session.close());
+    const fakeSession = pi.latestSession();
+    fakeSession.compactError = error;
+    const compactionFinished = Promise.withResolvers<void>();
+    fakeSession.compactCompletion = compactionFinished.promise;
+    const handler = session.tryHandleOutOfBand("/compact");
+    const compaction = handler!.run({ emit: () => {} });
+
+    const { turnId } = await session.startTurn("continue after compaction");
+    await flushTurnScheduling();
+    expect(fakeSession.compactRequests).toEqual([{}]);
+    expect(fakeSession.prompts).toEqual([]);
+
+    compactionFinished.resolve();
+    await compaction;
+    await flushTurnScheduling();
+    expect(fakeSession.prompts).toEqual([{ message: "continue after compaction", imageCount: 0 }]);
+
+    fakeSession.finishTurn();
+    await events.nextTurnCompletion();
+    expect(events.turnLifecycleEvents()).toContainEqual({ type: "turn_completed", turnId });
+    expect(events.eventTypes()).not.toContain("turn_failed");
+  });
+
+  test.each(["interrupt", "close"] as const)(
+    "does not send a prompt waiting on compaction after %s",
+    async (action) => {
+      const { pi, session } = await createSession();
+      onTestFinished(() => session.close());
+      const fakeSession = pi.latestSession();
+      const compactionFinished = Promise.withResolvers<void>();
+      fakeSession.compactCompletion = compactionFinished.promise;
+      const handler = session.tryHandleOutOfBand("/compact");
+      const compaction = handler!.run({ emit: () => {} });
+
+      await session.startTurn("do not send this prompt");
+      await session[action]();
+      compactionFinished.resolve();
+      await compaction;
+      await flushTurnScheduling();
+
+      expect(fakeSession.prompts).toEqual([]);
+    },
+  );
+
+  test("does not send a prompt when compaction finishes during interruption", async () => {
+    const { pi, session, events } = await createSession();
+    onTestFinished(() => session.close());
+    const fakeSession = pi.latestSession();
+    const compactionFinished = Promise.withResolvers<void>();
+    fakeSession.compactCompletion = compactionFinished.promise;
+    const clearQueueFinished = Promise.withResolvers<void>();
+    fakeSession.clearQueueCompletion = clearQueueFinished.promise;
+    const handler = session.tryHandleOutOfBand("/compact");
+    const compaction = handler!.run({ emit: () => {} });
+    const { turnId } = await session.startTurn("do not send this prompt");
+
+    const interruption = session.interrupt();
+    compactionFinished.resolve();
+    await compaction;
+    await flushTurnScheduling();
+    expect(fakeSession.controlRequests).toEqual(["clear_queue"]);
+    expect(fakeSession.prompts).toEqual([]);
+
+    clearQueueFinished.resolve();
+    await interruption;
+    expect(events.turnLifecycleEvents()).toEqual([{ type: "turn_canceled", turnId }]);
+    expect(fakeSession.prompts).toEqual([]);
+  });
+
   test("closes Pi compact loading marker when RPC rejects after compaction starts", async () => {
     const { pi, session } = await createSession();
     const fakeSession = pi.latestSession();

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

@@ -1189,6 +1189,7 @@ export class PiRpcAgentSession implements AgentSession {
   private readonly contextUserEntries: PiCapturedEntry[] = [];
   private readonly treeUserEntriesById = new Map<string, PiCapturedEntry>();
   private readonly pendingExtensionResults = new Map<string, PendingExtensionResult>();
+  private compactionPromise: Promise<void> | null = null;
   private outOfBandCompactionEmit: ((event: AgentStreamEvent) => void) | null = null;
   private outOfBandCompactionStarted = false;
   private outOfBandCompactionCompleted = false;
@@ -1296,6 +1297,15 @@ export class PiRpcAgentSession implements AgentSession {
 
     void (async () => {
       try {
+        if (this.compactionPromise) {
+          // Pi RPC rejects prompts until the manual compaction request settles.
+          await this.compactionPromise.catch(() => undefined);
+          const turnCanceled =
+            this.closed || this.activeTurnId !== turnId || this.interruptingTurn?.turnId === turnId;
+          if (turnCanceled) {
+            return;
+          }
+        }
         const ack = await this.runtimeSession.prompt(payload.text, payload.images);
         this.activePromptRequestId = ack.requestId ?? null;
         const correlatedResult = ack.requestId
@@ -1804,7 +1814,8 @@ export class PiRpcAgentSession implements AgentSession {
     this.outOfBandCompactionStarted = false;
     this.outOfBandCompactionCompleted = false;
     try {
-      await this.runtimeSession.compact(customInstructions);
+      this.compactionPromise = this.runtimeSession.compact(customInstructions);
+      await this.compactionPromise;
     } catch (error) {
       const message = error instanceof Error ? error.message : String(error);
       if (
@@ -1830,6 +1841,7 @@ export class PiRpcAgentSession implements AgentSession {
         },
       });
     } finally {
+      this.compactionPromise = null;
       if (this.outOfBandCompactionEmit === emit && !this.outOfBandCompactionStarted) {
         this.outOfBandCompactionEmit = null;
         this.outOfBandCompactionStarted = false;

packages/server/src/server/agent/providers/pi/test-utils/fake-pi.ts

@@ -108,6 +108,7 @@ export class FakePiSession implements PiRuntimeSession {
   steerError: Error | null = null;
   readonly controlRequests: string[] = [];
   clearQueueError: Error | null = null;
+  clearQueueCompletion: Promise<void> | null = null;
   readonly compactRequests: Array<{ customInstructions?: string }> = [];
   readonly setAutoCompactionRequests: boolean[] = [];
   readonly subagentSubscriptionRequests: FakePiSubagentSubscriptionLevel[] = [];
@@ -140,6 +141,7 @@ export class FakePiSession implements PiRuntimeSession {
   subagentSubscriptionError: Error | null = null;
   setSessionNameError: Error | null = null;
   compactError: Error | null = null;
+  compactCompletion: Promise<void> | null = null;
   emitCompactEnd = true;
   getStateError: Error | null = null;
   getSessionStatsError: Error | null = null;
@@ -233,6 +235,7 @@ export class FakePiSession implements PiRuntimeSession {
 
   async clearQueue(): Promise<void> {
     this.controlRequests.push("clear_queue");
+    await this.clearQueueCompletion;
     if (this.clearQueueError) {
       throw this.clearQueueError;
     }
@@ -241,6 +244,7 @@ export class FakePiSession implements PiRuntimeSession {
   async compact(customInstructions?: string): Promise<void> {
     this.compactRequests.push(customInstructions === undefined ? {} : { customInstructions });
     this.emit({ type: "compaction_start", reason: "manual" });
+    await this.compactCompletion;
     if (this.emitCompactEnd) {
       this.emit({ type: "compaction_end", reason: "manual" });
     }

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([]);
   },

packages/server/src/server/daemon-e2e/pi.real.e2e.test.ts

@@ -219,7 +219,7 @@ test(
 );
 
 test(
-  "real Pi daemon executes manual compact out-of-band instead of prompt text",
+  "real Pi daemon accepts a prompt after manual compact",
   async () => {
     const cwd = tmpCwd("pi-manual-compact-");
 
@@ -234,20 +234,24 @@ test(
           cwd,
           title: "pi-manual-compact",
           provider: "pi",
-          model: PI_COMPACTION_TEST_MODEL,
+          model: PI_REAL_TEST_MODEL,
         });
 
         await client.sendMessage(agent.id, "Reply exactly: compact-ready");
         const finish = await client.waitForFinish(agent.id, PI_TEST_TIMEOUT_MS);
-        expect(finish.status).toBe("idle");
+        expect(finish.status, JSON.stringify(await fetchCanonicalTimeline(client, agent.id))).toBe(
+          "idle",
+        );
 
         await client.sendMessage(agent.id, "/compact summarize this e2e run");
-
         await waitForTimelineItem(
           client,
           agent.id,
           (item) => item.type === "compaction" && item.status === "completed",
         );
+        await client.sendMessage(agent.id, "Reply exactly: after-compaction");
+        const followUpFinish = await client.waitForFinish(agent.id, PI_TEST_TIMEOUT_MS);
+        expect(followUpFinish.status).toBe("idle");
 
         const items = await fetchCanonicalTimeline(client, agent.id);
         expect(items).toEqual(
@@ -263,9 +267,12 @@ test(
             }),
           ]),
         );
-        expect(
-          items.some((item) => item.type === "user_message" && item.text.includes("/compact")),
-        ).toBe(false);
+        expect(items).toContainEqual(
+          expect.objectContaining({
+            type: "assistant_message",
+            text: "after-compaction",
+          }),
+        );
         expect(
           items
             .filter((item) => item.type === "assistant_message")

scripts/ci-workflow.test.mjs

@@ -1,5 +1,8 @@
 import assert from "node:assert/strict";
-import { readFileSync, readdirSync } from "node:fs";
+import { readFileSync, readdirSync, mkdtempSync, writeFileSync, rmSync } from "node:fs";
+import { spawnSync } from "node:child_process";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
 import { relative as relativePath } from "node:path";
 import test from "node:test";
 import { runInNewContext } from "node:vm";
@@ -530,3 +533,111 @@ test("Windows units and integration run independently with complete routed cover
     assert.equal(selected(integrationCondition, outputs, true), false);
   }
 });
+
+test("macOS release retries only transient stapling failures without rebuilding the signed app", async (t) => {
+  const workflow = readFileSync(new URL(".github/workflows/desktop-release.yml", repoRoot), "utf8");
+  const wrapper = workflow
+    .match(/<<'STAPLER'\n([\s\S]*?)\n          STAPLER/)[1]
+    .replace(/^          /gm, "");
+  const networkError =
+    'Error Domain=NSURLErrorDomain Code=-1005 "The network connection was lost."';
+  const cases = [
+    {
+      name: "success",
+      outcomes: [0],
+      output: "stapled",
+      command: "stapler",
+      expected: 0,
+      attempts: 1,
+    },
+    {
+      name: "transient then success",
+      outcomes: [68, 0],
+      output: networkError,
+      command: "stapler",
+      expected: 0,
+      attempts: 2,
+    },
+    {
+      name: "exhausted network retries",
+      outcomes: [68, 68, 68, 68, 0],
+      output: networkError,
+      command: "stapler",
+      expected: 68,
+      attempts: 4,
+    },
+    {
+      name: "non-network staple error",
+      outcomes: [68, 0],
+      output: "ticket not found",
+      command: "stapler",
+      expected: 68,
+      attempts: 1,
+    },
+    {
+      name: "different exit code",
+      outcomes: [1, 0],
+      output: networkError,
+      command: "stapler",
+      expected: 1,
+      attempts: 1,
+    },
+    {
+      name: "notary submission is not retried",
+      outcomes: [68, 0],
+      output: networkError,
+      command: "notarytool",
+      expected: 68,
+      attempts: 1,
+    },
+  ];
+  for (const scenario of cases) {
+    await t.test(scenario.name, () => {
+      const directory = mkdtempSync(join(tmpdir(), "stapler-retry-"));
+      try {
+        const statePath = join(directory, "state.json");
+        const realXcrun = join(directory, "real-xcrun");
+        const wrapperPath = join(directory, "xcrun");
+        writeFileSync(wrapperPath, wrapper, { mode: 0o755 });
+        writeFileSync(join(directory, "sleep"), "#!/bin/sh\nexit 0\n", { mode: 0o755 });
+        writeFileSync(
+          statePath,
+          JSON.stringify({ outcomes: scenario.outcomes, output: scenario.output, calls: [] }),
+        );
+        writeFileSync(
+          realXcrun,
+          `#!/usr/bin/env node
+const fs = require("node:fs");
+const state = JSON.parse(fs.readFileSync(process.env.STAPLER_STATE, "utf8"));
+state.calls.push(process.argv.slice(2));
+const code = state.outcomes.shift();
+fs.writeFileSync(process.env.STAPLER_STATE, JSON.stringify(state));
+console.log(state.output);
+process.exit(code);
+`,
+          { mode: 0o755 },
+        );
+        const args = [scenario.command, "staple", "-v", "Paseo signed.app"];
+        const result = spawnSync("bash", [wrapperPath, ...args], {
+          encoding: "utf8",
+          env: {
+            ...process.env,
+            PASEO_REAL_XCRUN: realXcrun,
+            STAPLER_STATE: statePath,
+            PATH: `${directory}:${process.env.PATH}`,
+          },
+          timeout: 5000,
+        });
+        assert.equal(result.status, scenario.expected, result.stdout + result.stderr);
+        const state = JSON.parse(readFileSync(statePath, "utf8"));
+        assert.deepEqual(
+          state.calls,
+          Array.from({ length: scenario.attempts }, () => args),
+        );
+        assert.match(result.stdout, /stapled|network connection was lost|ticket not found/);
+      } finally {
+        rmSync(directory, { recursive: true, force: true });
+      }
+    });
+  }
+});