fix(coding-agent): snapshot transfer ids from the materialized cursor; mismatches settle the transfer, not the worker channel

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.

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<DaemonOutbound, { type: "session_replaced" }>,
 	): 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<DaemonOutbound, { type: "session_replaced" }>,
-		snapshotId: string,
 		snapshotSignal: AbortSignal,
 	): Promise<void> {
 		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,16 @@ const ROSTER_SESSION_EVENT_TRIGGERS = new Set([
 	"thinking_level_changed",
 ]);
 
+/**
+ * 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.
+	const cursor = snapshot.lastEventCursor!;
+	return `${snapshot.activeSessionId}-${cursor.generation}-${cursor.sequence}`;
+}
+
 function hasDaemonOutboundActiveSessionId(
 	message: DaemonOutbound,
 ): message is DaemonOutbound & { activeSessionId: string } {

packages/coding-agent/src/modes/daemon/daemon-supervisor.ts

@@ -3705,6 +3705,33 @@ export class DaemonSupervisor {
 		}
 	}
 
+	/** 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,
+		snapshotId: string,
+		error: Error,
+		snapshotPurpose: Extract<DaemonWorkerFrameHeader, { kind: "outbound" }>["snapshotPurpose"],
+	): void {
+		const published = worker.transcriptCaches.get(activeSessionId)?.snapshotId === snapshotId;
+		this.failWorkerSnapshotCache(worker, activeSessionId, error, false, snapshotId);
+		// 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");
+		}
+	}
+
+	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 +5538,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 +5560,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 +5674,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 +5731,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 +5746,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 +5772,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,

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;

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<DaemonSocketClient>(),
 			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<string>(),
 		} 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<string, ActiveSessionState>;
 			createAttachResult(): DaemonAttachResult;
-			streamWorkerSnapshot: typeof streamWorkerSnapshot;
-			log: typeof log;
+			streamWorkerSnapshot(): Promise<void>;
+			log(message: string): void;
 			handleCommand(client: DaemonSocketClient, command: DaemonCommand): Promise<unknown>;
 		};
 		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<void>((resolve) => setImmediate(resolve));
 			await new Promise<void>((resolve) => setImmediate(resolve));
 		} finally {
@@ -232,6 +239,18 @@ describe("ENG-4602 snapshot transfer containment", () => {
 		expect(unhandled).not.toHaveBeenCalled();
 	});
 
+	it("derives the chunked transfer id from the materialized snapshot cursor", async () => {
+		// 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 attach()) 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 +358,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 +435,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<void>((resolve) => setImmediate(resolve));
+		await new Promise<void>((resolve) => setImmediate(resolve));
+		expect(request).toHaveBeenCalledWith(expect.objectContaining({ type: "attach" }));
 	});
 
 	it("holds catch-up behind duplicate validation and rejects it on mismatch", async () => {
@@ -476,11 +499,13 @@ describe("ENG-4602 snapshot transfer containment", () => {
 		);
 		await failedCatchup;
 
-		expect(request).not.toHaveBeenCalled();
+		// 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);
 		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,26 +628,26 @@ describe("ENG-4602 snapshot transfer containment", () => {
 		internals.handleWorkerFrame(reentrant.worker, frame(frames.begin));
 		internals.handleWorkerFrame(reentrant.worker, frame(frames.begin));
 		await new Promise<void>((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();
 		for (const message of [frames.begin, frames.chunk, frames.end, frames.begin, frames.chunk]) {
 			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();