feat(client): observe streamed diffs with atomic revision updates

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.2",

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.ts

@@ -1,3 +1,9 @@
+import {
+  DiffStreamRouter,
+  DiffSubscriptionError,
+  type DiffCompare,
+  type DiffState,
+} from "./diff/index.js";
 import {
   ChunkedTransport,
   ChunkedTransportProtocolError,
@@ -1411,6 +1417,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>();
@@ -1760,6 +1780,7 @@ 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"));
@@ -2663,7 +2684,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>>(
@@ -2689,12 +2715,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;
         }
       },
@@ -4660,6 +4690,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 },
@@ -6936,6 +6986,7 @@ export class DaemonClient {
     this.chunkedTransport?.close();
     this.chunkedTransport = null;
     this.owned.disconnected();
+    this.diffStreams.disconnected();
     this.providerSnapshotUpdates.clear();
     this.stopLivenessHeartbeat();
     this.cleanupTransport();
@@ -7126,6 +7177,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;
@@ -7326,6 +7381,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
@@ -7485,6 +7541,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,
@@ -7761,3 +7818,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,256 @@
+import {
+  decodeDiffFrame,
+  type DiffEntry as WireEntry,
+} from "@getpaseo/protocol/binary-frames/index";
+import type {
+  DiffEndpoint,
+  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 interface DiffCompare {
+  cwd: string;
+  base: DiffEndpoint;
+  head: DiffEndpoint;
+  mergeBase?: boolean;
+  ignoreWhitespace?: boolean;
+  highlight?: boolean;
+}
+export type DiffEntry = Omit<WireEntry, "body">;
+export interface ResolvedShas {
+  baseSha: string;
+  headSha: string;
+  mergeBaseSha: string;
+}
+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 };
+interface LegacyCompare {
+  mode: "uncommitted" | "base" | "all";
+  baseRef?: string;
+  ignoreWhitespace?: boolean;
+}
+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

@@ -1212,3 +1212,5 @@ function createGeneratedClientId(): string {
       : Math.random().toString(36).slice(2);
   return `paseo-sdk-${randomId}`;
 }
+
+export type { DiffCompare, DiffState, DiffEntry, ResolvedShas } from "./diff/index.js";

packages/server/src/server/daemon-e2e/checkout-diff-content.e2e.test.ts

@@ -416,3 +416,73 @@ test("streams whole-file batches and answers a same-socket ping before the revis
     rmSync(cwd, { recursive: true, force: true });
   }
 }, 60_000);
+
+test("SDK diff journey publishes a tag tree, commits a one-file refresh and shares two sockets", async () => {
+  const ctx = await createDaemonTestContext();
+  const cwd = mkdtempSync(join(tmpdir(), "sdk-stream-diff-"));
+  const { DaemonClient } = await import("../test-utils/daemon-client.js");
+  const { startGitCommandMetrics, stopGitCommandMetrics } =
+    await import("../../utils/run-git-command.js");
+  const second = new DaemonClient({ url: `ws://127.0.0.1:${ctx.daemon.port}/ws` });
+  try {
+    git(cwd, "init", "-b", "main");
+    git(cwd, "config", "user.name", "Test");
+    git(cwd, "config", "user.email", "test@example.com");
+    writeFileSync(join(cwd, "a.ts"), "export const a = 1;\n");
+    writeFileSync(join(cwd, "b.ts"), "export const b = 1;\n");
+    git(cwd, "add", ".");
+    git(cwd, "-c", "commit.gpgsign=false", "commit", "-m", "base");
+    git(cwd, "tag", "v1");
+    writeFileSync(join(cwd, "a.ts"), "export const a = 2;\n");
+    writeFileSync(join(cwd, "b.ts"), "export const b = 2;\n");
+    const compare = {
+      cwd,
+      base: { kind: "rev", rev: "refs/tags/v1" },
+      head: { kind: "worktree" },
+      mergeBase: true,
+    } as const;
+    const states: import("@getpaseo/client").DiffState[] = [];
+    const observation = ctx.client.observeDiff(compare);
+    observation.subscribe({ snapshot: (state) => states.push(state), update: () => {} });
+    await expect.poll(() => states.at(-1)?.phase, { timeout: 15000 }).toBe("ready");
+    expect(states[0]).toMatchObject({
+      phase: "loading",
+      entries: [{ path: "a.ts" }, { path: "b.ts" }],
+    });
+    const initial = states.at(-1);
+    if (initial?.phase !== "ready") throw new Error("Expected ready");
+    await second.connect();
+    startGitCommandMetrics();
+    const duplicateStates: import("@getpaseo/client").DiffState[] = [];
+    const duplicate = second.observeDiff(compare);
+    duplicate.subscribe({ snapshot: (state) => duplicateStates.push(state), update: () => {} });
+    await expect.poll(() => duplicateStates.at(-1)?.phase).toBe("ready");
+    const metrics = stopGitCommandMetrics();
+    expect(duplicateStates.at(-1)).toMatchObject({
+      phase: "ready",
+      files: [{ path: "a.ts" }, { path: "b.ts" }],
+    });
+    expect(metrics.commands.filter((command) => command.args.includes("--raw"))).toHaveLength(0);
+    startGitCommandMetrics();
+    writeFileSync(join(cwd, "a.ts"), "export const a = 3;\n");
+    await expect
+      .poll(() => states.at(-1)?.revision, { timeout: 15000 })
+      .toBeGreaterThan(initial.revision);
+    await expect.poll(() => duplicateStates.at(-1)?.revision).toBeGreaterThan(initial.revision);
+    const refreshMetrics = stopGitCommandMetrics();
+    expect(
+      refreshMetrics.commands.filter((command) => command.args.includes("--raw")),
+    ).toHaveLength(1);
+    const refreshed = states.at(-1);
+    if (refreshed?.phase !== "ready") throw new Error("Expected ready");
+    expect(refreshed.files[0]).not.toBe(initial.files[0]);
+    expect(refreshed.files[1]).toBe(initial.files[1]);
+    expect(states.filter((state) => state.phase === "loading")).toHaveLength(1);
+    await duplicate.release();
+    await observation.release();
+  } finally {
+    await second.close();
+    await ctx.cleanup();
+    rmSync(cwd, { recursive: true, force: true });
+  }
+}, 45000);