diff --git a/.pylon/features.yaml b/.pylon/features.yaml index 8fa12f8edc..a8cc132686 100644 --- a/.pylon/features.yaml +++ b/.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 diff --git a/.pylon/upstream-review.md b/.pylon/upstream-review.md index d8428f3d66..633f485a63 100644 --- a/.pylon/upstream-review.md +++ b/.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. diff --git a/packages/coding-agent/.changes/snapshot-catchup-isolation.md b/packages/coding-agent/.changes/snapshot-catchup-isolation.md new file mode 100644 index 0000000000..7a5c12f4f7 --- /dev/null +++ b/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)). diff --git a/packages/coding-agent/src/modes/agent-connection/daemon-agent-connection.ts b/packages/coding-agent/src/modes/agent-connection/daemon-agent-connection.ts index 9005aa0f65..45b23fa1cd 100644 --- a/packages/coding-agent/src/modes/agent-connection/daemon-agent-connection.ts +++ b/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({ 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,7 +1986,7 @@ export class DaemonAgentConnection implements AgentConnection { : { type: "attach", activeSessionId: targetActiveSessionId, - snapshotGenerationNonce: randomUUID(), + ...(this.supportsSnapshotGenerationNonce() ? { snapshotGenerationNonce: randomUUID() } : {}), supportsExtensionUi, clientId: this.clientId, capabilities, @@ -1982,7 +1994,7 @@ export class DaemonAgentConnection implements AgentConnection { 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 { + private async handleDaemonMessage(message: DaemonOutbound, routeAttachmentEpoch: number): Promise { 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 { diff --git a/packages/coding-agent/src/modes/daemon/daemon-mode.ts b/packages/coding-agent/src/modes/daemon/daemon-mode.ts index d79f7ef6ee..cce909f09d 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-mode.ts +++ b/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( { 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 } : {}), }, diff --git a/packages/coding-agent/src/modes/daemon/daemon-protocol.ts b/packages/coding-agent/src/modes/daemon/daemon-protocol.ts index d5a420947a..a4e3e4d4eb 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-protocol.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-protocol.ts @@ -78,8 +78,9 @@ export const DAEMON_COMMAND_ENVELOPE_MIN_PROTOCOL_VERSION = 7; // Revision 25 negotiates immutable snapshot transfer identities with session workers. // Revision 26 correlates snapshot failures that occur before a begin frame can be emitted. // Revision 27 adds a capability-gated authoritative owned-session cleanup query. -export const DAEMON_SCHEMA_REVISION = 27; -export const DAEMON_SCHEMA_ID = "protocol-7-schema-27-7c0c21a689b5"; +// Revision 28 negotiates fresh snapshot generations and adds private worker chunk routing metadata. +export const DAEMON_SCHEMA_REVISION = 28; +export const DAEMON_SCHEMA_ID = "protocol-7-schema-28-7c0c21a689b5"; export type DaemonProtocolName = typeof DAEMON_PROTOCOL_NAME; export type DaemonProtocolVersion = number; @@ -131,6 +132,7 @@ export type DaemonServerCapability = | "session_input_pause" | "owned_prompt_cancellation" | "acp_mcp_servers" + | "snapshot_generation_nonce_v1" | "authoritative_owned_session_cleanup_v1"; export type DaemonReplayStatus = "complete" | "partial" | "unavailable"; @@ -178,6 +180,7 @@ export const DAEMON_DEFAULT_SERVER_CAPABILITIES: readonly DaemonServerCapability "rlm_quiescence_barrier", "session_input_pause", "acp_mcp_servers", + "snapshot_generation_nonce_v1", ]; /** Supervisor-only offers. Private session workers must not advertise these commands. */ @@ -773,6 +776,12 @@ const CORRELATED_PROMPT_LIFECYCLE_COMMAND = { minSchemaRevision: 24, capability: "correlated_prompt_lifecycle_v1", } as const; +export const DAEMON_SNAPSHOT_GENERATION_NONCE_MIN_SCHEMA_REVISION = 28; +const SNAPSHOT_GENERATION_NONCE_COMMAND = { + minProtocol: 7, + minSchemaRevision: DAEMON_SNAPSHOT_GENERATION_NONCE_MIN_SCHEMA_REVISION, + capability: "snapshot_generation_nonce_v1", +} as const; export const DAEMON_COMMAND_COMPATIBILITY = { ack_result: LEGACY_DAEMON_COMMAND, @@ -889,6 +898,9 @@ export function getDaemonCommandCompatibilities(command: DaemonCommand): readonl if ((command.type === "attach" || command.type === "reattach") && command.recoveryConfig !== undefined) { requirements.push(OWNED_SESSION_RECOVERY_CONTEXT); } + if (command.type === "attach" && command.snapshotGenerationNonce !== undefined) { + requirements.push(SNAPSHOT_GENERATION_NONCE_COMMAND); + } const carriesTelemetryPolicy = ((command.type === "attach" || command.type === "reattach") && command.telemetryDisabled !== undefined) || (command.type === "create" && command.config?.telemetryDisabled !== undefined); diff --git a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts index 4d53497da1..1927471677 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts @@ -74,6 +74,7 @@ import { DAEMON_PROTOCOL_INFO, DAEMON_SCHEMA_ID, DAEMON_SCHEMA_REVISION, + DAEMON_SNAPSHOT_GENERATION_NONCE_MIN_SCHEMA_REVISION, DAEMON_SUPERVISOR_SERVER_CAPABILITIES, DAEMON_UPDATE_RESTART_FORMAT_VERSION, type DaemonAttachResult, @@ -170,6 +171,24 @@ function waitForAbortable(promise: Promise, signal: AbortSignal | undefine void promise.then(resolve, reject).finally(() => signal.removeEventListener("abort", onAbort)); }); } + +function hasCanonicalSnapshotChunkEnvelope( + payload: Buffer, + activeSessionId: string, + snapshotId: string, + index: number, +): boolean { + const prefix = Buffer.from( + `{"type":"session_snapshot_chunk","activeSessionId":${JSON.stringify(activeSessionId)},` + + `"snapshotId":${JSON.stringify(snapshotId)},"index":${index},"messages":[`, + ); + const suffix = Buffer.from("]}\n"); + return ( + payload.length >= prefix.length + suffix.length && + payload.subarray(0, prefix.length).equals(prefix) && + payload.subarray(payload.length - suffix.length).equals(suffix) + ); +} const UPDATE_RESTART_MUTATION_DRAIN_TIMEOUT_MS = 80_000; const UPDATE_RESTART_WORKER_REQUEST_TIMEOUT_MS = 90_000; // The whole pre-commit prepare (drain + worker fencing) must finish inside the @@ -4550,7 +4569,12 @@ export class DaemonSupervisor { { type: "attach", activeSessionId, - snapshotGenerationNonce: randomUUID(), + ...(workerClient.supportsServerCapability?.( + "snapshot_generation_nonce_v1", + DAEMON_SNAPSHOT_GENERATION_NONCE_MIN_SCHEMA_REVISION, + ) + ? { snapshotGenerationNonce: randomUUID() } + : {}), capabilities: workerCapabilities, supportsExtensionUi: false, env: command.env ?? collectDaemonClientEnv(), @@ -5447,25 +5471,47 @@ export class DaemonSupervisor { return; } if (outboundType === "session_snapshot_chunk" && activeSessionId) { - let chunk: Extract; + const routedIndex = frame.header.snapshotChunkIndex; + const routedMessageCount = frame.header.snapshotChunkMessageCount; + let snapshotId = frameSnapshotId; + let chunkIndex: number; + let chunkMessageCount: number; + if (snapshotId && !this.snapshotGeneration(worker, activeSessionId, snapshotId)?.incoming) return; try { - chunk = JSON.parse(frame.payload.toString("utf8")) as typeof chunk; + if (snapshotId && routedIndex !== undefined && routedMessageCount !== undefined) { + if (!hasCanonicalSnapshotChunkEnvelope(frame.payload, activeSessionId, snapshotId, routedIndex)) { + throw new Error(`Snapshot ${snapshotId} returned an invalid canonical chunk envelope`); + } + chunkIndex = routedIndex; + chunkMessageCount = routedMessageCount; + } else { + const chunk = JSON.parse(frame.payload.toString("utf8")) as Extract< + DaemonOutbound, + { type: "session_snapshot_chunk" } + >; + snapshotId ??= typeof chunk.snapshotId === "string" ? chunk.snapshotId : undefined; + if ( + chunk.type !== "session_snapshot_chunk" || + chunk.activeSessionId !== activeSessionId || + chunk.snapshotId !== snapshotId || + !Number.isSafeInteger(chunk.index) || + chunk.index < 0 || + !Array.isArray(chunk.messages) + ) { + throw new Error(`Snapshot ${snapshotId ?? "unknown"} returned an invalid chunk frame`); + } + chunkIndex = chunk.index; + chunkMessageCount = chunk.messages.length; + } } catch (error) { const frameError = error instanceof Error ? error : new Error(String(error)); - if (frameSnapshotId) { - this.failCorrelatedSnapshotAttempt( - worker, - activeSessionId, - frameSnapshotId, - frameError, - snapshotPurpose, - ); + if (snapshotId) { + this.failCorrelatedSnapshotAttempt(worker, activeSessionId, snapshotId, frameError, snapshotPurpose); } else { this.failWorkerSnapshotCache(worker, activeSessionId, frameError, true); } return; } - const snapshotId = frameSnapshotId ?? (typeof chunk.snapshotId === "string" ? chunk.snapshotId : undefined); if (!snapshotId) { this.failWorkerSnapshotCache( worker, @@ -5476,23 +5522,15 @@ export class DaemonSupervisor { return; } const generation = this.snapshotGeneration(worker, activeSessionId, snapshotId); - if (!generation?.incoming) { - return; - } + if (!generation?.incoming) return; try { const duplicateIndex = generation.duplicateChunkIndex; const expectedIndex = duplicateIndex ?? generation.transcript.chunkCount; - if ( - chunk.type !== "session_snapshot_chunk" || - chunk.activeSessionId !== activeSessionId || - chunk.snapshotId !== generation.transcript.snapshotId || - chunk.index !== expectedIndex || - !Array.isArray(chunk.messages) - ) { + if (chunkIndex !== expectedIndex) { throw new Error(`Snapshot ${generation.transcript.snapshotId} returned an invalid chunk frame`); } if (duplicateIndex === undefined) { - const receivedMessageCount = (generation.receivedMessageCount ?? 0) + chunk.messages.length; + const receivedMessageCount = (generation.receivedMessageCount ?? 0) + chunkMessageCount; if (receivedMessageCount > (generation.result.snapshotStream?.messageCount ?? Number.MAX_SAFE_INTEGER)) { throw new Error(`Snapshot ${generation.transcript.snapshotId} exceeded its declared message count`); } diff --git a/packages/coding-agent/src/modes/daemon/daemon-worker-client.ts b/packages/coding-agent/src/modes/daemon/daemon-worker-client.ts index 9c435b2e72..8b8e9c130e 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-worker-client.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-worker-client.ts @@ -78,8 +78,11 @@ export class DaemonWorkerClient { socket.on("close", () => this.notifyClosed(socket, new Error("Daemon worker socket closed"))); } - supportsServerCapability(capability: DaemonServerCapability): boolean { - return this.hello?.serverCapabilities.includes(capability) === true; + supportsServerCapability(capability: DaemonServerCapability, minSchemaRevision = 0): boolean { + return ( + this.hello?.serverCapabilities.includes(capability) === true && + (this.hello.schemaRevision ?? 0) >= minSchemaRevision + ); } waitForHello(timeoutMs = 3000): Promise { diff --git a/packages/coding-agent/src/modes/daemon/daemon-worker-protocol.ts b/packages/coding-agent/src/modes/daemon/daemon-worker-protocol.ts index 7ce97bc04e..8c4fed73d9 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-worker-protocol.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-worker-protocol.ts @@ -30,6 +30,9 @@ export type DaemonWorkerFrameHeader = sessionEventType?: string; payloadEncoding?: "jsonl" | "assistant-delta"; snapshotPurpose?: "attach" | "replacement" | "catchup"; + /** Additive routing metadata avoids reparsing large canonical snapshot chunks. */ + snapshotChunkIndex?: number; + snapshotChunkMessageCount?: number; }; export type DaemonCreateCommand = Extract; @@ -212,6 +215,11 @@ export function isDaemonWorkerFrameHeader(value: unknown): value is DaemonWorker (candidate.activeSessionId === undefined || typeof candidate.activeSessionId === "string") && (candidate.snapshotId === undefined || typeof candidate.snapshotId === "string") && (candidate.sessionEventType === undefined || typeof candidate.sessionEventType === "string") && + (candidate.snapshotChunkIndex === undefined || + (Number.isSafeInteger(candidate.snapshotChunkIndex) && (candidate.snapshotChunkIndex as number) >= 0)) && + (candidate.snapshotChunkMessageCount === undefined || + (Number.isSafeInteger(candidate.snapshotChunkMessageCount) && + (candidate.snapshotChunkMessageCount as number) >= 0)) && (candidate.snapshotPurpose === undefined || candidate.snapshotPurpose === "attach" || candidate.snapshotPurpose === "replacement" || diff --git a/packages/coding-agent/src/modes/daemon/snapshot-transcript-cache.ts b/packages/coding-agent/src/modes/daemon/snapshot-transcript-cache.ts index 5dc2092883..1e24109c9c 100644 --- a/packages/coding-agent/src/modes/daemon/snapshot-transcript-cache.ts +++ b/packages/coding-agent/src/modes/daemon/snapshot-transcript-cache.ts @@ -167,7 +167,7 @@ interface SnapshotTranscriptChunkLifecycle { dispose?(): void; } -export type SnapshotTranscriptWireChunk = Buffer | readonly Buffer[]; +export type SnapshotTranscriptWireChunk = Buffer | (readonly Buffer[] & { readonly snapshotMessageCount?: number }); export type SnapshotTranscriptChunkSource = ( | Iterable | AsyncIterable @@ -203,17 +203,10 @@ export function createImmutableSnapshotMessages(messages: readonly AgentMessage[ return cloneSnapshotValue([...messages]); } -function createGenerationGuard( - source: readonly AgentMessage[], - selected: readonly AgentMessage[], - validate?: () => void, -): () => void { - return () => { - validate?.(); - if (!isDeepStrictEqual(source, selected)) { - throw new Error("Session snapshot generation changed during transcript preparation"); - } - }; +function assertSnapshotSourceUnchanged(source: readonly AgentMessage[], selected: readonly AgentMessage[]): void { + if (!isDeepStrictEqual(source, selected)) { + throw new Error("Session snapshot generation changed during transcript preparation"); + } } interface JsonParts { @@ -221,21 +214,25 @@ interface JsonParts { bytes: number; } +interface IncrementalJsonYieldState { + bytesSinceYield: number; +} + class IncrementalJsonEncoder { private readonly parts: Buffer[] = []; private bytes = 0; - private bytesSinceYield = 0; private readonly ancestors = new Set(); constructor( private readonly signal: AbortSignal | undefined, - private readonly assertGeneration: () => void, + private readonly validateGeneration: () => void, + private readonly yieldState: IncrementalJsonYieldState, ) {} async encode(value: unknown): Promise { const encoded = await this.encodeValue(value, "", false); if (!encoded) throw new Error("Snapshot message is not JSON serializable"); - await this.yieldIfNeeded(true); + await this.yieldIfNeeded(); return { parts: this.parts, bytes: this.bytes }; } @@ -244,15 +241,15 @@ class IncrementalJsonEncoder { if (buffer.length === 0) return; this.parts.push(buffer); this.bytes += buffer.length; - this.bytesSinceYield += buffer.length; + this.yieldState.bytesSinceYield += buffer.length; } - private async yieldIfNeeded(force = false): Promise { - if (!force && this.bytesSinceYield < SNAPSHOT_PREPARATION_YIELD_BYTES) return; - this.bytesSinceYield = 0; + private async yieldIfNeeded(): Promise { + if (this.yieldState.bytesSinceYield < SNAPSHOT_PREPARATION_YIELD_BYTES) return; + this.yieldState.bytesSinceYield = 0; await new Promise((resolveYield) => setImmediate(resolveYield)); this.signal?.throwIfAborted(); - this.assertGeneration(); + this.validateGeneration(); } private resolveJsonValue(value: unknown, key: string): unknown { @@ -412,9 +409,12 @@ export class SnapshotTranscriptPayloadCache { signal?: AbortSignal, validateGeneration?: () => void, ): Promise { - signal?.throwIfAborted(); - const assertGeneration = createGenerationGuard(sourceMessages, selectedMessages, validateGeneration); - assertGeneration(); + const validatePreparation = () => { + signal?.throwIfAborted(); + validateGeneration?.(); + }; + validatePreparation(); + const yieldState: IncrementalJsonYieldState = { bytesSinceYield: 0 }; let currentParts: Buffer[] = []; let currentBytes = 0; let currentMessages = 0; @@ -426,8 +426,8 @@ export class SnapshotTranscriptPayloadCache { currentMessages = 0; }; for (const message of selectedMessages) { - signal?.throwIfAborted(); - const encoded = await new IncrementalJsonEncoder(signal, assertGeneration).encode(message); + validatePreparation(); + const encoded = await new IncrementalJsonEncoder(signal, validatePreparation, yieldState).encode(message); const separatorBytes = currentMessages > 0 ? 1 : 0; if (currentMessages > 0 && currentBytes + separatorBytes + encoded.bytes > this.targetChunkBytes) { await flush(); @@ -440,10 +440,11 @@ export class SnapshotTranscriptPayloadCache { currentBytes += encoded.bytes; currentMessages++; this.messageTotal++; - assertGeneration(); + validatePreparation(); } await flush(); - assertGeneration(); + validatePreparation(); + assertSnapshotSourceUnchanged(sourceMessages, selectedMessages); } async readPayloadChunk(index: number): Promise { @@ -496,7 +497,11 @@ export class SnapshotTranscriptPayloadCache { const prefix = `{"type":"session_snapshot_chunk","activeSessionId":${JSON.stringify(activeSessionId)},` + `"snapshotId":${JSON.stringify(snapshotId)},"index":${index},"messages":[`; - yield [Buffer.from(prefix), payload, Buffer.from("]}\n")]; + const parts = [Buffer.from(prefix), payload, Buffer.from("]}\n")] as Buffer[] & { + snapshotMessageCount: number; + }; + parts.snapshotMessageCount = this.chunkMessageCount(index); + yield parts; } } finally { release(); @@ -822,7 +827,7 @@ export class SnapshotTranscriptCache { const prefix = `{"type":"session_snapshot_chunk","activeSessionId":${JSON.stringify(this.activeSessionId)},` + `"snapshotId":${JSON.stringify(snapshotId)},"index":${index},"messages":[`; - return Buffer.concat([Buffer.from(prefix), chunk.subarray(payloadStart, payloadEnd), Buffer.from("]}\n")]); + return [Buffer.from(prefix), chunk.subarray(payloadStart, payloadEnd), Buffer.from("]}\n")]; } retain(): () => void { @@ -912,8 +917,7 @@ export class SnapshotTranscriptCache { const index = this.chunks.length - 1; const waiters = this.chunkWaiters.get(index); if (waiters) { - const stored = this.readChunk(index); - for (const waiter of waiters) waiter.resolve(stored); + for (const waiter of waiters) waiter.resolve(buffer); this.chunkWaiters.delete(index); } } diff --git a/packages/coding-agent/src/modes/session-worker/private-framing.ts b/packages/coding-agent/src/modes/session-worker/private-framing.ts index 993ed61031..4c4329b88a 100644 --- a/packages/coding-agent/src/modes/session-worker/private-framing.ts +++ b/packages/coding-agent/src/modes/session-worker/private-framing.ts @@ -1,6 +1,7 @@ import type { Duplex } from "node:stream"; const FRAME_PREFIX_BYTES = 8; +const FRAME_BUFFER_COMPACTION_MIN_HEAD = 1024; export interface PrivateFrameLimits { maxHeaderBytes: number; @@ -69,6 +70,7 @@ export function encodePrivateFrame( export class PrivateFrameDecoder { private readonly buffers: Buffer[] = []; + private headIndex = 0; private headOffset = 0; private totalBufferedBytes = 0; private totalCoalescedBytes = 0; @@ -139,24 +141,21 @@ export class PrivateFrameDecoder { private consume(length: number): Buffer { if (length === 0) return Buffer.alloc(0); - const first = this.buffers[0]; + const first = this.buffers[this.headIndex]; if (!first) throw new Error("Private frame decoder buffer underflow"); const firstAvailable = first.length - this.headOffset; if (firstAvailable >= length) { const value = first.subarray(this.headOffset, this.headOffset + length); this.headOffset += length; this.totalBufferedBytes -= length; - if (this.headOffset === first.length) { - this.buffers.shift(); - this.headOffset = 0; - } + if (this.headOffset === first.length) this.releaseHeadBuffer(); return value; } const value = Buffer.allocUnsafe(length); this.totalCoalescedBytes += length; let written = 0; while (written < length) { - const current = this.buffers[0]; + const current = this.buffers[this.headIndex]; if (!current) throw new Error("Private frame decoder buffer underflow"); const available = current.length - this.headOffset; const selected = Math.min(available, length - written); @@ -164,13 +163,24 @@ export class PrivateFrameDecoder { written += selected; this.headOffset += selected; this.totalBufferedBytes -= selected; - if (this.headOffset === current.length) { - this.buffers.shift(); - this.headOffset = 0; - } + if (this.headOffset === current.length) this.releaseHeadBuffer(); } return value; } + + private releaseHeadBuffer(): void { + this.headIndex++; + this.headOffset = 0; + if (this.headIndex === this.buffers.length) { + this.buffers.length = 0; + this.headIndex = 0; + return; + } + if (this.headIndex >= FRAME_BUFFER_COMPACTION_MIN_HEAD && this.headIndex * 2 >= this.buffers.length) { + this.buffers.splice(0, this.headIndex); + this.headIndex = 0; + } + } } export type PrivateFrameListener = (frame: PrivateFrame) => void; diff --git a/packages/coding-agent/test/agent-connection-daemon.test.ts b/packages/coding-agent/test/agent-connection-daemon.test.ts index 2a63c60498..fd412cea6a 100644 --- a/packages/coding-agent/test/agent-connection-daemon.test.ts +++ b/packages/coding-agent/test/agent-connection-daemon.test.ts @@ -5533,18 +5533,87 @@ describe("DaemonAgentConnection", () => { expect((connection as unknown as { snapshotAssemblies: Map }).snapshotAssemblies.size).toBe(0); }); - it("retries one failed attach request and clears attempt state after success", async () => { + it("retains the retry request created while a runtime snapshot close is being rejected", async () => { const fakeClient = new FakeDaemonClient(); + const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-1"); + let resolveRecovery!: () => void; + const recovery = new Promise((resolve) => { + resolveRecovery = resolve; + }); + const internals = connection as unknown as { + attachmentEpoch: number; + snapshotRequestAttempts: Map; + getSnapshotAssembly(snapshotId: string): { + begin?: Extract; + }; + startSnapshotRequestAttempt( + activeSessionId: string, + attachmentEpoch: number, + logicalPurpose: "replacement", + wirePurpose: "attach", + ): unknown; + recoverFailedSnapshot(): Promise; + rejectSnapshotAssemblies(error: Error, recoverRuntime: boolean): void; + }; + internals.recoverFailedSnapshot = () => { + internals.startSnapshotRequestAttempt("active-1", internals.attachmentEpoch, "replacement", "attach"); + return recovery; + }; + const assembly = internals.getSnapshotAssembly("runtime-close"); + const result = createAttachResult("active-1", "client-1", undefined, 12); + const { messages: _messages, ...snapshot } = result.snapshot; + assembly.begin = { + type: "session_snapshot_begin", + activeSessionId: "active-1", + snapshotId: "runtime-close", + snapshot, + messageCount: result.snapshot.messages.length, + targetChunkBytes: 512 * 1024, + purpose: "replacement", + }; + + internals.rejectSnapshotAssemblies(new Error("transport closed"), true); + + expect([...internals.snapshotRequestAttempts.values()]).toEqual([ + expect.objectContaining({ logicalPurpose: "replacement" }), + ]); + resolveRecovery(); + await recovery; + await Promise.resolve(); + await connection.dispose(); + }); + + it("retries one failed attach request with a fresh negotiated generation nonce", async () => { + const fakeClient = new FakeDaemonClient(); + fakeClient.serverCapabilities.add("snapshot_generation_nonce_v1"); fakeClient.attachFailures = 1; const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-1"); await expect(connection.attach()).resolves.toBeUndefined(); - expect(fakeClient.requests.filter((request) => request.type === "attach")).toHaveLength(2); + const attachRequests = fakeClient.requests.filter((request) => request.type === "attach"); + expect(attachRequests).toHaveLength(2); + expect(attachRequests[0]?.snapshotGenerationNonce).toEqual(expect.any(String)); + expect(attachRequests[1]?.snapshotGenerationNonce).toEqual(expect.any(String)); + expect(attachRequests[1]?.snapshotGenerationNonce).not.toBe(attachRequests[0]?.snapshotGenerationNonce); expect( (connection as unknown as { snapshotRequestAttempts: Map }).snapshotRequestAttempts.size, ).toBe(0); }); + it("omits generation nonces from a revision 27 daemon even if it advertises the newer capability", async () => { + const fakeClient = new FakeDaemonClient(); + fakeClient.serverCapabilities.add("snapshot_generation_nonce_v1"); + fakeClient.hello = { ...fakeClient.hello!, schemaRevision: 27 }; + const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-1"); + + await connection.attach(); + + const attachRequests = fakeClient.requests.filter((request) => request.type === "attach"); + expect(attachRequests).toHaveLength(1); + expect(attachRequests[0]?.snapshotGenerationNonce).toBeUndefined(); + await connection.dispose(); + }); + it("reuses a settled same-id assembly with a fresh timer and resolves the original attach caller", async () => { const fakeClient = new FakeDaemonClient(); let attachAttempt = 0; @@ -5607,6 +5676,60 @@ describe("DaemonAgentConnection", () => { ).toBe(0); }); + it("rejects a streamed begin that mismatches its response descriptor and retries fresh", async () => { + const fakeClient = new FakeDaemonClient(); + fakeClient.serverCapabilities.add("snapshot_generation_nonce_v1"); + let attempt = 0; + fakeClient.attachResultFactory = (command) => { + const currentAttempt = ++attempt; + const full = createAttachResult(command.activeSessionId, command.clientId, command.capabilities, 12, { + messages: [{ role: "user", content: `generation-${currentAttempt}`, timestamp: currentAttempt }], + }); + const snapshotId = `descriptor-${currentAttempt}`; + const { messages: _messages, ...snapshot } = full.snapshot; + queueMicrotask(() => { + fakeClient.emitMessage({ + type: "session_snapshot_begin", + activeSessionId: command.activeSessionId, + snapshotId, + snapshot, + messageCount: currentAttempt === 1 ? 2 : 1, + targetChunkBytes: 512 * 1024, + }); + if (currentAttempt === 1) return; + fakeClient.emitMessage({ + type: "session_snapshot_chunk", + activeSessionId: command.activeSessionId, + snapshotId, + index: 0, + messages: full.snapshot.messages, + }); + fakeClient.emitMessage({ + type: "session_snapshot_end", + activeSessionId: command.activeSessionId, + snapshotId, + chunkCount: 1, + lastEventSequence: full.lastEventSequence, + lastEventCursor: full.lastEventCursor, + }); + }); + return { + ...full, + snapshot: { ...full.snapshot, messages: [] }, + snapshotStream: { id: snapshotId, messageCount: 1, targetChunkBytes: 512 * 1024 }, + }; + }; + const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-1"); + + await expect(connection.attach()).resolves.toBeUndefined(); + await expect(connection.getInitialSnapshot()).resolves.toMatchObject({ + messages: [{ role: "user", content: "generation-2", timestamp: 2 }], + }); + const attachRequests = fakeClient.requests.filter((request) => request.type === "attach"); + expect(attachRequests).toHaveLength(2); + expect(attachRequests[1]?.snapshotGenerationNonce).not.toBe(attachRequests[0]?.snapshotGenerationNonce); + }); + it("aliases a failed command reattach to one fresh attach and resolves the switch caller", async () => { const fakeClient = new FakeDaemonClient(); const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-source", { @@ -5687,6 +5810,23 @@ describe("DaemonAgentConnection", () => { ); }); + it("does not attach-fallback after an invalid reattach response", async () => { + const fakeClient = new FakeDaemonClient(); + const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-source"); + await connection.attach(); + fakeClient.requests.length = 0; + fakeClient.switchSessionActiveSessionId = "active-target"; + fakeClient.reattachResultFactory = () => createAttachResult("active-wrong", "client-1", undefined, 1); + fakeClient.attachResultFactory = () => { + throw new Error("fallback must not run"); + }; + + await expect(connection.switchSession("/tmp/target.jsonl")).rejects.toThrow( + "Daemon returned an invalid session snapshot", + ); + expect(fakeClient.requests.map((request) => request.type)).toEqual(["switch_session", "reattach"]); + }); + it("rejects one failed snapshot without interrupting another session on the shared client", async () => { const fakeClient = new FakeDaemonClient(); const sibling = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-2"); @@ -5742,25 +5882,123 @@ describe("DaemonAgentConnection", () => { await sibling.dispose(); }); - it("does not start runtime recovery from an unfenced failure-before-begin frame", async () => { + it.each(["replacement", "resync"] as const)( + "recovers a current %s failure-before-begin frame without an assembly", + async (purpose) => { + const fakeClient = new FakeDaemonClient(); + const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-1"); + await connection.attach(); + const events: AgentConnectionEvent[] = []; + connection.subscribe((event) => { + events.push(event); + }); + const snapshotId = `pre-begin-${purpose}`; + + fakeClient.emitMessage({ + type: "session_snapshot_failed", + activeSessionId: "active-1", + snapshotId, + purpose, + error: `${purpose} preparation failed before begin`, + }); + + await vi.waitFor(() => + expect( + events.some( + (event) => event.type === (purpose === "replacement" ? "session_replaced" : "session_resynced"), + ), + ).toBe(true), + ); + expect(fakeClient.requests.filter((request) => request.type === "attach")).toHaveLength(2); + expect((connection as unknown as { runtimeSnapshotAttempt?: unknown }).runtimeSnapshotAttempt).toBeUndefined(); + expect((connection as unknown as { snapshotAssemblies: Map }).snapshotAssemblies.size).toBe( + 0, + ); + + fakeClient.emitMessage({ + type: "session_snapshot_failed", + activeSessionId: "active-1", + snapshotId, + purpose, + error: "duplicate stale failure", + }); + await Promise.resolve(); + expect(fakeClient.requests.filter((request) => request.type === "attach")).toHaveLength(2); + await connection.dispose(); + }, + ); + + it("ignores a pre-begin runtime failure queued under a stale attachment epoch", async () => { const fakeClient = new FakeDaemonClient(); const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-1"); await connection.attach(); + let releaseRoute!: () => void; + const routeGate = new Promise((resolve) => { + releaseRoute = resolve; + }); + const internals = connection as unknown as { + attachmentEpoch: number; + replacementMessageTail?: Promise; + runtimeSnapshotAttempt?: unknown; + }; + internals.replacementMessageTail = routeGate; fakeClient.emitMessage({ type: "session_snapshot_failed", activeSessionId: "active-1", - snapshotId: "unfenced-runtime-failure", + snapshotId: "stale-epoch-failure", purpose: "resync", - error: "private unfenced failure", + error: "stale preparation failed", }); - await Promise.resolve(); + internals.attachmentEpoch++; + releaseRoute(); + await new Promise((resolve) => setImmediate(resolve)); expect(fakeClient.requests.filter((request) => request.type === "attach")).toHaveLength(1); - expect((connection as unknown as { runtimeSnapshotAttempt?: unknown }).runtimeSnapshotAttempt).toBeUndefined(); - expect((connection as unknown as { snapshotAssemblies: Map }).snapshotAssemblies.size).toBe(0); + expect(internals.runtimeSnapshotAttempt).toBeUndefined(); + await connection.dispose(); }); + it.each(["replacement", "resync"] as const)( + "terminates one failed retry for a pre-begin %s failure and fences later failures", + async (purpose) => { + const fakeClient = new FakeDaemonClient(); + const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-1"); + await connection.attach(); + fakeClient.attachResultFactory = () => { + throw new Error("retry preparation also failed"); + }; + const events: AgentConnectionEvent[] = []; + connection.subscribe((event) => { + events.push(event); + }); + + fakeClient.emitMessage({ + type: "session_snapshot_failed", + activeSessionId: "active-1", + snapshotId: `terminal-${purpose}`, + purpose, + error: "pre-begin preparation failed", + }); + await vi.waitFor(() => expect(events.filter((event) => event.type === "closed")).toHaveLength(1)); + expect(fakeClient.requests.filter((request) => request.type === "attach")).toHaveLength(2); + + for (const snapshotId of [`terminal-${purpose}`, `late-${purpose}`]) { + fakeClient.emitMessage({ + type: "session_snapshot_failed", + activeSessionId: "active-1", + snapshotId, + purpose, + error: "late terminal failure", + }); + } + await Promise.resolve(); + expect(fakeClient.requests.filter((request) => request.type === "attach")).toHaveLength(2); + expect(events.filter((event) => event.type === "closed")).toHaveLength(1); + await connection.dispose(); + }, + ); + it("binds a failure-before-begin frame to the exact current attach attempt", async () => { const fakeClient = new FakeDaemonClient(); let releaseAttach!: () => void; diff --git a/packages/coding-agent/test/daemon-mode.test.ts b/packages/coding-agent/test/daemon-mode.test.ts index fd50ed3344..05a7125ddf 100644 --- a/packages/coding-agent/test/daemon-mode.test.ts +++ b/packages/coding-agent/test/daemon-mode.test.ts @@ -3292,7 +3292,7 @@ describe("daemon mode helpers", () => { expect(client.catchupActiveSessionIds).toEqual(new Set()); }); - it("terminates a failed catch-up attempt without an unbounded self-retry", async () => { + it("emits one worker pre-begin catch-up failure without an unbounded self-retry", async () => { const daemon = new AgentDaemon("/tmp/prime-agent-test.sock", { defaultSessionConfig: { agentDir: "/tmp/prime-agent-test-agent", cwd: "/tmp" }, createRuntime: async () => { @@ -3355,6 +3355,18 @@ describe("daemon mode helpers", () => { }), expect.objectContaining({ type: "session_resynced", activeSessionId: secondState.activeSessionId }), ]); + expect( + messages.filter( + (message) => + message.activeSessionId === firstState.activeSessionId && message.type === "session_snapshot_failed", + ), + ).toHaveLength(1); + expect( + messages.some( + (message) => + message.activeSessionId === firstState.activeSessionId && message.type === "session_snapshot_begin", + ), + ).toBe(false); }); it("clears a scheduled catch-up retry when the client disconnects", async () => { diff --git a/packages/coding-agent/test/daemon-protocol.test.ts b/packages/coding-agent/test/daemon-protocol.test.ts index 455a6a67c2..526f00eeab 100644 --- a/packages/coding-agent/test/daemon-protocol.test.ts +++ b/packages/coding-agent/test/daemon-protocol.test.ts @@ -15,6 +15,7 @@ import { DAEMON_PROTOCOL_VERSION, DAEMON_SCHEMA_ID, DAEMON_SCHEMA_REVISION, + DAEMON_SNAPSHOT_GENERATION_NONCE_MIN_SCHEMA_REVISION, DAEMON_SUPERVISOR_SERVER_CAPABILITIES, DAEMON_SUPPORTED_CLIENT_CAPABILITIES, type DaemonCommand, @@ -171,6 +172,22 @@ describe("daemon protocol helpers", () => { expect(DAEMON_COMMAND_COMPATIBILITY.set_rlm_max_depth).toEqual({ minProtocol: 7, minSchemaRevision: 11 }); }); + it("capability- and schema-gates fresh snapshot generation nonces", () => { + expect(DAEMON_SCHEMA_REVISION).toBe(28); + expect(DAEMON_SNAPSHOT_GENERATION_NONCE_MIN_SCHEMA_REVISION).toBe(28); + expect( + getDaemonCommandCompatibilities({ + type: "attach", + activeSessionId: "active-1", + snapshotGenerationNonce: "nonce-1", + }), + ).toEqual([ + { minProtocol: 7, minSchemaRevision: 28, capability: "snapshot_generation_nonce_v1" }, + { minProtocol: 7 }, + ]); + expect(DAEMON_DEFAULT_SERVER_CAPABILITIES).toContain("snapshot_generation_nonce_v1"); + }); + it("schema-gates session commands that carry the telemetry policy", () => { expect(getDaemonCommandCompatibilities({ type: "create", config: { cwd: "/tmp" } })).toEqual([ { minProtocol: 7 }, diff --git a/packages/coding-agent/test/daemon-supervisor-process.test.ts b/packages/coding-agent/test/daemon-supervisor-process.test.ts index 346d83d126..c919d82d59 100644 --- a/packages/coding-agent/test/daemon-supervisor-process.test.ts +++ b/packages/coding-agent/test/daemon-supervisor-process.test.ts @@ -85,13 +85,17 @@ function spawnSupervisor( extraEnv: NodeJS.ProcessEnv = {}, ): ChildProcess { daemonSockets.add(socketPath); + const inheritedEnv = { ...process.env }; + for (const name of Object.keys(inheritedEnv)) { + if (name.startsWith("PRIME_AGENT_INTERNAL_DAEMON_") || name.startsWith("RLM_")) delete inheritedEnv[name]; + } const child = spawn( process.execPath, [tsxPath, cliPath, "--mode", "daemon", "--daemon-socket", socketPath, "--offline", ...extraArgs], { cwd, env: { - ...process.env, + ...inheritedEnv, ...extraEnv, [ENV_AGENT_DIR]: agentDir, PI_OFFLINE: "1", @@ -209,6 +213,7 @@ async function connectEventually(socketPath: string, child?: ChildProcess): Prom async function createSnapshotRetryProxy( proxyPath: string, targetPath: string, + fault: "corrupt-first-chunk" | "corrupt-every-first-chunk" | "drop-first-end" = "corrupt-first-chunk", ): Promise<{ server: Server; attachRequests: () => number; @@ -223,6 +228,7 @@ async function createSnapshotRetryProxy( let firstTransferId: string | undefined; let firstChunk: Buffer | undefined; let firstEnd: Buffer | undefined; + const corruptedTransferIds = new Set(); downstream.on("data", (chunk: Buffer) => { requestBuffer = Buffer.concat([requestBuffer, chunk]); while (true) { @@ -260,16 +266,17 @@ async function createSnapshotRetryProxy( if (!observedTransferIds.includes(parsed.snapshotId)) observedTransferIds.push(parsed.snapshotId); firstTransferId ??= parsed.snapshotId; } - if ( - parsed.type === "session_snapshot_chunk" && - typeof parsed.snapshotId === "string" && - parsed.snapshotId === firstTransferId && - !firstChunk - ) { - firstChunk = original; - parsed.snapshotId = `${parsed.snapshotId}-corrupted`; - downstream.write(`${JSON.stringify(parsed)}\n`); - continue; + if (parsed.type === "session_snapshot_chunk" && typeof parsed.snapshotId === "string") { + if (parsed.snapshotId === firstTransferId && !firstChunk) firstChunk = original; + const corruptsThisTransfer = + (fault === "corrupt-first-chunk" && parsed.snapshotId === firstTransferId) || + fault === "corrupt-every-first-chunk"; + if (corruptsThisTransfer && !corruptedTransferIds.has(parsed.snapshotId)) { + corruptedTransferIds.add(parsed.snapshotId); + parsed.snapshotId = `${parsed.snapshotId}-corrupted`; + downstream.write(`${JSON.stringify(parsed)}\n`); + continue; + } } if ( parsed.type === "session_snapshot_end" && @@ -277,6 +284,7 @@ async function createSnapshotRetryProxy( parsed.snapshotId === firstTransferId ) { firstEnd = original; + if (fault === "drop-first-end") continue; } downstream.write(original); if ( @@ -310,6 +318,67 @@ async function createSnapshotRetryProxy( }; } +async function createSnapshotDrainAbortProxy( + proxyPath: string, + targetPath: string, +): Promise<{ server: Server; aborted: Promise }> { + let resolveAborted!: (bufferedBytes: number) => void; + const aborted = new Promise((resolve) => { + resolveAborted = resolve; + }); + const server = createServer((downstream: Socket) => { + const upstream = createConnection(targetPath); + let responseBuffer = Buffer.alloc(0); + let paused = false; + downstream.on("data", (chunk: Buffer) => { + upstream.write(chunk); + }); + upstream.on("data", (chunk: Buffer) => { + if (paused) return; + responseBuffer = Buffer.concat([responseBuffer, chunk]); + while (true) { + const newline = responseBuffer.indexOf(0x0a); + if (newline < 0) break; + const original = Buffer.from(responseBuffer.subarray(0, newline + 1)); + responseBuffer = responseBuffer.subarray(newline + 1); + downstream.write(original); + let type: unknown; + try { + type = (JSON.parse(original.toString("utf8")) as { type?: unknown }).type; + } catch { + continue; + } + if (type !== "session_snapshot_begin") continue; + paused = true; + upstream.pause(); + let settled = false; + const abort = () => { + if (settled) return; + settled = true; + clearTimeout(deadline); + const bufferedBytes = responseBuffer.length + upstream.readableLength; + resolveAborted(bufferedBytes); + upstream.destroy(); + downstream.destroy(); + }; + const deadline = setTimeout(abort, 1_000); + deadline.unref(); + upstream.once("readable", abort); + return; + } + }); + upstream.on("error", () => downstream.destroy()); + downstream.on("error", () => upstream.destroy()); + upstream.on("close", () => downstream.destroy()); + downstream.on("close", () => upstream.destroy()); + }); + await new Promise((resolveListen, reject) => { + server.once("error", reject); + server.listen(proxyPath, () => resolveListen()); + }); + return { server, aborted }; +} + async function waitForSocketGone(socketPath: string): Promise { const deadline = Date.now() + 10_000; while (Date.now() < deadline) { @@ -333,6 +402,31 @@ function requireSummary(responseData: unknown): SessionSummary { return responseData as SessionSummary; } +function createSnapshotSessionFile(agentDir: string, projectDir: string, label: string): string { + const sessionManager = SessionManager.create(projectDir, join(agentDir, "sessions")); + sessionManager.appendMessage({ role: "user", content: label, timestamp: 1 }); + sessionManager.appendMessage({ + role: "assistant", + content: [{ type: "text", text: "fixture complete" }], + api: "openai-responses", + provider: "faux", + model: "faux", + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop", + timestamp: 2, + }); + const sessionFile = sessionManager.getSessionFile(); + if (!sessionFile) throw new Error("Snapshot fixture did not persist"); + return sessionFile; +} + function requireSessionList(responseData: unknown): SessionSummary[] { if (!responseData || typeof responseData !== "object" || !("sessions" in responseData)) { throw new Error("Missing daemon session list"); @@ -632,6 +726,184 @@ describe("daemon supervisor resident workers", () => { await waitForSocketGone(socketPath); }, 60_000); + it("retries a real worker snapshot whose first generation never sends an end frame", async () => { + const root = tempDir(); + const agentDir = join(root, "agent"); + const projectDir = join(root, "project"); + const suffix = `${process.pid}-${randomUUID().slice(0, 8)}`; + const socketPath = join(tmpdir(), `prime-snapshot-missing-end-${suffix}.sock`); + const proxyPath = join(tmpdir(), `prime-snapshot-missing-end-proxy-${suffix}.sock`); + mkdirSync(projectDir, { recursive: true }); + const fixtureLabel = "real missing-end snapshot fixture"; + const sessionFile = createSnapshotSessionFile(agentDir, projectDir, fixtureLabel); + + const supervisor = spawnSupervisor(agentDir, socketPath, projectDir); + const creator = await connectEventually(socketPath, supervisor); + const created = await creator.request({ + type: "create", + sessionPath: sessionFile, + config: { + cwd: projectDir, + agentDir, + sessionDir: join(agentDir, "sessions"), + noTools: true, + noExtensions: true, + }, + }); + if (!created.success) throw new Error(created.error); + const summary = requireSummary(created.data); + if (!summary.activeSessionId || !summary.workerPid) throw new Error("Missing-end worker was incomplete"); + workerPids.add(summary.workerPid); + creator.close(); + + const proxy = await createSnapshotRetryProxy(proxyPath, socketPath, "drop-first-end"); + const client = new DaemonClient(proxyPath); + try { + await client.connect(3_000); + await client.waitForHello(3_000); + const connection = await DaemonAgentConnection.attach(client, summary.activeSessionId, { + supportsExtensionUi: false, + snapshotTimeoutMs: 250, + }); + await expect(connection.getInitialSnapshot()).resolves.toMatchObject({ + messages: expect.arrayContaining([expect.objectContaining({ role: "user", content: fixtureLabel })]), + }); + expect(proxy.attachRequests()).toBe(2); + expect(proxy.transferIds()).toHaveLength(2); + expect(new Set(proxy.transferIds()).size).toBe(2); + await connection.dispose(); + } finally { + client.close(); + await new Promise((resolveClose) => proxy.server.close(() => resolveClose())); + } + + const shutdownClient = await connectEventually(socketPath); + await shutdownClient.request({ type: "shutdown" }); + shutdownClient.close(); + await waitForProcessGone(summary.workerPid); + workerPids.delete(summary.workerPid); + await waitForSocketGone(socketPath); + }, 60_000); + + it("terminates after two corrupted real worker snapshot generations without self-requeue", async () => { + const root = tempDir(); + const agentDir = join(root, "agent"); + const projectDir = join(root, "project"); + const suffix = `${process.pid}-${randomUUID().slice(0, 8)}`; + const socketPath = join(tmpdir(), `prime-snapshot-terminal-${suffix}.sock`); + const proxyPath = join(tmpdir(), `prime-snapshot-terminal-proxy-${suffix}.sock`); + mkdirSync(projectDir, { recursive: true }); + const sessionFile = createSnapshotSessionFile(agentDir, projectDir, "real terminal snapshot fixture"); + + const supervisor = spawnSupervisor(agentDir, socketPath, projectDir); + const creator = await connectEventually(socketPath, supervisor); + const created = await creator.request({ + type: "create", + sessionPath: sessionFile, + config: { + cwd: projectDir, + agentDir, + sessionDir: join(agentDir, "sessions"), + noTools: true, + noExtensions: true, + }, + }); + if (!created.success) throw new Error(created.error); + const summary = requireSummary(created.data); + if (!summary.activeSessionId || !summary.workerPid) throw new Error("Terminal worker was incomplete"); + workerPids.add(summary.workerPid); + creator.close(); + + const proxy = await createSnapshotRetryProxy(proxyPath, socketPath, "corrupt-every-first-chunk"); + const client = new DaemonClient(proxyPath); + try { + await client.connect(3_000); + await client.waitForHello(3_000); + await expect( + DaemonAgentConnection.attach(client, summary.activeSessionId, { + supportsExtensionUi: false, + snapshotTimeoutMs: 1_000, + }), + ).rejects.toThrow(); + expect(proxy.attachRequests()).toBe(2); + expect(proxy.transferIds()).toHaveLength(2); + expect(new Set(proxy.transferIds()).size).toBe(2); + } finally { + client.close(); + await new Promise((resolveClose) => proxy.server.close(() => resolveClose())); + } + + const shutdownClient = await connectEventually(socketPath); + await shutdownClient.request({ type: "shutdown" }); + shutdownClient.close(); + await waitForProcessGone(summary.workerPid); + workerPids.delete(summary.workerPid); + await waitForSocketGone(socketPath); + }, 60_000); + + it("releases a real public snapshot drain when the client detaches", async () => { + const root = tempDir(); + const agentDir = join(root, "agent"); + const projectDir = join(root, "project"); + const suffix = `${process.pid}-${randomUUID().slice(0, 8)}`; + const socketPath = join(tmpdir(), `prime-snapshot-drain-${suffix}.sock`); + const proxyPath = join(tmpdir(), `prime-snapshot-drain-proxy-${suffix}.sock`); + mkdirSync(projectDir, { recursive: true }); + const largeFixture = `drain:${"x".repeat(8 * 1024 * 1024)}`; + const sessionFile = createSnapshotSessionFile(agentDir, projectDir, largeFixture); + + const supervisor = spawnSupervisor(agentDir, socketPath, projectDir); + const creator = await connectEventually(socketPath, supervisor); + const created = await creator.request({ + type: "create", + sessionPath: sessionFile, + config: { + cwd: projectDir, + agentDir, + sessionDir: join(agentDir, "sessions"), + noTools: true, + noExtensions: true, + }, + }); + if (!created.success) throw new Error(created.error); + const summary = requireSummary(created.data); + if (!summary.activeSessionId || !summary.workerPid) throw new Error("Drain worker was incomplete"); + workerPids.add(summary.workerPid); + creator.close(); + + const proxy = await createSnapshotDrainAbortProxy(proxyPath, socketPath); + const blockedClient = new DaemonClient(proxyPath); + try { + await blockedClient.connect(3_000); + await blockedClient.waitForHello(3_000); + const failedAttach = DaemonAgentConnection.attach(blockedClient, summary.activeSessionId, { + supportsExtensionUi: false, + snapshotTimeoutMs: 3_000, + }); + const bufferedBytes = await proxy.aborted; + expect(bufferedBytes).toBeGreaterThan(0); + await expect(failedAttach).rejects.toThrow(); + } finally { + blockedClient.close(); + await new Promise((resolveClose) => proxy.server.close(() => resolveClose())); + } + + const recoveryClient = await connectEventually(socketPath); + const connection = await DaemonAgentConnection.attach(recoveryClient, summary.activeSessionId, { + supportsExtensionUi: false, + snapshotTimeoutMs: 5_000, + }); + const recovered = await connection.getInitialSnapshot(); + const recoveredUser = recovered.messages.find((message) => message.role === "user"); + expect(recoveredUser?.content).toBe(largeFixture); + await connection.dispose(); + await recoveryClient.request({ type: "shutdown" }); + recoveryClient.close(); + await waitForProcessGone(summary.workerPid); + workerPids.delete(summary.workerPid); + await waitForSocketGone(socketPath); + }, 60_000); + it("keeps client-owned workers hidden and removes them without archiving", async () => { const root = tempDir(); const agentDir = join(root, "agent"); diff --git a/packages/coding-agent/test/daemon-version-compatibility.test.ts b/packages/coding-agent/test/daemon-version-compatibility.test.ts index 51940da6fb..056f0ec583 100644 --- a/packages/coding-agent/test/daemon-version-compatibility.test.ts +++ b/packages/coding-agent/test/daemon-version-compatibility.test.ts @@ -15,7 +15,7 @@ import type { SessionSummary } from "../src/modes/daemon/daemon-session-list.js" * The artifact must be built from the pinned git tag v0.8.1, not a protocol mock. */ const historicalCli = process.env.PRIME_AGENT_HISTORICAL_081_CLI; -const currentCli = resolve(__dirname, "../dist/bundle/cli.js"); +const currentCli = process.env.PRIME_AGENT_CURRENT_CLI ?? resolve(__dirname, "../dist/bundle/cli.js"); const compiledCli = process.env.PRIME_AGENT_COMPILED_CLI; const processTests = historicalCli ? describe : describe.skip; const children = new Set(); @@ -24,14 +24,8 @@ const roots: string[] = []; function scrubbedEnvironment(agentDir: string): NodeJS.ProcessEnv { const environment = { ...process.env, [ENV_AGENT_DIR]: agentDir, PI_OFFLINE: "1" }; - for (const name of [ - "PRIME_AGENT_INTERNAL_DAEMON_WORKER", - "PRIME_AGENT_INTERNAL_DAEMON_WORKER_TOKEN", - "PRIME_AGENT_INTERNAL_DAEMON_WORKER_ACTIVE_SESSION_ID", - "PRIME_AGENT_INTERNAL_DAEMON_SUPERVISOR_SOCKET", - "PRIME_AGENT_INTERNAL_DAEMON_WORKER_RECOVERY_JOURNAL", - ]) { - delete environment[name]; + for (const name of Object.keys(environment)) { + if (name.startsWith("PRIME_AGENT_INTERNAL_DAEMON_") || name.startsWith("RLM_")) delete environment[name]; } return environment; } @@ -165,6 +159,7 @@ afterEach(async () => { processTests("stock v0.8.1 daemon entrypoint compatibility", () => { it.each([ + ["new supervisor adopts a new worker", currentCli, currentCli], ["new supervisor adopts a stock 0.8.1 worker", historicalCli!, currentCli], ["stock 0.8.1 supervisor adopts a new worker", currentCli, historicalCli!], ] as const)( @@ -282,9 +277,11 @@ performanceTests("compiled snapshot process performance", () => { const supervisor = launch(compiledCli!, agentDir, socketPath, projectDir); const client = await connectEventually(socketPath, supervisor); + const transferClient = await connectEventually(socketPath, supervisor); + const probeClient = await connectEventually(socketPath, supervisor); const created = await client.request({ type: "create", - sessionPath: sessionFile, + noSession: true, config: { cwd: projectDir, agentDir, sessionDir, noTools: true, noExtensions: true }, }); if (!created.success || !created.data || typeof created.data !== "object") { @@ -294,30 +291,95 @@ performanceTests("compiled snapshot process performance", () => { if (!summary.activeSessionId || !summary.workerPid || !supervisor.pid) { throw new Error("Compiled snapshot processes were incomplete"); } - const baselineRss = residentSetBytes(supervisor.pid) + residentSetBytes(summary.workerPid); + const connection = await DaemonAgentConnection.attach(client, summary.activeSessionId, { + supportsExtensionUi: false, + }); + let replacementTimeout: NodeJS.Timeout | undefined; + const replacement = new Promise((resolveReplacement, rejectReplacement) => { + const unsubscribe = connection.subscribe((event) => { + if (event.type !== "session_replaced") return; + unsubscribe(); + if (replacementTimeout) clearTimeout(replacementTimeout); + resolveReplacement(); + }); + replacementTimeout = setTimeout(() => { + unsubscribe(); + rejectReplacement(new Error("Compiled snapshot replacement did not finish")); + }, 30_000); + replacementTimeout.unref(); + }); + await connection.switchSession(sessionFile); + await replacement; + const initialLargeSnapshot = await connection.getInitialSnapshot(); + expect(initialLargeSnapshot.messages[0]).toMatchObject({ role: "user", content: largePrompt }); + + const corruptedCacheFiles = readdirSync(agentDir, { recursive: true, encoding: "utf8" }).filter( + (entry) => entry.includes("snapshot-cache") && entry.endsWith(".jsonl"), + ); + expect(corruptedCacheFiles.length).toBeGreaterThan(0); + for (const entry of corruptedCacheFiles) rmSync(join(agentDir, entry)); + + const baselineSupervisorRss = residentSetBytes(supervisor.pid); + const baselineWorkerRss = residentSetBytes(summary.workerPid); + const baselineRss = baselineSupervisorRss + baselineWorkerRss; + let peakSupervisorRss = baselineSupervisorRss; + let peakWorkerRss = baselineWorkerRss; let peakRss = baselineRss; - let maxEventLoopDelayMs = 0; + let maxClientEventLoopDelayMs = 0; + let maxWorkerRoundTripMs = 0; + let monitorActive = true; let previousTick = performance.now(); const monitor = setInterval(() => { const current = performance.now(); - maxEventLoopDelayMs = Math.max(maxEventLoopDelayMs, current - previousTick - 25); + maxClientEventLoopDelayMs = Math.max(maxClientEventLoopDelayMs, current - previousTick - 25); previousTick = current; - peakRss = Math.max(peakRss, residentSetBytes(supervisor.pid!) + residentSetBytes(summary.workerPid!)); + const supervisorRss = residentSetBytes(supervisor.pid!); + const workerRss = residentSetBytes(summary.workerPid!); + peakSupervisorRss = Math.max(peakSupervisorRss, supervisorRss); + peakWorkerRss = Math.max(peakWorkerRss, workerRss); + peakRss = Math.max(peakRss, supervisorRss + workerRss); }, 25); + const probe = (async () => { + while (monitorActive) { + const startedAt = performance.now(); + const state = await probeClient.request( + { type: "get_state", activeSessionId: summary.activeSessionId! }, + 3_000, + ); + if (!state.success) throw new Error(state.error); + maxWorkerRoundTripMs = Math.max(maxWorkerRoundTripMs, performance.now() - startedAt); + await new Promise((resolveYield) => setImmediate(resolveYield)); + } + })(); + let retriedConnection: DaemonAgentConnection | undefined; try { - const connection = await DaemonAgentConnection.attach(client, summary.activeSessionId, { + retriedConnection = await DaemonAgentConnection.attach(transferClient, summary.activeSessionId, { supportsExtensionUi: false, }); - const snapshot = await connection.getInitialSnapshot(); + const snapshot = await retriedConnection.getInitialSnapshot(); expect(snapshot.messages[0]).toMatchObject({ role: "user", content: largePrompt }); - await connection.dispose(); } finally { + monitorActive = false; clearInterval(monitor); + await probe; } - expect(peakRss - baselineRss).toBeLessThan(192 * 1024 * 1024); - expect(maxEventLoopDelayMs).toBeLessThan(500); + const rssDeltaBytes = peakRss - baselineRss; + console.info( + `snapshot process metrics: rssDeltaMiB=${(rssDeltaBytes / 1024 / 1024).toFixed(1)} ` + + `supervisorDeltaMiB=${((peakSupervisorRss - baselineSupervisorRss) / 1024 / 1024).toFixed(1)} ` + + `workerDeltaMiB=${((peakWorkerRss - baselineWorkerRss) / 1024 / 1024).toFixed(1)} ` + + `clientEventLoopDelayMs=${maxClientEventLoopDelayMs.toFixed(1)} ` + + `workerRoundTripMs=${maxWorkerRoundTripMs.toFixed(1)}`, + ); + expect(rssDeltaBytes).toBeLessThan(192 * 1024 * 1024); + expect(maxClientEventLoopDelayMs).toBeLessThan(500); + expect(maxWorkerRoundTripMs).toBeLessThan(1_000); + await retriedConnection?.dispose(); + await connection.dispose(); await client.request({ type: "shutdown" }); client.close(); + transferClient.close(); + probeClient.close(); await stopSupervisor(supervisor); }, 120_000); }); diff --git a/packages/coding-agent/test/session-worker-private-framing.test.ts b/packages/coding-agent/test/session-worker-private-framing.test.ts index 32f9a61973..bc78cc8da0 100644 --- a/packages/coding-agent/test/session-worker-private-framing.test.ts +++ b/packages/coding-agent/test/session-worker-private-framing.test.ts @@ -1,6 +1,7 @@ import { spawn } from "node:child_process"; import { once } from "node:events"; import { createServer, type Socket } from "node:net"; +import { performance } from "node:perf_hooks"; import { PassThrough } from "node:stream"; import { describe, expect, it } from "vitest"; import { @@ -46,6 +47,24 @@ describe("private worker framing", () => { ]); }); + it("decodes a frame fragmented into more than 100,000 one-byte chunks without quadratic queue work", () => { + const payload = Buffer.alloc(128 * 1024, 0x61); + const frame = encodePrivateFrame({ type: "snapshot", requestId: "heavily-fragmented" }, payload); + const decoder = new PrivateFrameDecoder(isTestHeader); + const frames = []; + const started = performance.now(); + + for (let offset = 0; offset < frame.length; offset++) { + frames.push(...decoder.push(frame.subarray(offset, offset + 1))); + } + const elapsedMs = performance.now() - started; + decoder.finish(); + + expect(frames).toEqual([{ header: { type: "snapshot", requestId: "heavily-fragmented" }, payload }]); + expect(decoder.coalescedBytes).toBeLessThanOrEqual(frame.length); + expect(elapsedMs).toBeLessThan(5_000); + }, 10_000); + it("coalesces a 36 MiB fragmented payload at most once", () => { const payload = Buffer.alloc(36 * 1024 * 1024, 0x61); const frame = encodePrivateFrame({ type: "snapshot", requestId: "large" }, payload); diff --git a/packages/coding-agent/test/snapshot-transcript-cache.test.ts b/packages/coding-agent/test/snapshot-transcript-cache.test.ts index 4b0ceda6ed..b750803314 100644 --- a/packages/coding-agent/test/snapshot-transcript-cache.test.ts +++ b/packages/coding-agent/test/snapshot-transcript-cache.test.ts @@ -88,6 +88,15 @@ describe("snapshot transcript cache", () => { const end = cache.waitForChunk(1); cache.markComplete(); await expect(end).resolves.toBeUndefined(); + const reEnveloped = await cache.waitForTransferChunk("snapshot-fresh", 0); + expect(Array.isArray(reEnveloped)).toBe(true); + const reEnvelopedParts = reEnveloped as readonly Buffer[]; + expect(reEnvelopedParts[1]?.buffer).toBe(encoded.buffer); + expect(JSON.parse(Buffer.concat(reEnvelopedParts).toString("utf8"))).toMatchObject({ + snapshotId: "snapshot-fresh", + index: 0, + messages: [], + }); cache.dispose(); }); @@ -334,6 +343,24 @@ describe("snapshot transcript cache", () => { expect(() => createSnapshotCacheProcessRoot(symlinkParent)).toThrow("private directory"); }); + it("prepares 10,000 small messages with linear scaling", async () => { + const cacheRoot = tempDir(); + const measure = async (count: number) => { + const started = performance.now(); + const payload = await prepareSnapshotTranscriptPayload({ messages: messages(count, 8), cacheRoot }); + const elapsedMs = performance.now() - started; + payload.dispose(); + return elapsedMs; + }; + + await measure(1_000); + const halfElapsedMs = await measure(5_000); + const fullElapsedMs = await measure(10_000); + + expect(fullElapsedMs).toBeLessThan(5_000); + expect(fullElapsedMs).toBeLessThan(halfElapsedMs * 3 + 250); + }, 15_000); + it("bounds 36 MiB single and multi-message preparation and shares one payload across three purposes", async () => { const cacheRoot = tempDir(); const run = async (source: AgentMessage[]) => { 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 e14dded443..ab8eabd0ac 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 @@ -511,6 +511,42 @@ describe("ENG-4602 snapshot transfer containment", () => { expect(worker.snapshotGenerations.has(activeSessionId)).toBe(false); }); + it("uses routed chunk metadata without reparsing a large canonical payload", () => { + const supervisor = new DaemonSupervisor("/tmp/eng-4602-supervisor-routed-chunk.sock", { + defaultSessionConfig: { agentDir: "/tmp", cwd: "/tmp" }, + }); + const { worker } = workerHarness(); + registerWorker(supervisor, worker); + const messages: AgentMessage[] = [{ role: "user", content: "x".repeat(1024 * 1024), timestamp: 1 }]; + const frames = snapshotFrames(messages); + const internals = supervisor as unknown as { + handleWorkerFrame(worker: WorkerHarness, frame: PrivateFrame): void; + }; + internals.handleWorkerFrame(worker, frame(frames.begin)); + const payload = Buffer.from(`${JSON.stringify(frames.chunk)}\n`); + const parse = vi.spyOn(JSON, "parse"); + try { + internals.handleWorkerFrame(worker, { + header: { + kind: "outbound", + outboundType: "session_snapshot_chunk", + activeSessionId, + snapshotId, + payloadEncoding: "jsonl", + snapshotChunkIndex: 0, + snapshotChunkMessageCount: 1, + }, + payload, + }); + expect(parse).not.toHaveBeenCalled(); + } finally { + parse.mockRestore(); + } + expect(worker.transcriptCaches.get(activeSessionId)?.chunkCount).toBe(1); + internals.handleWorkerFrame(worker, frame(frames.end)); + worker.transcriptCaches.get(activeSessionId)?.dispose(); + }); + it("keeps a multi-session worker connected after a scoped snapshot failure frame", () => { const supervisor = new DaemonSupervisor("/tmp/eng-4602-supervisor-failure.sock", { defaultSessionConfig: { agentDir: "/tmp", cwd: "/tmp" }, @@ -923,6 +959,49 @@ describe("ENG-4602 snapshot transfer containment", () => { expect(replaced.worker.transcriptCaches.get(activeSessionId)?.snapshotId).toBe("snapshot-4602-new"); }); + it("emits one supervisor pre-begin catch-up failure without a self-requeue", async () => { + const supervisor = new DaemonSupervisor("/tmp/eng-4602-supervisor-pre-begin.sock", { + defaultSessionConfig: { agentDir: "/tmp", cwd: "/tmp" }, + descriptorDir: "/tmp/eng-4602-supervisor-pre-begin-state", + }); + const socket = new PassThrough(); + socket.resume(); + const client = socketClient("pre-begin", socket); + const records: DaemonOutbound[] = []; + const preparationError = new Error("supervisor snapshot preparation failed"); + const attachClient = vi.fn(async () => { + throw preparationError; + }); + const internals = supervisor as unknown as { + clients: Set; + attachClient: typeof attachClient; + writeSnapshotRecord(client: DaemonSocketClient, message: DaemonOutbound): Promise; + queueCatchup(client: DaemonSocketClient, activeSessionId: string, purpose: "replacement" | "resync"): void; + catchUpClient(client: DaemonSocketClient): Promise; + }; + internals.clients.add(client); + internals.attachClient = attachClient; + internals.writeSnapshotRecord = async (_client, message) => { + records.push(message); + return true; + }; + + internals.queueCatchup(client, activeSessionId, "resync"); + await internals.catchUpClient(client); + + expect(attachClient).toHaveBeenCalledOnce(); + expect(records).toEqual([ + expect.objectContaining({ + type: "session_snapshot_failed", + activeSessionId, + error: preparationError.message, + purpose: "resync", + }), + ]); + expect(client.catchupActiveSessionIds).toEqual(new Set()); + socket.destroy(); + }); + it("emits one terminal failure without a supervisor-side retry loop", async () => { const supervisor = new DaemonSupervisor("/tmp/eng-4602-supervisor-bounded-retry.sock", { defaultSessionConfig: { agentDir: "/tmp", cwd: "/tmp" },