fix(agent): bound managed staging by reclaiming superseded deltas

packages/agent/CHANGELOG.md

@@ -6,6 +6,7 @@
 
 ### Fixed
 
+- Managed fallback no longer kills a long turn with `Managed fallback attempt exceeded the provisional event buffer limit`. Every staged streaming frame carries the whole accumulated partial (once as `message`, once as `assistantMessageEvent.partial`), so staged bytes grew quadratically with the response length and a reasoning-heavy turn of a few thousand tokens crossed the 16 MiB cap even though no single event came close to it. Reaching the cap now first reclaims the staged `*_delta` increments, whose complete value is re-published by the retained `*_end` and terminal `message_end`/`done` frames, and only a batch that still cannot fit fails. Attempt atomicity is unchanged: nothing is published early, so a discarded attempt stays unobservable, and a single oversized event keeps its pre-clone rejection with no provider-fallback authority.
 - Non-managed lossless response staging now commits its buffered lifecycle and switches to ordinary pass-through publication when the provisional event cap is reached, instead of turning a large reasoning-only response into a fatal `local_snapshot_failure`. Managed fallback attempts keep the strict bounded-buffer rejection required for atomic retry and provider-fallback isolation.
 - Managed snapshot machinery no longer fails runs on benign payload-class or readable-proxy roots: an assistant message or stream event whose fields live on prototype getters (which `structuredClone` drops — it copies only own enumerable properties) or behind a proxy whose gets are readable is repaired through the existing guarded-read path instead of throwing a deterministic `shell.role`/`event.unknownType`/`event.snapshot` local snapshot failure. The run-loop message_update replay also builds its event through the managed event snapshot instead of a naive `{ ...event }` spread, which silently dropped prototype-carried fields before the snapshot boundary could see them. Hostile shapes (throwing get traps, sentinel-marked degraded content, malformed non-string event types) keep their named fail-fast diagnostics with no retry authority.
 

packages/agent/src/agent-loop.ts

@@ -1204,9 +1204,28 @@ function warnManagedSnapshotFailure(
  * commits the transaction.
  */
 type ManagedAttemptBatchItem =
-	| { type: "event"; event: AgentEvent }
+	| { type: "event"; event: AgentEvent; bytes?: number }
 	| { type: "assistant_event"; message: AssistantMessage; event: AssistantMessageEvent };
 
+/**
+ * Streaming increments whose complete value is re-published by the block's own
+ * `*_end` frame and by the terminal `message_end` / `done` frames. Those
+ * terminal frames are never reclaimed, so dropping the increments loses no
+ * content — only the intermediate frames that carried it on the way there.
+ */
+const MANAGED_SUPERSEDED_DELTA_EVENT_TYPES: ReadonlySet<string> = new Set([
+	"text_delta",
+	"thinking_delta",
+	"reasoning_summary_delta",
+	"toolcall_delta",
+]);
+
+function isSupersededStreamingDelta(item: ManagedAttemptBatchItem): boolean {
+	if (item.type === "assistant_event") return MANAGED_SUPERSEDED_DELTA_EVENT_TYPES.has(item.event.type);
+	if (item.event.type !== "message_update") return false;
+	return MANAGED_SUPERSEDED_DELTA_EVENT_TYPES.has(item.event.assistantMessageEvent.type);
+}
+
 class ManagedAttemptTransaction {
 	#batch: ManagedAttemptBatchItem[] = [];
 	#stagedEventCount = 0;
@@ -1369,6 +1388,48 @@ class ManagedAttemptTransaction {
 		);
 	}
 
+	/**
+	 * Reclaim staged frames that later staged frames already supersede.
+	 *
+	 * Every staged streaming frame carries the WHOLE accumulated partial (once as
+	 * `message`, once as `assistantMessageEvent.partial`), so a turn that streams
+	 * N increments stages ~N * length bytes: quadratic in the response length. A
+	 * reasoning-heavy turn of a few thousand tokens therefore used to exhaust the
+	 * provisional cap and kill the whole run, even though the attempt itself was
+	 * healthy and the cap exists only to bound memory.
+	 *
+	 * Each `*_delta` increment is re-published in full by its block's `*_end`
+	 * frame and by the terminal `message_end` / `done` frames, and those are
+	 * retained, so dropping the increments reclaims the growth without inventing
+	 * or losing content. Nothing is published here: the batch stays
+	 * all-or-nothing, so a discarded attempt remains unobservable and the
+	 * fallback chain is still untouched.
+	 *
+	 * Returns whether anything was reclaimed, so the caller can re-test the cap
+	 * and keep failing fast on a single payload that cannot fit on its own.
+	 */
+	#compactSupersededFrames(): boolean {
+		if (this.#batch.length === 0) return false;
+		const retained: ManagedAttemptBatchItem[] = [];
+		let reclaimedBytes = 0;
+		let reclaimedEvents = 0;
+		for (const item of this.#batch) {
+			if (!isSupersededStreamingDelta(item)) {
+				retained.push(item);
+				continue;
+			}
+			if (item.type === "event") {
+				reclaimedBytes += item.bytes ?? 0;
+				reclaimedEvents += 1;
+			}
+		}
+		if (retained.length === this.#batch.length) return false;
+		this.#batch = retained;
+		this.#stagedBytes -= reclaimedBytes;
+		this.#stagedEventCount -= reclaimedEvents;
+		return true;
+	}
+
 	#stage(event: AgentEvent): void {
 		if (this.snapshotMode === "lossless") {
 			const snapshot = this.#repairAssistantEvent(event);
@@ -1402,7 +1463,7 @@ class ManagedAttemptTransaction {
 				this.push(detached);
 				return;
 			}
-			this.#batch.push({ type: "event", event: detached });
+			this.#batch.push({ type: "event", event: detached, bytes: detachedBytes });
 			this.#stagedEventCount++;
 			this.#stagedBytes += detachedBytes;
 			return;
@@ -1420,8 +1481,14 @@ class ManagedAttemptTransaction {
 			bytes = undefined;
 		}
 		if (bytes !== undefined && this.#wouldOverflow(bytes)) {
-			this.discard();
-			throw new ManagedAttemptBufferOverflowError("overflow.preMeasure");
+			// A long turn reaches the cap through accumulated streaming increments,
+			// not through one oversized payload. Reclaim the superseded increments
+			// first; only a batch that still cannot fit is a real local overflow.
+			this.#compactSupersededFrames();
+			if (this.#wouldOverflow(bytes)) {
+				this.discard();
+				throw new ManagedAttemptBufferOverflowError("overflow.preMeasure");
+			}
 		}
 		const repaired = this.#repairAssistantEvent(event);
 		const detailed = managedAttemptSnapshotDetailed(repaired);
@@ -1441,10 +1508,15 @@ class ManagedAttemptTransaction {
 			throw new ManagedAttemptSnapshotError("staging.sanitize");
 		}
 		if (this.#wouldOverflow(bytes)) {
-			this.discard();
-			throw new ManagedAttemptBufferOverflowError("overflow.staged");
+			this.#compactSupersededFrames();
+			if (this.#wouldOverflow(bytes)) {
+				this.discard();
+				throw new ManagedAttemptBufferOverflowError("overflow.staged");
+			}
 		}
-		this.#batch.push({ type: "event", event: snapshot });
+		// Retain each frame's accounted size so compaction can debit exactly what
+		// it reclaims instead of re-measuring the whole batch.
+		this.#batch.push({ type: "event", event: snapshot, bytes });
 		this.#stagedEventCount += 1;
 
 		this.#stagedBytes += bytes;

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

@@ -1144,6 +1144,75 @@ describe("managed attempt transaction", () => {
 		expect(surfaced[0]?.transportFailure).toBeUndefined();
 	});
 
+	it("completes a long managed stream by reclaiming superseded increments instead of failing", async () => {
+		// Regression: every staged frame carries the WHOLE accumulated partial
+		// (once as `message`, once as `assistantMessageEvent.partial`), so staged
+		// bytes grow quadratically with the streamed length. A reasoning-heavy
+		// turn of a few thousand tokens used to cross the 16 MiB cap and kill the
+		// run with "exceeded the provisional event buffer limit", even though no
+		// single event was anywhere near the cap and the attempt itself was
+		// healthy. 200 increments of 1 KiB stage ~40 MiB uncompacted.
+		const deltaCount = 200;
+		const delta = "x".repeat(1024);
+		const fullText = delta.repeat(deltaCount);
+		const mock = createMockModel();
+		const streamFn = () => {
+			const stream = new AssistantMessageEventStream();
+			queueMicrotask(() => {
+				const partial = assistantMessage(mock.model);
+				stream.push({ type: "start", partial });
+				partial.content.push({ type: "text", text: "" });
+				stream.push({ type: "text_start", contentIndex: 0, partial });
+				for (let index = 0; index < deltaCount; index++) {
+					const block = partial.content[0] as { type: "text"; text: string };
+					block.text += delta;
+					stream.push({ type: "text_delta", contentIndex: 0, delta, partial });
+				}
+				stream.push({ type: "text_end", contentIndex: 0, content: fullText, partial });
+				stream.push({ type: "done", reason: "stop", message: partial });
+			});
+			return stream;
+		};
+		const callbacks: AssistantMessageEvent[] = [];
+		const agent = new Agent({
+			initialState: { model: mock.model, systemPrompt: ["test"], tools: [], messages: [] },
+			streamFn,
+			onAssistantMessageEvent: (_message, event) => callbacks.push(event),
+		});
+		const replayedUpdates: string[] = [];
+		agent.subscribe(event => {
+			if (event.type !== "message_update") return;
+			replayedUpdates.push(event.assistantMessageEvent.type);
+		});
+		let outcomeCalls = 0;
+
+		await agent.prompt("run", {
+			fallbackManaged: true,
+			onManagedAttemptOutcome: () => {
+				outcomeCalls += 1;
+				return { type: "terminal", terminal: { stopReason: "exhausted" } };
+			},
+		} as any);
+		await agent.waitForIdle();
+
+		// The turn completes and commits its whole response.
+		expect(agent.state.error).toBeUndefined();
+		const committed = agent.state.messages.at(-1) as AssistantMessage;
+		expect(committed.role).toBe("assistant");
+		expect(committed.content).toEqual([{ type: "text", text: fullText }]);
+		// A local staging limit is not provider evidence either way: reclaiming
+		// must not report an outcome or consume the fallback chain.
+		expect(outcomeCalls).toBe(0);
+		// Superseded increments were actually reclaimed rather than all replayed.
+		expect(replayedUpdates.filter(type => type === "text_delta").length).toBeLessThan(deltaCount);
+		// Structural frames survive, so the block's complete content is still
+		// delivered on the retained path.
+		expect(replayedUpdates).toContain("text_start");
+		expect(replayedUpdates).toContain("text_end");
+		const textEnd = callbacks.find(event => event.type === "text_end");
+		expect(textEnd).toMatchObject({ type: "text_end", contentIndex: 0, content: fullText });
+	});
+
 	it("retains queued follow-up input when its managed attempt is discarded for retry", async () => {
 		const mock = createMockModel({ responses: [{ content: ["initial"] }, { content: ["retried"] }] });
 		let calls = 0;