From 07035427c7b29452e2124bce683bd08262d927f1 Mon Sep 17 00:00:00 2001 From: Sebastian Date: Fri, 4 Sep 2026 13:33:03 +0200 Subject: [PATCH 1/4] fix(coding-agent): derive snapshot ids from the materialized cursor and settle transfer mismatches transfer-scoped Worker side, the chunked-snapshot transfer id was computed from the live session cursor at a different time than the message-array capture it labels, so two transfers could carry the same id for different bytes whenever events arrived during snapshot materialization. Derive the id from the snapshot's own lastEventCursor at all three call sites (attach, replacement, catchup) so the id always names the captured cut. Supervisor side, a duplicate-transfer disagreement (reentrant begin, mismatched duplicate envelope, mismatched duplicate chunk bytes, or mismatched duplicate end metadata) was punished with closeWorkerChannel=true, bouncing a healthy worker and every session on it. Settle these anomalies with the transfer as the blast radius instead: fail the one transfer and queue attached clients for a resync from a fresh snapshot, mirroring the existing session_snapshot_failed handling, which now shares the same helper. Malformed-frame handling (undecodable payloads) still closes the channel. --- .../.changes/worker-snapshot-cursor.md | 1 + .../src/modes/daemon/daemon-mode.ts | 21 +++-- .../src/modes/daemon/daemon-supervisor.ts | 84 +++++++++++-------- ...4602-snapshot-transfer-idempotency.test.ts | 70 ++++++++++++++-- 4 files changed, 128 insertions(+), 48 deletions(-) create mode 100644 packages/coding-agent/.changes/worker-snapshot-cursor.md diff --git a/packages/coding-agent/.changes/worker-snapshot-cursor.md b/packages/coding-agent/.changes/worker-snapshot-cursor.md new file mode 100644 index 0000000000..4af2717659 --- /dev/null +++ b/packages/coding-agent/.changes/worker-snapshot-cursor.md @@ -0,0 +1 @@ +- Fixed chunked session-snapshot transfers so the transfer id names the exact materialized snapshot cut, and a mismatched or restarted transfer now fails only that transfer (clients resync) instead of bouncing the whole worker channel. diff --git a/packages/coding-agent/src/modes/daemon/daemon-mode.ts b/packages/coding-agent/src/modes/daemon/daemon-mode.ts index a052ee863f..54cf809c11 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-mode.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-mode.ts @@ -3956,7 +3956,7 @@ export class AgentDaemon { }); } if (streamsSnapshot) { - const snapshotId = `${state.activeSessionId}-${state.eventGeneration}-${state.lastEventSequence}`; + const snapshotId = snapshotTransferId(result.snapshot); let transcript: SnapshotTranscriptChunkSource; try { transcript = createSnapshotTranscriptChunks({ @@ -6674,10 +6674,9 @@ export class AgentDaemon { state: ActiveSessionState, message: Extract, ): void { - const snapshotId = `${state.activeSessionId}-${state.eventGeneration}-${state.lastEventSequence}`; // Mark before the registry read so later events queue behind this snapshot. const snapshotSignal = markClientSnapshotStreaming(client, state.activeSessionId); - void this.prepareReplacementSnapshot(client, state, message, snapshotId, snapshotSignal).catch((error) => { + void this.prepareReplacementSnapshot(client, state, message, snapshotSignal).catch((error) => { finishClientSnapshotStreaming(client, state.activeSessionId); this.log(`could not prepare replacement snapshot: ${String(error)}`); if (!client.socket.destroyed && this.sessions.get(state.activeSessionId) === state) { @@ -6695,13 +6694,13 @@ export class AgentDaemon { client: DaemonSocketClient, state: ActiveSessionState, message: Extract, - snapshotId: string, snapshotSignal: AbortSignal, ): Promise { const result = await this.createAttachResult(client, state, { type: "attach", activeSessionId: state.activeSessionId, }); + const snapshotId = snapshotTransferId(result.snapshot); if (this.sessions.get(state.activeSessionId) !== state) { finishClientSnapshotStreaming(client, state.activeSessionId); if (!client.snapshotStreaming && client.catchupActiveSessionIds?.size) { @@ -7103,7 +7102,7 @@ export class AgentDaemon { ), }); } - const snapshotId = `${activeSessionId}-${state.eventGeneration}-${state.lastEventSequence}`; + const snapshotId = snapshotTransferId(result.snapshot); const snapshotSignal = markClientSnapshotStreaming(client, activeSessionId); let transcript: SnapshotTranscriptChunkSource; try { @@ -7385,6 +7384,18 @@ const ROSTER_SESSION_EVENT_TRIGGERS = new Set([ "thinking_level_changed", ]); +/** + * A chunked-snapshot transfer id must name the cursor observed when the message + * array was materialized, not the live session cursor: events appended between + * materialization and id computation would let two different byte streams share + * one snapshot id, which the supervisor rejects as a mismatched transfer. + */ +function snapshotTransferId(snapshot: DaemonSessionSnapshot): string { + // createSessionSnapshot always sets lastEventCursor; it is optional only on the wire. + const cursor = snapshot.lastEventCursor!; + return `${snapshot.activeSessionId}-${cursor.generation}-${cursor.sequence}`; +} + function hasDaemonOutboundActiveSessionId( message: DaemonOutbound, ): message is DaemonOutbound & { activeSessionId: string } { diff --git a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts index 55ade4be1e..04f90bcb12 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts @@ -3705,6 +3705,36 @@ export class DaemonSupervisor { } } + /** + * Settle a snapshot transfer anomaly with the transfer itself as the blast + * radius: the worker channel (and every other session on it) stays up, and + * attached clients resync from a fresh snapshot instead of riding a worker + * reconnect. + */ + private failSnapshotTransfer( + worker: ResidentWorker, + activeSessionId: string, + snapshotId: string, + error: Error, + snapshotPurpose: Extract["snapshotPurpose"], + ): void { + const published = worker.transcriptCaches.get(activeSessionId)?.snapshotId === snapshotId; + this.failWorkerSnapshotCache(worker, activeSessionId, error, false, snapshotId); + if (published && (snapshotPurpose === "replacement" || snapshotPurpose === "catchup")) { + this.queueSnapshotResync(activeSessionId, snapshotPurpose); + } + } + + private queueSnapshotResync(activeSessionId: string, snapshotPurpose: "replacement" | "catchup"): void { + for (const client of this.clients) { + if (!client.attachedActiveSessionIds.has(activeSessionId)) continue; + this.queueCatchup(client, activeSessionId, snapshotPurpose === "replacement" ? "replacement" : "resync"); + void this.catchUpClient(client).catch((error) => + this.log(`Failed to catch up client ${client.id}: ${String(error)}`), + ); + } + } + private retireWorkerSnapshotCache( worker: ResidentWorker, activeSessionId: string, @@ -5511,12 +5541,12 @@ export class DaemonSupervisor { const generations = this.snapshotGenerationsFor(worker, activeSessionId); let generation = generations.get(begin.snapshotId); if (generation?.incoming) { - this.failWorkerSnapshotCache( + this.failSnapshotTransfer( worker, activeSessionId, - new Error(`Snapshot ${begin.snapshotId} restarted before completion`), - true, begin.snapshotId, + new Error(`Snapshot ${begin.snapshotId} restarted before completion`), + snapshotPurpose, ); return; } @@ -5533,12 +5563,12 @@ export class DaemonSupervisor { generation.result.snapshot.lastEventCursor?.generation === result.snapshot.lastEventCursor?.generation && generation.result.snapshot.lastEventCursor?.sequence === result.snapshot.lastEventCursor?.sequence; if (generation?.transcript.complete && !duplicate) { - this.failWorkerSnapshotCache( + this.failSnapshotTransfer( worker, activeSessionId, - new Error(`Snapshot ${begin.snapshotId} did not match the cached transfer`), - true, begin.snapshotId, + new Error(`Snapshot ${begin.snapshotId} did not match the cached transfer`), + snapshotPurpose, ); return; } @@ -5647,12 +5677,12 @@ export class DaemonSupervisor { generation.duplicateChunkIndex = duplicateIndex + 1; } } catch (error) { - this.failWorkerSnapshotCache( + this.failSnapshotTransfer( worker, activeSessionId, - error instanceof Error ? error : new Error(String(error)), - true, generation.transcript.snapshotId, + error instanceof Error ? error : new Error(String(error)), + snapshotPurpose, ); } } @@ -5704,12 +5734,12 @@ export class DaemonSupervisor { generation.duplicateChunkIndex = undefined; generation.duplicateResult = undefined; } catch (error) { - this.failWorkerSnapshotCache( + this.failSnapshotTransfer( worker, activeSessionId, - error instanceof Error ? error : new Error(String(error)), - true, transcript.snapshotId, + error instanceof Error ? error : new Error(String(error)), + snapshotPurpose, ); return; } @@ -5719,13 +5749,7 @@ export class DaemonSupervisor { transcript.dispose(); } if (published && (snapshotPurpose === "replacement" || snapshotPurpose === "catchup")) { - for (const client of this.clients) { - if (!client.attachedActiveSessionIds.has(activeSessionId)) continue; - this.queueCatchup(client, activeSessionId, snapshotPurpose === "replacement" ? "replacement" : "resync"); - void this.catchUpClient(client).catch((error) => - this.log(`Failed to catch up client ${client.id}: ${String(error)}`), - ); - } + this.queueSnapshotResync(activeSessionId, snapshotPurpose); } return; } @@ -5751,21 +5775,13 @@ export class DaemonSupervisor { if (!generation) { return; } - const published = worker.transcriptCaches.get(activeSessionId) === generation.transcript; - this.failWorkerSnapshotCache(worker, activeSessionId, new Error(failed.error), false, failed.snapshotId); - if (published && (snapshotPurpose === "replacement" || snapshotPurpose === "catchup")) { - for (const client of this.clients) { - if (!client.attachedActiveSessionIds.has(activeSessionId)) continue; - this.queueCatchup( - client, - activeSessionId, - snapshotPurpose === "replacement" ? "replacement" : "resync", - ); - void this.catchUpClient(client).catch((error) => - this.log(`Failed to catch up client ${client.id}: ${String(error)}`), - ); - } - } + this.failSnapshotTransfer( + worker, + activeSessionId, + failed.snapshotId, + new Error(failed.error), + snapshotPurpose, + ); } catch (error) { this.failWorkerSnapshotCache( worker, diff --git a/packages/coding-agent/test/suite/regressions/4602-snapshot-transfer-idempotency.test.ts b/packages/coding-agent/test/suite/regressions/4602-snapshot-transfer-idempotency.test.ts index 56b99ebe0b..c555b9c55f 100644 --- a/packages/coding-agent/test/suite/regressions/4602-snapshot-transfer-idempotency.test.ts +++ b/packages/coding-agent/test/suite/regressions/4602-snapshot-transfer-idempotency.test.ts @@ -232,6 +232,53 @@ describe("ENG-4602 snapshot transfer containment", () => { expect(unhandled).not.toHaveBeenCalled(); }); + it("derives the chunked transfer id from the materialized snapshot cursor", async () => { + const daemon = new AgentDaemon("/tmp/eng-4602-worker-cursor.sock", { + defaultSessionConfig: { agentDir: "/tmp", cwd: "/tmp" }, + createRuntime: async () => { + throw new Error("unexpected runtime creation"); + }, + }); + // The session cursor has advanced past the materialized snapshot cut. + const state = { + activeSessionId, + clients: new Set(), + eventGeneration: "generation-4602", + lastEventSequence: 5, + runtime: { metadata: { kind: "top-level", createdAt: 1 } }, + } as unknown as ActiveSessionState; + const socket = new PassThrough(); + const client = { + id: "supervisor", + socket: socket as unknown as Socket, + transport: "private-framed", + attachedActiveSessionIds: new Set(), + detachInput: () => {}, + supportsExtensionUi: false, + capabilities: new Set(), + } as DaemonSocketClient; + const streamWorkerSnapshot = vi.fn(async () => {}); + const internals = daemon as unknown as { + sessions: Map; + createAttachResult(): DaemonAttachResult; + streamWorkerSnapshot: typeof streamWorkerSnapshot; + handleCommand(client: DaemonSocketClient, command: DaemonCommand): Promise; + }; + internals.sessions.set(activeSessionId, state); + internals.createAttachResult = () => streamedResult([]); + internals.streamWorkerSnapshot = streamWorkerSnapshot; + try { + const response = (await internals.handleCommand(client, { + type: "attach", + activeSessionId, + capabilities: ["attach_snapshot", "event_sequence", "slim_attach", "chunked_snapshot"], + })) as { data?: DaemonAttachResult }; + expect(response.data?.snapshotStream?.id).toBe(`${activeSessionId}-generation-4602-1`); + } finally { + socket.destroy(); + } + }); + it("fails one worker snapshot without dropping another session on the supervisor channel", async () => { const daemon = new AgentDaemon("/tmp/eng-4602-stream.sock", { defaultSessionConfig: { agentDir: "/tmp", cwd: "/tmp" }, @@ -339,7 +386,7 @@ describe("ENG-4602 snapshot transfer containment", () => { defaultSessionConfig: { agentDir: "/tmp", cwd: "/tmp" }, descriptorDir: "/tmp/eng-4602-supervisor-state", }); - const { close, worker } = workerHarness(); + const { close, request, worker } = workerHarness(); const client = socketClient("public", new PassThrough()); const streamSnapshot = vi.fn(async () => {}); const internals = supervisor as unknown as { @@ -416,7 +463,11 @@ describe("ENG-4602 snapshot transfer containment", () => { expect(worker.transcriptCaches.has(activeSessionId)).toBe(false); expect(worker.snapshotCache.has(activeSessionId)).toBe(false); expect(streamSnapshot).not.toHaveBeenCalled(); - expect(close).toHaveBeenCalledOnce(); + expect(close).not.toHaveBeenCalled(); + expect(worker.descriptor.lifecycle).toBe("ready"); + await new Promise((resolve) => setImmediate(resolve)); + await new Promise((resolve) => setImmediate(resolve)); + expect(request).toHaveBeenCalledWith(expect.objectContaining({ type: "attach" })); }); it("holds catch-up behind duplicate validation and rejects it on mismatch", async () => { @@ -480,7 +531,8 @@ describe("ENG-4602 snapshot transfer containment", () => { expect(streamSnapshot).not.toHaveBeenCalled(); expect(worker.snapshotCache.has(activeSessionId)).toBe(false); expect(worker.transcriptCaches.has(activeSessionId)).toBe(false); - expect(close).toHaveBeenCalledOnce(); + expect(close).not.toHaveBeenCalled(); + expect(worker.descriptor.lifecycle).toBe("ready"); }); it("rejects a quarantined catch-up before intentional worker stop", async () => { @@ -603,18 +655,18 @@ describe("ENG-4602 snapshot transfer containment", () => { internals.handleWorkerFrame(reentrant.worker, frame(frames.begin)); internals.handleWorkerFrame(reentrant.worker, frame(frames.begin)); await new Promise((resolve) => setImmediate(resolve)); - expect(reentrant.close).toHaveBeenCalledOnce(); + expect(reentrant.close).not.toHaveBeenCalled(); expect(reentrant.worker.transcriptCaches.has(activeSessionId)).toBe(false); - expect(reentrant.worker.client).toBeUndefined(); - expect(reentrant.worker.descriptor.lifecycle).toBe("recovering"); - expect(recoverWorker).toHaveBeenCalledWith(reentrant.worker); + expect(reentrant.worker.client).toBeDefined(); + expect(reentrant.worker.descriptor.lifecycle).toBe("ready"); + expect(recoverWorker).not.toHaveBeenCalled(); const completed = workerHarness(); for (const message of [frames.begin, frames.chunk, frames.end]) { internals.handleWorkerFrame(completed.worker, frame(message)); } internals.handleWorkerFrame(completed.worker, frame({ ...frames.begin, messageCount: 2 })); - expect(completed.close).toHaveBeenCalledOnce(); + expect(completed.close).not.toHaveBeenCalled(); expect(completed.worker.transcriptCaches.has(activeSessionId)).toBe(false); const mismatchedEnd = workerHarness(); @@ -622,7 +674,7 @@ describe("ENG-4602 snapshot transfer containment", () => { internals.handleWorkerFrame(mismatchedEnd.worker, frame(message)); } internals.handleWorkerFrame(mismatchedEnd.worker, frame({ ...frames.end, lastEventSequence: 2 })); - expect(mismatchedEnd.close).toHaveBeenCalledOnce(); + expect(mismatchedEnd.close).not.toHaveBeenCalled(); expect(mismatchedEnd.worker.transcriptCaches.has(activeSessionId)).toBe(false); const replaced = workerHarness(); From 43cc51ecd582d1cf224ea18f42c7ef293a841a38 Mon Sep 17 00:00:00 2001 From: Sebastian Date: Fri, 4 Sep 2026 13:57:31 +0200 Subject: [PATCH 2/4] test(coding-agent): give stubbed attach snapshots the materialization cursor --- packages/coding-agent/test/daemon-mode.test.ts | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/packages/coding-agent/test/daemon-mode.test.ts b/packages/coding-agent/test/daemon-mode.test.ts index 0638f19895..aeb2f1610f 100644 --- a/packages/coding-agent/test/daemon-mode.test.ts +++ b/packages/coding-agent/test/daemon-mode.test.ts @@ -3650,7 +3650,14 @@ describe("daemon mode helpers", () => { client.transport = "private-framed"; const result = { activeSessionId: state.activeSessionId, - snapshot: { summary: {}, state: {}, messages: [] }, + snapshot: { + activeSessionId: state.activeSessionId, + summary: {}, + state: {}, + messages: [], + lastEventSequence: 0, + lastEventCursor: { generation: state.eventGeneration, sequence: 0 }, + }, lastEventSequence: 0, } as unknown as DaemonAttachResult; const streamWorkerSnapshot = vi.fn(async () => undefined); @@ -3717,9 +3724,12 @@ describe("daemon mode helpers", () => { const result = { activeSessionId: state.activeSessionId, snapshot: { + activeSessionId: state.activeSessionId, summary: {}, state: {}, messages: [{ role: "user", content: "x".repeat(4 * 1024 * 1024 + 1), timestamp: 0 }], + lastEventSequence: 0, + lastEventCursor: { generation: state.eventGeneration, sequence: 0 }, }, lastEventSequence: 0, } as unknown as DaemonAttachResult; From b92fa5fc2b4f063c1c7285c6b32d03d08d6d3ee7 Mon Sep 17 00:00:00 2001 From: Sebastian Date: Fri, 4 Sep 2026 16:10:58 +0200 Subject: [PATCH 3/4] fix(coding-agent): requeue snapshot resyncs from the published-cache drop, not the failing frame's purpose MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A failed published transfer can be serving any attached client's catch-up wait: drainClientCatchups clears its queue entry before draining, so when duplicate validation rejects the waiter it was logged and dropped, and an attach-purpose failure never requeued it — the client stayed attached with a stale transcript while the worker stayed ready. Queue the resync whenever the published cache is dropped; the frame purpose only picks the replacement/resync flavor. --- .../coding-agent/src/modes/daemon/daemon-supervisor.ts | 7 +++++-- .../regressions/4602-snapshot-transfer-idempotency.test.ts | 4 +++- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts index 04f90bcb12..f4a7286913 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts @@ -3720,8 +3720,11 @@ export class DaemonSupervisor { ): void { const published = worker.transcriptCaches.get(activeSessionId)?.snapshotId === snapshotId; this.failWorkerSnapshotCache(worker, activeSessionId, error, false, snapshotId); - if (published && (snapshotPurpose === "replacement" || snapshotPurpose === "catchup")) { - this.queueSnapshotResync(activeSessionId, snapshotPurpose); + // The resync is driven by the published-cache drop, not the failing frame's + // purpose: a published transfer can be serving any attached client's + // catch-up wait, whose queue entry drainClientCatchups already cleared. + if (published) { + this.queueSnapshotResync(activeSessionId, snapshotPurpose === "replacement" ? "replacement" : "catchup"); } } diff --git a/packages/coding-agent/test/suite/regressions/4602-snapshot-transfer-idempotency.test.ts b/packages/coding-agent/test/suite/regressions/4602-snapshot-transfer-idempotency.test.ts index c555b9c55f..1f7aedd20f 100644 --- a/packages/coding-agent/test/suite/regressions/4602-snapshot-transfer-idempotency.test.ts +++ b/packages/coding-agent/test/suite/regressions/4602-snapshot-transfer-idempotency.test.ts @@ -527,7 +527,9 @@ describe("ENG-4602 snapshot transfer containment", () => { ); await failedCatchup; - expect(request).not.toHaveBeenCalled(); + // The rejected catch-up waiter is requeued by the published-cache drop and + // retries with a fresh worker snapshot request instead of staying stale. + expect(request).toHaveBeenCalledWith(expect.objectContaining({ type: "attach", activeSessionId })); expect(streamSnapshot).not.toHaveBeenCalled(); expect(worker.snapshotCache.has(activeSessionId)).toBe(false); expect(worker.transcriptCaches.has(activeSessionId)).toBe(false); From 5af3bbe8090240507d6b0297d0da007159e15305 Mon Sep 17 00:00:00 2001 From: Sebastian Date: Fri, 4 Sep 2026 18:54:30 +0200 Subject: [PATCH 4/4] refactor(coding-agent): share the worker-attach harness and tighten comments One workerAttachHarness builder serves both worker-side attach pins, and multi-line comment blocks collapse toward one-line invariant guards. No behavior or coverage change. --- .../src/modes/daemon/daemon-mode.ts | 6 +- .../src/modes/daemon/daemon-supervisor.ts | 12 +-- ...4602-snapshot-transfer-idempotency.test.ts | 83 ++++++------------- 3 files changed, 32 insertions(+), 69 deletions(-) diff --git a/packages/coding-agent/src/modes/daemon/daemon-mode.ts b/packages/coding-agent/src/modes/daemon/daemon-mode.ts index 54cf809c11..7bce8852cf 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-mode.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-mode.ts @@ -7385,10 +7385,8 @@ const ROSTER_SESSION_EVENT_TRIGGERS = new Set([ ]); /** - * A chunked-snapshot transfer id must name the cursor observed when the message - * array was materialized, not the live session cursor: events appended between - * materialization and id computation would let two different byte streams share - * one snapshot id, which the supervisor rejects as a mismatched transfer. + * The transfer id must name the cursor observed at materialization, not the live session cursor: + * events appended in between would let two different byte streams share one snapshot id. */ function snapshotTransferId(snapshot: DaemonSessionSnapshot): string { // createSessionSnapshot always sets lastEventCursor; it is optional only on the wire. diff --git a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts index f4a7286913..8cf94b7ee4 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts @@ -3705,12 +3705,7 @@ export class DaemonSupervisor { } } - /** - * Settle a snapshot transfer anomaly with the transfer itself as the blast - * radius: the worker channel (and every other session on it) stays up, and - * attached clients resync from a fresh snapshot instead of riding a worker - * reconnect. - */ + /** Settle a transfer anomaly with the transfer as the blast radius: the worker channel stays up and clients resync fresh. */ private failSnapshotTransfer( worker: ResidentWorker, activeSessionId: string, @@ -3720,9 +3715,8 @@ export class DaemonSupervisor { ): void { const published = worker.transcriptCaches.get(activeSessionId)?.snapshotId === snapshotId; this.failWorkerSnapshotCache(worker, activeSessionId, error, false, snapshotId); - // The resync is driven by the published-cache drop, not the failing frame's - // purpose: a published transfer can be serving any attached client's - // catch-up wait, whose queue entry drainClientCatchups already cleared. + // The published-cache drop drives the resync, not the frame's purpose: a published transfer + // can be serving any client's catch-up wait, whose queue entry drainClientCatchups already cleared. if (published) { this.queueSnapshotResync(activeSessionId, snapshotPurpose === "replacement" ? "replacement" : "catchup"); } diff --git a/packages/coding-agent/test/suite/regressions/4602-snapshot-transfer-idempotency.test.ts b/packages/coding-agent/test/suite/regressions/4602-snapshot-transfer-idempotency.test.ts index 1f7aedd20f..4a5f91a3eb 100644 --- a/packages/coding-agent/test/suite/regressions/4602-snapshot-transfer-idempotency.test.ts +++ b/packages/coding-agent/test/suite/regressions/4602-snapshot-transfer-idempotency.test.ts @@ -172,8 +172,8 @@ function snapshotFrames(messages: AgentMessage[]) { } describe("ENG-4602 snapshot transfer containment", () => { - it("observes the deferred attach snapshot promise", async () => { - const daemon = new AgentDaemon("/tmp/eng-4602-worker.sock", { + function workerAttachHarness(socketPath: string, lastEventSequence: number) { + const daemon = new AgentDaemon(socketPath, { defaultSessionConfig: { agentDir: "/tmp", cwd: "/tmp" }, createRuntime: async () => { throw new Error("unexpected runtime creation"); @@ -183,7 +183,7 @@ describe("ENG-4602 snapshot transfer containment", () => { activeSessionId, clients: new Set(), eventGeneration: "generation-4602", - lastEventSequence: 1, + lastEventSequence, runtime: { metadata: { kind: "top-level", createdAt: 1 } }, } as unknown as ActiveSessionState; const socket = new PassThrough(); @@ -196,30 +196,37 @@ describe("ENG-4602 snapshot transfer containment", () => { supportsExtensionUi: false, capabilities: new Set(), } as DaemonSocketClient; - const streamError = new Error("encoder failed after begin"); - const log = vi.fn(); - const streamWorkerSnapshot = vi.fn(async () => { - throw streamError; - }); const internals = daemon as unknown as { sessions: Map; createAttachResult(): DaemonAttachResult; - streamWorkerSnapshot: typeof streamWorkerSnapshot; - log: typeof log; + streamWorkerSnapshot(): Promise; + log(message: string): void; handleCommand(client: DaemonSocketClient, command: DaemonCommand): Promise; }; internals.sessions.set(activeSessionId, state); internals.createAttachResult = () => streamedResult([]); + const attach = () => + internals.handleCommand(client, { + type: "attach", + activeSessionId, + capabilities: ["attach_snapshot", "event_sequence", "slim_attach", "chunked_snapshot"], + }); + return { internals, client, socket, attach }; + } + + it("observes the deferred attach snapshot promise", async () => { + const { internals, socket, attach } = workerAttachHarness("/tmp/eng-4602-worker.sock", 1); + const streamError = new Error("encoder failed after begin"); + const log = vi.fn(); + const streamWorkerSnapshot = vi.fn(async () => { + throw streamError; + }); internals.streamWorkerSnapshot = streamWorkerSnapshot; internals.log = log; const unhandled = vi.fn(); process.on("unhandledRejection", unhandled); try { - await internals.handleCommand(client, { - type: "attach", - activeSessionId, - capabilities: ["attach_snapshot", "event_sequence", "slim_attach", "chunked_snapshot"], - }); + await attach(); await new Promise((resolve) => setImmediate(resolve)); await new Promise((resolve) => setImmediate(resolve)); } finally { @@ -233,46 +240,11 @@ describe("ENG-4602 snapshot transfer containment", () => { }); it("derives the chunked transfer id from the materialized snapshot cursor", async () => { - const daemon = new AgentDaemon("/tmp/eng-4602-worker-cursor.sock", { - defaultSessionConfig: { agentDir: "/tmp", cwd: "/tmp" }, - createRuntime: async () => { - throw new Error("unexpected runtime creation"); - }, - }); - // The session cursor has advanced past the materialized snapshot cut. - const state = { - activeSessionId, - clients: new Set(), - eventGeneration: "generation-4602", - lastEventSequence: 5, - runtime: { metadata: { kind: "top-level", createdAt: 1 } }, - } as unknown as ActiveSessionState; - const socket = new PassThrough(); - const client = { - id: "supervisor", - socket: socket as unknown as Socket, - transport: "private-framed", - attachedActiveSessionIds: new Set(), - detachInput: () => {}, - supportsExtensionUi: false, - capabilities: new Set(), - } as DaemonSocketClient; - const streamWorkerSnapshot = vi.fn(async () => {}); - const internals = daemon as unknown as { - sessions: Map; - createAttachResult(): DaemonAttachResult; - streamWorkerSnapshot: typeof streamWorkerSnapshot; - handleCommand(client: DaemonSocketClient, command: DaemonCommand): Promise; - }; - internals.sessions.set(activeSessionId, state); - internals.createAttachResult = () => streamedResult([]); - internals.streamWorkerSnapshot = streamWorkerSnapshot; + // The live session cursor (sequence 5) has advanced past the materialized snapshot cut (sequence 1). + const { internals, socket, attach } = workerAttachHarness("/tmp/eng-4602-worker-cursor.sock", 5); + internals.streamWorkerSnapshot = vi.fn(async () => {}); try { - const response = (await internals.handleCommand(client, { - type: "attach", - activeSessionId, - capabilities: ["attach_snapshot", "event_sequence", "slim_attach", "chunked_snapshot"], - })) as { data?: DaemonAttachResult }; + const response = (await attach()) as { data?: DaemonAttachResult }; expect(response.data?.snapshotStream?.id).toBe(`${activeSessionId}-generation-4602-1`); } finally { socket.destroy(); @@ -527,8 +499,7 @@ describe("ENG-4602 snapshot transfer containment", () => { ); await failedCatchup; - // The rejected catch-up waiter is requeued by the published-cache drop and - // retries with a fresh worker snapshot request instead of staying stale. + // The published-cache drop requeued the rejected waiter: it retries with a fresh snapshot request. expect(request).toHaveBeenCalledWith(expect.objectContaining({ type: "attach", activeSessionId })); expect(streamSnapshot).not.toHaveBeenCalled(); expect(worker.snapshotCache.has(activeSessionId)).toBe(false);