fix(agent,ai): degrade primitive stream increments instead of killing managed turns

packages/agent/CHANGELOG.md

@@ -11,6 +11,7 @@
 
 ## [0.14.1] - 2026-08-18
 - Compaction pruning no longer kills the turn when a persisted `toolCall.arguments` is `null`. Sessions written by an earlier cold-spill eviction path store `null` where the spill sentinel belongs, and the staleness index dereferenced that payload unguarded, so reloading such a session threw `null is not an object (evaluating 'args.path')` as a turn-fatal error instead of skipping the one unusable call. `ToolCall.arguments` is typed non-nullable, so no type check flagged the gap. Every read of a persisted argument bag — path extraction, `apply_patch` header parsing, idempotent-bash keys, and search target keys — now treats a non-object payload as absent. The original arguments are not lost: the eviction marker still names the blob and rehydration restores them.
+- Managed fallback attempt snapshots no longer fail the whole run on benign provider shape variations: an assistant message whose `content` is a bare string, is missing, or is a primitive scalar (null/number/boolean) now degrades to an empty content array; staged `*_delta`/`*_end` events whose `delta`/`content` is missing or a primitive scalar degrade to an empty string; and staged assistant events with out-of-vocabulary `done`/`error` reasons or an unknown string `type` degrade to schema-valid values instead of throwing a non-retryable `ManagedAttemptSnapshotError`. Object-shaped or other plain-object `content`/`delta` stays fail-closed under the named `shell.content`/`event.delta`/`event.content` diagnostic, as does sanitizer-sentinel string content (`[unserializable]`/`[accessor]`/`[truncated]`/`[Circular]`, which marks a non-cloneable original rather than provider string variance — degrading those would silently drop real tool-call or streamed content behind a successful empty turn), and hostile inputs keep failing fast with no retry authority: a live proxy root, a throwing `get`/`getOwnPropertyDescriptor` trap, and a non-string event `type` all remain local snapshot failures.
 
 ### Added
 

packages/agent/src/agent-loop.ts

@@ -735,6 +735,26 @@ const SANITIZER_SENTINELS: ReadonlySet<string> = new Set([
 	"[truncated]",
 	"[Circular]",
 ]);
+/**
+ * Bounded diagnostic for a degraded primitive at the shared managed-snapshot
+ * boundary. Every provider/custom stream that still forwards a malformed
+ * primitive increment degrades here (to "" / []), so the degradation stays
+ * observable. The caller supplies a run-scoped set so repeated malformed
+ * increments emit at most one payload-free warning per field name, naming
+ * only the field and the received typeof — never the payload.
+ */
+function warnManagedDegradedPrimitive(
+	field: string,
+	received: unknown,
+	diagnostics: Set<string> = new Set<string>(),
+): void {
+	if (diagnostics.has(field)) return;
+	diagnostics.add(field);
+	logger.warn("agent: managed snapshot degraded a non-string primitive to an empty value", {
+		field,
+		receivedType: received === null ? "null" : typeof received,
+	});
+}
 
 /**
  * Cycle-aware deep clone that always returns a detached, JSON-serializable
@@ -887,16 +907,16 @@ function managedSnapshotJsonBytes(value: unknown): number | undefined {
 	}
 }
 
-function managedAttemptSnapshotDetailed<T>(value: T): { snapshot: T; jsonBytes?: number } {
+function managedAttemptSnapshotDetailed<T>(value: T): { snapshot: T; jsonBytes?: number; sanitized: boolean } {
 	try {
 		const snapshot = structuredClone(value);
 		const jsonBytes = managedSnapshotJsonBytes(snapshot);
-		if (jsonBytes !== undefined) return { snapshot, jsonBytes };
+		if (jsonBytes !== undefined) return { snapshot, jsonBytes, sanitized: false };
 		const sanitized = sanitizedDetachedClone(snapshot);
-		return { snapshot: sanitized, jsonBytes: managedSnapshotJsonBytes(sanitized) };
+		return { snapshot: sanitized, jsonBytes: managedSnapshotJsonBytes(sanitized), sanitized: true };
 	} catch {
 		const snapshot = sanitizedDetachedClone(value);
-		return { snapshot, jsonBytes: managedSnapshotJsonBytes(snapshot) };
+		return { snapshot, jsonBytes: managedSnapshotJsonBytes(snapshot), sanitized: true };
 	}
 }
 
@@ -997,7 +1017,11 @@ function losslessDetachedClone<T>(value: T): T {
  * are read, and executable content is retained only when it has its complete
  * discriminant shape.
  */
-function managedAssistantShell(value: unknown, model: AgentLoopConfig["model"]): AssistantMessage {
+function managedAssistantShell(
+	value: unknown,
+	model: AgentLoopConfig["model"],
+	degradedFieldDiagnostics: Set<string> = new Set<string>(),
+): AssistantMessage {
 	const detailed = managedAttemptSnapshotDetailed(value);
 	const snapshotRecord = isManagedPlainRecord(detailed.snapshot) ? detailed.snapshot : undefined;
 	// Two benign root degradations are repaired by reading through the
@@ -1015,22 +1039,26 @@ function managedAssistantShell(value: unknown, model: AgentLoopConfig["model"]):
 		snapshotRecord !== undefined && managedProperty(snapshotRecord, "role") === "assistant" ? snapshotRecord : value;
 	if (managedProperty(source, "role") !== "assistant") throw new ManagedAttemptSnapshotError("shell.role");
 	const rawContent = managedAttemptSnapshot(managedProperty(source, "content"));
-	// Benign providers occasionally deliver a string or missing content value.
-	// Degrade those to an empty content array — an empty assistant turn —
-	// instead of failing the whole managed run: the staged shell must stay
-	// schema-valid, and empty content is the neutral, side-effect-free
-	// degradation. A string is benign ONLY when the provider actually sent a
-	// string: when the whole-message snapshot degraded, the sanitizer replaces
-	// a non-cloneable content value (proxy, function, accessor) with one of
-	// its own sentinel strings, and mistaking that sentinel for provider
-	// variance would silently drop real content (tool calls) behind a
-	// successful empty turn. Sentinel-string content therefore stays
-	// fail-closed, as does every other non-array shape, so the named-site
-	// diagnostic can report shell.content.
+	// Providers may deliver `content` as a string, missing value, or a primitive
+	// scalar — all benign variance that degrades to an empty content array
+	// (an empty, side-effect-free assistant turn). A plain-object `content`
+	// is NOT degraded: it can carry array-like toolCall payloads
+	// (`{0:{type:"toolCall"}}`) and silently dropping them would lose
+	// executable content behind a successful empty turn. Only sentinel
+	// strings produced by the sanitizer itself (`[unserializable]` etc.)
+	// also stay fail-closed for the same reason, plus any plain object.
 	const rawArray = Array.isArray(rawContent) ? rawContent : undefined;
-	const benignContent =
-		rawContent === undefined || (typeof rawContent === "string" && !SANITIZER_SENTINELS.has(rawContent));
-	if (rawArray === undefined && !benignContent) throw new ManagedAttemptSnapshotError("shell.content");
+	if (rawArray === undefined) {
+		if (typeof rawContent === "string" && SANITIZER_SENTINELS.has(rawContent)) {
+			throw new ManagedAttemptSnapshotError("shell.content");
+		}
+		if (rawContent !== null && typeof rawContent === "object") {
+			throw new ManagedAttemptSnapshotError("shell.content");
+		}
+		if (rawArray === undefined && rawContent !== undefined && !SANITIZER_SENTINELS.has(rawContent as string)) {
+			warnManagedDegradedPrimitive("shell.content", rawContent, degradedFieldDiagnostics);
+		}
+	}
 	const content = rawArray === undefined ? [] : rawArray.flatMap(managedContentBlock);
 	const usage = managedAssistantUsage(managedAttemptSnapshot(managedProperty(source, "usage")));
 	const api = managedProperty(source, "api");
@@ -1153,8 +1181,45 @@ function managedAssistantUsage(value: unknown): AssistantMessage["usage"] {
 export function managedAssistantEventSnapshot(
 	event: AssistantMessageEvent,
 	message: AssistantMessage,
+	degradedFieldDiagnostics: Set<string> = new Set<string>(),
 ): AssistantMessageEvent {
-	const detached = managedAttemptSnapshot(event);
+	const directType = managedProperty(event, "type");
+	if (
+		directType === "text_delta" ||
+		directType === "thinking_delta" ||
+		directType === "reasoning_summary_delta" ||
+		directType === "toolcall_delta"
+	) {
+		// Delta events are snapshotted field-by-field so unrelated event metadata
+		// cannot erase provenance. The delta is read once, detached once, then the
+		// same captured value is both validated and emitted.
+		const contentIndex = managedAttemptSnapshot(managedProperty(event, "contentIndex"));
+		if (!Number.isInteger(contentIndex) || (contentIndex as number) < 0) {
+			throw new ManagedAttemptSnapshotError("event.contentIndex");
+		}
+		const deltaSnapshot = managedAttemptSnapshotDetailed(managedProperty(event, "delta"));
+		const delta = deltaSnapshot.snapshot;
+		if (directType === "toolcall_delta" && (deltaSnapshot.sanitized || typeof delta !== "string")) {
+			throw new ManagedAttemptSnapshotError("event.delta");
+		}
+		if (deltaSnapshot.sanitized && typeof delta === "string" && SANITIZER_SENTINELS.has(delta)) {
+			throw new ManagedAttemptSnapshotError("event.delta");
+		}
+		if (delta !== undefined && delta !== null && typeof delta === "object") {
+			throw new ManagedAttemptSnapshotError("event.delta");
+		}
+		if (typeof delta !== "string") {
+			warnManagedDegradedPrimitive("event.delta", delta, degradedFieldDiagnostics);
+		}
+		return {
+			type: directType,
+			contentIndex: contentIndex as number,
+			delta: typeof delta === "string" ? delta : "",
+			partial: message,
+		};
+	}
+	const eventSnapshot = managedAttemptSnapshotDetailed(event);
+	const detached = eventSnapshot.snapshot;
 	const record = isManagedPlainRecord(detached) ? detached : undefined;
 	// Root repair, mirroring the shell: two benign degradations are re-read
 	// through the original event with guarded reads (`managedProperty`) —
@@ -1184,20 +1249,20 @@ export function managedAssistantEventSnapshot(
 		type === "toolcall_start"
 	)
 		return { type, contentIndex: indexed(), partial: message };
-	if (
-		type === "text_delta" ||
-		type === "thinking_delta" ||
-		type === "reasoning_summary_delta" ||
-		type === "toolcall_delta"
-	) {
-		const delta = managedProperty(source, "delta");
-		if (typeof delta !== "string") throw new ManagedAttemptSnapshotError("event.delta");
-		return { type, contentIndex: indexed(), delta, partial: message };
-	}
 	if (type === "text_end" || type === "thinking_end" || type === "reasoning_summary_end") {
-		const content = managedProperty(source, "content");
-		if (typeof content !== "string") throw new ManagedAttemptSnapshotError("event.content");
-		return { type, contentIndex: indexed(), content, partial: message };
+		const contentSnapshot = managedAttemptSnapshotDetailed(managedProperty(source, "content"));
+		const content = contentSnapshot.snapshot;
+		if (contentSnapshot.sanitized && typeof content === "string" && SANITIZER_SENTINELS.has(content)) {
+			throw new ManagedAttemptSnapshotError("event.content");
+		}
+		if (content !== undefined && content !== null && typeof content === "object") {
+			throw new ManagedAttemptSnapshotError("event.content");
+		}
+		if (typeof content !== "string") {
+			warnManagedDegradedPrimitive("event.content", content, degradedFieldDiagnostics);
+		}
+		const safeContent = typeof content === "string" ? content : "";
+		return { type, contentIndex: indexed(), content: safeContent, partial: message };
 	}
 	if (type === "toolcall_end") {
 		const toolCall = managedAssistantContent(managedAttemptSnapshot(managedProperty(source, "toolCall")));
@@ -1304,6 +1369,7 @@ class ManagedAttemptTransaction {
 	#lastStagedShape: { stagedEventCount: number; stagedBytes: number; contentBlockCount: number } | undefined;
 	#discarded = false;
 	#committed = false;
+	#degradedFieldDiagnostics = new Set<string>();
 
 	constructor(
 		private readonly stream: EventStream<AgentEvent, AgentMessage[]>,
@@ -1596,22 +1662,28 @@ class ManagedAttemptTransaction {
 		if (this.snapshotMode === "lossless") return event;
 		if (event.type === "message_start" || event.type === "message_end" || event.type === "turn_end") {
 			return event.message.role === "assistant"
-				? { ...event, message: managedAssistantShell(event.message, this.model) }
+				? { ...event, message: managedAssistantShell(event.message, this.model, this.#degradedFieldDiagnostics) }
 				: event;
 		}
 		if (event.type === "message_update") {
-			const message = managedAssistantShell(event.message, this.model);
+			const message = managedAssistantShell(event.message, this.model, this.#degradedFieldDiagnostics);
 			return {
 				...event,
 				message,
-				assistantMessageEvent: managedAssistantEventSnapshot(event.assistantMessageEvent, message),
+				assistantMessageEvent: managedAssistantEventSnapshot(
+					event.assistantMessageEvent,
+					message,
+					this.#degradedFieldDiagnostics,
+				),
 			};
 		}
 		if (event.type === "agent_end") {
 			return {
 				...event,
 				messages: event.messages.map(message =>
-					message.role === "assistant" ? managedAssistantShell(message, this.model) : message,
+					message.role === "assistant"
+						? managedAssistantShell(message, this.model, this.#degradedFieldDiagnostics)
+						: message,
 				),
 			};
 		}
@@ -1647,11 +1719,13 @@ class ManagedAttemptTransaction {
 	#assistantSnapshot(message: AssistantMessage): AssistantMessage {
 		return this.snapshotMode === "lossless"
 			? this.#losslessSnapshot(message)
-			: managedAssistantShell(message, this.model);
+			: managedAssistantShell(message, this.model, this.#degradedFieldDiagnostics);
 	}
 
 	#assistantEventSnapshot(event: AssistantMessageEvent, message: AssistantMessage): AssistantMessageEvent {
-		if (this.snapshotMode === "managed") return managedAssistantEventSnapshot(event, message);
+		if (this.snapshotMode === "managed") {
+			return managedAssistantEventSnapshot(event, message, this.#degradedFieldDiagnostics);
+		}
 		const snapshot = this.#losslessSnapshot(event);
 		if (snapshot.type === "done") return { ...snapshot, message };
 		if (snapshot.type === "error") return { ...snapshot, error: message };
@@ -2819,6 +2893,7 @@ async function streamAssistantResponse(
 	provisionalToolTransaction?: ManagedAttemptTransaction,
 	toolChoiceOverride?: { value: ToolChoice | undefined },
 ): Promise<AssistantMessage> {
+	const managedDegradedFieldDiagnostics = new Set<string>();
 	// Apply context transform if configured (AgentMessage[] → AgentMessage[])
 	let messages = context.messages;
 	if (config.transformContext) {
@@ -3143,7 +3218,7 @@ async function streamAssistantResponse(
 					switch (event.type) {
 						case "start":
 							partialMessage = config.fallbackManaged
-								? managedAssistantShell(event.partial, config.model)
+								? managedAssistantShell(event.partial, config.model, managedDegradedFieldDiagnostics)
 								: event.partial;
 							context.messages.push(partialMessage);
 							addedPartial = true;
@@ -3171,7 +3246,7 @@ async function streamAssistantResponse(
 						case "toolcall_end":
 							if (partialMessage) {
 								partialMessage = config.fallbackManaged
-									? managedAssistantShell(event.partial, config.model)
+									? managedAssistantShell(event.partial, config.model, managedDegradedFieldDiagnostics)
 									: event.partial;
 								// Normalize through the managed event snapshot instead of a
 								// naive `{ ...event }` spread: spreading copies only own
@@ -3181,7 +3256,7 @@ async function streamAssistantResponse(
 								// downstream. The snapshot repairs benign class/prototype shapes
 								// and keeps the named fail-fast diagnostics for hostile ones.
 								const partialEvent = config.fallbackManaged
-									? managedAssistantEventSnapshot(event, partialMessage)
+									? managedAssistantEventSnapshot(event, partialMessage, managedDegradedFieldDiagnostics)
 									: event;
 								context.messages[context.messages.length - 1] = partialMessage;
 								if (provisionalToolTransaction) {
@@ -3210,7 +3285,7 @@ async function streamAssistantResponse(
 						case "done":
 						case "error": {
 							const finalMessage = config.fallbackManaged
-								? managedAssistantShell(await finishResponse(), config.model)
+								? managedAssistantShell(await finishResponse(), config.model, managedDegradedFieldDiagnostics)
 								: await finishResponse();
 							promoteTypedEmptyResponseStop(finalMessage);
 							if (addedPartial) {
@@ -3233,7 +3308,7 @@ async function streamAssistantResponse(
 			}
 
 			const trailing = config.fallbackManaged
-				? managedAssistantShell(await finishResponse(), config.model)
+				? managedAssistantShell(await finishResponse(), config.model, managedDegradedFieldDiagnostics)
 				: await finishResponse();
 			await finishChat(trailing);
 			return trailing;

packages/agent/test/agent-loop-anthropic-truncated-toolcall.test.ts

@@ -182,7 +182,7 @@ describe("agentLoop with Anthropic truncated tool calls", () => {
 		expect(toolResults[1]).toEqual({ isError: false, text: "wrote" });
 	});
 
-	it("executes only the complete same-ID replacement after a malformed duplicate index", async () => {
+	it("fails closed before executing a malformed duplicate index", async () => {
 		const responses = [duplicateToolResponse("tool_shared"), textResponse("done")];
 		let responseIndex = 0;
 		vi.spyOn(Messages.prototype, "create").mockImplementation(() => {
@@ -223,11 +223,9 @@ describe("agentLoop with Anthropic truncated tool calls", () => {
 			}
 		}
 
-		expect(responseIndex).toBe(2);
-		expect(executed).toEqual([{ path: "b.ts", content: "ok" }]);
+		expect(responseIndex).toBe(1);
+		expect(executed).toHaveLength(0);
 		expect(toolResults).toHaveLength(2);
-		expect(toolResults[0].isError).toBe(true);
-		expect(toolResults[0].text).toContain("cut off");
-		expect(toolResults[1]).toEqual({ isError: false, text: "wrote" });
+		expect(toolResults.every(result => result.isError)).toBe(true);
 	});
 });