fix(server): reject oversized provider stdout before parsing

Admit raw UTF-8 line bytes before accumulation and fail the provider session explicitly on overflow. Share line framing so fragmented JSONL input no longer repeatedly scans and copies the complete prefix.

.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/codex/app-server-transport.test.ts

@@ -183,3 +183,29 @@ describe("Codex app-server transport", () => {
     codex.child.stdin.end();
   });
 });
+
+describe("Codex stdout byte admission", () => {
+  test("fails pending requests and the session before joining an oversized line", async () => {
+    const child = createCodexAppServerChildProcess();
+    const client = new CodexAppServerClient(child, createTestLogger());
+    const terminations: Error[] = [];
+    client.setUnexpectedTerminationHandler((error) => terminations.push(error));
+    const request = client.request("thread/resume", {}, 1000).catch((error: Error) => error);
+    const chunk = Buffer.alloc(64 * 1024, "x");
+    child.stdout.write('{"id":1,"result":"');
+    for (let index = 0; index < 2048; index++) child.stdout.write(chunk);
+    const error = await request;
+    expect(error).toMatchObject({
+      name: "ProviderStreamLimitError",
+      code: "PROVIDER_STREAM_LINE_TOO_LARGE",
+      limitBytes: 128 * 1024 * 1024,
+    });
+    expect(terminations).toEqual([error]);
+    expect(client.hasPendingRequests()).toBe(false);
+    child.stdout.write('"}\n{"id":1,"result":"late"}\n');
+    await expect(client.request("model/list")).rejects.toThrow("closed");
+    await client.dispose();
+    child.stdout.end();
+    child.stderr.end();
+  });
+});

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

@@ -1,10 +1,15 @@
 import type { ChildProcessWithoutNullStreams } from "node:child_process";
-import { StringDecoder } from "node:string_decoder";
 import type { Logger } from "pino";
 import { z } from "zod";
 
 import { terminateWithTreeKill } from "../../../../utils/tree-kill.js";
 
+import { ProviderLineReader, ProviderStreamLimitError } from "../jsonl-frame-decoder.js";
+
+// A measured long-thread resume was 19.75 MB; preserve the existing 32 MiB fixture
+// with 4x margin for longer threads. This is provider stdout, not the client protocol.
+export const CODEX_PROVIDER_LINE_BYTES = 128 * 1024 * 1024;
+
 const DEFAULT_TIMEOUT_MS = 14 * 24 * 60 * 60 * 1000;
 const APP_SERVER_GRACEFUL_SHUTDOWN_TIMEOUT_MS = 2_000;
 const APP_SERVER_FORCE_SHUTDOWN_TIMEOUT_MS = 1_000;
@@ -167,39 +172,6 @@ function readProviderTurnId(params: unknown): string | undefined {
   return isRecord(turn) && typeof turn.id === "string" ? turn.id : undefined;
 }
 
-/**
- * Splits stdout into lines on "\n" only. `node:readline` additionally treats U+2028 and U+2029
- * as line terminators, but Codex emits those characters raw inside JSON strings, so readline
- * would cut a single JSON-RPC message into pieces that fail to parse.
- */
-function createNewlineLineReader(
-  input: NodeJS.ReadableStream,
-  onLine: (line: string) => void,
-): { close(): void } {
-  const decoder = new StringDecoder("utf8");
-  // Text after the last "\n", kept as parts and joined once its line ends, so a large line
-  // arriving in many chunks is scanned and copied once instead of once per chunk.
-  let unterminated: string[] = [];
-  const onData = (chunk: Buffer | string): void => {
-    const text = typeof chunk === "string" ? chunk : decoder.write(chunk);
-    let lineStart = 0;
-    let index = text.indexOf("\n");
-    while (index !== -1) {
-      unterminated.push(text.slice(lineStart, index));
-      const line = unterminated.join("");
-      unterminated = [];
-      onLine(line.endsWith("\r") ? line.slice(0, -1) : line);
-      lineStart = index + 1;
-      index = text.indexOf("\n", lineStart);
-    }
-    if (lineStart < text.length) {
-      unterminated.push(text.slice(lineStart));
-    }
-  };
-  input.on("data", onData);
-  return { close: () => void input.off("data", onData) };
-}
-
 export class CodexAppServerClient {
   private readonly stdoutLineReader: { close(): void };
   private activeProviderRequests = 0;
@@ -216,11 +188,24 @@ export class CodexAppServerClient {
     private readonly logger: Logger,
     private readonly getTraceContext: () => CodexAppServerTraceContext = () => ({}),
   ) {
-    this.stdoutLineReader = createNewlineLineReader(child.stdout, (line) => {
-      void this.handleLine(line).catch((error) => {
-        this.logger.warn({ error, line }, "Failed to handle Codex app-server stdout line");
-      });
-    });
+    // Split only on LF: U+2028/U+2029 are valid raw characters inside Codex JSON.
+    const reader = new ProviderLineReader(
+      CODEX_PROVIDER_LINE_BYTES,
+      (line) => {
+        void this.handleLine(line).catch((error) => {
+          this.logger.warn({ error, line }, "Failed to handle Codex app-server stdout line");
+        });
+      },
+      (error) => this.handleStreamLimit(error),
+    );
+    const onData = (chunk: Buffer | string) => reader.write(chunk);
+    child.stdout.on("data", onData);
+    this.stdoutLineReader = {
+      close: () => {
+        reader.close();
+        child.stdout.off("data", onData);
+      },
+    };
 
     child.stderr.on("data", (chunk) => {
       this.stderrBuffer += chunk.toString();
@@ -318,6 +303,15 @@ export class CodexAppServerClient {
     }
   }
 
+  private handleStreamLimit(failure: ProviderStreamLimitError): void {
+    // A rejected response/notification cannot be replayed safely. Fail the session,
+    // including live turns with no pending RPC, rather than dropping a stream unit.
+    this.handleUnexpectedTermination(failure);
+    void this.dispose().catch((err) => {
+      this.logger.warn({ err }, "Failed to terminate oversized Codex stdout producer");
+    });
+  }
+
   private handleUnexpectedTermination(error: Error): void {
     if (this.disposed) {
       return;

packages/server/src/server/agent/providers/jsonl-frame-decoder.test.ts

@@ -1,3 +1,5 @@
+import { performance } from "node:perf_hooks";
+import { setTimeout as delay } from "node:timers/promises";
 import { describe, expect, it } from "vitest";
 import { JsonlFrameDecoder, type JsonlFrameProblem } from "./jsonl-frame-decoder.js";
 
@@ -36,6 +38,16 @@ describe("JsonlFrameDecoder", () => {
     expect(problems).toEqual([]);
   });
 
+  it("preserves UTF-8 characters split across raw stdout buffers", () => {
+    const { decoder, frames, problems } = decoderHarness();
+    const bytes = Buffer.from('{"type":"notice","text":"é🙂"}\n');
+    const split = bytes.indexOf(Buffer.from("🙂")) + 1;
+    decoder.write(bytes.subarray(0, split));
+    decoder.write(bytes.subarray(split));
+    expect(frames).toEqual([{ type: "notice", text: "é🙂" }]);
+    expect(problems).toEqual([]);
+  });
+
   it("reassembles a large ordered v2 frame", () => {
     const { decoder, frames, problems } = decoderHarness();
     const frame = { type: "models", value: "x".repeat(1024 * 1024) };
@@ -55,3 +67,57 @@ describe("JsonlFrameDecoder", () => {
     expect(problems).toEqual(["out-of-order-chunk"]);
   });
 });
+
+describe("provider line admission", () => {
+  it("keeps the loop responsive while accumulating a fragmented provider response", async () => {
+    const { decoder, frames, problems } = decoderHarness();
+    const part = "x".repeat(512);
+    let last = performance.now();
+    let longestGap = 0;
+    const probe = setInterval(() => {
+      const now = performance.now();
+      longestGap = Math.max(longestGap, now - last);
+      last = now;
+    }, 1);
+    try {
+      await delay(5);
+      decoder.write('{"type":"notice","text":"');
+      // 1 MiB is enough to expose the former rescan of every accumulated prefix.
+      for (let index = 0; index < 2048; index++) decoder.write(part);
+      decoder.write('"}\n');
+      await delay(5);
+      expect(frames).toEqual([{ type: "notice", text: part.repeat(2048) }]);
+      expect(problems).toEqual([]);
+      expect(longestGap).toBeLessThan(50);
+    } finally {
+      clearInterval(probe);
+    }
+  });
+
+  it("rejects an unterminated over-cap UTF-8 line and stops accepting frames", () => {
+    const frames: Record<string, unknown>[] = [];
+    const failures: Array<{ problem: string; detail: unknown }> = [];
+    const decoder = new JsonlFrameDecoder({
+      frame: (frame) => frames.push(frame),
+      problem: (problem, detail) => failures.push({ problem, detail }),
+    });
+    // 64 MiB of UTF-8, only 32 Mi UTF-16 code units. No newline or JSON parse needed.
+    const chunk = "é".repeat(32 * 1024);
+    for (let index = 0; index < 1024; index++) decoder.write(chunk);
+    expect(failures).toEqual([]);
+    decoder.write("é");
+    expect(failures).toEqual([
+      {
+        problem: "line-too-large",
+        detail: expect.objectContaining({
+          name: "ProviderStreamLimitError",
+          code: "PROVIDER_STREAM_LINE_TOO_LARGE",
+          limitBytes: 64 * 1024 * 1024,
+        }),
+      },
+    ]);
+    decoder.write('\n{"type":"ready"}\n');
+    expect(frames).toEqual([]);
+    expect(failures).toHaveLength(1);
+  });
+});

packages/server/src/server/agent/providers/jsonl-frame-decoder.ts

@@ -1,3 +1,60 @@
+// Provider-process stdout only; these are not app/daemon wire limits.
+export const JSONL_PROVIDER_LINE_BYTES = 64 * 1024 * 1024;
+
+export class ProviderStreamLimitError extends Error {
+  readonly code = "PROVIDER_STREAM_LINE_TOO_LARGE";
+
+  constructor(readonly limitBytes: number) {
+    super(
+      `Provider stdout line exceeds ${limitBytes} bytes. Start a new thread or reduce the provider response size.`,
+    );
+    this.name = "ProviderStreamLimitError";
+  }
+}
+
+/** Admits raw UTF-8 bytes before retaining, decoding or joining a complete line. */
+export class ProviderLineReader {
+  private parts: Buffer[] = [];
+  private bytes = 0;
+  private stopped = false;
+
+  constructor(
+    private readonly limitBytes: number,
+    private readonly onLine: (line: string) => void,
+    private readonly onLimit: (error: ProviderStreamLimitError) => void,
+  ) {}
+
+  write(chunk: Buffer | string): void {
+    if (this.stopped) return;
+    const buffer = typeof chunk === "string" ? Buffer.from(chunk, "utf8") : chunk;
+    let start = 0;
+    while (start < buffer.byteLength && !this.stopped) {
+      const newline = buffer.indexOf(10, start);
+      const end = newline === -1 ? buffer.byteLength : newline;
+      const length = end - start;
+      if (length > this.limitBytes - this.bytes) {
+        this.close();
+        this.onLimit(new ProviderStreamLimitError(this.limitBytes));
+        return;
+      }
+      if (length) this.parts.push(buffer.subarray(start, end));
+      this.bytes += length;
+      if (newline === -1) return;
+      const line = Buffer.concat(this.parts, this.bytes).toString("utf8");
+      this.parts = [];
+      this.bytes = 0;
+      this.onLine(line.endsWith("\r") ? line.slice(0, -1) : line);
+      start = newline + 1;
+    }
+  }
+
+  close(): void {
+    this.stopped = true;
+    this.parts = [];
+    this.bytes = 0;
+  }
+}
+
 const V2_FRAME_BYTES = 1024 * 1024;
 const V2_REASSEMBLED_BYTES = 64 * 1024 * 1024;
 const V2_CHUNK_BYTES = 256 * 1024;
@@ -5,6 +62,7 @@ const V2_BASE64_CHARS = Math.ceil(V2_CHUNK_BYTES / 3) * 4;
 const V2_MAX_CHUNKS = Math.ceil(V2_REASSEMBLED_BYTES / V2_CHUNK_BYTES);
 
 export type JsonlFrameProblem =
+  | "line-too-large"
   | "invalid-json"
   | "invalid-chunk"
   | "invalid-base64"
@@ -78,7 +136,7 @@ function chunkBytes(data: string): Buffer | null {
 }
 
 export class JsonlFrameDecoder {
-  private lineBuffer = "";
+  private readonly lines: ProviderLineReader;
   private assembly: ChunkAssembly | null = null;
 
   constructor(
@@ -86,17 +144,18 @@ export class JsonlFrameDecoder {
       frame(message: Record<string, unknown>): void;
       problem(problem: JsonlFrameProblem, detail?: unknown): void;
     },
-  ) {}
+  ) {
+    this.lines = new ProviderLineReader(
+      JSONL_PROVIDER_LINE_BYTES,
+      (line) => {
+        if (line.trim()) this.consumeLine(line);
+      },
+      (error) => this.drop("line-too-large", error),
+    );
+  }
 
-  write(text: string): void {
-    this.lineBuffer += text;
-    let newline = this.lineBuffer.indexOf("\n");
-    while (newline !== -1) {
-      const line = this.lineBuffer.slice(0, newline).replace(/\r$/, "");
-      this.lineBuffer = this.lineBuffer.slice(newline + 1);
-      if (line.trim()) this.consumeLine(line);
-      newline = this.lineBuffer.indexOf("\n");
-    }
+  write(chunk: Buffer | string): void {
+    this.lines.write(chunk);
   }
 
   private consumeLine(line: string): void {

packages/server/src/server/agent/providers/jsonl-rpc-process.test.ts

@@ -471,3 +471,25 @@ describe("JsonlRpcProcess", () => {
     }
   });
 });
+
+describe("JSONL stdout byte admission", () => {
+  test("fails pending requests and publishes the typed failure to live-turn consumers", async () => {
+    const child = createInMemoryChildProcess();
+    const process = startProcess({ child });
+    const exit = new Promise<JsonlRpcExit>((resolve) => process.onExit(resolve));
+    const request = process.request({ type: "hang" }, 1000).catch((error: Error) => error);
+    const chunk = Buffer.alloc(64 * 1024, "x");
+    child.stdout.write('{"type":"notice","text":"');
+    for (let index = 0; index < 1024; index++) child.stdout.write(chunk);
+    const error = await request;
+    expect(error).toMatchObject({
+      name: "ProviderStreamLimitError",
+      code: "PROVIDER_STREAM_LINE_TOO_LARGE",
+      limitBytes: 64 * 1024 * 1024,
+    });
+    expect((await exit).error).toBe(error);
+    await process.close();
+    child.stdout.end();
+    child.stderr.end();
+  });
+});

packages/server/src/server/agent/providers/jsonl-rpc-process.ts

@@ -3,7 +3,7 @@ import type { Logger } from "pino";
 
 import { spawnProcess } from "../../../utils/spawn.js";
 import { terminateWithTreeKill } from "../../../utils/tree-kill.js";
-import { JsonlFrameDecoder } from "./jsonl-frame-decoder.js";
+import { JsonlFrameDecoder, ProviderStreamLimitError } from "./jsonl-frame-decoder.js";
 export { supportsJsonlRpcProtocolV2 } from "./jsonl-frame-decoder.js";
 
 /** Default wall-clock timeout for control-plane / short RPC calls. */
@@ -81,6 +81,7 @@ export class JsonlRpcProcess {
   private stderrBuffer = "";
   private nextRequestId = 1;
   private disposed = false;
+  private streamFailure: ProviderStreamLimitError | null = null;
   private exited = false;
   private closing: Promise<void> | null = null;
   private readonly frameDecoder: JsonlFrameDecoder;
@@ -90,6 +91,14 @@ export class JsonlRpcProcess {
     this.frameDecoder = new JsonlFrameDecoder({
       frame: (message) => this.dispatchFrame(message),
       problem: (problem, detail) => {
+        if (problem === "line-too-large" && detail instanceof ProviderStreamLimitError) {
+          // Stop the producer: silently skipping a response or turn event loses state.
+          this.streamFailure = detail;
+          void this.close(detail).catch((err) => {
+            this.options.logger.warn({ err }, `Failed to close ${this.diagnosticName}`);
+          });
+          return;
+        }
         this.options.logger.warn(
           { problem, detail },
           `Ignoring invalid ${this.diagnosticName} frame`,
@@ -98,7 +107,7 @@ export class JsonlRpcProcess {
     });
     this.child = (options.spawn ?? spawnJsonlRpcProcess)(options.launch);
     this.child.stdout.on("data", (chunk) => {
-      this.handleStdoutChunk(chunk.toString());
+      this.handleStdoutChunk(chunk);
     });
     this.child.stderr.on("data", (chunk) => {
       this.stderrBuffer += chunk.toString();
@@ -114,9 +123,11 @@ export class JsonlRpcProcess {
     });
     this.child.on("exit", (code, signal) => {
       this.exited = true;
-      const error = new Error(
-        `${this.diagnosticName} process exited with code ${code ?? "null"} and signal ${signal ?? "null"}\n${this.stderrBuffer}`.trim(),
-      );
+      const error =
+        this.streamFailure ??
+        new Error(
+          `${this.diagnosticName} process exited with code ${code ?? "null"} and signal ${signal ?? "null"}\n${this.stderrBuffer}`.trim(),
+        );
       const exit = { code, signal, error };
       for (const subscriber of this.exitSubscribers) {
         subscriber(exit);
@@ -254,7 +265,7 @@ export class JsonlRpcProcess {
     }
   }
 
-  private handleStdoutChunk(chunk: string): void {
+  private handleStdoutChunk(chunk: Buffer | string): void {
     this.frameDecoder.write(chunk);
   }
 

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/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")

packages/server/src/server/workspace-registry.test.ts

@@ -116,6 +116,35 @@ describe("workspace registries", () => {
     expect(await projectRegistry.list()).toEqual([]);
   });
 
+  test("writes once per mutation and skips missing-record mutations", async () => {
+    let writes = 0;
+    const registry = new FileBackedProjectRegistry(
+      path.join(tmpDir, "projects", "write-count.json"),
+      logger,
+      {
+        writeRecords: async (filePath, records) => {
+          writes++;
+          await writeJsonFileAtomic(filePath, records);
+        },
+      },
+    );
+    const project = createPersistedProjectRecord({
+      projectId: "counted",
+      rootPath: tmpDir,
+      kind: "git",
+      displayName: "Counted",
+      createdAt: "2026-03-01T00:00:00.000Z",
+      updatedAt: "2026-03-01T00:00:00.000Z",
+    });
+    await registry.upsert(project);
+    expect(writes).toBe(1);
+    await registry.update("missing", (current) => current);
+    await registry.remove("missing");
+    expect(writes).toBe(1);
+    await registry.update(project.projectId, (current) => ({ ...current, customName: "Changed" }));
+    expect(writes).toBe(2);
+  });
+
   test("preserves a concurrent project update when archiving", async () => {
     let pauseNextWrite = false;
     let releaseWrite!: () => void;

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 });
+      }
+    });
+  }
+});