fix(coding-agent): isolate snapshot catch-up failures

.pylon/features.yaml

@@ -194,17 +194,19 @@ decisions:
     pylon_refs:
       - https://github.com/pylon-code/prime-agent/issues/11
       - https://github.com/pylon-code/prime-agent/pull/14
+      - https://github.com/pylon-code/prime-agent/issues/21
+      - https://github.com/pylon-code/prime-agent/pull/19
     upstream_refs:
       - https://github.com/PrimeIntellect-ai/prime-agent/issues/1229
       - https://github.com/PrimeIntellect-ai/prime-agent/pull/1845
       - https://github.com/PrimeIntellect-ai/prime-agent/pull/1847
       - https://github.com/PrimeIntellect-ai/prime-agent/pull/1756
       - https://github.com/PrimeIntellect-ai/prime-agent/pull/1864
-    fork_change: immutable-snapshot-transfer-and-isolated-retry
-    upstream_support: Prime now provides one RLM child projection, tolerant legacy topology recovery, recovery-aware session reuse, and ownership checks for in-flight opens, but still reuses the event cursor as snapshot transfer identity, serializes selected message objects lazily, and closes a healthy worker on an isolated duplicate-transfer mismatch.
+    fork_change: immutable-snapshot-transfer-isolated-retry-and-fresh-generation-v1
+    upstream_support: Prime through c382f09856d4 still lacks a capability-gated fresh snapshot-generation identity, attachment-local retry containment, and bounded large-transcript framing that preserve a healthy resident worker across mixed-version catch-up failures.
     revisit_when:
-      - Prime gives each transfer immutable bytes and a unique identity while preserving the event cursor as ordering metadata.
-      - Prime retries an isolated bad snapshot generation without closing the resident worker control channel.
+      - Prime gives every mixed-version transfer immutable bytes and a unique generation identity while preserving the event cursor as ordering metadata.
+      - Prime isolates one malformed or changing snapshot generation without recycling the healthy worker and bounds preparation, framing, drain waits, and spill ownership.
 
   authoritative-owned-session-cleanup:
     area: runtime-reliability
@@ -247,12 +249,14 @@ decisions:
 
   negotiated-daemon-session-capabilities:
     area: runtime-reliability
-    state: candidate
+    state: shipped
     owner: shared
     decision: retain
     pylon_refs:
       - https://github.com/pylon-code/prime-agent/issues/17
+      - https://github.com/pylon-code/prime-agent/pull/18
       - https://github.com/pylon-code/pylon/issues/190
+      - https://github.com/pylon-code/pylon/pull/191
       - https://github.com/rynfar/comet/issues/7
     upstream_refs:
       - https://github.com/PrimeIntellect-ai/prime-agent/tree/a903d4b6768f484bd6d459b7b0aa7dee38e461e2

.pylon/upstream-review.md

@@ -107,3 +107,14 @@ This ledger records Prime upstream evidence and the decision taken for each over
 - Repair validation: focused daemon connection/client tests pass 219/219.
 - Hosted exact-head CI on `c27e1b6ea329aabe19ec70f768bcbf37c79a6bef` exposed a deterministic `ResumeDaemonClient` fixture incompatibility in shard 1/3: the cast fake lacked `getTransportGeneration()` and returned a stale fixed attachment identity/capability echo. The test-only repair gives the fake one stable transport generation and echoes each request's exact client ID/capabilities; the exact ENG-4656 file passes 3/3 and local shard 1/3 passes 1,466/1,466. They cover exact post-commit proof; same-stack queued and active B/C attach admission-token races; a superseded active B/C reattach with commit-time proof observation; synchronous admission fencing of every outstanding snapshot assembly and headerless old-route begin/end/failed frames; current-admission revision binding for the only legitimate request-attempt exception; pre-begin failure admission only for an existing assembly, exact active request, or current route-local replacement fence; retired same-route begin and failure tombstones across duplicate terminal frames, later request attempts, and post-commit replay; permanent fixed fail-close instead of tombstone eviction on the 129th distinct ignored ID while active B and queued C remain uncommitted; exact current-request same-ID begin reuse without allowing a pre-descriptor runtime commit; terminal and update close retirement of ordinary begin/chunk/end tails before close or authoritative restoration publication; full tombstone-budget message-first and socket-first update, killed/shutdown, and explicit owner-close classification before pending-fence or dead-transport retirement, with idempotent duplicate closes, first-owner contradictory update/terminal ordering, exactly one authoritative outcome, and no per-snapshot recovery attach; direct replacement/resync suppression throughout newer cross-route inline/streamed admission, terminal/update ownership, and disposal await windows, with immediate fresh-route close delivery and no stale post-close publication; prompt release after a newer admission retires a stalled streamed attach; cumulative and individual pre-proof weight overflow below the count cap; successful release/discard accounting resets; payload-free fail-close; peer-only unshaped replacement commit; full stale same-connection route discard after B reattach; bounded chunked-replacement queue overflow; header-only A-fence retirement before B reattach and physical reconnect; matching session-close bypass/retirement; delayed retired-A begin/end and pre-begin failed-frame discard during gated B admission; exact current-attempt binding for legitimate failure-before-begin; stale queued-frame discard and fresh B-route event delivery; permanent fail-closed retry rejection; transport-reset fences; target-reattach invalidation; shared-client serialization and global overwrite invalidation; terminal direct-close generation, active/socketless notification, normal/update in-flight recovery suppression, duplicate-event prevention, and payload-free public close; disposal re-entry; unproved owned cleanup reattach; valid same-generation resync; offer-without-echo resync and frame suppression; fixed malformed/wrong-client errors; duplicate/unrequested/unknown/non-array/non-string proof rejection; replacement/socket-close clearing; and frozen public-root export. The renewed root build, `npm run check`, `git diff --check`, six isolated process files (79/79), and clean real supervisor process suite pass; the latter records 13 passes with 8 fixture-gated skips. With the installed pinned stock 0.8.1 CLI (`package.json` version 0.8.1; launcher SHA-256 `c5dffcd16a401551986023134fbc05457775e1eeed618a5a6ae0d0cee1772d5f`), both runtime adoption directions pass against the rebuilt current CLI. Namespace probes confirm stock 0.8.1 has neither token nor accessor while the rebuilt candidate exports both. Exact committed-head reviews remain mandatory.
 - Cross-repository merge order: Prime issue #17 and a reproducible artifact, Pylon issue #190 consuming the exact post-attach proof, then Comet issue #7. Exact committed-head API/security/test review and trusted hosted CI remain mandatory. Revisit when upstream offers an equivalent generation-scoped proof and both consumers can remove this token without weakening fail-closed negotiation.
+
+
+## 2026-08-31 — negotiated proof shipped and mixed-version snapshot catch-up follow-up
+
+- Upstream evidence remains fully audited through `PrimeIntellect-ai/prime-agent@a903d4b6768f484bd6d459b7b0aa7dee38e461e2`, the product base used by PR #19. The only later upstream-main commit currently visible is `c382f09856d4a8c8d2b765179657047d58691f25` (PR #1893, terminal Mermaid rendering); its changed paths do not overlap daemon snapshot, worker, supervisor, framing, or recovery code and it does not supersede this boundary.
+- Corrected `negotiated-daemon-session-capabilities` from candidate to shipped. Fork PR [#18](https://github.com/pylon-code/prime-agent/pull/18) merged as `91e13b6798343995291ccca6f523fba81ff96cd6`, and Pylon consumed its generation-scoped post-attach proof in [pylon-code/pylon#191](https://github.com/pylon-code/pylon/pull/191). The prior candidate text was stale metadata, not an incomplete implementation.
+- Follow-up issue [#21](https://github.com/pylon-code/prime-agent/issues/21) separates the remaining catch-up work from the foundation shipped by #11 and PR #14. `snapshot-recovery-integrity` remains **hybridize**: keep upstream child projection, topology, reuse, and ownership primitives while retaining Pylon's immutable transfer and isolated-retry contract.
+- Candidate PR [#19](https://github.com/pylon-code/prime-agent/pull/19) adds daemon schema revision 28 and negotiated `snapshot_generation_nonce_v1`. The nonce is sent only when protocol, schema, and capability agree. Begin/chunk/end identity, progress, attachment epoch, alias routing, and late failure frames are validated before publication. One affected attachment may request one bounded fresh generation without recycling or self-requeueing an otherwise healthy resident worker.
+- Large-transcript handling prepares one immutable payload, bounds framing and drain waits, avoids quadratic private-buffer shifting, and preserves spill ownership and cleanup across cancellation, crash, and stale generations. Stock/current `v0.8.1` supervisor and worker directions retain their mixed-version fallback.
+- The pre-ledger source candidate `8b504e3774875c241c5d0d3b4b588a09f4aa3f8e` passed `npm run check`, package build, 246 conflict-affected exact-head tests after rebase, 16 real supervisor-process tests with 8 fixture-gated skips, stock/current compatibility in both directions, a 36 MiB exact-package transfer, a 10,000-message preparation probe, a 131,000-fragment framing probe, and two independent adversarial reviews. The ledger correction changes the exact head and therefore requires renewed targeted checks and hosted CI before merge.
+- Revisit when Prime upstream supplies the same capability-gated fresh-generation identity, attachment-local retry containment, mixed-version behavior, and bounded preparation/framing guarantees without weakening Pylon's correlated lifecycle or cleanup contracts.

packages/coding-agent/.changes/snapshot-catchup-isolation.md

@@ -0,0 +1 @@
+- Isolated daemon snapshot catch-up failures so corrupt or stalled transfers retry once without closing healthy worker sessions ([#21](https://github.com/pylon-code/prime-agent/issues/21)).

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

@@ -41,6 +41,7 @@ import { deserializeDaemonError } from "../daemon/daemon-errors.js";
 import {
 	collectDaemonClientEnv,
 	collectDaemonLaunchEnv,
+	DAEMON_SNAPSHOT_GENERATION_NONCE_MIN_SCHEMA_REVISION,
 	DAEMON_SUPPORTED_CLIENT_CAPABILITIES,
 	type DaemonAttachResult,
 	type DaemonClientCapability,
@@ -504,12 +505,15 @@ export class DaemonAgentConnection implements AgentConnection {
 			message.type === "session_replaced" &&
 			!message.snapshotFollows &&
 			this.supportsNegotiatedRuntimeCapability("correlated_prompt_lifecycle_v1");
+		const routeAttachmentEpoch = this.attachmentEpoch;
 		const previous = this.replacementMessageTail;
 		const handling = previous
 			? previous.then(() => {
-					if (!this.replacementReconciliationFailed) return this.handleDaemonMessage(message);
+					if (!this.replacementReconciliationFailed) {
+						return this.handleDaemonMessage(message, routeAttachmentEpoch);
+					}
 				})
-			: this.handleDaemonMessage(message);
+			: this.handleDaemonMessage(message, routeAttachmentEpoch);
 		const settled = handling.catch((error: unknown) => {
 			try {
 				appendRotatingLog(
@@ -705,7 +709,7 @@ export class DaemonAgentConnection implements AgentConnection {
 				result = await this.requestData<SessionSummary | DaemonAttachResult>({
 					type: "attach",
 					activeSessionId: requestedActiveSessionId,
-					snapshotGenerationNonce: randomUUID(),
+					...(this.supportsSnapshotGenerationNonce() ? { snapshotGenerationNonce: randomUUID() } : {}),
 					supportsExtensionUi,
 					clientId: this.clientId,
 					capabilities,
@@ -994,6 +998,13 @@ export class DaemonAgentConnection implements AgentConnection {
 		);
 	}
 
+	private supportsSnapshotGenerationNonce(): boolean {
+		return (
+			this.client.supportsServerCapability("snapshot_generation_nonce_v1") &&
+			(this.client.hello?.schemaRevision ?? 0) >= DAEMON_SNAPSHOT_GENERATION_NONCE_MIN_SCHEMA_REVISION
+		);
+	}
+
 	/** Server-offer evidence used to construct the pre-attach capability list. */
 	supportsCorrelatedPromptLifecycle(): boolean {
 		return this.client.supportsServerCapability("correlated_prompt_lifecycle_v1");
@@ -1930,6 +1941,7 @@ export class DaemonAgentConnection implements AgentConnection {
 			throw new Error("Daemon connection replacement reconciliation has failed");
 		}
 		this.pendingReattachActiveSessionIds.add(targetActiveSessionId);
+		let attachmentResponseReceived = false;
 		let reattached = false;
 		let attachmentEpoch: number | undefined;
 		let invalidationRevision: number | undefined;
@@ -1974,15 +1986,15 @@ export class DaemonAgentConnection implements AgentConnection {
 							: {
 									type: "attach",
 									activeSessionId: targetActiveSessionId,
-									snapshotGenerationNonce: randomUUID(),
+									...(this.supportsSnapshotGenerationNonce() ? { snapshotGenerationNonce: randomUUID() } : {}),
 									supportsExtensionUi,
 									clientId: this.clientId,
 									capabilities,
 									env: this.options.sendClientEnv ? collectDaemonClientEnv() : undefined,
 									telemetryDisabled: this.options.telemetryDisabled,
 								},
 					);
-					reattached = true;
+					attachmentResponseReceived = true;
 					this.assertAttachmentCommit(
 						targetActiveSessionId,
 						attachmentEpoch,
@@ -1993,6 +2005,7 @@ export class DaemonAgentConnection implements AgentConnection {
 					if (result.activeSessionId !== targetActiveSessionId) {
 						throw new Error("Daemon returned an invalid session snapshot");
 					}
+					reattached = true;
 					if (result.snapshotStream) {
 						this.bindSnapshotExpectation(requestAttempt, result);
 						requestAttempt.state = "receiving";
@@ -2066,12 +2079,12 @@ export class DaemonAgentConnection implements AgentConnection {
 				this.attachmentEpoch === attachmentEpoch &&
 				this.attachmentInvalidationRevision === invalidationRevision &&
 				sharedAttachmentOwners.get(this.client)?.get(targetActiveSessionId) === this;
-			if (reattached && stillCurrent) {
+			if (attachmentResponseReceived && stillCurrent) {
 				if (this.supportsCorrelatedPromptLifecycle()) {
 					await this.emit({ type: "correlated_prompt_protocol_violation" });
 				}
 				this.failClosedReplacementReconciliation();
-			} else if (!reattached && stillCurrent) {
+			} else if (!attachmentResponseReceived && stillCurrent) {
 				this.invalidateNegotiatedCapabilityProof();
 			}
 			throw error;
@@ -2345,7 +2358,7 @@ export class DaemonAgentConnection implements AgentConnection {
 		return response.data as T;
 	}
 
-	private async handleDaemonMessage(message: DaemonOutbound): Promise<void> {
+	private async handleDaemonMessage(message: DaemonOutbound, routeAttachmentEpoch: number): Promise<void> {
 		if (message.type === "heartbeats_changed") {
 			await this.emit({ type: "heartbeats_changed" });
 			return;
@@ -2513,11 +2526,25 @@ export class DaemonAgentConnection implements AgentConnection {
 				}
 				return;
 			}
+			const explicitPurpose = message.purpose;
 			if (!existingAssembly && !this.pendingChunkedReplacement && !activeRequestAttempt) {
 				this.ignoreSnapshotId(message.snapshotId);
+				if (
+					(explicitPurpose === "replacement" || explicitPurpose === "resync") &&
+					this.attachedSessionId !== undefined &&
+					routeAttachmentEpoch === this.attachmentEpoch &&
+					message.activeSessionId === this.activeSessionId &&
+					!this.pendingReattachActiveSessionIds.has(message.activeSessionId)
+				) {
+					await this.transitionRuntimeSnapshotFailure(
+						explicitPurpose,
+						new Error(message.error),
+						message.snapshotId,
+						message.activeSessionId,
+					);
+				}
 				return;
 			}
-			const explicitPurpose = message.purpose;
 			const assembly =
 				existingAssembly ??
 				(this.pendingChunkedReplacement || activeRequestAttempt
@@ -3284,7 +3311,9 @@ export class DaemonAgentConnection implements AgentConnection {
 	}
 
 	private rejectSnapshotAssemblies(error: Error, recoverRuntime = false): void {
-		for (const [snapshotId, assembly] of [...this.snapshotAssemblies]) {
+		const assemblies = [...this.snapshotAssemblies];
+		const requestAttempts = [...this.snapshotRequestAttempts];
+		for (const [snapshotId, assembly] of assemblies) {
 			clearTimeout(assembly.timeout);
 			if (recoverRuntime) {
 				void this.transitionSnapshotFailure(
@@ -3297,13 +3326,17 @@ export class DaemonAgentConnection implements AgentConnection {
 			} else {
 				assembly.reject(error);
 			}
+			if (this.snapshotAssemblies.get(snapshotId) === assembly) this.snapshotAssemblies.delete(snapshotId);
 		}
-		this.snapshotAssemblies.clear();
 		this.completedSnapshots.clear();
 		this.completedSnapshotAttemptIds.clear();
-		this.snapshotRequestAttempts.clear();
-		this.ignoredSnapshotIds.clear();
-		if (!recoverRuntime) this.runtimeSnapshotAttempt = undefined;
+		for (const [attemptId, attempt] of requestAttempts) {
+			if (this.snapshotRequestAttempts.get(attemptId) === attempt) this.snapshotRequestAttempts.delete(attemptId);
+		}
+		if (!recoverRuntime) {
+			this.ignoredSnapshotIds.clear();
+			this.runtimeSnapshotAttempt = undefined;
+		}
 	}
 
 	private ignoreSnapshotId(snapshotId: string): void {

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

@@ -5465,12 +5465,16 @@ export class AgentDaemon {
 	}
 
 	private encodeWorkerPrivateFrameParts(
-		line: string | Buffer | readonly Buffer[],
+		line: string | SnapshotTranscriptWireChunk,
 		message: DaemonOutbound,
 		payloadEncoding: "jsonl" | "assistant-delta",
 		snapshotPurpose?: "attach" | "replacement" | "catchup",
 	): Buffer[] {
 		const payloadParts = typeof line === "string" ? [Buffer.from(line)] : Buffer.isBuffer(line) ? [line] : [...line];
+		const snapshotChunkMessageCount =
+			message.type === "session_snapshot_chunk" && typeof line !== "string" && !Buffer.isBuffer(line)
+				? line.snapshotMessageCount
+				: undefined;
 		return encodePrivateFrameParts<DaemonWorkerFrameHeader>(
 			{
 				kind: "outbound",
@@ -5481,6 +5485,12 @@ export class AgentDaemon {
 					? { snapshotId: message.snapshotId }
 					: {}),
 				...(message.type === "session_event" ? { sessionEventType: message.event.type } : {}),
+				...(message.type === "session_snapshot_chunk"
+					? {
+							snapshotChunkIndex: message.index,
+							...(snapshotChunkMessageCount === undefined ? {} : { snapshotChunkMessageCount }),
+						}
+					: {}),
 				payloadEncoding,
 				...(snapshotPurpose ? { snapshotPurpose } : {}),
 			},