fix(agent): serialize detached managed snapshots

docs/non-compaction-retry-policy.md

@@ -42,22 +42,20 @@ Current retryable inputs are regex/string-classified:
 - provider-suggested retry wording, including OpenAI `retry your request` failures
 - network/connection/socket failures, refused/closed connections, upstream connect/reset-before-headers, socket hang up, timeout/timed out, fetch failed, terminated, retry delay wording, and unexpected socket close messages
 - canonical idle-stream watchdog stalls (`stream stalled while waiting for the next event`); in the legacy single-model path these remain retryable but use the bounded `retry.maxRetries` budget
-- local snapshot failures (`errorKind: "local_snapshot_failure"`, or the stable `Managed fallback attempt could not produce a serializable event snapshot` message prefix for restored sessions): the staged managed attempt was discarded before publication, so a content-free same-model re-issue is replay-safe
+- canonical local snapshot failure classification (`errorKind: "local_snapshot_failure"`, or the stable `Managed fallback attempt could not produce a serializable event snapshot` message prefix for restored sessions) is recognized, but only so the failure can be routed to its immediate-surface policy below — it is never re-issued
 
 Managed fallback uses structured transport facts and typed provider error codes when available. A structured classification of `other` becomes the bounded `unknown` fallback class; error prose cannot promote it to quota or transient. Regex classification is retained only as a legacy fallback.
 
-### Local snapshot failures (bounded same-model session retry)
+### Local snapshot failures (surface immediately, no retry)
 
-`local_snapshot_failure` is a local machinery fault, not provider evidence, so it gets a dedicated session retry class that deliberately bypasses the managed provider-fallback chain:
+`local_snapshot_failure` is a local machinery fault, not provider evidence. The retained producer shape is deterministic, so re-streaming the same request only reproduces the same local defect; it is surfaced immediately instead of being amplified across identical retries:
 
-- Bounded by `retry.maxRetries` (same budget in managed and single-model paths); on exhaustion the original local diagnostic surfaces unchanged.
+- Surfaces immediately with the original producer-boundary diagnostic, regardless of `retry.*` settings.
 - Never charges the fallback controller (the started attempt's provisional charge is discarded), never advances models, never emits `model_fallback_switched`, and never mutates or rotates credentials.
-- Only content-free failures are eligible; a failure that carries visible or tool content surfaces immediately.
-- `retry.enabled: false` disables this retry even when a managed fallback chain is configured — the managed chain keeps its own availability policy, but the local-snapshot path is a session retry governed by `retry.*` settings.
 
 ### Local buffer overflows (surface immediately, no retry)
 
-`local_buffer_overflow` (`errorKind`, or the stable `Managed fallback attempt exceeded the provisional event buffer limit` message prefix for restored sessions) is the sibling local staging fault: the provisional managed-attempt buffer exceeded its cap. Unlike snapshot failures, re-streaming the same request reproduces the same oversized response, so it is never retried:
+`local_buffer_overflow` (`errorKind`, or the stable `Managed fallback attempt exceeded the provisional event buffer limit` message prefix for restored sessions) is the sibling local staging fault: the provisional managed-attempt buffer exceeded its cap. Like snapshot failures, re-streaming the same request reproduces the same oversized response, so it is never retried:
 
 - Surfaces immediately with the original local diagnostic, regardless of `retry.*` settings.
 - Like snapshot failures, it never charges the fallback controller (the started attempt's provisional charge is discarded), never advances models, never emits `model_fallback_switched`, and never mutates or rotates credentials.
@@ -74,7 +72,7 @@ Session state used by retry:
 Flow (`#handleRetryableError`):
 
 1. Read `retry` settings group.
-2. If `retry.enabled === false`, stop immediately (`false`, no retry started). This opt-out also covers local snapshot failures on managed chains; other managed provider-fallback failures keep their own chain policy.
+2. If `retry.enabled === false`, stop immediately (`false`, no retry started). Managed provider-fallback failures keep their own chain policy; local snapshot and buffer-overflow failures surface immediately regardless of this setting.
 3. Increment `#retryAttempt`.
 4. Create `#retryPromise` once (first attempt in a chain).
 5. In the legacy single-model path, ordinary transient errors retry without an attempt limit, while canonical idle-stream watchdog stalls and unknown/no-code errors stop after `retry.maxRetries`. Managed fallback instead uses its controller's per-entry `fallback.maxAttempts` budget.

packages/agent/CHANGELOG.md

@@ -4,7 +4,7 @@
 
 ### Fixed
 
-- Managed fallback snapshot failures now retain a typed `local_snapshot_failure` error kind when they reach the agent terminal event, so session retry policy can auto-recover them with a bounded same-model retry instead of treating the local failure as provider fallback evidence or surfacing it as terminal.
+- Managed fallback now validates and byte-measures the detached event snapshot rather than trusting the live payload's JSON result. Custom payload classes whose prototype `toJSON()` hides bigint state are sanitized after `structuredClone` removes that serializer, so every accepted snapshot stays detached, JSON-serializable, and bounded; residual typed `local_snapshot_failure` diagnostics remain outside provider fallback authority and surface without deterministic retry amplification.
 - Managed fallback buffer overflows now retain a typed `local_buffer_overflow` error kind on the terminal assistant message, so session retry policy surfaces them immediately without provider-fallback attribution instead of admitting them to the bounded `unknown` retry class.
 - Managed local-failure diagnostics: `ManagedAttemptSnapshotError` and `ManagedAttemptBufferOverflowError` now carry a stable `stage` discriminator naming the exact rejecting site (`shell.role`, `shell.content`, `event.snapshot`, `event.contentIndex`, `event.delta`, `event.content`, `event.toolcall`, `event.done.reason`, `event.error.reason`, `event.unknownType`, `staging.losslessSnapshot`, `staging.measure`, `staging.sanitize`, `staging.overflow`, `overflow.preMeasure`, `overflow.staged`), and the run-loop failure boundary emits ONE bounded shape-only `logger.warn` per stream invocation (stage, error kind, model, provider, snapshot mode, staged event count/bytes, and content block count for the content stage). The diagnostic is gated on the module-private local error identities and its stage is whitelisted against the closed vocabulary, so neither a foreign error that self-labels a local failure kind nor an in-module regression can route arbitrary text into the log; it never records raw text, thinking, tool arguments, or any provider payload, and the user-facing message string is unchanged so session-side classification keeps matching. Previously all 14 rejecting sites shared one static message, leaving no way to identify which provider shape a normalizer must be taught to accept.
 - A turn whose tool arguments arrive flagged `escapedNonAsciiArguments` is now resampled instead of being reported as a tool failure: the defective assistant turn is dropped from history and the request is re-issued, up to twice per turn, before the terminal per-call rejection takes over. Hand-spelled `\uXXXX` arguments decode into valid-looking but silently wrong text (observed as garbled Hangul in `ask` prompts) and no post-parse repair can recover them, but the defect is a wire-format accident that resampling clears - surfacing it as a tool error instead burned the whole turn and fed the literal escape syntax back into the context the model samples from next. Scoped to the non-managed session path, matching the existing `invalid_prompt` and reasoning-content repairs; managed fallback keeps owning its own retry policy.

packages/agent/src/agent-loop.ts

@@ -144,9 +144,8 @@ class ManagedAttemptBufferOverflowError extends Error {
  * or status, so managed fallback classification never treats it as a provider
  * retry trigger — it never burns the fallback chain, advances models, or
  * mutates credentials. The typed `local_snapshot_failure` kind lets session
- * retry policy recover it as a bounded same-model retry instead: the staged
- * attempt was discarded before publication, so a content-free re-issue is
- * replay-safe and, in practice, a fresh stream clears the failure.
+ * policy surface the producer-boundary diagnostic immediately instead of
+ * amplifying one deterministic local defect across identical retries.
  */
 class ManagedAttemptSnapshotError extends Error {
 	readonly errorKind = "local_snapshot_failure" as const;
@@ -801,12 +800,33 @@ export function sanitizedDetachedClone<T>(value: T, maxNodes: number = MANAGED_S
  * (e.g. a live `Headers` inside a provider error's `transportFailure` from a
  * legacy payload), and a thrown `DataCloneError` here would mask the real
  * provider outcome and burn the whole fallback chain.
+ *
+ * `structuredClone` success is not sufficient: it can erase a custom
+ * prototype `toJSON()` while retaining an own bigint field. The live value is
+ * JSON-safe, but the detached clone is not. Validate and measure the DETACHED
+ * value with the exact serialization operation used by staging; sanitize the
+ * detached clone when that validation fails so every accepted snapshot is
+ * both isolated and JSON-serializable.
  */
-function managedAttemptSnapshotDetailed<T>(value: T): { snapshot: T; degraded: boolean } {
+function managedSnapshotJsonBytes(value: unknown): number | undefined {
+	try {
+		const serialized = JSON.stringify(value);
+		return serialized === undefined ? undefined : managedAttemptTextEncoder.encode(serialized).byteLength;
+	} catch {
+		return undefined;
+	}
+}
+
+function managedAttemptSnapshotDetailed<T>(value: T): { snapshot: T; jsonBytes?: number } {
 	try {
-		return { snapshot: structuredClone(value), degraded: false };
+		const snapshot = structuredClone(value);
+		const jsonBytes = managedSnapshotJsonBytes(snapshot);
+		if (jsonBytes !== undefined) return { snapshot, jsonBytes };
+		const sanitized = sanitizedDetachedClone(snapshot);
+		return { snapshot: sanitized, jsonBytes: managedSnapshotJsonBytes(sanitized) };
 	} catch {
-		return { snapshot: sanitizedDetachedClone(value), degraded: true };
+		const snapshot = sanitizedDetachedClone(value);
+		return { snapshot, jsonBytes: managedSnapshotJsonBytes(snapshot) };
 	}
 }
 
@@ -1377,34 +1397,24 @@ class ManagedAttemptTransaction {
 		}
 		const repaired = this.#repairAssistantEvent(event);
 		const detailed = managedAttemptSnapshotDetailed(repaired);
-		let snapshot = detailed.snapshot;
-		if (bytes === undefined || detailed.degraded) {
-			// Account the bytes of what is actually retained: a degraded
-			// snapshot replaces non-JSON leaves with placeholders, so the raw
-			// pre-measure (which omits e.g. function-valued properties) can
-			// undercount the staged form.
-			try {
-				bytes = managedAttemptTextEncoder.encode(JSON.stringify(snapshot)).byteLength;
-			} catch {
-				try {
-					snapshot = sanitizedDetachedClone(snapshot);
-					bytes = managedAttemptTextEncoder.encode(JSON.stringify(snapshot)).byteLength;
-				} catch {
-					bytes = undefined;
-				}
-			}
-			if (bytes === undefined) {
-				// The sanitizer's output is total (detached, JSON-safe), so this
-				// is unreachable unless the sanitizer itself regresses. Fail as a
-				// dedicated local error: it carries no transport facts, so it is
-				// non-retryable and can never be misattributed to the provider.
-				this.discard();
-				throw new ManagedAttemptSnapshotError("staging.sanitize");
-			}
-			if (this.#wouldOverflow(bytes)) {
-				this.discard();
-				throw new ManagedAttemptBufferOverflowError("overflow.staged");
-			}
+		const snapshot = detailed.snapshot;
+		// Always account the exact detached value. A live custom class can use
+		// prototype `toJSON()` to serialize compactly while structuredClone
+		// removes that serializer and exposes a larger or JSON-hostile own value.
+		// Reusing the live pre-measure would therefore accept an unserializable
+		// snapshot or undercount the retained bytes.
+		bytes = detailed.jsonBytes;
+		if (bytes === undefined) {
+			// The sanitizer's output is total (detached, JSON-safe), so this is
+			// unreachable unless the sanitizer itself regresses. Fail as a
+			// dedicated local error: it carries no transport facts, so it is
+			// non-retryable and can never be misattributed to the provider.
+			this.discard();
+			throw new ManagedAttemptSnapshotError("staging.sanitize");
+		}
+		if (this.#wouldOverflow(bytes)) {
+			this.discard();
+			throw new ManagedAttemptBufferOverflowError("overflow.staged");
 		}
 		this.#batch.push({ type: "event", event: snapshot });
 		this.#stagedEventCount += 1;

packages/agent/test/managed-attempt-transaction.test.ts

@@ -49,6 +49,14 @@ function assistantMessage(model: ReturnType<typeof createMockModel>["model"]): A
 	};
 }
 
+class JsonSafeBigIntEnvelope {
+	sequence = 1n;
+
+	toJSON(): { sequence: string } {
+		return { sequence: this.sequence.toString() };
+	}
+}
+
 function expectManagedRunStart(events: string[]): void {
 	expect(events.filter(type => type === "agent_start")).toHaveLength(1);
 	const start = events.indexOf("agent_start");
@@ -157,6 +165,83 @@ describe("managed attempt transaction", () => {
 		}
 	});
 
+	it("publishes JSON-serializable snapshots when structuredClone removes a payload class serializer", async () => {
+		const mock = createMockModel();
+		const liveEnvelope = new JsonSafeBigIntEnvelope();
+		const callbackValues: Array<{ path: string; value: unknown }> = [];
+		const publicValues: Array<{ path: string; value: unknown }> = [];
+		const streamFn = () => {
+			const stream = new AssistantMessageEventStream();
+			queueMicrotask(() => {
+				const partial = assistantMessage(mock.model);
+				(partial as unknown as Record<string, unknown>).providerPayload = {
+					envelope: liveEnvelope,
+				};
+				stream.push({ type: "start", partial });
+				partial.content.push({ type: "text", text: "accepted" });
+				stream.push({ type: "text_start", contentIndex: 0, partial });
+				stream.push({ type: "done", reason: "stop", message: partial });
+			});
+			return stream;
+		};
+		const agent = new Agent({
+			initialState: { model: mock.model, systemPrompt: ["test"], tools: [], messages: [] },
+			streamFn,
+			onAssistantMessageEvent: (message, event) => {
+				callbackValues.push({ path: `callback.${event.type}.message`, value: message });
+				callbackValues.push({ path: `callback.${event.type}.event`, value: event });
+			},
+		});
+		agent.subscribe(event => publicValues.push({ path: `public.${event.type}`, value: event }));
+
+		await agent.prompt("run", { fallbackManaged: true });
+		liveEnvelope.sequence = 2n;
+
+		const failures = [...callbackValues, ...publicValues].flatMap(candidate => {
+			try {
+				JSON.stringify(candidate.value);
+				return [];
+			} catch {
+				return [
+					{
+						path: `${candidate.path}.providerPayload.envelope.sequence`,
+						valueClass: JsonSafeBigIntEnvelope.name,
+						valueType: "bigint",
+					},
+				];
+			}
+		});
+		expect(failures).toEqual([]);
+		const callbackMessage = callbackValues.find(candidate => candidate.path === "callback.text_start.message")!
+			.value as Record<string, unknown>;
+		const callbackEvent = callbackValues.find(candidate => candidate.path === "callback.text_start.event")!
+			.value as Extract<AssistantMessageEvent, { type: "text_start" }>;
+		const turnEnd = publicValues.find(candidate => candidate.path === "public.turn_end")!.value as Extract<
+			AgentEvent,
+			{ type: "turn_end" }
+		>;
+		const agentEnd = publicValues.find(candidate => candidate.path === "public.agent_end")!.value as Extract<
+			AgentEvent,
+			{ type: "agent_end" }
+		>;
+		const agentEndAssistant = agentEnd.messages.find(message => message.role === "assistant");
+		const sequence = (value: unknown): unknown => {
+			if (value === null || typeof value !== "object") return undefined;
+			const providerPayload = (value as Record<string, unknown>).providerPayload;
+			if (providerPayload === null || typeof providerPayload !== "object") return undefined;
+			const envelope = (providerPayload as Record<string, unknown>).envelope;
+			return envelope !== null && typeof envelope === "object"
+				? (envelope as Record<string, unknown>).sequence
+				: undefined;
+		};
+		expect([
+			sequence(callbackMessage),
+			sequence(callbackEvent.partial),
+			sequence(turnEnd.message),
+			sequence(agentEndAssistant),
+		]).toEqual(["1", "1", "1", "1"]);
+	});
+
 	it("replays mutating provider partials as event-time snapshots with callbacks first", async () => {
 		const mock = createMockModel();
 		const streamFn = () => {

packages/coding-agent/CHANGELOG.md

@@ -6,7 +6,7 @@
 - Session endpoints hosted on the notification-adapter transport now deliver every ring-retained session event live to attached SDK subscribers as the same positioned `event` envelope (`generation`/`seq`) that `event_replay` returns, sent per connection over the validated directed leg with the same capability gating replay applies. Previously the live leg only pushed raw side-channel frames — the native broadcast enum reduced non-native kinds (including terminal `agent_end` lifecycle) to empty `unknown` frames, and correlated lifecycle reached only the submitting connection — so an already-attached direct SDK subscriber could observe a later positioned event, including a turn's terminal lifecycle, only by issuing another replay. Each connection's directed writer now bounds queued host frames to the replay-ring capacity; a lagged subscriber rejects additional best-effort live sends and recovers through replay (including the existing sequence-gap contract) instead of growing an unbounded backlog. Ring persistence, replay ordering, event positions, correlated requester delivery, and native notification frames are unchanged.
 - Telegram notification delivery now carries an explicit per-update inbound acknowledgement contract: user messages are acked `accepted` at session preflight acceptance (before the turn starts, so a fast turn can no longer out-race the pending-update registration), late admission failures ack `rejected`, and genuinely discarded frames ack `dropped`. Policy-suspended control commands are deferred to activation instead of being acked as dropped, per-update reaction transitions are serialized with terminal states monotonic (a slow queued 👀 can no longer overwrite a later ✅), and retraction sends the empty reaction list the Bot API requires. Daemon generation bumped 167→168. (#4528)
 
-- Managed fallback local snapshot failures now auto-recover with a bounded same-model retry (capped at `retry.maxRetries`) instead of terminating the turn: the discarded attempt is replay-safe and content-free, so the session re-issues the request without charging the provider fallback chain, advancing models, or mutating credentials. Exhausted retries still surface the explicit local diagnostic.
+- Managed fallback local snapshot failures now surface their one producer-boundary diagnostic immediately instead of re-issuing the identical request up to three times. The failure still never charges the provider fallback chain, advances models, or mutates credentials.
 
 - Managed fallback local buffer overflows (`local_buffer_overflow`) now surface immediately with the original local diagnostic instead of entering the bounded `unknown` retry class: re-streaming the same request reproduces the same oversized response, and a local staging failure must never charge or advance the provider fallback chain, emit `model_fallback_switched`, or rotate credentials.