Merge remote-tracking branch 'origin/main' into plan-plugin-api-phase7-composers-by-ref

packages/client/package.json

@@ -45,7 +45,8 @@
     "prepack": "npm run build:clean",
     "typecheck": "tsgo --noEmit",
     "typecheck:examples": "tsgo -p tsconfig.examples.json --noEmit",
-    "test": "vitest run"
+    "test": "vitest run",
+    "measure:diff-materialization": "tsx scripts/measure-diff-materialization.ts"
   },
   "dependencies": {
     "@getpaseo/protocol": "0.15.3",

packages/client/scripts/measure-diff-materialization.ts

@@ -0,0 +1,38 @@
+import { readFile, writeFile } from "node:fs/promises";
+import { resolve } from "node:path";
+import { performance } from "node:perf_hooks";
+import { decodeDiffFrame } from "@getpaseo/protocol/binary-frames/index";
+import { validateWSOutboundMessage } from "@getpaseo/protocol/validation/ws-outbound";
+import { DiffStream } from "../src/diff/stream.js";
+const artifacts = resolve(import.meta.dirname, "../../../.artifacts/diff-subscription");
+const results = [];
+for (const count of [1, 30]) {
+  const corpus = JSON.parse(await readFile(resolve(artifacts, `corpus-${count}.json`), "utf8")) as {
+    frames: string[];
+    legacy: string;
+  };
+  const frames = corpus.frames.map((value) => Buffer.from(value, "base64"));
+  const binary: number[] = [];
+  const json: number[] = [];
+  for (let iteration = 0; iteration < 15; iteration++) {
+    let files = 0;
+    const stream = new DiffStream((state) => {
+      if (state.phase === "ready") files = state.files.length;
+    });
+    let start = performance.now();
+    for (const frame of frames) stream.receive(decodeDiffFrame(frame));
+    binary.push(performance.now() - start);
+    start = performance.now();
+    validateWSOutboundMessage(JSON.parse(corpus.legacy));
+    json.push(performance.now() - start);
+    if (!files) throw new Error("No materialized files");
+  }
+  const median = (values: number[]) =>
+    values.slice(3).sort((a, b) => a - b)[Math.floor((values.length - 3) / 2)];
+  const result = { count, materializationMs: median(binary), jsonParseValidationMs: median(json) };
+  results.push(result);
+  console.log(JSON.stringify(result));
+}
+await writeFile(resolve(artifacts, "client-metrics.json"), JSON.stringify(results, null, 2));
+if (results.some((result) => result.materializationMs > result.jsonParseValidationMs))
+  process.exitCode = 1;

packages/client/src/daemon-client-transport.test.ts

@@ -1,5 +1,9 @@
 import { EventEmitter } from "node:events";
 import { describe, expect, test, vi } from "vitest";
+import { WebSocketServer, type WebSocket as ServerSocket } from "ws";
+import { exportPublicKey, generateKeyPair } from "@getpaseo/relay/e2ee";
+import { DaemonClient } from "./daemon-client.js";
+import { defaultWebSocketFactory } from "./daemon-client-websocket-transport.js";
 import {
   createEncryptedTransport,
   createWebSocketTransportFactory,
@@ -190,3 +194,103 @@ test("native drain capability survives WebSocket and encryption wrappers", async
   });
   expect(encrypted.supportsDrain).toBe(false);
 });
+
+interface ObservedSocket {
+  closeFrame: { code: number; reason: string } | null;
+}
+
+/** A daemon endpoint that accepts sockets and never answers, so every attempt times out. */
+async function silentDaemon() {
+  const server = new WebSocketServer({ host: "127.0.0.1", port: 0 });
+  await new Promise<void>((resolve) => server.once("listening", () => resolve()));
+  const sockets: ObservedSocket[] = [];
+  server.on("connection", (socket: ServerSocket) => {
+    const observed: ObservedSocket = { closeFrame: null };
+    sockets.push(observed);
+    socket.on("close", (code, reason) => {
+      observed.closeFrame = { code, reason: reason.toString() };
+    });
+  });
+  const address = server.address();
+  if (!address || typeof address === "string") throw new Error("Missing port");
+  return {
+    url: `ws://127.0.0.1:${address.port}/ws`,
+    sockets,
+    openSockets: () => sockets.filter((entry) => entry.closeFrame === null),
+    stop: () =>
+      new Promise<void>((resolve) => {
+        for (const client of server.clients) client.terminate();
+        server.close(() => resolve());
+      }),
+  };
+}
+
+// Node's global WebSocket applies the WHATWG close rules Chromium and Electron apply:
+// close() accepts only 1000 or 3000-4999 and a reason of at most 123 UTF-8 bytes.
+describe("abandoned connections close their WebSocket", () => {
+  test.each([
+    { name: "direct", e2ee: false },
+    { name: "relay E2EE", e2ee: true },
+  ])(
+    "$name: a timed-out attempt sends a close frame before the client reconnects",
+    async ({ e2ee }) => {
+      const daemon = await silentDaemon();
+      const client = new DaemonClient({
+        url: daemon.url,
+        clientId: "cid_abandoned_close_test",
+        webSocketFactory: defaultWebSocketFactory,
+        connectTimeoutMs: 1_500,
+        reconnect: { enabled: true, baseDelayMs: 50, maxDelayMs: 50 },
+        logger: { debug() {}, info() {}, warn() {}, error() {} },
+        ...(e2ee
+          ? {
+              e2ee: {
+                enabled: true,
+                daemonPublicKeyB64: exportPublicKey(generateKeyPair().publicKey),
+              },
+            }
+          : {}),
+      });
+      try {
+        void client.connect().catch(() => {});
+        await vi.waitFor(() => expect(daemon.sockets.length).toBeGreaterThanOrEqual(2), {
+          timeout: 4_000,
+        });
+        const [abandoned] = daemon.sockets;
+        await vi.waitFor(() => expect(abandoned?.closeFrame).not.toBeNull(), { timeout: 500 });
+        expect(abandoned?.closeFrame?.reason).toBe("Connection timed out");
+
+        // The newest attempt may still be connecting; aborting it closes without a frame.
+        await client.close();
+        await vi.waitFor(() => expect(daemon.openSockets()).toEqual([]), { timeout: 500 });
+      } finally {
+        await client.close();
+        await daemon.stop();
+      }
+    },
+    10_000,
+  );
+
+  test("a code or reason the WebSocket API rejects still closes the socket", async () => {
+    const daemon = await silentDaemon();
+    try {
+      const transport = createWebSocketTransportFactory(defaultWebSocketFactory)({
+        url: daemon.url,
+      });
+      await new Promise<void>((resolve) => transport.onOpen(() => resolve()));
+      const reason = `Decryption failed: ${"é".repeat(100)}`;
+
+      expect(() => transport.close(1011, reason)).not.toThrow();
+
+      await vi.waitFor(() => expect(daemon.sockets[0]?.closeFrame).not.toBeNull(), {
+        timeout: 500,
+      });
+      const frame = daemon.sockets[0]?.closeFrame;
+      expect(frame?.code).toBe(1000);
+      expect(reason.startsWith(frame?.reason ?? "")).toBe(true);
+      expect(Buffer.byteLength(frame?.reason ?? "")).toBe(123);
+    } finally {
+      await daemon.stop();
+    }
+  });
+});

packages/client/src/daemon-client-websocket-transport.ts

@@ -82,7 +82,7 @@ export function createWebSocketTransportFactory(factory: WebSocketFactory): Daem
         // connect timeouts does not crash the CLI with an unhandled error event.
         const suppressEarlyCloseError = bindTemporaryEarlyCloseErrorHandler(ws);
         try {
-          ws.close(code, reason);
+          ws.close(toSendableCloseCode(code), toSendableCloseReason(reason));
         } finally {
           if (typeof ws.on !== "function" && typeof ws.addEventListener !== "function") {
             suppressEarlyCloseError();
@@ -97,6 +97,32 @@ export function createWebSocketTransportFactory(factory: WebSocketFactory): Daem
   };
 }
 
+// WHATWG WebSocket.close throws for a code other than 1000 or 3000-4999 and for a
+// reason over 123 UTF-8 bytes, and the throw leaves the socket open: the peer keeps
+// streaming into a connection nobody reads. Browsers, Electron and Node's global
+// WebSocket enforce this; Node's `ws` accepts more. Translate once here so callers
+// keep their reasons and every runtime sends the same close frame.
+const MAX_CLOSE_REASON_BYTES = 123;
+
+function toSendableCloseCode(code: number | undefined): number {
+  if (code === 1000 || (code !== undefined && code >= 3000 && code <= 4999)) return code;
+  return 1000;
+}
+
+function toSendableCloseReason(reason: string | undefined): string | undefined {
+  if (reason === undefined) return undefined;
+  const encoder = new TextEncoder();
+  if (encoder.encode(reason).byteLength <= MAX_CLOSE_REASON_BYTES) return reason;
+  let truncated = "";
+  let bytes = 0;
+  for (const character of reason) {
+    bytes += encoder.encode(character).byteLength;
+    if (bytes > MAX_CLOSE_REASON_BYTES) break;
+    truncated += character;
+  }
+  return truncated;
+}
+
 function bindWsMessageHandler(
   ws: WebSocketLike,
   handler: (data: unknown, isBinary: boolean) => void,

packages/client/src/daemon-client.test.ts

@@ -728,6 +728,10 @@ class FakeDaemon {
     this.onClose({ reason });
   }
 
+  daemonSendsPong(): void {
+    this.onMessage(JSON.stringify({ type: "pong" }));
+  }
+
   triggerError(event?: unknown): void {
     this.onError(event);
   }
@@ -2605,7 +2609,7 @@ test("keeps default connect timeout shorter than session RPC waiters", async ()
   }
 });
 
-test("foreground checks a silently broken socket through two normal heartbeat misses", async () => {
+test("foreground tears down a silently dead socket after two fresh five-second windows and reconnects at once", async () => {
   useHeartbeatClock();
   const first = new FakeDaemon();
   const second = new FakeDaemon();
@@ -2623,14 +2627,23 @@ test("foreground checks a silently broken socket through two normal heartbeat mi
   await connection;
 
   client.ensureConnected({ verify: true });
-  await vi.advanceTimersByTimeAsync(15_000);
+  expect(first.pingTimestamps()).toEqual(["0s"]);
+  await vi.advanceTimersByTimeAsync(5_000);
+  // The first silent window is followed by a second check at once, not ten seconds later.
   expect(first.closesFromClient()).toEqual([]);
-  await vi.advanceTimersByTimeAsync(25_000);
+  expect(first.pingTimestamps()).toEqual(["0s", "5s"]);
+  await vi.advanceTimersByTimeAsync(4_999);
+  expect(first.closesFromClient()).toEqual([]);
+  await vi.advanceTimersByTimeAsync(1);
   expect(first.closesFromClient()).toEqual([{ code: 1001, reason: "Liveness check timed out" }]);
-  client.ensureConnected();
+  // The reconnect is immediate: no 1.5s backoff after a foreground liveness teardown.
   expect(attempts).toBe(2);
   second.openConnection();
   expect(client.getConnectionState()).toEqual({ status: "connected" });
+  // The replacement runs the normal heartbeat cadence.
+  await vi.advanceTimersByTimeAsync(10_000);
+  expect(second.pingTimestamps()).toEqual(["20s"]);
+  expect(second.closesFromClient()).toEqual([]);
 });
 
 test("foreground verification preserves a healthy socket and deduplicates simultaneous checks", async () => {
@@ -2675,22 +2688,26 @@ test("chunks prevent heartbeat misses until silence causes two misses", async ()
   void client.measureLatency({ timeoutMs: 5_000 }).catch((error: unknown) => {
     measurementError = error;
   });
-  await vi.advanceTimersByTimeAsync(5_000);
+  // Chunks complete every four seconds for the first 36 seconds; no pong ever arrives.
+  for (let index = 0; index < 9; index++) {
+    await vi.advanceTimersByTimeAsync(4_000);
+    daemon.receiveChunk(index === 0);
+    expect(daemon.closesFromClient()).toEqual([]);
+  }
   expect(measurementError).toEqual(
     expect.objectContaining({
       message: "Latency measurement timed out (5000ms)",
     }),
   );
-  for (let index = 0; index < 4; index++) {
-    await vi.advanceTimersByTimeAsync(index === 0 ? 5_000 : 10_000);
-    daemon.receiveChunk(index === 0);
-    expect(daemon.closesFromClient()).toEqual([]);
-  }
-  // No complete message or pong has arrived, but progress prevents misses.
-  expect(daemon.pingTimestamps()).toEqual(["0s", "25s"]);
-  await vi.advanceTimersByTimeAsync(25_000);
+  // Progress inside the foreground window returned the heartbeat to its normal cadence.
+  expect(daemon.pingTimestamps()).toEqual(["0s", "15s"]);
+  await vi.advanceTimersByTimeAsync(19_000);
   expect(daemon.closesFromClient()).toEqual([]);
-  await vi.advanceTimersByTimeAsync(25_000);
+  expect(daemon.pingTimestamps()).toEqual(["0s", "15s", "40s"]);
+  await vi.advanceTimersByTimeAsync(24_999);
+  expect(daemon.closesFromClient()).toEqual([]);
+  await vi.advanceTimersByTimeAsync(1);
+  expect(daemon.pingTimestamps()).toEqual(["0s", "15s", "40s", "65s"]);
   expect(daemon.closesFromClient()).toEqual([{ code: 1001, reason: "Liveness check timed out" }]);
 });
 
@@ -2712,13 +2729,14 @@ test("foreground preserves a multi-MB transfer with its pong queued behind chunk
   client.ensureConnected({ verify: true });
   await vi.advanceTimersByTimeAsync(3_001);
   expect(daemon.closesFromClient()).toEqual([]);
-  for (let index = 0; index < 4; index++) {
-    await vi.advanceTimersByTimeAsync(10_000);
+  for (let index = 0; index < 5; index++) {
+    await vi.advanceTimersByTimeAsync(4_000);
     daemon.receiveChunk(index === 0);
     expect(daemon.closesFromClient()).toEqual([]);
   }
-  // No complete message or pong has arrived, but progress prevents misses.
-  expect(daemon.pingTimestamps()).toEqual(["0s", "25s"]);
+  // The first chunk landed seven seconds after foreground: the first window was silent, the
+  // second saw progress, so the cycle ended without a teardown and the cadence is normal again.
+  expect(daemon.pingTimestamps()).toEqual(["0s", "5s", "20s"]);
   daemon.finishTransfer();
   await vi.advanceTimersByTimeAsync(30_000);
   expect(client.getConnectionState()).toEqual({ status: "connected" });
@@ -2742,13 +2760,14 @@ test("chunk progress beyond the daemon lease keeps sending heartbeat pings", asy
   await connection;
   client.ensureConnected({ verify: true });
   for (let index = 0; index < 12; index++) {
-    await vi.advanceTimersByTimeAsync(5_000);
+    await vi.advanceTimersByTimeAsync(4_000);
     daemon.receiveChunk(index === 0);
     // Each completed window starts the next ping ten seconds later,
     // even though the response and all queued pongs are still in flight.
+    const now = (index + 1) * 4;
     const expected = ["0s"];
-    if (index >= 4) expected.push("25s");
-    if (index >= 9) expected.push("50s");
+    if (now >= 15) expected.push("15s");
+    if (now >= 40) expected.push("40s");
     expect(daemon.pingTimestamps()).toEqual(expected);
     expect(daemon.closesFromClient()).toEqual([]);
   }
@@ -2758,6 +2777,393 @@ test("chunk progress beyond the daemon lease keeps sending heartbeat pings", asy
   expect(daemon.closesFromClient()).toEqual([]);
 });
 
+function connectThroughReplacements(
+  clientId: string,
+  daemons: FakeDaemon[],
+  options?: { reconnect?: { enabled: boolean } },
+): { client: DaemonClient; attempts: () => number } {
+  let attempts = 0;
+  const client = new DaemonClient({
+    url: "ws://test",
+    clientId,
+    logger: noopLogger,
+    ...(options?.reconnect ? { reconnect: options.reconnect } : {}),
+    transportFactory: () => {
+      const daemon = daemons[Math.min(attempts, daemons.length - 1)]!;
+      attempts += 1;
+      return daemon.transport;
+    },
+  });
+  clients.push(client);
+  return { client, attempts: () => attempts };
+}
+
+test("foreground supersedes a heartbeat that straddled the suspension without counting a miss", async () => {
+  useHeartbeatClock();
+  const first = new FakeDaemon();
+  const second = new FakeDaemon();
+  const { client, attempts } = connectThroughReplacements("resume-before-expiry", [first, second]);
+  const connection = client.connect();
+  first.openConnection();
+  await connection;
+  first.daemonGoesSilent();
+  await vi.advanceTimersByTimeAsync(10_000); // heartbeat ping outstanding when the OS suspends JS
+  await vi.advanceTimersByTimeAsync(14_990); // resume: AppState lands before the stale timer
+  client.ensureConnected({ verify: true });
+  expect(first.pingTimestamps()).toEqual(["10s", "24.99s"]);
+  await vi.advanceTimersByTimeAsync(10); // the stale deadline passes without a miss
+  await vi.advanceTimersByTimeAsync(4_990);
+  expect(first.closesFromClient()).toEqual([]);
+  expect(first.pingTimestamps()).toEqual(["10s", "24.99s", "29.99s"]);
+  await vi.advanceTimersByTimeAsync(5_000);
+  expect(first.closesFromClient()).toEqual([{ code: 1001, reason: "Liveness check timed out" }]);
+  expect(attempts()).toBe(2);
+});
+
+test("foreground does not inherit a miss recorded before the app was suspended", async () => {
+  useHeartbeatClock();
+  const first = new FakeDaemon();
+  const second = new FakeDaemon();
+  const { client, attempts } = connectThroughReplacements("resume-after-expiry", [first, second]);
+  const connection = client.connect();
+  first.openConnection();
+  await connection;
+  first.daemonGoesSilent();
+  await vi.advanceTimersByTimeAsync(25_000); // ping at 10s missed at 25s, before or during suspension
+  client.ensureConnected({ verify: true });
+  expect(first.pingTimestamps()).toEqual(["10s", "25s"]);
+  await vi.advanceTimersByTimeAsync(5_000);
+  // Two fresh windows: the old miss would have torn down here.
+  expect(first.closesFromClient()).toEqual([]);
+  expect(first.pingTimestamps()).toEqual(["10s", "25s", "30s"]);
+  await vi.advanceTimersByTimeAsync(5_000);
+  expect(first.closesFromClient()).toEqual([{ code: 1001, reason: "Liveness check timed out" }]);
+  expect(attempts()).toBe(2);
+});
+
+test("a slow but healthy socket survives foreground after a miss before the suspension", async () => {
+  useHeartbeatClock();
+  const daemon = new FakeDaemon();
+  const { client, attempts } = connectThroughReplacements("resume-slow-pong", [daemon]);
+  const connection = client.connect();
+  daemon.openConnection();
+  await connection;
+  daemon.daemonGoesSilent();
+  await vi.advanceTimersByTimeAsync(25_000); // one miss
+  daemon.daemonAnswersPingsAfter("5.5s");
+  client.ensureConnected({ verify: true });
+  await vi.advanceTimersByTimeAsync(60_000);
+  expect(daemon.closesFromClient()).toEqual([]);
+  expect(client.getConnectionState()).toEqual({ status: "connected" });
+  expect(attempts()).toBe(1);
+});
+
+test("a pong that arrives after the first foreground window keeps the connection and restores the cadence", async () => {
+  useHeartbeatClock();
+  const daemon = new FakeDaemon();
+  daemon.daemonAnswersPingsAfter("5.5s");
+  const { client, attempts } = connectThroughReplacements("slow-pong", [daemon]);
+  const connection = client.connect();
+  daemon.openConnection();
+  await connection;
+  client.ensureConnected({ verify: true });
+  await vi.advanceTimersByTimeAsync(5_000);
+  expect(daemon.pingTimestamps()).toEqual(["0s", "5s"]);
+  await vi.advanceTimersByTimeAsync(55_000);
+  expect(client.getConnectionState()).toEqual({ status: "connected" });
+  expect(daemon.closesFromClient()).toEqual([]);
+  expect(attempts()).toBe(1);
+  // After the late pong the heartbeat runs ten seconds after each settled probe again.
+  expect(daemon.pingTimestamps()).toEqual(["0s", "5s", "15.5s", "31s", "46.5s"]);
+});
+
+test("a latency measurement outstanding at foreground keeps its own deadline, then the cycle runs", async () => {
+  useHeartbeatClock();
+  const first = new FakeDaemon();
+  const second = new FakeDaemon();
+  first.daemonGoesSilent();
+  const { client, attempts } = connectThroughReplacements("latency-then-foreground", [
+    first,
+    second,
+  ]);
+  const connection = client.connect();
+  first.openConnection();
+  await connection;
+  const measurement = client.measureLatency({ timeoutMs: 1_000 });
+  const measurementError = measurement.then(
+    () => null,
+    (error: unknown) => error,
+  );
+  client.ensureConnected({ verify: true });
+  expect(first.pingTimestamps()).toEqual(["0s"]);
+  await vi.advanceTimersByTimeAsync(1_000);
+  await expect(measurementError).resolves.toEqual(
+    expect.objectContaining({ message: "Latency measurement timed out (1000ms)" }),
+  );
+  expect(first.closesFromClient()).toEqual([]);
+  await vi.advanceTimersByTimeAsync(5_000);
+  expect(first.pingTimestamps()).toEqual(["0s", "1s", "6s"]);
+  expect(first.closesFromClient()).toEqual([]);
+  await vi.advanceTimersByTimeAsync(5_000);
+  expect(first.closesFromClient()).toEqual([{ code: 1001, reason: "Liveness check timed out" }]);
+  expect(attempts()).toBe(2);
+});
+
+test("a shared latency probe's activity from before foreground does not satisfy the fresh check", async () => {
+  useHeartbeatClock();
+  const first = new FakeDaemon();
+  const second = new FakeDaemon();
+  first.daemonGoesSilent();
+  const { client, attempts } = connectThroughReplacements("stale-activity", [first, second]);
+  const connection = client.connect();
+  first.openConnection();
+  await connection;
+  void client.measureLatency({ timeoutMs: 5_000 }).catch(() => {});
+  first.receiveChunk(true); // a frame arrived while the measurement was outstanding, then the socket died
+  client.ensureConnected({ verify: true });
+  expect(first.pingTimestamps()).toEqual(["0s"]);
+  await vi.advanceTimersByTimeAsync(5_000);
+  // The measurement's own deadline passed; the foreground cycle starts its first fresh window now.
+  expect(first.closesFromClient()).toEqual([]);
+  expect(first.pingTimestamps()).toEqual(["0s", "5s"]);
+  await vi.advanceTimersByTimeAsync(5_000);
+  expect(first.pingTimestamps()).toEqual(["0s", "5s", "10s"]);
+  await vi.advanceTimersByTimeAsync(5_000);
+  expect(first.closesFromClient()).toEqual([{ code: 1001, reason: "Liveness check timed out" }]);
+  expect(attempts()).toBe(2);
+});
+
+test("a latency measurement that joined a heartbeat probe survives the foreground check", async () => {
+  useHeartbeatClock();
+  const daemon = new FakeDaemon();
+  daemon.daemonAnswersPingsAfter("2s");
+  const { client, attempts } = connectThroughReplacements("latency-joins-heartbeat", [daemon]);
+  const connection = client.connect();
+  daemon.openConnection();
+  await connection;
+  await vi.advanceTimersByTimeAsync(10_000); // the first heartbeat ping is outstanding
+  const measurement = client.measureLatency({ timeoutMs: 5_000 });
+  await vi.advanceTimersByTimeAsync(1_000);
+  client.ensureConnected({ verify: true });
+  expect(daemon.pingTimestamps()).toEqual(["10s", "11s"]);
+  // The heartbeat's pong lands at 12s and answers the measurement that joined it.
+  await vi.advanceTimersByTimeAsync(1_000);
+  await expect(measurement).resolves.toBe(2000);
+  await vi.advanceTimersByTimeAsync(19_000);
+  expect(client.getConnectionState()).toEqual({ status: "connected" });
+  expect(daemon.closesFromClient()).toEqual([]);
+  expect(attempts()).toBe(1);
+  expect(daemon.pingTimestamps()).toEqual(["10s", "11s", "22s"]);
+});
+
+test("a joined latency measurement outlives a silent foreground window until its own deadline", async () => {
+  useHeartbeatClock();
+  const daemon = new FakeDaemon();
+  daemon.daemonAnswersPingsAfter("7s");
+  const { client, attempts } = connectThroughReplacements("latency-outlives-window", [daemon]);
+  const connection = client.connect();
+  daemon.openConnection();
+  await connection;
+  await vi.advanceTimersByTimeAsync(10_000); // the first heartbeat ping is outstanding
+  const measurement = client.measureLatency({ timeoutMs: 10_000 });
+  await vi.advanceTimersByTimeAsync(1_000);
+  client.ensureConnected({ verify: true });
+  await vi.advanceTimersByTimeAsync(5_000); // the first window closes silent at 16s
+  expect(daemon.pingTimestamps()).toEqual(["10s", "11s", "16s"]);
+  // The heartbeat's pong lands at 17s, inside the second window and before the 20s deadline.
+  await vi.advanceTimersByTimeAsync(1_000);
+  await expect(measurement).resolves.toBe(7000);
+  await vi.advanceTimersByTimeAsync(10_000);
+  expect(client.getConnectionState()).toEqual({ status: "connected" });
+  expect(daemon.closesFromClient()).toEqual([]);
+  expect(attempts()).toBe(1);
+  expect(daemon.pingTimestamps()).toEqual(["10s", "11s", "16s", "27s"]);
+});
+
+// Fake timers cannot suspend: a suspension is modelled as the wall clock jumping forward while
+// the pending window timer stays queued, then firing late on resume.
+async function suspendFor(ms: number): Promise<void> {
+  vi.setSystemTime(Date.now() + ms);
+}
+
+test("a foreground window that spanned a second suspension is discarded when the timer fires first", async () => {
+  useHeartbeatClock();
+  const daemon = new FakeDaemon();
+  daemon.daemonGoesSilent();
+  const { client, attempts } = connectThroughReplacements("suspended-mid-cycle-timer", [daemon]);
+  const connection = client.connect();
+  daemon.openConnection();
+  await connection;
+  client.ensureConnected({ verify: true });
+  await vi.advanceTimersByTimeAsync(5_000); // window 1 silent, window 2 outstanding
+  expect(daemon.pingTimestamps()).toEqual(["0s", "5s"]);
+  await suspendFor(60_000);
+  daemon.daemonAnswersPingsAfter("fast"); // the socket is healthy on resume
+  await vi.advanceTimersByTimeAsync(5_000); // window 2's timer fires late
+  client.ensureConnected({ verify: true }); // then AppState lands
+  await vi.advanceTimersByTimeAsync(1_000);
+  expect(daemon.closesFromClient()).toEqual([]);
+  expect(client.getConnectionState()).toEqual({ status: "connected" });
+  expect(attempts()).toBe(1);
+});
+
+test("a foreground window that spanned a second suspension is discarded when AppState fires first", async () => {
+  useHeartbeatClock();
+  const daemon = new FakeDaemon();
+  daemon.daemonGoesSilent();
+  const { client, attempts } = connectThroughReplacements("suspended-mid-cycle-appstate", [daemon]);
+  const connection = client.connect();
+  daemon.openConnection();
+  await connection;
+  client.ensureConnected({ verify: true });
+  await vi.advanceTimersByTimeAsync(5_000);
+  expect(daemon.pingTimestamps()).toEqual(["0s", "5s"]);
+  await suspendFor(60_000);
+  daemon.daemonAnswersPingsAfter("fast");
+  client.ensureConnected({ verify: true }); // AppState lands before the queued timer
+  await vi.advanceTimersByTimeAsync(5_000); // window 2's timer fires late
+  await vi.advanceTimersByTimeAsync(1_000);
+  expect(daemon.closesFromClient()).toEqual([]);
+  expect(client.getConnectionState()).toEqual({ status: "connected" });
+  expect(attempts()).toBe(1);
+});
+
+test("a dead socket after a second suspension mid-cycle still gets two fresh windows", async () => {
+  useHeartbeatClock();
+  const first = new FakeDaemon();
+  const second = new FakeDaemon();
+  first.daemonGoesSilent();
+  const { client, attempts } = connectThroughReplacements("suspended-mid-cycle-dead", [
+    first,
+    second,
+  ]);
+  const connection = client.connect();
+  first.openConnection();
+  await connection;
+  client.ensureConnected({ verify: true });
+  await vi.advanceTimersByTimeAsync(5_000);
+  first.receiveChunk(true); // a frame before the second suspension is not evidence after it
+  await suspendFor(60_000);
+  client.ensureConnected({ verify: true });
+  await vi.advanceTimersByTimeAsync(5_000); // the late timer discards window 2 and starts over
+  expect(first.pingTimestamps()).toEqual(["0s", "5s", "10s"]);
+  await vi.advanceTimersByTimeAsync(5_000);
+  expect(first.closesFromClient()).toEqual([]);
+  expect(first.pingTimestamps()).toEqual(["0s", "5s", "10s", "15s"]);
+  await vi.advanceTimersByTimeAsync(5_000);
+  expect(first.closesFromClient()).toEqual([{ code: 1001, reason: "Liveness check timed out" }]);
+  expect(attempts()).toBe(2);
+});
+
+test("repeated foreground events share one cycle and do not restart its windows", async () => {
+  useHeartbeatClock();
+  const first = new FakeDaemon();
+  const second = new FakeDaemon();
+  first.daemonGoesSilent();
+  const { client, attempts } = connectThroughReplacements("repeated-foreground", [first, second]);
+  const connection = client.connect();
+  first.openConnection();
+  await connection;
+  client.ensureConnected({ verify: true });
+  for (let elapsedSeconds = 1; elapsedSeconds <= 9; elapsedSeconds++) {
+    await vi.advanceTimersByTimeAsync(1_000);
+    client.ensureConnected({ verify: true });
+    expect(first.closesFromClient()).toEqual([]);
+  }
+  expect(first.pingTimestamps()).toEqual(["0s", "5s"]);
+  await vi.advanceTimersByTimeAsync(1_000);
+  expect(first.closesFromClient()).toEqual([{ code: 1001, reason: "Liveness check timed out" }]);
+  expect(attempts()).toBe(2);
+  second.openConnection();
+  expect(client.getConnectionState()).toEqual({ status: "connected" });
+});
+
+test("a late pong from the replaced socket does not count for the replacement", async () => {
+  useHeartbeatClock();
+  const first = new FakeDaemon();
+  const second = new FakeDaemon();
+  first.daemonGoesSilent();
+  second.daemonGoesSilent();
+  const { client, attempts } = connectThroughReplacements("late-pong", [first, second]);
+  const connection = client.connect();
+  first.openConnection();
+  await connection;
+  client.ensureConnected({ verify: true });
+  await vi.advanceTimersByTimeAsync(10_000);
+  expect(attempts()).toBe(2);
+  second.openConnection();
+  client.ensureConnected({ verify: true });
+  expect(second.pingTimestamps()).toEqual(["10s"]);
+  await vi.advanceTimersByTimeAsync(2_000);
+  first.daemonSendsPong();
+  await vi.advanceTimersByTimeAsync(8_000);
+  expect(second.pingTimestamps()).toEqual(["10s", "15s"]);
+  expect(second.closesFromClient()).toEqual([{ code: 1001, reason: "Liveness check timed out" }]);
+  expect(first.closesFromClient()).toEqual([{ code: 1001, reason: "Liveness check timed out" }]);
+});
+
+test("closing the client in the middle of a foreground cycle stops it", async () => {
+  useHeartbeatClock();
+  const first = new FakeDaemon();
+  const second = new FakeDaemon();
+  first.daemonGoesSilent();
+  const { client, attempts } = connectThroughReplacements("close-mid-cycle", [first, second]);
+  const connection = client.connect();
+  first.openConnection();
+  await connection;
+  client.ensureConnected({ verify: true });
+  await vi.advanceTimersByTimeAsync(5_000);
+  expect(first.pingTimestamps()).toEqual(["0s", "5s"]);
+  await client.close();
+  await vi.advanceTimersByTimeAsync(60_000);
+  expect(first.pingTimestamps()).toEqual(["0s", "5s"]);
+  expect(first.closesFromClient()).toEqual([{ code: 1000, reason: "Client closed" }]);
+  expect(attempts()).toBe(1);
+  expect(client.getConnectionState()).toEqual({ status: "disposed" });
+});
+
+test("a foreground teardown with reconnect disabled reports the timeout and stays down", async () => {
+  useHeartbeatClock();
+  const daemon = new FakeDaemon();
+  daemon.daemonGoesSilent();
+  const { client, attempts } = connectThroughReplacements("no-reconnect", [daemon], {
+    reconnect: { enabled: false },
+  });
+  const connection = client.connect();
+  daemon.openConnection();
+  await connection;
+  client.ensureConnected({ verify: true });
+  await vi.advanceTimersByTimeAsync(10_000);
+  expect(client.getConnectionState()).toEqual({
+    status: "disconnected",
+    reason: "Liveness check timed out (5000ms)",
+  });
+  await vi.advanceTimersByTimeAsync(10_000);
+  expect(attempts()).toBe(1);
+});
+
+test("the immediate reconnect after a foreground teardown backs off normally if it fails", async () => {
+  useHeartbeatClock();
+  const first = new FakeDaemon();
+  const second = new FakeDaemon();
+  const third = new FakeDaemon();
+  first.daemonGoesSilent();
+  const { client, attempts } = connectThroughReplacements("no-storm", [first, second, third]);
+  const connection = client.connect();
+  first.openConnection();
+  await connection;
+  client.ensureConnected({ verify: true });
+  await vi.advanceTimersByTimeAsync(10_000);
+  expect(attempts()).toBe(2);
+  second.daemonClosesWith("connection refused");
+  await vi.advanceTimersByTimeAsync(1_499);
+  expect(attempts()).toBe(2);
+  await vi.advanceTimersByTimeAsync(1);
+  expect(attempts()).toBe(3);
+  third.openConnection();
+  expect(client.getConnectionState()).toEqual({ status: "connected" });
+});
+
 test("an obsolete heartbeat cannot close a replacement connection", async () => {
   useHeartbeatClock();
   const first = new FakeDaemon();
@@ -2887,6 +3293,18 @@ test("a heartbeat sharing a latency measurement still saves the RTT", async () =
   expect(session.lastLivenessRttMs()).toBe(2000);
 });
 
+test("a measurement that joins an in-flight ping reports that ping's round trip", async () => {
+  useHeartbeatClock();
+  const session = new DaemonClientSession();
+  session.daemonAnswersPingsAfter("2s");
+  await session.connect();
+  await session.advance(11_000); // the heartbeat ping sent at 10s is still outstanding
+  const measurement = session.measureLatency({ timeoutMs: 5_000 });
+  await session.advance(1_000);
+  await expect(measurement).resolves.toBe(2000);
+  expect(session.lastLivenessRttMs()).toBe(2000);
+});
+
 test("reports the round-trip time of the last successful heartbeat", async () => {
   useHeartbeatClock();
   const session = new DaemonClientSession();

packages/client/src/daemon-client.ts

@@ -1,3 +1,9 @@
+import {
+  DiffStreamRouter,
+  DiffSubscriptionError,
+  type DiffCompare,
+  type DiffState,
+} from "./diff/index.js";
 import {
   ChunkedTransport,
   ChunkedTransportProtocolError,
@@ -1149,6 +1155,12 @@ const DEFAULT_CONNECT_TIMEOUT_MS = 15_000;
 const DEFAULT_LIVENESS_TIMEOUT_MS = 5000;
 const LIVENESS_HEARTBEAT_INTERVAL_MS = 10_000;
 const LIVENESS_HEARTBEAT_TIMEOUT_MS = 15_000;
+// Window of each check after the app returns to the foreground. A socket that silently died
+// while the app was suspended stays OPEN and accepts sends, so only inbound frames prove it.
+const FOREGROUND_LIVENESS_TIMEOUT_MS = 5_000;
+// A foreground window whose timer fires this much later than scheduled (wall clock) spanned
+// another suspension or a blocked JS thread, so its silence is no evidence either way.
+const SUSPENDED_WINDOW_SLACK_MS = 2_000;
 const LIVENESS_FAILURE_RECONNECT_THRESHOLD = 2;
 
 /** Default timeout for waiting for connection before sending queued messages */
@@ -1311,13 +1323,27 @@ interface PingProbe {
   reject: (error: Error) => void;
   timeoutHandle: ReturnType<typeof setTimeout>;
   startedAt: number;
+  // Wall clock, which keeps running while the OS has JS suspended; perfNow may not.
+  startedAtWallMs: number;
   receivedActivity: boolean;
-  // Whether a timeout on this ping should be recorded as a liveness failure. Only the
-  // heartbeat sets this; a latency measurement never drives teardown, even when a
-  // heartbeat tick shares (dedupes onto) an in-flight measurement ping.
-  drivesLivenessFailure: boolean;
+  role: PingProbeRole;
 }
 
+// A latency measurement waits for the next pong, not for a particular probe: the liveness
+// cycle may replace the probe the measurement joined, and a pong that answers an expired window
+// still proves the socket. It settles on a pong, a connection failure, or its own deadline.
+interface PongWaiter {
+  startedAt: number;
+  resolve: (rttMs: number) => void;
+  reject: (error: Error) => void;
+}
+
+// Who started the probe. A "latency" probe never drives teardown, even when a heartbeat
+// tick shares (dedupes onto) it. A "heartbeat" probe runs the regular cadence. A
+// "foreground" probe belongs to the verification cycle that ensureConnected({ verify })
+// runs after the app resumes: short window, immediate second check, reconnect at once.
+type PingProbeRole = "latency" | "heartbeat" | "foreground";
+
 // COMPAT(providerUsageList): added in v0.1.98, remove after 2027-03-26.
 export function supportsUsageReports(features: ServerInfoStatusPayload["features"]): boolean {
   return features?.usageSources === true || features?.providerUsageList === true;
@@ -1426,6 +1452,20 @@ export class DaemonClient {
   private connectionFailure: Error | undefined;
   private authFailureReasonValue: DaemonAuthFailureReason | null = null;
   private connectionState: ConnectionState = { status: "idle" };
+  private readonly diffStreams = new DiffStreamRouter({
+    modern: (compare) =>
+      this.observe(
+        "diff.subscribe.response",
+        { type: "diff.subscribe.request", ...compare },
+        {
+          validateSnapshot: (payload) => {
+            if (payload.error) throw new DiffSubscriptionError(payload.error);
+          },
+        },
+      ),
+    legacy: (cwd, compare) => this.observeCheckoutDiff(cwd, compare),
+    failed: (error) => this.logger.error({ err: error }, "Diff observer failed"),
+  });
   private readonly terminalStreams = new TerminalStreamRouter();
   private pendingBinaryFileReads = new Map<string, PendingBinaryFileRead>();
   private activeBinaryFileTransfers = new Map<string, BinaryFileTransferState>();
@@ -1455,6 +1495,7 @@ export class DaemonClient {
   private runtimeMetricsInterval: ReturnType<typeof setInterval> | null = null;
   private runtimeMetrics: DaemonClientRuntimeMetrics | null = null;
   private pingProbe: PingProbe | null = null;
+  private readonly pongWaiters = new Set<PongWaiter>();
   private livenessHeartbeatTimer: ReturnType<typeof setTimeout> | null = null;
   private lastLivenessRttMs: number | null = null;
   private consecutiveLivenessFailures = 0;
@@ -1775,10 +1816,12 @@ export class DaemonClient {
     this.disposeTransport(1000, "Client closed");
     this.providerSnapshotUpdates.clear();
     this.clearWaiters(new Error("Daemon client closed"));
+    await this.diffStreams.close();
     await this.owned.close();
     this.creations.close();
     this.rejectPendingSendQueue(new Error("Daemon client closed"));
     this.rejectPingProbe(new Error("Daemon client closed"));
+    this.rejectPongWaiters(new Error("Daemon client closed"));
     this.terminalStreams.clearSlots();
     this.lastServerInfoMessage = null;
     if (this.runtimeMetricsInterval) {
@@ -1801,7 +1844,7 @@ export class DaemonClient {
       this.shouldReconnect = true;
     }
     if (this.connectionState.status === "connected") {
-      if (options?.verify) this.tickLivenessHeartbeat();
+      if (options?.verify) this.verifyConnectionOnForeground();
       return;
     }
     if (this.connectionState.status === "connecting") return;
@@ -2548,43 +2591,41 @@ export class DaemonClient {
 
   measureLatency(params?: { timeoutMs?: number }): Promise<number> {
     const timeoutMs = Math.max(1, params?.timeoutMs ?? DEFAULT_LIVENESS_TIMEOUT_MS);
-    let deadline: ReturnType<typeof setTimeout>;
-    const timeout = new Promise<never>((_resolve, reject) => {
-      deadline = setTimeout(() => reject(new PingTimeoutError(timeoutMs)), timeoutMs);
-    });
-    return Promise.race([
-      this.sendPingAwaitRtt({ timeoutMs, drivesLivenessFailure: false }),
-      timeout,
-    ])
-      .catch((error) => {
-        throw toTimeoutError(error, "Latency measurement", timeoutMs);
-      })
-      .finally(() => clearTimeout(deadline));
-  }
-
-  private async livenessPing(params?: { timeoutMs?: number }): Promise<number> {
-    const timeoutMs = Math.max(1, params?.timeoutMs ?? DEFAULT_LIVENESS_TIMEOUT_MS);
-    try {
-      const rttMs = await this.sendPingAwaitRtt({ timeoutMs, drivesLivenessFailure: true });
-      this.lastLivenessRttMs = rttMs;
-      return rttMs;
-    } catch (error) {
-      throw toTimeoutError(error, "Liveness check", timeoutMs);
-    }
-  }
-
-  private sendPingAwaitRtt(params: {
-    timeoutMs: number;
-    drivesLivenessFailure: boolean;
-  }): Promise<number> {
     if (this.connectionState.status !== "connected" || !this.transport) {
       return Promise.reject(
         new Error(`Transport not connected (status: ${this.connectionState.status})`),
       );
     }
+    const waiter: PongWaiter = { startedAt: perfNow(), resolve: () => {}, reject: () => {} };
+    const pong = new Promise<number>((resolve, reject) => {
+      waiter.resolve = resolve;
+      waiter.reject = reject;
+    });
+    let deadline: ReturnType<typeof setTimeout>;
+    const timeout = new Promise<never>((_resolve, reject) => {
+      deadline = setTimeout(() => reject(new PingTimeoutError(timeoutMs)), timeoutMs);
+    });
+    this.pongWaiters.add(waiter);
+    // The probe only puts a ping on the wire; its own outcome is the heartbeat's concern. A
+    // measurement that joins an in-flight ping reports that ping's round trip, so the clock
+    // starts at the send, not at the join.
+    const probe = this.ensurePingProbe({ timeoutMs, role: "latency" });
+    void probe.promise.catch(() => {});
+    waiter.startedAt = probe.startedAt;
+    return Promise.race([pong, timeout])
+      .catch((error) => {
+        throw toTimeoutError(error, "Latency measurement", timeoutMs);
+      })
+      .finally(() => {
+        clearTimeout(deadline);
+        this.pongWaiters.delete(waiter);
+      });
+  }
 
+  // One ping is outstanding at a time; a caller that finds one shares it.
+  private ensurePingProbe(params: { timeoutMs: number; role: PingProbeRole }): PingProbe {
     if (this.pingProbe) {
-      return this.pingProbe.promise;
+      return this.pingProbe;
     }
 
     const startedAt = perfNow();
@@ -2600,10 +2641,20 @@ export class DaemonClient {
       this.pingProbe = null;
       const error = new PingTimeoutError(timeoutMs);
       probe.reject(error);
+      if (
+        probe.role === "foreground" &&
+        Date.now() - probe.startedAtWallMs > timeoutMs + SUSPENDED_WINDOW_SLACK_MS
+      ) {
+        // The app was suspended again inside this window. Discard it, any frame it saw and
+        // the misses before it; the heartbeat tick that owns this probe starts the cycle over.
+        probe.receivedActivity = false;
+        this.consecutiveLivenessFailures = 0;
+        return;
+      }
       // Keep the deadline fixed so subsequent pings renew the daemon socket lease
       // during a long download. Progress proves life even when its pong is queued.
-      if (probe.drivesLivenessFailure && !probe.receivedActivity) {
-        this.recordLivenessFailure(toTimeoutError(error, "Liveness check", timeoutMs));
+      if (probe.role !== "latency" && !probe.receivedActivity) {
+        this.recordLivenessFailure(toTimeoutError(error, "Liveness check", timeoutMs), probe.role);
       }
     };
     const probe: PingProbe = {
@@ -2613,7 +2664,8 @@ export class DaemonClient {
       timeoutHandle: setTimeout(expire, timeoutMs),
       receivedActivity: false,
       startedAt,
-      drivesLivenessFailure: params.drivesLivenessFailure,
+      startedAtWallMs: Date.now(),
+      role: params.role,
     };
     this.pingProbe = probe;
 
@@ -2621,7 +2673,8 @@ export class DaemonClient {
       if (this.pingProbe !== probe) return;
       const sendError = error instanceof Error ? error : new Error(String(error));
       this.rejectPingProbe(sendError);
-      if (probe.drivesLivenessFailure) this.recordLivenessFailure(sendError);
+      this.rejectPongWaiters(sendError);
+      if (probe.role !== "latency") this.recordLivenessFailure(sendError, probe.role);
     };
     try {
       void this.sendJsonMessage("ping", "ping", { type: "ping" }).catch(rejectSend);
@@ -2629,7 +2682,7 @@ export class DaemonClient {
       rejectSend(error);
     }
 
-    return promise;
+    return probe;
   }
 
   private startLivenessHeartbeat(): void {
@@ -2656,14 +2709,55 @@ export class DaemonClient {
     }, LIVENESS_HEARTBEAT_INTERVAL_MS);
   }
 
-  private tickLivenessHeartbeat(): void {
+  /**
+   * The app came back to the foreground. Start a fresh verification cycle: two
+   * FOREGROUND_LIVENESS_TIMEOUT_MS windows back to back, then teardown and an immediate
+   * reconnect. Misses recorded before the app was suspended do not count, and a heartbeat
+   * probe that straddled the suspension measured the suspension, not the socket, so it is
+   * dropped without a miss. Latency measurements wait for the next pong, so replacing the
+   * probe they joined does not fail them.
+   */
+  private verifyConnectionOnForeground(): void {
+    // A cycle already running keeps its deadline and its window count.
+    if (this.pingProbe?.role === "foreground") return;
+    if (this.pingProbe?.role === "heartbeat") {
+      this.rejectPingProbe(new Error("Heartbeat superseded by foreground check"));
+    }
+    // A shared latency probe keeps its deadline, but only frames that arrive from now on count
+    // for this check; a latency probe never reads this flag for itself.
+    if (this.pingProbe) this.pingProbe.receivedActivity = false;
+    this.consecutiveLivenessFailures = 0;
+    this.tickLivenessHeartbeat("foreground");
+  }
+
+  private tickLivenessHeartbeat(check: "heartbeat" | "foreground" = "heartbeat"): void {
     this.stopLivenessHeartbeat();
     const transport = this.transport;
-    const tick = this.livenessPing({ timeoutMs: LIVENESS_HEARTBEAT_TIMEOUT_MS });
-    void tick
-      .catch(() => {})
-      .finally(() => {
-        if (this.transport === transport) this.scheduleNextLivenessHeartbeat();
+    if (this.connectionState.status !== "connected" || !transport) return;
+    const probe = this.ensurePingProbe({
+      timeoutMs:
+        check === "foreground" ? FOREGROUND_LIVENESS_TIMEOUT_MS : LIVENESS_HEARTBEAT_TIMEOUT_MS,
+      role: check,
+    });
+    void probe.promise
+      .then(
+        (rttMs) => {
+          this.lastLivenessRttMs = rttMs;
+          return true;
+        },
+        () => probe.receivedActivity,
+      )
+      .then((alive) => {
+        // A replaced transport, or a probe that superseded this one, owns the next step.
+        if (this.transport !== transport || this.pingProbe) return;
+        if (!alive && check === "foreground") {
+          // Zero inbound frames in a foreground window: check again now. The second silent
+          // window reaches the threshold inside recordLivenessFailure.
+          this.tickLivenessHeartbeat("foreground");
+          return;
+        }
+        this.scheduleNextLivenessHeartbeat();
+        return;
       });
   }
 
@@ -2674,7 +2768,12 @@ export class DaemonClient {
   private observe<T extends CorrelatedResponseType>(
     responseType: T,
     message: { type: SessionInboundMessage["type"] } & Record<string, unknown>,
-    options?: { requestId?: string; timeout?: number; signal?: AbortSignal },
+    options?: {
+      requestId?: string;
+      timeout?: number;
+      signal?: AbortSignal;
+      validateSnapshot?: (payload: CorrelatedResponsePayload<T>) => void;
+    },
   ): OwnedSubscription<CorrelatedResponsePayload<T>> {
     let resetSource = false;
     return this.owned.observeRequest<CorrelatedResponsePayload<T>>(
@@ -2700,12 +2799,16 @@ export class DaemonClient {
                   requestType: message.type,
                   error: payload.error,
                 });
+              options?.validateSnapshot?.(payload);
               accept(payload);
               return payload;
             },
           });
         } catch (error) {
-          resetSource = !(error instanceof DaemonRpcError) && this.isConnected;
+          resetSource =
+            !(error instanceof DaemonRpcError) &&
+            !(error instanceof DiffSubscriptionError) &&
+            this.isConnected;
           throw error;
         }
       },
@@ -4856,6 +4959,26 @@ export class DaemonClient {
     });
   }
 
+  observeDiff(compare: DiffCompare): OwnedSubscription<DiffState> {
+    return this.diffStreams.observe(compare);
+  }
+
+  async readDiff(compare: DiffCompare): Promise<DiffState> {
+    const observation = this.observeDiff(compare);
+    try {
+      return await new Promise<DiffState>((resolve) => {
+        observation.subscribe({
+          snapshot: (state) => {
+            if (state.phase !== "loading") resolve(state);
+          },
+          update: () => {},
+        });
+      });
+    } finally {
+      await observation.release();
+    }
+  }
+
   observeCheckoutDiff(
     cwd: string,
     compare: { mode: "uncommitted" | "base" | "all"; baseRef?: string; ignoreWhitespace?: boolean },
@@ -7180,6 +7303,7 @@ export class DaemonClient {
     this.chunkedTransport?.close();
     this.chunkedTransport = null;
     this.owned.disconnected();
+    this.diffStreams.disconnected();
     this.providerSnapshotUpdates.clear();
     this.stopLivenessHeartbeat();
     this.cleanupTransport();
@@ -7312,6 +7436,7 @@ export class DaemonClient {
         messageType: "pong",
       });
       this.resolvePingProbe();
+      this.resolvePongWaiters();
       this.runtimeMetrics?.recordMessage("pong", bytes, perfNow() - startMs);
       return;
     }
@@ -7370,6 +7495,10 @@ export class DaemonClient {
       for (const handler of this.hostFrameListeners) handler(hostFrame);
       return true;
     }
+    if (rawBytes[0] >= 0x40 && rawBytes[0] <= 0x4f) {
+      this.diffStreams.receive(rawBytes);
+      return true;
+    }
     if (isPluginFrameOpcode(rawBytes[0])) {
       this.handlePluginFrame(rawBytes);
       return true;
@@ -7557,6 +7686,8 @@ export class DaemonClient {
     reason?: string;
     event?: string;
     reasonCode?: string;
+    // Attempt now instead of after the backoff delay. Failures of that attempt back off as usual.
+    immediate?: boolean;
   }): void {
     if (this.reconnectTimeout) {
       clearTimeout(this.reconnectTimeout);
@@ -7570,6 +7701,7 @@ export class DaemonClient {
     }
 
     this.owned.disconnected();
+    this.diffStreams.disconnected();
     this.providerSnapshotUpdates.clear();
 
     // Clear all pending waiters and queued sends since the connection was lost
@@ -7586,6 +7718,7 @@ export class DaemonClient {
     this.clearWaiters(connectionError, true);
     this.rejectPendingSendQueue(connectionError);
     this.rejectPingProbe(connectionError);
+    this.rejectPongWaiters(connectionError);
     this.terminalStreams.clearSlots();
     this.lastServerInfoMessage = null;
 
@@ -7605,6 +7738,10 @@ export class DaemonClient {
       return;
     }
 
+    if (input?.immediate) {
+      void this.attemptConnect();
+      return;
+    }
     this.armReconnectTimer();
   }
 
@@ -7679,7 +7816,18 @@ export class DaemonClient {
     probe.reject(error);
   }
 
-  private recordLivenessFailure(error: Error): void {
+  private resolvePongWaiters(): void {
+    const now = perfNow();
+    for (const waiter of this.pongWaiters) waiter.resolve(now - waiter.startedAt);
+    this.pongWaiters.clear();
+  }
+
+  private rejectPongWaiters(error: Error): void {
+    for (const waiter of this.pongWaiters) waiter.reject(error);
+    this.pongWaiters.clear();
+  }
+
+  private recordLivenessFailure(error: Error, role: PingProbeRole): void {
     this.consecutiveLivenessFailures += 1;
     if (this.consecutiveLivenessFailures < LIVENESS_FAILURE_RECONNECT_THRESHOLD) {
       return;
@@ -7691,6 +7839,8 @@ export class DaemonClient {
       reason: error.message,
       event: "LIVENESS_TIMEOUT",
       reasonCode: "liveness_timeout",
+      // The foreground cycle already waited two windows; the user is looking at the app.
+      immediate: role === "foreground",
     });
   }
 
@@ -7730,6 +7880,7 @@ export class DaemonClient {
           this.reconnectAttempt = 0;
           this.updateConnectionState({ status: "connected" }, { event: "HELLO_SERVER_INFO" });
           this.startLivenessHeartbeat();
+          this.diffStreams.connected(serverInfo.features?.diffSubscription === true);
           this.owned.restore(serverInfo, {
             ...DEFAULT_CLIENT_CAPABILITIES,
             ...this.config.capabilities,
@@ -8007,3 +8158,5 @@ function resolveAgentConfig(options: CreateAgentRequestOptions): AgentSessionCon
     cwd: merged.cwd,
   };
 }
+
+export type { DiffCompare, DiffState, DiffEntry, ResolvedShas } from "./diff/index.js";

packages/client/src/diff/client.test.ts

@@ -0,0 +1,210 @@
+import { expect, test } from "vitest";
+import { DaemonClient, type DaemonTransport } from "../daemon-client.js";
+import {
+  encodeDiffFrame,
+  DiffOpcode,
+  type DiffFrame,
+} from "@getpaseo/protocol/binary-frames/index";
+import type { DiffState } from "./index.js";
+function socket() {
+  let message: (data: unknown, binary?: boolean) => void = () => {};
+  let open: () => void = () => {};
+  let close: (event?: unknown) => void = () => {};
+  const sent: Record<string, unknown>[] = [];
+  const transport: DaemonTransport = {
+    send: (data) => {
+      if (typeof data === "string") {
+        const value = JSON.parse(data);
+        if (value.type === "session") {
+          sent.push(value.message);
+          if (value.message.type === "subscription.release.request")
+            queueMicrotask(() =>
+              receive({
+                type: "subscription.release.response",
+                payload: {
+                  requestId: value.message.requestId,
+                  subscriptionId: value.message.subscriptionId,
+                  released: true,
+                },
+              }),
+            );
+        }
+      }
+    },
+    close: () => {},
+    onMessage: (handler) => {
+      message = handler;
+      return () => {};
+    },
+    onOpen: (handler) => {
+      open = handler;
+      return () => {};
+    },
+    onClose: (handler) => {
+      close = handler;
+      return () => {};
+    },
+    onError: () => () => {},
+  };
+  const receive = (value: unknown) =>
+    message(JSON.stringify({ type: "session", message: value }), false);
+  return {
+    transport,
+    sent,
+    receive,
+    raw: (bytes: Uint8Array) => message(bytes, true),
+    frame: (frame: DiffFrame) => message(encodeDiffFrame(frame), true),
+    disconnect: () => close({ code: 1001 }),
+    open: (modern = true) => {
+      open();
+      receive({
+        type: "status",
+        payload: {
+          status: "server_info",
+          serverId: "srv_test",
+          hostname: null,
+          version: null,
+          features: { ownedSubscriptions: true, diffSubscription: modern, checkoutDiffAll: true },
+        },
+      });
+    },
+  };
+}
+const compare = {
+  cwd: "/repo",
+  base: { kind: "rev", rev: "HEAD" },
+  head: { kind: "worktree" },
+} as const;
+const begin: DiffFrame = {
+  opcode: DiffOpcode.Begin,
+  slot: 3,
+  revision: 1,
+  baseRevision: 0,
+  baseSha: "a",
+  headSha: "b",
+  mergeBaseSha: "",
+  entries: [],
+};
+const end: DiffFrame = {
+  opcode: DiffOpcode.End,
+  slot: 3,
+  revision: 1,
+  diffTooLarge: false,
+  omitted: [],
+  error: null,
+};
+async function setup(modern = true) {
+  const wire = socket();
+  const client = new DaemonClient({
+    url: "ws://test",
+    clientId: "diff-test",
+    transportFactory: () => wire.transport,
+    reconnect: { enabled: false },
+  });
+  const connecting = client.connect();
+  wire.open(modern);
+  await connecting;
+  return { wire, client };
+}
+function acknowledge(wire: ReturnType<typeof socket>, ordinal = 1) {
+  const request = wire.sent.findLast((value) => value.type === "diff.subscribe.request");
+  wire.receive({
+    type: "diff.subscribe.response",
+    payload: {
+      requestId: request?.requestId,
+      subscriptionId: `diff-${ordinal}`,
+      slot: 3,
+      error: null,
+    },
+  });
+}
+test("attaches binary routing during handshake and resubscribes with a full state on reconnect", async () => {
+  const { wire, client } = await setup();
+  try {
+    const states: DiffState[] = [];
+    const observation = client.observeDiff(compare);
+    observation.subscribe({ snapshot: (state) => states.push(state), update: () => {} });
+    acknowledge(wire);
+    wire.frame(begin);
+    wire.frame(end);
+    expect(states.map((state) => state.phase)).toEqual(["loading", "ready"]);
+    wire.disconnect();
+    const connected = client.connect();
+    wire.open();
+    await connected;
+    acknowledge(wire, 2);
+    wire.frame(begin);
+    wire.frame(end);
+    expect(states.map((state) => state.phase)).toEqual(["loading", "ready", "loading", "ready"]);
+    await observation.release();
+  } finally {
+    await client.close();
+  }
+});
+test("an unknown version fails only its subscription and readDiff resolves typed subscribe errors", async () => {
+  const { wire, client } = await setup();
+  try {
+    const observation = client.observeDiff(compare);
+    const states: DiffState[] = [];
+    observation.subscribe({ snapshot: (state) => states.push(state), update: () => {} });
+    acknowledge(wire);
+    const malformed = { ...begin, slot: 3 };
+    const bytes = encodeDiffFrame(malformed);
+    bytes[2] = 255;
+    wire.raw(bytes);
+    expect(states.at(-1)).toMatchObject({ phase: "error", error: { code: "UNKNOWN" } });
+    const next = client.readDiff(compare);
+    const request = wire.sent.findLast((value) => value.type === "diff.subscribe.request");
+    wire.receive({
+      type: "diff.subscribe.response",
+      payload: {
+        requestId: request?.requestId,
+        subscriptionId: "",
+        slot: 0,
+        error: { code: "INVALID_COMPARE", message: "invalid" },
+      },
+    });
+    expect(await next).toMatchObject({ phase: "error", error: { code: "INVALID_COMPARE" } });
+    expect(client.isConnected).toBe(true);
+    await observation.release();
+  } finally {
+    await client.close();
+  }
+});
+test.each(["uncommitted", "base", "all"] as const)(
+  "old hosts serve the %s comparison through the tagged compatibility route",
+  async (mode) => {
+    const { wire, client } = await setup(false);
+    try {
+      const comparison =
+        mode === "uncommitted"
+          ? compare
+          : {
+              ...compare,
+              base: { kind: "rev", rev: "refs/tags/v1" } as const,
+              head: mode === "base" ? ({ kind: "rev", rev: "HEAD" } as const) : compare.head,
+              mergeBase: true,
+            };
+      const result = client.readDiff(comparison);
+      const request = wire.sent.find((value) => value.type === "subscribe_checkout_diff_request");
+      expect(request?.compare).toMatchObject({ mode });
+      wire.receive({
+        type: "subscribe_checkout_diff_response",
+        payload: {
+          requestId: request?.requestId,
+          subscriptionId: "legacy",
+          cwd: "/repo",
+          files: [],
+          error: null,
+        },
+      });
+      wire.receive({
+        type: "checkout_diff_update",
+        payload: { subscriptionId: "legacy", cwd: "/repo", files: [], error: null },
+      });
+      expect(await result).toMatchObject({ phase: "ready", files: [] });
+    } finally {
+      await client.close();
+    }
+  },
+);

packages/client/src/diff/index.ts

@@ -0,0 +1,246 @@
+import {
+  decodeDiffFrame,
+  type DiffEntry as WireEntry,
+  type DiffBegin,
+} from "@getpaseo/protocol/binary-frames/index";
+import type {
+  DiffSubscribeRequest,
+  SubscribeCheckoutDiffRequest,
+  DiffError,
+  DiffSubscribeResponse,
+  SubscribeCheckoutDiffResponse,
+  ParsedDiffFile,
+} from "@getpaseo/protocol/messages";
+import type { OwnedSubscription, SubscriptionObserver } from "../connection/index.js";
+import { DiffStream } from "./stream.js";
+export type { DiffEndpoint } from "@getpaseo/protocol/messages";
+export type DiffCompare = Omit<DiffSubscribeRequest, "type" | "requestId">;
+
+export type DiffEntry = Omit<WireEntry, "body">;
+export type ResolvedShas = Pick<DiffBegin, "baseSha" | "headSha" | "mergeBaseSha">;
+
+export type DiffState =
+  | { phase: "loading"; revision: number; entries: DiffEntry[] }
+  | {
+      phase: "ready";
+      revision: number;
+      resolved: ResolvedShas;
+      diffTooLarge: boolean;
+      files: ParsedDiffFile[];
+    }
+  | { phase: "error"; revision: number; error: DiffError };
+type LegacyCompare = SubscribeCheckoutDiffRequest["compare"];
+
+interface Port {
+  modern(compare: DiffCompare): OwnedSubscription<DiffSubscribeResponse["payload"]>;
+  legacy(
+    cwd: string,
+    compare: LegacyCompare,
+  ): OwnedSubscription<SubscribeCheckoutDiffResponse["payload"]>;
+  failed(error: unknown): void;
+}
+export class DiffSubscriptionError extends Error {
+  constructor(readonly detail: DiffError) {
+    super(detail.message);
+  }
+}
+interface Registration {
+  start(modern: boolean): void;
+  disconnected(): void;
+  release(): Promise<void>;
+}
+/** Capability selection, slot ownership and reconnect belong to this domain. */
+export class DiffStreamRouter {
+  private registrations = new Set<Registration>();
+  private slots = new Map<number, { receive(bytes: Uint8Array): void }>();
+  private mode: boolean | null = null;
+  constructor(private port: Port) {}
+  connected(modern: boolean): void {
+    this.mode = modern;
+    for (const registration of this.registrations) registration.start(modern);
+  }
+  disconnected(): void {
+    this.mode = null;
+    this.slots.clear();
+    for (const registration of this.registrations) registration.disconnected();
+  }
+  async close(): Promise<void> {
+    await Promise.all([...this.registrations].map((registration) => registration.release()));
+  }
+  receive(bytes: Uint8Array): void {
+    const owner = this.slots.get(bytes[1]);
+    owner?.receive(bytes);
+  }
+  private releaseBackend(backend: OwnedSubscription<unknown> | null): void {
+    void backend?.release().catch(this.port.failed);
+  }
+  observe(compare: DiffCompare): OwnedSubscription<DiffState> {
+    let backend: OwnedSubscription<unknown> | null = null;
+    let state: (DiffState & { subscriptionId: string }) | null = null;
+    let slot: number | null = null;
+    let released = false;
+    let generation = 0;
+    let stream: DiffStream;
+    const observers = new Set<SubscriptionObserver<DiffState & { subscriptionId: string }>>();
+    let resolve!: (state: DiffState & { subscriptionId: string }) => void;
+    const ready = new Promise<DiffState & { subscriptionId: string }>((accept) => {
+      resolve = accept;
+    });
+    const publish = (value: DiffState) => {
+      if (released) return;
+      state = { ...value, subscriptionId: backend?.subscriptionId ?? "" };
+      resolve(state);
+      for (const observer of observers) {
+        try {
+          observer.snapshot(state);
+        } catch (error) {
+          this.port.failed(error);
+        }
+      }
+    };
+    const detach = () => {
+      if (slot !== null) this.slots.delete(slot);
+      slot = null;
+    };
+    const fail = (error: unknown) => {
+      detach();
+      publish({
+        phase: "error",
+        revision: 0,
+        error:
+          error instanceof DiffSubscriptionError
+            ? error.detail
+            : { code: "UNKNOWN", message: error instanceof Error ? error.message : String(error) },
+      });
+      this.releaseBackend(backend);
+    };
+    const registration: Registration = {
+      start: (modern) => {
+        if (released || backend) return;
+        const current = ++generation;
+        const failed = (error: unknown) => {
+          if (current === generation) fail(error);
+        };
+        stream = new DiffStream(publish);
+        if (modern) {
+          const observation = this.port.modern(compare);
+          backend = observation;
+          observation.subscribe({
+            snapshot: (payload) => {
+              if (current !== generation) return;
+              detach();
+              if (payload.error) {
+                fail(new DiffSubscriptionError(payload.error));
+                return;
+              }
+              slot = payload.slot;
+              const owner = {
+                receive: (bytes: Uint8Array) => {
+                  try {
+                    stream.receive(decodeDiffFrame(bytes));
+                  } catch (error) {
+                    fail(error);
+                  }
+                },
+              };
+              this.slots.set(slot, owner);
+            },
+            update: () => {},
+            error: failed,
+          });
+          void observation.ready.catch(failed);
+        } else {
+          // COMPAT(diffSubscription): added in v0.15.0, remove after 2027-05-06.
+          const legacy = legacyCompare(compare);
+          if (!legacy) {
+            fail(
+              new DiffSubscriptionError({
+                code: "INVALID_COMPARE",
+                message: "Update the host to use this comparison",
+              }),
+            );
+            return;
+          }
+          const observation = this.port.legacy(compare.cwd, legacy);
+          backend = observation;
+          const apply = (
+            payload: Pick<
+              SubscribeCheckoutDiffResponse["payload"],
+              "files" | "error" | "diffTooLarge"
+            >,
+          ) => {
+            if (current !== generation) return;
+            if (payload.error)
+              publish({
+                phase: "error",
+                revision: 0,
+                error: {
+                  code: payload.error.code === "NOT_GIT_REPO" ? "NOT_GIT_REPO" : "UNKNOWN",
+                  message: payload.error.message,
+                },
+              });
+            else
+              publish({
+                phase: "ready",
+                revision: (state?.revision ?? 0) + 1,
+                resolved: { baseSha: "", headSha: "", mergeBaseSha: "" },
+                diffTooLarge: payload.diffTooLarge === true,
+                files: payload.files,
+              });
+          };
+          observation.subscribe({
+            snapshot: apply,
+            update: (message) => {
+              if (message.type === "checkout_diff_update") apply(message.payload);
+            },
+            error: failed,
+          });
+          void observation.ready.catch(failed);
+        }
+      },
+      disconnected: () => {
+        generation++;
+        detach();
+        state = null;
+        const old = backend;
+        backend = null;
+        void old?.release().catch(this.port.failed);
+      },
+      release: async () => {
+        if (released) return;
+        released = true;
+        generation++;
+        detach();
+        this.registrations.delete(registration);
+        await backend?.release();
+        observers.clear();
+      },
+    };
+    this.registrations.add(registration);
+    if (this.mode !== null) registration.start(this.mode);
+    return {
+      get subscriptionId() {
+        return backend?.subscriptionId ?? null;
+      },
+      ready,
+      subscribe: (observer) => {
+        observers.add(observer);
+        if (state) observer.snapshot(state);
+        return () => observers.delete(observer);
+      },
+      release: () => registration.release(),
+    };
+  }
+}
+function legacyCompare(compare: DiffCompare): LegacyCompare | null {
+  if (compare.highlight === false) return null;
+  const { base, head } = compare;
+  if (base.kind !== "rev") return null;
+  if (base.rev === "HEAD" && head.kind === "worktree")
+    return { mode: "uncommitted", ignoreWhitespace: compare.ignoreWhitespace };
+  if (compare.mergeBase && head.kind === "rev" && head.rev === "HEAD")
+    return { mode: "base", baseRef: base.rev, ignoreWhitespace: compare.ignoreWhitespace };
+  if (compare.mergeBase && head.kind === "worktree")
+    return { mode: "all", baseRef: base.rev, ignoreWhitespace: compare.ignoreWhitespace };
+  return null;
+}

packages/client/src/diff/stream.test.ts

@@ -0,0 +1,86 @@
+import { expect, test } from "vitest";
+import { DiffStream } from "./stream.js";
+import { DiffOpcode, type DiffBegin } from "@getpaseo/protocol/binary-frames/index";
+import type { DiffState } from "./index.js";
+const entry = {
+  path: "a.ts",
+  oldPath: null,
+  change: "modified" as const,
+  content: "text" as const,
+  additions: 1,
+  deletions: 0,
+  body: true,
+};
+const begin = (revision: number, baseRevision = 0): DiffBegin => ({
+  opcode: DiffOpcode.Begin,
+  slot: 0,
+  revision,
+  baseRevision,
+  baseSha: "a",
+  headSha: "b",
+  mergeBaseSha: "",
+  entries: [entry],
+});
+const end = (revision: number) =>
+  ({
+    opcode: DiffOpcode.End,
+    slot: 0,
+    revision,
+    diffTooLarge: false,
+    omitted: [],
+    error: null,
+  }) as const;
+test("publishes the initial tree, commits on End and buffers refreshes with stable equal files", () => {
+  const states: DiffState[] = [];
+  const stream = new DiffStream((state) => states.push(state));
+  stream.receive(begin(1));
+  expect(states[0]).toMatchObject({ phase: "loading", entries: [{ path: "a.ts" }] });
+  stream.receive({
+    opcode: DiffOpcode.Bodies,
+    slot: 0,
+    revision: 1,
+    files: [{ path: "a.ts", hunks: [] }],
+  });
+  expect(states).toHaveLength(1);
+  stream.receive(end(1));
+  const ready = states[1];
+  if (ready.phase !== "ready") throw new Error("Expected ready");
+  stream.receive(begin(2, 1));
+  stream.receive({
+    opcode: DiffOpcode.Bodies,
+    slot: 0,
+    revision: 2,
+    files: [{ path: "a.ts", hunks: [] }],
+  });
+  expect(states).toHaveLength(2);
+  stream.receive(end(2));
+  const refreshed = states[2];
+  if (refreshed.phase !== "ready") throw new Error("Expected ready");
+  expect(refreshed.files[0]).toBe(ready.files[0]);
+});
+test("superseded and stale frames cannot leave the client loading; End is authoritative", () => {
+  const states: DiffState[] = [];
+  const stream = new DiffStream((s) => states.push(s));
+  stream.receive(begin(1));
+  stream.receive(begin(2));
+  stream.receive(end(1));
+  expect(() => stream.receive(end(2))).toThrow(/Missing/);
+  stream.receive(begin(3));
+  stream.receive({ ...end(3), error: { code: "REF_NOT_FOUND", message: "gone" } });
+  expect(states.at(-1)).toMatchObject({ phase: "error", error: { code: "REF_NOT_FOUND" } });
+  stream.receive({ ...begin(4), entries: [] });
+  stream.receive(end(4));
+  expect(states.at(-1)).toMatchObject({ phase: "ready", files: [] });
+});
+test("unchanged entries retain hunks, omissions override cache, removed paths disappear", () => {
+  const states: DiffState[] = [];
+  const stream = new DiffStream((s) => states.push(s));
+  stream.receive(begin(1));
+  stream.receive({ ...end(1), omitted: ["a.ts"], diffTooLarge: true });
+  stream.receive({ ...begin(2, 1), entries: [{ ...entry, body: false }] });
+  stream.receive(end(2));
+  expect(states.at(-1)).toMatchObject({ phase: "ready", files: [{ status: "too_large" }] });
+  stream.receive({ ...begin(3, 2), entries: [] });
+  stream.receive(end(3));
+  expect(states.at(-1)).toMatchObject({ phase: "ready", files: [] });
+});

packages/client/src/diff/stream.ts

@@ -0,0 +1,94 @@
+import {
+  DiffOpcode,
+  materializeDiffFile,
+  type DiffBegin,
+  type DiffBody,
+  type DiffFrame,
+} from "@getpaseo/protocol/binary-frames/index";
+import type { ParsedDiffFile } from "@getpaseo/protocol/messages";
+import type { DiffState } from "./index.js";
+
+/** Owns revision assembly. End, rather than a local body count, commits the snapshot. */
+export class DiffStream {
+  private pending: { begin: DiffBegin; bodies: Map<string, DiffBody> } | null = null;
+  private committed: Extract<DiffState, { phase: "ready" }> | null = null;
+  constructor(private publish: (state: DiffState) => void) {}
+  receive(frame: DiffFrame): void {
+    if (frame.opcode === DiffOpcode.Begin) {
+      if (frame.baseRevision !== 0 && frame.baseRevision !== this.committed?.revision)
+        throw new Error("Unknown diff base revision");
+      if (new Set(frame.entries.map((entry) => entry.path)).size !== frame.entries.length)
+        throw new Error("Duplicate diff entry");
+      this.pending = { begin: frame, bodies: new Map() };
+      if (!this.committed)
+        this.publish({
+          phase: "loading",
+          revision: frame.revision,
+          entries: frame.entries.map(({ body: _body, ...entry }) => entry),
+        });
+      return;
+    }
+    const pending = this.pending;
+    if (!pending || frame.revision !== pending.begin.revision) return;
+    if (frame.opcode === DiffOpcode.Bodies) {
+      for (const body of frame.files) {
+        if (pending.bodies.has(body.path)) throw new Error("Duplicate diff body");
+        const entry = pending.begin.entries.find((candidate) => candidate.path === body.path);
+        if (!entry?.body) throw new Error("Unexpected diff body");
+        pending.bodies.set(body.path, body);
+      }
+      return;
+    }
+    this.pending = null;
+    if (frame.error) {
+      this.committed = null;
+      this.publish({ phase: "error", revision: frame.revision, error: frame.error });
+      return;
+    }
+    const previous = new Map(this.committed?.files.map((file) => [file.path, file]));
+    const omitted = new Set(frame.omitted);
+    for (const path of omitted)
+      if (!pending.begin.entries.some((entry) => entry.path === path))
+        throw new Error("Unknown omitted diff path");
+    const files = pending.begin.entries.map((entry) => {
+      const old = previous.get(entry.path);
+      const body = pending.bodies.get(entry.path);
+      if (entry.content === "text" && !omitted.has(entry.path) && entry.body && !body)
+        throw new Error(`Missing diff body: ${entry.path}`);
+      if (
+        entry.content === "text" &&
+        !entry.body &&
+        !omitted.has(entry.path) &&
+        (!old || pending.begin.baseRevision === 0)
+      )
+        throw new Error(`Missing inherited diff: ${entry.path}`);
+      let file = materializeDiffFile(entry, body ?? null, omitted.has(entry.path));
+      if (entry.content === "text" && !entry.body && !omitted.has(entry.path) && old)
+        file = { ...file, hunks: old.hunks, status: old.status };
+      return old && sameFile(old, file) ? old : file;
+    });
+    const { revision, baseSha, headSha, mergeBaseSha } = pending.begin;
+    this.committed = {
+      phase: "ready",
+      revision,
+      resolved: { baseSha, headSha, mergeBaseSha },
+      files,
+      diffTooLarge: frame.diffTooLarge,
+    };
+    this.publish(this.committed);
+  }
+}
+function sameFile(a: ParsedDiffFile, b: ParsedDiffFile): boolean {
+  if (
+    a.path !== b.path ||
+    a.oldPath !== b.oldPath ||
+    a.isNew !== b.isNew ||
+    a.isDeleted !== b.isDeleted ||
+    a.additions !== b.additions ||
+    a.deletions !== b.deletions ||
+    a.status !== b.status
+  )
+    return false;
+  if (a.hunks === b.hunks) return true;
+  return JSON.stringify(a.hunks) === JSON.stringify(b.hunks);
+}

packages/client/src/index.ts

@@ -1377,6 +1377,8 @@ function createGeneratedClientId(): string {
   return `paseo-sdk-${randomId}`;
 }
 
+export type { DiffCompare, DiffState, DiffEntry, ResolvedShas } from "./diff/index.js";
+
 interface WorkspaceLabelAssignment {
   client: DaemonClient;
   workspaceId: string;