From 236270768db1e85703e66d7ff8c48431fb00bb27 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 6 Nov 2026 05:56:31 +0100 Subject: [PATCH] feat(client): observe streamed diffs with atomic revision updates --- packages/client/package.json | 3 +- .../scripts/measure-diff-materialization.ts | 38 +++ packages/client/src/daemon-client.ts | 63 ++++- packages/client/src/diff/client.test.ts | 210 ++++++++++++++ packages/client/src/diff/index.ts | 256 ++++++++++++++++++ packages/client/src/diff/stream.test.ts | 86 ++++++ packages/client/src/diff/stream.ts | 94 +++++++ packages/client/src/index.ts | 2 + .../checkout-diff-content.e2e.test.ts | 70 +++++ 9 files changed, 819 insertions(+), 3 deletions(-) create mode 100644 packages/client/scripts/measure-diff-materialization.ts create mode 100644 packages/client/src/diff/client.test.ts create mode 100644 packages/client/src/diff/index.ts create mode 100644 packages/client/src/diff/stream.test.ts create mode 100644 packages/client/src/diff/stream.ts diff --git a/packages/client/package.json b/packages/client/package.json index 62e5b8c664..7059ce5b9d 100644 --- a/packages/client/package.json +++ b/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", diff --git a/packages/client/scripts/measure-diff-materialization.ts b/packages/client/scripts/measure-diff-materialization.ts new file mode 100644 index 0000000000..54f7b1f53a --- /dev/null +++ b/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; diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index b29a34111c..f03295ef97 100644 --- a/packages/client/src/daemon-client.ts +++ b/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(); private activeBinaryFileTransfers = new Map(); @@ -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( responseType: T, message: { type: SessionInboundMessage["type"] } & Record, - options?: { requestId?: string; timeout?: number; signal?: AbortSignal }, + options?: { + requestId?: string; + timeout?: number; + signal?: AbortSignal; + validateSnapshot?: (payload: CorrelatedResponsePayload) => void; + }, ): OwnedSubscription> { let resetSource = false; return this.owned.observeRequest>( @@ -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 { + return this.diffStreams.observe(compare); + } + + async readDiff(compare: DiffCompare): Promise { + const observation = this.observeDiff(compare); + try { + return await new Promise((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"; diff --git a/packages/client/src/diff/client.test.ts b/packages/client/src/diff/client.test.ts new file mode 100644 index 0000000000..04f06d227f --- /dev/null +++ b/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[] = []; + 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, 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(); + } + }, +); diff --git a/packages/client/src/diff/index.ts b/packages/client/src/diff/index.ts new file mode 100644 index 0000000000..b1945f325a --- /dev/null +++ b/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; +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; + legacy( + cwd: string, + compare: LegacyCompare, + ): OwnedSubscription; + 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; +} +/** Capability selection, slot ownership and reconnect belong to this domain. */ +export class DiffStreamRouter { + private registrations = new Set(); + private slots = new Map(); + 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 { + 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 | null): void { + void backend?.release().catch(this.port.failed); + } + observe(compare: DiffCompare): OwnedSubscription { + let backend: OwnedSubscription | 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>(); + let resolve!: (state: DiffState & { subscriptionId: string }) => void; + const ready = new Promise((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; +} diff --git a/packages/client/src/diff/stream.test.ts b/packages/client/src/diff/stream.test.ts new file mode 100644 index 0000000000..b740469029 --- /dev/null +++ b/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: [] }); +}); diff --git a/packages/client/src/diff/stream.ts b/packages/client/src/diff/stream.ts new file mode 100644 index 0000000000..ac55f681ec --- /dev/null +++ b/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 } | null = null; + private committed: Extract | 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); +} diff --git a/packages/client/src/index.ts b/packages/client/src/index.ts index ead91e719c..ecd099e7da 100644 --- a/packages/client/src/index.ts +++ b/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"; diff --git a/packages/server/src/server/daemon-e2e/checkout-diff-content.e2e.test.ts b/packages/server/src/server/daemon-e2e/checkout-diff-content.e2e.test.ts index c1be3b873c..4fa64312d7 100644 --- a/packages/server/src/server/daemon-e2e/checkout-diff-content.e2e.test.ts +++ b/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);