diff --git a/docs/environment-variables.md b/docs/environment-variables.md index 833c4f8011..047a751eec 100644 --- a/docs/environment-variables.md +++ b/docs/environment-variables.md @@ -464,6 +464,8 @@ Extra conditional behavior: | `GJC_SUBPROCESS_CMD` | Overrides subagent spawn command (`gjc` / `gjc.cmd` resolution bypass) | | `GJC_TASK_MAX_OUTPUT_BYTES` | Max captured output bytes per subagent (default `500000`) | | `GJC_TASK_MAX_OUTPUT_LINES` | Max captured output lines per subagent (default `5000`) | +| `GJC_FALLBACK_MAX_STAGED_EVENTS` | Positive-integer cap on events staged by the provisional staging transaction before it is rejected as a local overflow (default `10000`, hard ceiling `2000000`). Surrounding whitespace is ignored by the trusted environment resolver. Applies to both managed fallback and ordinary (non-managed lossless) sessions; in non-managed sessions the cap only decides how much reasoning buffers before the batch flushes and streams through. Invalid or non-positive values fall back to the default; values above the ceiling clamp to it with a warning — the staging guard stays bounded. Resolved from trusted environment sources only (process/agent/user config); a project `.env` cannot change these guardrails. | +| `GJC_FALLBACK_MAX_STAGED_BYTES` | Positive-integer byte cap on the provisional staging transaction (default `16777216` = 16 MiB, hard ceiling `1073741824` = 1 GiB). Surrounding whitespace is ignored by the trusted environment resolver. Applies to both managed fallback and ordinary (non-managed lossless) sessions; in non-managed sessions the cap only decides how much reasoning buffers before the batch flushes and streams through; raising it raises peak memory of ordinary runs by delaying that flush. A staged streaming frame is counted once as the message and once as the event's partial snapshot of that message, so a reasoning-heavy turn is charged roughly twice its retained volume — size the cap accordingly. Invalid or non-positive values fall back to the default; values above the ceiling clamp to it with a warning — the staging guard stays bounded. Resolved from trusted environment sources only (process/agent/user config); a project `.env` cannot change these guardrails. | | `GJC_TIMING` | If set (any non-empty value), prints a hierarchical timing-span tree to **stderr** via `logger.printTimings()`. In interactive mode the tree prints once the agent is ready (before the TUI starts); in print mode it prints after the whole prompt batch completes. Print-mode prompts are wrapped in `print:prompt:initial` / `print:prompt:next` spans so each user message shows up as its own row. `GJC_TIMING=x` exits the process with code 0 right after printing in interactive mode (use to measure cold startup only). `GJC_TIMING=full` lists every module-load entry instead of just the top N. | | `GJC_PACKAGE_DIR` | Overrides package asset base dir resolution (docs/examples/changelog path lookup) | | `GJC_DISABLE_LSPMUX` | Canonical lspmux opt-out. A truthy value disables lspmux probing and wrapping; `PI_DISABLE_LSPMUX` is a supported compatibility alias with the same effect. | diff --git a/packages/agent/CHANGELOG.md b/packages/agent/CHANGELOG.md index a3d5facd2b..dd235f964b 100644 --- a/packages/agent/CHANGELOG.md +++ b/packages/agent/CHANGELOG.md @@ -3,11 +3,13 @@ ## [Unreleased] ### Fixed +- Staged-payload sizing no longer materializes what it is bounding (#4602 fix-forward of the exact-head 078e22c0 review). All staging measurements now walk the JSON surface directly: exact byte counts come from a code-point walk (quotes, escapes, separators, delimiters, nulls, array holes, and keys all charged) instead of building the full `JSON.stringify` string plus its UTF-8 encoding, and lone surrogates are charged as the six-byte `\udXXX` escape JSON emits rather than their three-byte UTF-8 form, closing a ~2x undercount on surrogate-heavy strings. `structuredClone` is additionally preflighted by a clone-surface walk that never dispatches `toJSON`, accessors, or proxy traps: a live payload class whose compact `toJSON` hides an oversized own payload is rejected as the typed `local_buffer_overflow` at `overflow.preMeasure` — before the duplicate is allocated — instead of being cloned first and rejected at `overflow.staged`. Accessors are no longer invoked at all while sizing (a staged witness getter is read zero times), `undefined`-valued record properties are skipped exactly as `JSON.stringify` omits them, an unmeasurable assistant pair now fails closed like its `#stage` twin instead of being retained with a zero-byte charge, the `overflow.preMeasure` diagnostic reports the incoming event's real bounded size instead of a constant fabricated after `discard()`, and above-ceiling clamp warnings are logged once per distinct knob value with a bounded digest. | - Provider safety-stop messages now retain their explicitly allowlisted `errorKind: "provider_safety_stop"` through managed assistant snapshots and remain terminal even when transport facts are present on a multi-model fallback chain, while provider payloads still cannot forge the runtime-owned local diagnostic kinds (#4777). - A foreign error that self-declares a local failure kind no longer gets one either (#4618). `errorKind` and the structured `bufferOverflow` shape now come from a single identity-checked extractor (`managedLocalErrorDiagnostic`) used by both terminal-message producers — `managedFailureMessage` and the `Agent` run catch. Previously the shape was identity-gated but the label was not, so a provider or custom-stream failure carrying `errorKind: "local_buffer_overflow"` reached the parent receipt preview as `Local staging-buffer overflow; structured diagnostic unavailable.` and pointed whoever read it at the wrong subsystem. - Local diagnostic authority fields are no longer foreign-settable through the managed snapshot shell (#4618). `managedAssistantShell` spreads the provider/stream message snapshot into the rebuilt assistant message; a payload that smuggled a local `errorKind` or `bufferOverflow` through that spread could masquerade as the runtime's own identity-checked diagnostic at the parent boundary. Local kinds and `bufferOverflow` remain stripped from the snapshot spread, while the provider-owned safety-stop kind is copied only through its explicit closed-literal guard. - `Agent.waitForSteeringArrival(signal)` resolves when steering is queued without consuming it, so wait-style tools can end their observation early. +- Managed fallback provisional-buffer caps are now operator-configurable: `GJC_FALLBACK_MAX_STAGED_EVENTS` (default 10000, hard ceiling 2000000) and `GJC_FALLBACK_MAX_STAGED_BYTES` (default 16 MiB, hard ceiling 1 GiB) bound the events/bytes staged by the provisional staging transaction in both managed fallback and ordinary (non-managed lossless) sessions; in non-managed sessions the cap only decides how much reasoning buffers before the batch flushes and streams through. Values are read once per attempt; the trusted environment resolver ignores surrounding whitespace, while invalid or non-positive values fall back to the defaults, and values above the ceiling clamp to it with a warning so the staging guard stays bounded instead of trading a typed `local_buffer_overflow` for a process OOM. Every retained batch item — including the assistant message/event pair staged for streaming callbacks — is measured and charged against the caps BEFORE it is retained, so actual retention can never exceed the counted bounds, and the ceilings are set from total retained memory (2,000,000 events / 1 GiB) at values an ordinary host survives. The knobs resolve from trusted environment sources only (`$credentialEnv`, which excludes the repository `cwd/.env` overlay), so a project cannot weaken or weaponize the staging guard. Raise both to survive reasoning-heavy streaming in long-running sessions and `gjc team` workers (#4602, #4618). ### Fixed - The escaped-non-ASCII argument guard keeps its fail-closed terminal rejection and its unconditional two-resample budget for every tool and every field. After the budget is spent, one narrowly scoped exemption applies: a tool that enumerated its user-facing display fields (`displaySafeEscapedArgFields`; `ask` exempts only `questions.question` and `questions.options.label`) executes when every non-ASCII character lives inside those fields and is benign typographic punctuation (curated set: U+2014 em-dash). Escaped non-ASCII anywhere else — ids, deep-interview metadata, persisted records, non-ASCII object keys — and every other tool stays rejected terminally (#4627, reduced per both maintainer reviews: guard retained, exemption post-budget and field-scoped). diff --git a/packages/agent/src/agent-loop.ts b/packages/agent/src/agent-loop.ts index 07638696db..4daddc8736 100644 --- a/packages/agent/src/agent-loop.ts +++ b/packages/agent/src/agent-loop.ts @@ -33,7 +33,7 @@ import { stripUnusableReasoningItems, } from "@gajae-code/ai/utils"; import { isCursorExecResolved } from "@gajae-code/ai/utils/block-symbols"; -import { logger, sanitizeText } from "@gajae-code/utils"; +import { $credentialEnv, logger, sanitizeText } from "@gajae-code/utils"; import type { AttemptScope } from "./attempt-scope"; import { createHarmonyAuditEvent, @@ -98,6 +98,154 @@ const intrinsicReflectApply = Reflect.apply; export const MANAGED_ATTEMPT_MAX_STAGED_EVENTS = 10_000; export const MANAGED_ATTEMPT_MAX_STAGED_BYTES = 16 * 1024 * 1024; +/** + * Hard ceilings for the operator overrides. The caps exist to bound memory, so + * an override may raise them only within a range that still leaves the guard + * meaningful — near-`MAX_SAFE_INTEGER` values would trade a typed, bounded + * `local_buffer_overflow` for a process OOM, which is strictly harder to + * diagnose. Above-ceiling overrides clamp to the ceiling with a warning + * instead of being honored. + * + * The ceilings are derived from a survivable PEAK-RSS budget, not from the + * counted-bytes number: peak resident memory holds the live payload, its + * detached snapshot, and the retained batch simultaneously, so it is a + * multiple of the counted bytes. Sizing itself is walk-based (no JSON string + * or UTF-8 copy is materialized to measure), which is why the factor below + * covers the live value plus one detached copy plus batch retention with + * headroom. The bytes ceiling is the peak budget divided by that multiplier, + * so an override at the ceiling still fits an ordinary host. The events + * ceiling is the object-count equivalent for the same budget at a + * conservative per-item floor. + */ +export const MANAGED_STAGED_PEAK_RSS_BUDGET_BYTES = 4 * 1024 * 1024 * 1024; +export const MANAGED_STAGED_PEAK_RSS_FACTOR = 4; +export const MANAGED_ATTEMPT_STAGED_EVENTS_CEILING = 2_000_000; +export const MANAGED_ATTEMPT_STAGED_BYTES_CEILING = Math.floor( + MANAGED_STAGED_PEAK_RSS_BUDGET_BYTES / MANAGED_STAGED_PEAK_RSS_FACTOR, +); + +/** + * Warn once per distinct (knob, requested value) per process. The caps are + * re-read for every managed transaction — every streaming turn — so an + * unmemoized warning would re-log the same oversized operator value once per + * turn for the life of the process, embedding the full requested string in + * every record (log amplification). A bounded digest is logged instead of + * the raw value for the same reason. + */ +const clampedCapWarnings = new Set(); + +function warnClampedStagedCap( + name: "GJC_FALLBACK_MAX_STAGED_EVENTS" | "GJC_FALLBACK_MAX_STAGED_BYTES", + requested: number | string, + ceiling: number, +): void { + // A parsed number is already bounded; only the raw decimal string (which + // the beyond-safe-integer path can supply at arbitrary length) is reduced + // to a length-and-prefix digest before it is embedded in a log record. + const requestedPayload = + typeof requested === "number" ? requested : `${requested.length} digits (starts ${requested.slice(0, 8)})`; + const key = `${name}:${String(requestedPayload)}`; + if (clampedCapWarnings.has(key)) return; + clampedCapWarnings.add(key); + logger.warn(`${name} clamped to ${ceiling}: the provisional staging guard must stay bounded`, { + requested: requestedPayload, + ceiling, + }); +} + +function clampedStagedCap( + name: "GJC_FALLBACK_MAX_STAGED_EVENTS" | "GJC_FALLBACK_MAX_STAGED_BYTES", + fallback: number, + ceiling: number, +): number { + // Resolve from TRUSTED environment sources only ($credentialEnv excludes the + // caller's cwd/.env overlay): these knobs ARE a defensive resource guard, so + // a repository-controlled .env must not be able to weaken (or tighten into + // failure) the staging bound. Values must be positive integers (digits only + // after the trusted resolver's surrounding-whitespace normalization); + // anything else falls back to the default. Any digits-only positive + // decimal that is at or below the ceiling is honored verbatim, and any + // digits-only positive decimal above the ceiling — including ones beyond + // Number.MAX_SAFE_INTEGER, which a numeric parse would misclassify — clamps + // to the ceiling with a warning, exactly as documented. + const raw = $credentialEnv(name)?.trim(); + if (raw === undefined) return fallback; + const parsed = parsePositiveEnvInt(raw); + if (parsed !== undefined) { + if (parsed <= ceiling) return parsed; + warnClampedStagedCap(name, parsed, ceiling); + return ceiling; + } + if (isPositiveDecimalDigits(raw) && decimalAtLeast(raw, ceiling + 1)) { + warnClampedStagedCap(name, raw, ceiling); + return ceiling; + } + return fallback; +} + +function parsePositiveEnvInt(raw: string): number | undefined { + if (!raw || !/^\d+$/.test(raw)) return undefined; + const parsed = Number(raw); + return Number.isSafeInteger(parsed) && parsed > 0 ? parsed : undefined; +} + +/** True when the value is a digits-only positive decimal string (no sign). */ +function isPositiveDecimalDigits(raw: string): boolean { + return raw.length > 0 && /^\d+$/.test(raw) && raw.replace(/^0+/, "") !== ""; +} + +/** + * Lexical comparison of a digits-only decimal against a numeric threshold, + * valid past Number.MAX_SAFE_INTEGER: compare stripped-leading-zero digit + * length first, then digit by digit. + */ +function decimalAtLeast(raw: string, threshold: number): boolean { + const digits = raw.replace(/^0+/, ""); + const thresholdDigits = String(threshold).replace(/^0+/, ""); + if (digits.length !== thresholdDigits.length) return digits.length > thresholdDigits.length; + return digits >= thresholdDigits; +} + +/** + * Max events staged by a provisional managed-attempt transaction before it is + * rejected. Configurable via `GJC_FALLBACK_MAX_STAGED_EVENTS` (default + * `MANAGED_ATTEMPT_MAX_STAGED_EVENTS`, ceiling + * `MANAGED_ATTEMPT_STAGED_EVENTS_CEILING`). Read once per transaction so + * operators can raise the cap without a rebuild and tests can exercise the + * knob in-process. Values must be positive integers after the trusted + * resolver ignores surrounding whitespace; invalid or + * non-positive values fall back to the default, and values above the ceiling + * clamp to it with a warning. + * + * @internal + */ +export function managedAttemptMaxStagedEvents(): number { + return clampedStagedCap( + "GJC_FALLBACK_MAX_STAGED_EVENTS", + MANAGED_ATTEMPT_MAX_STAGED_EVENTS, + MANAGED_ATTEMPT_STAGED_EVENTS_CEILING, + ); +} + +/** + * Max bytes staged by a provisional managed-attempt transaction before it is + * rejected. Configurable via `GJC_FALLBACK_MAX_STAGED_BYTES` (default + * `MANAGED_ATTEMPT_MAX_STAGED_BYTES`, ceiling + * `MANAGED_ATTEMPT_STAGED_BYTES_CEILING`). Read once per transaction; values + * must be positive integers after the trusted resolver ignores surrounding + * whitespace, anything else falls back to the + * default, and values above the ceiling clamp to it with a warning. + * + * @internal + */ +export function managedAttemptMaxStagedBytes(): number { + return clampedStagedCap( + "GJC_FALLBACK_MAX_STAGED_BYTES", + MANAGED_ATTEMPT_MAX_STAGED_BYTES, + MANAGED_ATTEMPT_STAGED_BYTES_CEILING, + ); +} + /** * Closed set of local-failure sites. A bounded diagnostic may name only these * literals: the log is shape-only, so no caller-supplied or provider-derived @@ -1017,25 +1165,418 @@ export function sanitizedDetachedClone(value: T, maxNodes: number = MANAGED_S * detached clone when that validation fails so every accepted snapshot is * both isolated and JSON-serializable. */ -function managedSnapshotJsonBytes(value: unknown): number | undefined { +/** + * Sentinel thrown from inside a size walk the moment the projected size + * crosses the budget. Returning a substituted value (e.g. "") would not + * abort a `JSON.stringify` walk (review finding at 2efaf269cd); throwing is + * the only way to terminate a traversal, and the walk-based oracles below + * rely on the same mechanism to stop before doing unbounded work. + */ +const MANAGED_SIZE_SENTINEL = Symbol("gjc.managed-staging-size-exceeded"); + +/** + * Walk `value`'s JSON surface — exactly the surface `JSON.stringify` sees, + * including `toJSON` dispatch — and return the exact UTF-8 byte length of + * its serialization WITHOUT materializing the JSON string or its UTF-8 + * encoding. Every serialized token is charged: quotes, escapes, separators, + * delimiters, nulls, array holes, and keys. + * + * A LONE surrogate (an unpaired UTF-16 unit; `codePointAt` yields the unit + * itself only when it is unpaired) is charged as the six-byte `\udXXX` + * escape `JSON.stringify` emits for it, not as its 3-byte UTF-8 encoding — + * the previous BMP charge undercounted surrogate-heavy strings by ~2x + * (exact-head 078e22c0 finding 2). + * + * Returns the byte count, `undefined` when the value cannot be serialized + * (cyclic or JSON-hostile), or throws {@link MANAGED_SIZE_SENTINEL} once + * the projected count crosses `limit`. + */ +/** + * Charge a string's exact serialized UTF-8 byte length: opening/closing + * quotes, per-code-point escaping, and lone-surrogate six-byte escapes. + * Printable-ASCII strings without `"` or `\` — the dominant case for + * streamed text, thinking, and tool-argument content — encode one byte per + * UTF-16 unit with no escapes, so they take a single native scan instead of + * a per-code-point JS loop. This keeps the walk-based oracles at native + * `JSON.stringify` cost for ordinary payloads instead of paying the slow + * path on every streaming delta. + */ +const MANAGED_PLAIN_ASCII = /[^\x20-\x21\x23-\x5b\x5d-\x7e]/; +function managedChargeStringBytes(text: string, add: (bytes: number) => void): void { + add(1); + if (!MANAGED_PLAIN_ASCII.test(text)) { + add(text.length); + add(1); + return; + } + for (let index = 0; index < text.length; ) { + const codePoint = text.codePointAt(index); + if (codePoint === undefined) throw new Error("missing string code point"); + if (codePoint === 0x22 || codePoint === 0x5c) add(2); + else if ( + codePoint === 0x08 || + codePoint === 0x09 || + codePoint === 0x0a || + codePoint === 0x0c || + codePoint === 0x0d + ) + add(2); + else if (codePoint <= 0x1f) add(6); + else if (codePoint >= 0xd800 && codePoint <= 0xdfff) add(6); + else if (codePoint <= 0x7f) add(1); + else if (codePoint <= 0x7ff) add(2); + else if (codePoint <= 0xffff) add(3); + else add(4); + index += codePoint > 0xffff ? 2 : 1; + } + add(1); +} + +function managedJsonByteLengthWithin(value: unknown, limit: number): number | undefined { + let seen = 0; + const add = (bytes: number): void => { + seen += bytes; + if (seen > limit) throw MANAGED_SIZE_SENTINEL; + }; + const addString = (text: string): void => managedChargeStringBytes(text, add); + const seenObjects = new WeakSet(); + const prepare = (input: unknown, key: string): { omitted: boolean; value?: unknown } => { + if ((typeof input !== "object" || input === null) && typeof input !== "function") { + return { omitted: false, value: input }; + } + try { + const toJSON = (input as { toJSON?: unknown }).toJSON; + const value = typeof toJSON === "function" ? toJSON.call(input, key) : input; + return { + omitted: value === undefined || typeof value === "function" || typeof value === "symbol", + value, + }; + } catch { + throw new Error("JSON toJSON failed"); + } + }; + const walkPrepared = (input: unknown, inArray: boolean): boolean => { + if (input === null) { + add(4); + return true; + } + if (input === undefined || typeof input === "function" || typeof input === "symbol") { + if (inArray) add(4); + return inArray; + } + if (typeof input === "string") { + addString(input); + return true; + } + if (typeof input === "boolean") { + add(input ? 4 : 5); + return true; + } + if (typeof input === "number") { + const encoded = JSON.stringify(input); + if (encoded === undefined) throw new Error("JSON number failed"); + add(managedAttemptTextEncoder.encode(encoded).byteLength); + return true; + } + if (typeof input === "bigint") throw new Error("JSON bigint failed"); + if (typeof input !== "object") throw new Error("JSON value failed"); + if (seenObjects.has(input)) throw new Error("JSON cycle detected"); + seenObjects.add(input); + try { + if (Array.isArray(input)) { + add(1); + for (let index = 0; index < input.length; index++) { + if (index > 0) add(1); + const prepared = prepare(input[index], String(index)); + if (prepared.omitted) add(4); + else walkPrepared(prepared.value, true); + } + add(1); + return true; + } + add(1); + let emitted = 0; + const record = input as Record; + for (const property of Object.keys(input)) { + const prepared = prepare(record[property], property); + // `JSON.stringify` omits undefined-valued record properties + // entirely (key, colon, and separator); charging them would + // overestimate and falsely reject healthy payloads. + if (prepared.omitted || prepared.value === undefined) continue; + if (emitted > 0) add(1); + emitted++; + addString(property); + add(1); + walkPrepared(prepared.value, false); + } + add(1); + return true; + } finally { + seenObjects.delete(input); + } + }; try { - const serialized = JSON.stringify(value); - return serialized === undefined ? undefined : managedAttemptTextEncoder.encode(serialized).byteLength; - } catch { + const prepared = prepare(value, ""); + if (prepared.omitted) return undefined; + walkPrepared(prepared.value, false); + return seen; + } catch (error) { + if (error === MANAGED_SIZE_SENTINEL) throw error; return undefined; } } -function managedAttemptSnapshotDetailed(value: T): { snapshot: T; jsonBytes?: number; sanitized: boolean } { +/** + * Pre-allocation size guard: reports whether serializing `value` as JSON + * would exceed `limit` bytes WITHOUT materializing the full JSON string or + * cloning the value. Walks the JSON surface directly and charges every + * token, including quotes, escapes, separators, delimiters, nulls, and + * array holes. Strings are charged by code point, so a large string never + * needs a second full-size escaped copy just to measure it. + * + * Returns "over" when the limit would be exceeded, "under" when it + * definitely is not, and "unknown" when the value cannot be serialized at + * all (cyclic or JSON-hostile), which callers treat exactly like the + * existing `undefined` measurement results. + */ +function managedSnapshotExceedsBytes(value: unknown, limit: number): "over" | "under" | "unknown" { + try { + const bytes = managedJsonByteLengthWithin(value, limit); + return bytes === undefined ? "unknown" : "under"; + } catch (error) { + if (error === MANAGED_SIZE_SENTINEL) return "over"; + return "unknown"; + } +} + +/** + * Per-node minimum charge for the structuredClone preflight. Cloning + * duplicates the object GRAPH — per-node headers, Map/Set entries, buffer + * contents — while immutable strings are only ever referenced, so a graph + * of millions of tiny nodes is cheap in counted JSON bytes yet allocates a + * large duplicate. Charging a per-node floor (a conservative minimum object + * size, far above the few JSON bytes such nodes serialize to) keeps the + * preflight an allocation bound, not just a serialization bound. + */ +const MANAGED_CLONE_NODE_OVERHEAD_BYTES = 64; + +/** Widest JSON literal any element of this typed-array kind can produce. */ +function managedTypedArrayJsonDigits(view: unknown): number { + if (view instanceof Uint8Array || view instanceof Int8Array || view instanceof Uint8ClampedArray) return 4; + if (view instanceof Uint16Array || view instanceof Int16Array) return 6; + if (view instanceof Uint32Array || view instanceof Int32Array || view instanceof Float32Array) return 11; + return 24; +} + +/** + * Preflight the CLONE-VISIBLE surface — the graph `structuredClone` would + * actually duplicate — against `limit` WITHOUT cloning. The JSON-surface + * walk dispatches `toJSON`, so a live class can serialize compactly while + * `structuredClone` (which drops the prototype serializer) would copy a + * large own payload; this walk never dispatches `toJSON` and charges the + * clone-visible graph instead. It also charges clone-only allocations the + * JSON walk cannot see — Map/Set entries, ArrayBuffer/TypedArray bytes, + * bigint magnitudes — plus the per-node header floor. + * + * Reads go through own-property descriptors only, so hostile accessors are + * never invoked: an accessor's cloned size cannot be known without invoking + * it, and a value `structuredClone` cannot duplicate at all (functions, + * symbols) fails the clone anyway. Both surface as `"degrade"`, which + * callers divert to the bounded sanitizer walk. + * + * Returns `"over"` when the clone-visible surface exceeds `limit`, + * `"degrade"` when the surface cannot be bounded by walking it, and + * `"under"` when the clone allocation is bounded. + */ +function managedCloneSurfaceExceedsBudget(value: unknown, limit: number): "over" | "under" | "degrade" { + let seen = 0; + const add = (bytes: number): void => { + seen += bytes; + if (seen > limit) throw MANAGED_SIZE_SENTINEL; + }; + const addString = (text: string): void => managedChargeStringBytes(text, add); + const readOwnValue = (input: object, key: string): { accessor: boolean; value?: unknown } => { + const descriptor = Object.getOwnPropertyDescriptor(input, key); + if (descriptor === undefined) return { accessor: false, value: undefined }; + if (!("value" in descriptor)) return { accessor: true }; + return { accessor: false, value: descriptor.value }; + }; + const seenObjects = new WeakSet(); + const walk = (input: unknown): void => { + if (input === null) { + add(4); + return; + } + if (typeof input === "function" || typeof input === "symbol") { + // structuredClone cannot duplicate these; degrade to the bounded + // sanitizer instead of discovering the failure by cloning. + throw new Error("clone surface uncloneable"); + } + if (typeof input === "undefined") return; + if (typeof input === "string") { + addString(input); + return; + } + if (typeof input === "number") { + add(managedAttemptTextEncoder.encode(JSON.stringify(input)).byteLength); + return; + } + if (typeof input === "boolean") { + add(input ? 4 : 5); + return; + } + if (typeof input === "bigint") { + add(MANAGED_CLONE_NODE_OVERHEAD_BYTES + input.toString().length); + return; + } + if (typeof input !== "object") return; + // Refuse proxies by internal-slot brand BEFORE any reflective operation: + // `Object.keys`/`getOwnPropertyDescriptor` on a live proxy would dispatch + // its `ownKeys`/descriptor traps, and on a revoked proxy would throw — + // either way the walk must not run payload-controlled code. The bounded + // sanitizer collapses proxies to a placeholder without dispatching traps. + if (nodeUtilTypes.isProxy(input)) throw new Error("clone surface proxy"); + if (seenObjects.has(input)) throw new Error("clone surface cycle"); + seenObjects.add(input); + try { + add(MANAGED_CLONE_NODE_OVERHEAD_BYTES); + if (nodeUtilTypes.isDate(input)) { + add(32); + return; + } + if (nodeUtilTypes.isRegExp(input)) { + add(2 + (input as RegExp).source.length); + return; + } + if (nodeUtilTypes.isMap(input)) { + for (const [entryKey, entryValue] of Map.prototype.entries.call(input as Map)) { + add(MANAGED_CLONE_NODE_OVERHEAD_BYTES); + walk(entryKey); + walk(entryValue); + } + return; + } + if (nodeUtilTypes.isSet(input)) { + for (const element of Set.prototype.values.call(input as Set)) { + add(MANAGED_CLONE_NODE_OVERHEAD_BYTES); + walk(element); + } + return; + } + if (nodeUtilTypes.isArrayBuffer(input)) { + add((input as ArrayBuffer).byteLength); + return; + } + if (nodeUtilTypes.isTypedArray(input)) { + const view = input as unknown as Uint8Array; + add(view.byteLength + view.length * managedTypedArrayJsonDigits(view)); + return; + } + if (nodeUtilTypes.isDataView(input)) { + add((input as DataView).byteLength); + return; + } + if (Array.isArray(input)) { + add(2); + for (let index = 0; index < input.length; index++) { + if (index > 0) add(1); + const read = readOwnValue(input, String(index)); + if (read.accessor) throw new Error("clone surface accessor"); + if (!Object.hasOwn(input, String(index))) + add(4); // array hole + else walk(read.value); + } + // Non-index own enumerable keys are cloned (allocation) even + // though JSON.stringify omits them from arrays; charge their + // graph so the preflight stays an allocation bound. + for (const key of Object.keys(input)) { + if (String(Number(key)) === key && Number(key) >= 0) continue; + const read = readOwnValue(input, key); + if (read.accessor) throw new Error("clone surface accessor"); + addString(key); + walk(read.value); + } + return; + } + add(2); + let emitted = 0; + for (const key of Object.keys(input)) { + const read = readOwnValue(input, key); + if (read.accessor) throw new Error("clone surface accessor"); + if (emitted > 0) add(1); + emitted++; + addString(key); + add(1); + walk(read.value); + } + return; + } finally { + seenObjects.delete(input); + } + }; + try { + walk(value); + return "under"; + } catch (error) { + if (error === MANAGED_SIZE_SENTINEL) return "over"; + return "degrade"; + } +} + +/** + * Ceiling for unbounded standalone snapshot sizing (the shell paths outside + * a transaction). Shared by the transaction-cap default so a standalone + * snapshot is never larger than the default transaction budget. + */ +function currentStagedBytesCap(): number { + return managedAttemptMaxStagedBytes(); +} + +/** + * Exact serialized size of a staged snapshot, computed by walking the JSON + * surface. Replaces the previous `JSON.stringify` + `TextEncoder.encode` + * measurement, which materialized a full copy of the serialized value — and + * a second copy of its UTF-8 encoding — BEFORE the cap check could reject + * it: the budget-sized transient allocation the memory guard exists to + * prevent (exact-head 078e22c0 finding 1). Returns `undefined` when the + * value cannot be serialized, matching the previous measurement's failure + * mode so callers keep their sanitize fallbacks. + * + * @internal + */ +export function managedSnapshotJsonByteLength(value: unknown): number | undefined { + return managedJsonByteLengthWithin(value, Number.POSITIVE_INFINITY); +} + +function managedAttemptSnapshotDetailed( + value: T, + byteLimit?: number, +): { + snapshot: T; + jsonBytes?: number; + sanitized: boolean; +} { + // Preflight the CLONE-VISIBLE surface so the structuredClone allocation + // itself is bounded: a live class can serialize compactly through a + // prototype `toJSON()` that structuredClone drops, so a JSON-surface + // precheck alone cannot bound what the clone would duplicate — the clone + // allocated the over-cap duplicate before the typed overflow could run + // (exact-head 078e22c0 finding 1). Degrade through the bounded sanitizer + // walk, which never clones, never dispatches accessors, and is + // node-budgeted, instead of allocating the duplicate. + if (managedCloneSurfaceExceedsBudget(value, byteLimit ?? currentStagedBytesCap()) !== "under") { + const bounded = sanitizedDetachedClone(value); + return { snapshot: bounded, jsonBytes: managedSnapshotJsonByteLength(bounded), sanitized: true }; + } try { const snapshot = structuredClone(value); - const jsonBytes = managedSnapshotJsonBytes(snapshot); + const jsonBytes = managedSnapshotJsonByteLength(snapshot); if (jsonBytes !== undefined) return { snapshot, jsonBytes, sanitized: false }; const sanitized = sanitizedDetachedClone(snapshot); - return { snapshot: sanitized, jsonBytes: managedSnapshotJsonBytes(sanitized), sanitized: true }; + return { snapshot: sanitized, jsonBytes: managedSnapshotJsonByteLength(sanitized), sanitized: true }; } catch { const snapshot = sanitizedDetachedClone(value); - return { snapshot, jsonBytes: managedSnapshotJsonBytes(snapshot), sanitized: true }; + return { snapshot, jsonBytes: managedSnapshotJsonByteLength(snapshot), sanitized: true }; } } @@ -1077,8 +1618,19 @@ const LOSSLESS_SNAPSHOT_KEYS = [ * A failed subtree is removed at its own property boundary; siblings retain * their exact structured-clone representation. The bounded recursive path is * used only after cloning the complete value fails. + * + * The `structuredClone` allocation is preflighted against the staged-bytes + * cap: a live payload class can serialize compactly through a prototype + * `toJSON()` the clone drops, so the JSON-surface budget alone does not + * bound what the clone duplicates. An over-budget clone surface collapses to + * the bounded sanitizer instead of allocating the duplicate — ordinary + * sessions never see this, because their transaction flushes and streams the + * live event through before reaching this clone. */ function losslessDetachedClone(value: T): T { + if (managedCloneSurfaceExceedsBudget(value, currentStagedBytesCap()) === "over") { + return sanitizedDetachedClone(value); + } try { const snapshot = structuredClone(value); // `structuredClone()` preserves own bigint fields while removing a @@ -1086,7 +1638,7 @@ function losslessDetachedClone(value: T): T { // serialize successfully while its detached clone cannot be staged. // Lossless staging still preserves every JSON-safe clone verbatim; only // the non-serializable detached form is sanitized. - return managedSnapshotJsonBytes(snapshot) !== undefined ? snapshot : sanitizedDetachedClone(snapshot); + return managedSnapshotJsonByteLength(snapshot) !== undefined ? snapshot : sanitizedDetachedClone(snapshot); } catch { // The managed sanitizer is explicitly bounded and total. Use it only to // identify which top-level assistant metadata surfaces are cloneable; the @@ -1127,7 +1679,7 @@ function losslessDetachedClone(value: T): T { } } } - return managedSnapshotJsonBytes(output) !== undefined ? (output as T) : sanitizedDetachedClone(output as T); + return managedSnapshotJsonByteLength(output) !== undefined ? (output as T) : sanitizedDetachedClone(output as T); } } @@ -1476,7 +2028,7 @@ function warnManagedSnapshotFailure( */ type ManagedAttemptBatchItem = | { type: "event"; event: AgentEvent; bytes?: number } - | { type: "assistant_event"; message: AssistantMessage; event: AssistantMessageEvent }; + | { type: "assistant_event"; message: AssistantMessage; event: AssistantMessageEvent; bytes?: number }; /** * Streaming increments whose complete value is re-published by the block's own @@ -1501,6 +2053,9 @@ class ManagedAttemptTransaction { #batch: ManagedAttemptBatchItem[] = []; #stagedEventCount = 0; #stagedBytes = 0; + /** Caps for this transaction, read once from the operator env knobs. */ + readonly #maxStagedEvents = managedAttemptMaxStagedEvents(); + readonly #maxStagedBytes = managedAttemptMaxStagedBytes(); /** Shape snapshot retained across discard() for bounded failure diagnostics. */ #lastStagedShape: { stagedEventCount: number; stagedBytes: number; contentBlockCount: number } | undefined; #discarded = false; @@ -1534,16 +2089,103 @@ class ManagedAttemptTransaction { } stageAssistantMessageEvent(message: AssistantMessage, event: AssistantMessageEvent): void { - const partial = this.#assistantSnapshot(message); if (this.#committed) { - this.onAssistantMessageEvent?.(partial, this.#assistantEventSnapshot(event, partial)); + // Already published: nothing is retained, so the live pair can go + // straight to the consumer without a staging measurement. One + // snapshot serves as BOTH the callback message and the event's + // `partial`, preserving the paired-snapshot identity the direct + // callbacks were built on and avoiding a second full clone of the + // growing message. + const committedPartial = this.#assistantSnapshot(message); + this.onAssistantMessageEvent?.(committedPartial, this.#assistantEventSnapshot(event, committedPartial)); return; } + // Every retained batch item must be charged against the caps BEFORE it + // is retained, including the assistant message/event pair: an uncharged + // snapshot would let actual retention exceed the caps while the counters + // still read under them. Two-phase guard so the allocation that could + // OOM never happens ahead of the check: + // 1. INCREMENTAL pre-check on the LIVE pair — a replacer walk that + // stops as soon as the projected size crosses the cap, without + // materializing the full JSON string or its UTF-8 encoding (review: + // "size incrementally so serialization can stop before + // materializing the whole value"). Only if the walk completes under + // the cap is any snapshot taken. + // 2. EXACT accounting of the retained detached pair — a live class can + // serialize compactly through a prototype `toJSON()` that + // `structuredClone` drops, so the live measurement may undercount + // what the retained snapshot actually holds (same convention as + // `#stage`). + const liveBudget = this.#maxStagedBytes - this.#stagedBytes; + const liveExcess = managedSnapshotExceedsBytes([message, event], liveBudget); + if (liveExcess === "over") { + this.#compactSupersededFrames(); + if (managedSnapshotExceedsBytes([message, event], this.#maxStagedBytes - this.#stagedBytes) === "over") { + if (this.snapshotMode === "lossless") { + this.flush(); + // One snapshot for the whole callback pair (see the committed + // branch above): the callback message and `event.partial` + // must be the same object. + const flushedPartial = this.#assistantSnapshot(message); + this.onAssistantMessageEvent?.(flushedPartial, this.#assistantEventSnapshot(event, flushedPartial)); + return; + } + // Report the POST-compaction remaining budget + 1 (a valid lower + // bound for the live pair's size, and arithmetically consistent + // with the post-compaction retained shape #overflowShape reports). + this.discard(); + throw new ManagedAttemptBufferOverflowError( + "overflow.staged", + this.#overflowShape("overflow.staged", this.#maxStagedBytes - this.#stagedBytes + 1), + ); + } + } + const partial = this.#assistantSnapshot(message); + const snapshotEvent = this.#assistantEventSnapshot(event, partial); + // Walk-based exact measure (no JSON string or UTF-8 copy materialized). + const retainedBytes = managedSnapshotJsonByteLength([partial, snapshotEvent]); + if (retainedBytes === undefined) { + // Fail CLOSED, exactly like the #stage twin on the same condition: an + // unmeasurable retained pair must not be retained uncharged (a 0-byte + // charge plus the skipped byte-cap gate would let actual retention + // exceed the caps while the counters still read under them). The + // snapshot forms are JSON-safe by construction, so this is only + // reachable if that construction regresses; it carries no transport + // facts and never burns the fallback chain. Lossless mode cannot + // fail the attempt: flush what is staged and publish the live pair. + if (this.snapshotMode === "lossless") { + this.flush(); + this.onAssistantMessageEvent?.(partial, snapshotEvent); + return; + } + this.discard(); + throw new ManagedAttemptSnapshotError("staging.measure"); + } + if (this.#wouldOverflow(retainedBytes)) { + this.#compactSupersededFrames(); + if (this.#wouldOverflow(retainedBytes)) { + if (this.snapshotMode === "lossless") { + this.flush(); + this.onAssistantMessageEvent?.(partial, snapshotEvent); + return; + } + this.discard(); + throw new ManagedAttemptBufferOverflowError( + "overflow.staged", + this.#overflowShape("overflow.staged", retainedBytes), + ); + } + } + // Each frame's exact accounted size is retained so compaction can debit + // exactly what it reclaims instead of re-measuring the whole batch. this.#batch.push({ type: "assistant_event", message: partial, - event: this.#assistantEventSnapshot(event, partial), + event: snapshotEvent, + bytes: retainedBytes, }); + this.#stagedEventCount += 1; + this.#stagedBytes += retainedBytes; } flush(): void { @@ -1654,10 +2296,7 @@ class ManagedAttemptTransaction { return 0; } #wouldOverflow(bytes: number): boolean { - return ( - this.#stagedEventCount + 1 > MANAGED_ATTEMPT_MAX_STAGED_EVENTS || - this.#stagedBytes + bytes > MANAGED_ATTEMPT_MAX_STAGED_BYTES - ); + return this.#stagedEventCount + 1 > this.#maxStagedEvents || this.#stagedBytes + bytes > this.#maxStagedBytes; } /** * Shape snapshot for a buffer-overflow diagnostic: the rejecting stage, @@ -1675,16 +2314,20 @@ class ManagedAttemptTransaction { incomingEventBytes: number, ): ManagedAttemptBufferOverflowError["overflow"] { const staged = this.stagedShape(); - const eventsExceeded = staged.stagedEventCount + 1 > MANAGED_ATTEMPT_MAX_STAGED_EVENTS; - const bytesExceeded = staged.stagedBytes + incomingEventBytes > MANAGED_ATTEMPT_MAX_STAGED_BYTES; + // Derive from the transaction's effective (operator-configurable) caps, + // not the module constants: the diagnostic must name the limits that + // actually tripped, which can differ from the defaults when an override + // is active. + const eventsExceeded = staged.stagedEventCount + 1 > this.#maxStagedEvents; + const bytesExceeded = staged.stagedBytes + incomingEventBytes > this.#maxStagedBytes; return { stage, exceeded: eventsExceeded && bytesExceeded ? "both" : eventsExceeded ? "events" : "bytes", stagedEventCount: staged.stagedEventCount, stagedBytes: staged.stagedBytes, incomingEventBytes, - maxStagedEvents: MANAGED_ATTEMPT_MAX_STAGED_EVENTS, - maxStagedBytes: MANAGED_ATTEMPT_MAX_STAGED_BYTES, + maxStagedEvents: this.#maxStagedEvents, + maxStagedBytes: this.#maxStagedBytes, }; } @@ -1718,10 +2361,8 @@ class ManagedAttemptTransaction { retained.push(item); continue; } - if (item.type === "event") { - reclaimedBytes += item.bytes ?? 0; - reclaimedEvents += 1; - } + reclaimedBytes += item.bytes ?? 0; + reclaimedEvents += 1; } if (retained.length === this.#batch.length) return false; this.#batch = retained; @@ -1733,17 +2374,31 @@ class ManagedAttemptTransaction { #stage(event: AgentEvent): void { if (this.snapshotMode === "lossless") { const snapshot = this.#repairAssistantEvent(event); - let rawBytes: number | undefined; - try { - rawBytes = managedAttemptTextEncoder.encode(JSON.stringify(snapshot)).byteLength; - } catch { - rawBytes = undefined; + const rawExcess = managedSnapshotExceedsBytes(snapshot, this.#maxStagedBytes - this.#stagedBytes); + if (rawExcess === "over") { + this.flush(); + this.push(event); + return; } + // Walk-based exact measure: no full JSON string or UTF-8 encoding is + // materialized just to size the candidate (exact-head 078e22c0 + // finding 1 applies to the lossless exact measure too). + const rawBytes = managedSnapshotJsonByteLength(snapshot); if (rawBytes !== undefined && this.#wouldOverflow(rawBytes)) { this.flush(); this.push(event); return; } + // Bound the structuredClone allocation inside the snapshot forms the + // same way the managed path does: a compact `toJSON()` surface can hide + // an over-budget clone-visible payload. Ordinary sessions flush and + // stream the LIVE event through rather than degrading it — the + // documented lossless contract — instead of staging a sanitized copy. + if (managedCloneSurfaceExceedsBudget(event, this.#maxStagedBytes - this.#stagedBytes) === "over") { + this.flush(); + this.push(event); + return; + } let detached: AgentEvent; try { detached = this.#losslessAgentEventSnapshot(snapshot); @@ -1751,10 +2406,8 @@ class ManagedAttemptTransaction { this.discard(); throw new ManagedAttemptSnapshotError("staging.losslessSnapshot"); } - let detachedBytes: number; - try { - detachedBytes = managedAttemptTextEncoder.encode(JSON.stringify(detached)).byteLength; - } catch { + const detachedBytes = managedSnapshotJsonByteLength(detached); + if (detachedBytes === undefined) { this.discard(); throw new ManagedAttemptSnapshotError("staging.measure"); } @@ -1768,19 +2421,22 @@ class ManagedAttemptTransaction { this.#stagedBytes += detachedBytes; return; } - // Measure the raw event FIRST so an oversized payload is rejected - // before the snapshot duplicates it — the staged-byte cap exists to - // bound memory, so cloning ahead of the check would defeat it. - // Cyclic/JSON-hostile events cannot be pre-measured; only those fall - // through to snapshot-then-measure, where the sanitized detached form - // is the cycle-safe estimator. - let bytes: number | undefined; - try { - bytes = managedAttemptTextEncoder.encode(JSON.stringify(event)).byteLength; - } catch { - bytes = undefined; - } - if (bytes !== undefined && this.#wouldOverflow(bytes)) { + // Walk the raw event FIRST so an oversized payload is rejected before the + // managed snapshot duplicates it. Both oracles run against the REMAINING + // budget: the JSON-surface walk bounds the serialized charge, and the + // clone-surface walk bounds the structuredClone ALLOCATION — a live class + // can serialize compactly through `toJSON()` while carrying a large own + // payload the clone would duplicate (exact-head 078e22c0 finding 1). + // Cyclic/JSON-hostile events fall through to the sanitized detached form + // below, which is the cycle-safe estimator. + const preflightOver = (): boolean => { + const remaining = this.#maxStagedBytes - this.#stagedBytes; + return ( + managedSnapshotExceedsBytes(event, remaining) === "over" || + managedCloneSurfaceExceedsBudget(event, remaining) === "over" + ); + }; + if (preflightOver()) { // 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. @@ -1788,23 +2444,49 @@ class ManagedAttemptTransaction { // volume that still cannot fit even after reclamation), because #4610 // made the pre-compaction shape describe deltas it already reclaimed. this.#compactSupersededFrames(); - if (this.#wouldOverflow(bytes)) { + if (preflightOver()) { + // Report the incoming event's real size (a bounded walk — the + // sentinel caps the count at twice the cap, so a hostile shared + // DAG cannot turn the diagnostic itself into unbounded work). + // The previous form passed `remaining + 1` evaluated AFTER + // discard() zeroed the counters, which fabricated the constant + // `maxStagedBytes + 1` and mislabeled every mixed overflow as a + // single-event blowout. + const remainingBytes = this.#maxStagedBytes - this.#stagedBytes; + const incomingBytes = (() => { + try { + // Bounded walk: the sentinel caps even this diagnostic's + // work at twice the cap. The remaining budget + 1 is the + // honest floor either way — the event demonstrably does + // not fit in what remains (that is why it is rejected), + // including when only the clone-visible surface tripped + // while the compact JSON surface reads small. + return Math.max( + managedJsonByteLengthWithin(event, this.#maxStagedBytes * 2) ?? 0, + remainingBytes + 1, + ); + } catch { + // Unserializable or beyond twice the cap: the floor alone + // still arithmetically explains the rejection. + return remainingBytes + 1; + } + })(); this.discard(); throw new ManagedAttemptBufferOverflowError( "overflow.preMeasure", - this.#overflowShape("overflow.preMeasure", bytes), + this.#overflowShape("overflow.preMeasure", incomingBytes), ); } } const repaired = this.#repairAssistantEvent(event); - const detailed = managedAttemptSnapshotDetailed(repaired); + const detailed = managedAttemptSnapshotDetailed(repaired, this.#maxStagedBytes - this.#stagedBytes); 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; + const 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 diff --git a/packages/agent/test/managed-attempt-transaction.test.ts b/packages/agent/test/managed-attempt-transaction.test.ts index 004d0c0ae3..353eb5da62 100644 --- a/packages/agent/test/managed-attempt-transaction.test.ts +++ b/packages/agent/test/managed-attempt-transaction.test.ts @@ -6,6 +6,8 @@ import { MANAGED_ATTEMPT_MAX_STAGED_BYTES, MANAGED_ATTEMPT_MAX_STAGED_EVENTS, managedAssistantEventSnapshot, + managedAttemptMaxStagedBytes, + managedAttemptMaxStagedEvents, sanitizedDetachedClone, } from "@gajae-code/agent-core/agent-loop"; import type { AgentContext, AgentEvent, AgentLoopConfig } from "@gajae-code/agent-core/types"; @@ -31,6 +33,21 @@ function captureSnapshotDiagnostics(): Record[] { return captured; } +/** + * Capture clamp-warning payloads for the staged-cap knobs. Kept separate from + * {@link captureSnapshotDiagnostics} so the two message streams stay + * independently assertable. + */ +function captureStagedCapClampWarnings(): Record[] { + const captured: Record[] = []; + vi.spyOn(logger, "warn").mockImplementation((message: string, payload?: unknown) => { + if (message.startsWith("GJC_FALLBACK_MAX_STAGED_") && message.includes("clamped to")) { + captured.push((payload ?? {}) as Record); + } + }); + return captured; +} + function assistantMessage(model: ReturnType["model"]): AssistantMessage { return { role: "assistant", @@ -77,8 +94,19 @@ function expectManagedRunStart(events: string[]): void { } describe("managed attempt transaction", () => { + // Snapshot the inherited knob values once: the staged-cap knobs change the + // transaction's provisional limits, so a host/CI export must be restored + // (not merely deleted) after each test; tests that need defaults clear the + // variables themselves inside the test. + const inheritedKnobEvents = process.env.GJC_FALLBACK_MAX_STAGED_EVENTS; + const inheritedKnobBytes = process.env.GJC_FALLBACK_MAX_STAGED_BYTES; + afterEach(() => { vi.restoreAllMocks(); + if (inheritedKnobEvents === undefined) delete process.env.GJC_FALLBACK_MAX_STAGED_EVENTS; + else process.env.GJC_FALLBACK_MAX_STAGED_EVENTS = inheritedKnobEvents; + if (inheritedKnobBytes === undefined) delete process.env.GJC_FALLBACK_MAX_STAGED_BYTES; + else process.env.GJC_FALLBACK_MAX_STAGED_BYTES = inheritedKnobBytes; }); it("flushes a successful assistant lifecycle once and in provider order", async () => { @@ -366,6 +394,128 @@ describe("managed attempt transaction", () => { expect(thinking.thinking).toHaveLength(MANAGED_ATTEMPT_MAX_STAGED_BYTES + 1); expect(lifecycle.slice(-5)).toEqual(["message_start", "message_update", "message_end", "turn_end", "agent_end"]); }); + it("honors a low GJC_FALLBACK_MAX_STAGED_EVENTS in ordinary lossless sessions by flushing through", async () => { + // Ordinary (non-managed) runs stage lossless snapshots behind the same + // limiter. With a 2-event cap the third staged frame must degrade to + // pass-through publication — callbacks preserved, lifecycle intact, no + // typed failure — instead of either failing locally or retaining + // unbounded state. + const previous = process.env.GJC_FALLBACK_MAX_STAGED_EVENTS; + process.env.GJC_FALLBACK_MAX_STAGED_EVENTS = "2"; + try { + const mock = createMockModel(); + const streamFn = () => { + const stream = new AssistantMessageEventStream(); + void (async () => { + const partial = assistantMessage(mock.model); + stream.push({ type: "start", partial }); + await Bun.sleep(0); + partial.content.push({ type: "thinking", thinking: "chunk-0" }); + stream.push({ type: "thinking_start", contentIndex: 0, partial }); + await Bun.sleep(0); + partial.content.push({ type: "text", text: "accepted" }); + stream.push({ type: "text_start", contentIndex: 0, partial }); + await Bun.sleep(0); + stream.push({ type: "done", reason: "stop", message: partial }); + })(); + return stream; + }; + const callbacks: AssistantMessageEvent[] = []; + const lifecycle: string[] = []; + const agent = new Agent({ + initialState: { model: mock.model, systemPrompt: ["test"], tools: [], messages: [] }, + streamFn, + onAssistantMessageEvent: (_message, event) => callbacks.push(event), + }); + agent.subscribe(event => lifecycle.push(event.type)); + + await agent.prompt("run"); + + expect(agent.state.error).toBeUndefined(); + const accepted = agent.state.messages.at(-1); + expect(accepted?.role).toBe("assistant"); + // The callback contract is preserved through the flush/pass-through. + expect(callbacks.map(event => event.type)).toContain("thinking_start"); + expect(callbacks.map(event => event.type)).toContain("text_start"); + expect(lifecycle.slice(-6)).toEqual([ + "message_start", + "message_update", + "message_update", + "message_end", + "turn_end", + "agent_end", + ]); + } finally { + if (previous === undefined) delete process.env.GJC_FALLBACK_MAX_STAGED_EVENTS; + else process.env.GJC_FALLBACK_MAX_STAGED_EVENTS = previous; + } + }); + + it("honors a low GJC_FALLBACK_MAX_STAGED_BYTES in ordinary lossless sessions by flushing through", async () => { + const previous = process.env.GJC_FALLBACK_MAX_STAGED_BYTES; + process.env.GJC_FALLBACK_MAX_STAGED_BYTES = "128"; + try { + const mock = createMockModel(); + const streamFn = () => { + const stream = new AssistantMessageEventStream(); + void (async () => { + const partial = assistantMessage(mock.model); + stream.push({ type: "start", partial }); + await Bun.sleep(0); + partial.content.push({ type: "thinking", thinking: "x".repeat(4096) }); + stream.push({ type: "thinking_start", contentIndex: 0, partial }); + await Bun.sleep(0); + stream.push({ type: "done", reason: "stop", message: partial }); + })(); + return stream; + }; + const agent = new Agent({ + initialState: { model: mock.model, systemPrompt: ["test"], tools: [], messages: [] }, + streamFn, + }); + const lifecycle: string[] = []; + agent.subscribe(event => lifecycle.push(event.type)); + + await agent.prompt("run"); + + // The 4 KiB frame exceeds the 128-byte cap: the lossless transaction + // flushes and streams through rather than failing the run. + expect(agent.state.error).toBeUndefined(); + const accepted = agent.state.messages.at(-1); + expect(accepted?.role).toBe("assistant"); + expect(lifecycle.slice(-5)).toEqual([ + "message_start", + "message_update", + "message_end", + "turn_end", + "agent_end", + ]); + } finally { + if (previous === undefined) delete process.env.GJC_FALLBACK_MAX_STAGED_BYTES; + else process.env.GJC_FALLBACK_MAX_STAGED_BYTES = previous; + } + }); + + it("ignores project .env values for the staged-cap knobs", async () => { + // The knobs are a defensive resource guard: a repository-controlled + // .env must not be able to weaken them. $credentialEnv excludes the + // cwd/.env overlay, so only a trusted (process/agent/user) source can + // move these caps. This regression pins the trust boundary by resolving + // through the same trusted resolver the limiter uses. + // A value set in the TRUSTED process environment is honored... + const previous = process.env.GJC_FALLBACK_MAX_STAGED_EVENTS; + process.env.GJC_FALLBACK_MAX_STAGED_EVENTS = "5000"; + try { + expect(managedAttemptMaxStagedEvents()).toBe(5000); + } finally { + if (previous === undefined) delete process.env.GJC_FALLBACK_MAX_STAGED_EVENTS; + else process.env.GJC_FALLBACK_MAX_STAGED_EVENTS = previous; + } + // With no trusted source set, the documented default applies regardless + // of what any project .env may contain. + delete process.env.GJC_FALLBACK_MAX_STAGED_EVENTS; + expect(managedAttemptMaxStagedEvents()).toBe(MANAGED_ATTEMPT_MAX_STAGED_EVENTS); + }); it("preserves lifecycle order when a compact live payload clones above the lossless cap", async () => { // Given a payload whose live serializer is compact but whose detached own @@ -1066,8 +1216,12 @@ describe("managed attempt transaction", () => { // The staged-byte cap exists to bound memory: an over-limit event must // be rejected from its measurement pass alone, WITHOUT first being // duplicated by structuredClone. The nested witness getter counts deep - // reads: measurement reads it exactly once; a snapshot taken before - // the cap check would read it a second time. + // reads. The clone-surface preflight reads only through own-property + // descriptors and refuses accessors outright, so the getter is NEVER + // invoked — zero reads. The previously blocked implementation cloned + // before the cap check, and structuredClone invokes accessors, so it + // read the witness exactly once: any read at all is the regression + // signal; zero is the strengthened invariant. const diagnostics = captureSnapshotDiagnostics(); const mock = createMockModel(); let witnessReads = 0; @@ -1109,7 +1263,7 @@ describe("managed attempt transaction", () => { expect(outcomeCalls).toBe(0); expect(agent.state.error).toContain("provisional event buffer limit"); expect((agent.state.messages.at(-1) as AssistantMessage).errorKind).toBe("local_buffer_overflow"); - expect(witnessReads).toBe(1); + expect(witnessReads).toBe(0); // One bounded diagnostic per stream invocation, shape-only. expect(diagnostics).toHaveLength(1); expect(diagnostics[0]).toMatchObject({ @@ -1119,6 +1273,106 @@ describe("managed attempt transaction", () => { snapshotMode: "managed", }); }); + it("rejects a toJSON-hidden clone-visible payload before any snapshot allocation", async () => { + // Exact-head 078e22c0 blocker: the live JSON surface (which dispatches + // `toJSON`) can be tiny while the clone-visible own payload the JSON + // walk never sees is huge. The rejection must come from the PRE-FLIGHT + // walks — stage `overflow.preMeasure`, before any snapshot work — not + // from `overflow.staged` after structuredClone has already duplicated + // the oversized payload. On the previously blocked head the JSON walk + // reads `{compact:true}` as under-budget, the clone allocates the full + // hidden payload, and only the detached measurement rejects it: the + // allocation the cap exists to prevent happened ahead of the guard. + const mock = createMockModel(); + const streamFn = () => { + const stream = new AssistantMessageEventStream(); + queueMicrotask(() => { + const partial = assistantMessage(mock.model); + (partial as unknown as Record).providerPayload = { + envelope: new CompactLargeEnvelope(), + }; + stream.push({ type: "start", partial }); + }); + return stream; + }; + const agent = new Agent({ + initialState: { model: mock.model, systemPrompt: ["test"], tools: [], messages: [] }, + streamFn, + }); + let outcomeCalls = 0; + const previous = process.env.GJC_FALLBACK_MAX_STAGED_BYTES; + process.env.GJC_FALLBACK_MAX_STAGED_BYTES = "2048"; + try { + await agent.prompt("run", { + fallbackManaged: true, + onManagedAttemptOutcome: () => { + outcomeCalls += 1; + return { type: "retry", continuation: () => {} }; + }, + }); + await agent.waitForIdle(); + expect(outcomeCalls).toBe(0); + expect(agent.state.error).toContain("provisional event buffer limit"); + const terminal = agent.state.messages.at(-1) as AssistantMessage; + expect(terminal.errorKind).toBe("local_buffer_overflow"); + const overflow = (terminal as unknown as { bufferOverflow?: { stage: string } }).bufferOverflow; + expect(overflow?.stage).toBe("overflow.preMeasure"); + } finally { + if (previous === undefined) delete process.env.GJC_FALLBACK_MAX_STAGED_BYTES; + else process.env.GJC_FALLBACK_MAX_STAGED_BYTES = previous; + } + }); + + it("counts lone surrogates as their six-byte JSON escape when bounding staging", async () => { + // Exact-head 078e22c0 blocker: a lone surrogate encodes to 3 UTF-8 + // bytes but `JSON.stringify` emits a six-byte `\udXXX` escape for it, + // so charging 3 undercounted surrogate-heavy strings by ~2x and let + // them pass the pre-check into a full serialization. The walk must + // charge the escape: 100 000 lone surrogates are 300 002 bytes at the + // old 3-byte charge (under the 400 KiB cap used here) but 600 002 as + // serialized JSON (over it). On the previously blocked head the + // pre-check passed and the detached measurement rejected the payload + // as `overflow.staged` after the fact; the walk must reject it up + // front as `overflow.preMeasure`. + const mock = createMockModel(); + const streamFn = () => { + const stream = new AssistantMessageEventStream(); + queueMicrotask(() => { + const partial = assistantMessage(mock.model); + partial.content.push({ type: "text", text: "\uD800".repeat(100_000) }); + stream.push({ type: "start", partial }); + }); + return stream; + }; + const agent = new Agent({ + initialState: { model: mock.model, systemPrompt: ["test"], tools: [], messages: [] }, + streamFn, + }); + let outcomeCalls = 0; + const previous = process.env.GJC_FALLBACK_MAX_STAGED_BYTES; + process.env.GJC_FALLBACK_MAX_STAGED_BYTES = String(400 * 1024); + try { + await agent.prompt("run", { + fallbackManaged: true, + onManagedAttemptOutcome: () => { + outcomeCalls += 1; + return { type: "retry", continuation: () => {} }; + }, + }); + await agent.waitForIdle(); + expect(outcomeCalls).toBe(0); + expect(agent.state.error).toContain("provisional event buffer limit"); + const terminal = agent.state.messages.at(-1) as AssistantMessage; + expect(terminal.errorKind).toBe("local_buffer_overflow"); + const overflow = (terminal as unknown as { bufferOverflow?: { stage: string; maxStagedBytes: number } }) + .bufferOverflow; + expect(overflow?.stage).toBe("overflow.preMeasure"); + expect(overflow?.maxStagedBytes).toBe(400 * 1024); + } finally { + if (previous === undefined) delete process.env.GJC_FALLBACK_MAX_STAGED_BYTES; + else process.env.GJC_FALLBACK_MAX_STAGED_BYTES = previous; + } + }); it("fails an over-limit provisional batch as a local error without consuming the chain", async () => { const mock = createMockModel(); @@ -1257,6 +1511,372 @@ describe("managed attempt transaction", () => { const textEnd = callbacks.find(event => event.type === "text_end"); expect(textEnd).toMatchObject({ type: "text_end", contentIndex: 0, content: fullText }); }); + it("honors GJC_FALLBACK_MAX_STAGED_EVENTS from the environment", async () => { + const previous = process.env.GJC_FALLBACK_MAX_STAGED_EVENTS; + process.env.GJC_FALLBACK_MAX_STAGED_EVENTS = "2"; + try { + const mock = createMockModel(); + const streamFn = () => { + const stream = new AssistantMessageEventStream(); + queueMicrotask(() => { + let last: AssistantMessage | undefined; + for (let i = 0; i < 5; i += 1) { + const partial = assistantMessage(mock.model); + partial.content.push({ type: "text", text: `chunk-${i}` }); + last = partial; + stream.push({ type: "start", partial }); + } + stream.end(last); + }); + return stream; + }; + const agent = new Agent({ + initialState: { model: mock.model, systemPrompt: ["test"], tools: [], messages: [] }, + streamFn, + }); + let outcomeCalls = 0; + await agent.prompt("run", { + fallbackManaged: true, + onManagedAttemptOutcome: () => { + outcomeCalls += 1; + return { type: "retry", continuation: () => {} }; + }, + }); + await agent.waitForIdle(); + // Each provider start stages a message_start event — the start frames + // carry no superseded delta to reclaim — so with a 2-event cap the + // third staged event trips the limit and the attempt fails as a local + // error (same behavior as the default cap): the overflow never carries + // provider evidence, so the fallback chain is not consumed. + expect(outcomeCalls).toBe(0); + expect(agent.state.error).toContain("provisional event buffer limit"); + expect((agent.state.messages.at(-1) as AssistantMessage).errorKind).toBe("local_buffer_overflow"); + } finally { + if (previous === undefined) delete process.env.GJC_FALLBACK_MAX_STAGED_EVENTS; + else process.env.GJC_FALLBACK_MAX_STAGED_EVENTS = previous; + } + }); + + it("honors GJC_FALLBACK_MAX_STAGED_BYTES from the environment", async () => { + const previous = process.env.GJC_FALLBACK_MAX_STAGED_BYTES; + process.env.GJC_FALLBACK_MAX_STAGED_BYTES = "128"; + try { + const mock = createMockModel(); + const streamFn = () => { + const stream = new AssistantMessageEventStream(); + queueMicrotask(() => { + const partial = assistantMessage(mock.model); + partial.content.push({ type: "text", text: "x".repeat(4096) }); + stream.push({ type: "start", partial }); + stream.end(); + }); + return stream; + }; + const agent = new Agent({ + initialState: { model: mock.model, systemPrompt: ["test"], tools: [], messages: [] }, + streamFn, + }); + let outcomeCalls = 0; + await agent.prompt("run", { + fallbackManaged: true, + onManagedAttemptOutcome: () => { + outcomeCalls += 1; + return { type: "retry", continuation: () => {} }; + }, + }); + await agent.waitForIdle(); + // A 4 KiB event exceeds the 128-byte cap and no reclaimable delta can + // shrink it, so the attempt fails as a local error without consuming + // the fallback chain. + expect(outcomeCalls).toBe(0); + expect(agent.state.error).toContain("provisional event buffer limit"); + } finally { + if (previous === undefined) delete process.env.GJC_FALLBACK_MAX_STAGED_BYTES; + else process.env.GJC_FALLBACK_MAX_STAGED_BYTES = previous; + } + }); + + it("falls back to the default cap for a non-positive GJC_FALLBACK_MAX_STAGED_EVENTS", async () => { + const previous = process.env.GJC_FALLBACK_MAX_STAGED_EVENTS; + process.env.GJC_FALLBACK_MAX_STAGED_EVENTS = "0"; + try { + const mock = createMockModel(); + const streamFn = () => { + const stream = new AssistantMessageEventStream(); + queueMicrotask(() => { + let last: AssistantMessage | undefined; + for (let i = 0; i < 5; i += 1) { + const partial = assistantMessage(mock.model); + partial.content.push({ type: "text", text: `chunk-${i}` }); + last = partial; + stream.push({ type: "start", partial }); + } + stream.end(last); + }); + return stream; + }; + const agent = new Agent({ + initialState: { model: mock.model, systemPrompt: ["test"], tools: [], messages: [] }, + streamFn, + }); + await agent.prompt("run", { fallbackManaged: true }); + await agent.waitForIdle(); + // "0" is not a positive integer, so the default 10_000-event cap + // applies and the small run completes without an overflow error. + expect(agent.state.error).toBeUndefined(); + } finally { + if (previous === undefined) delete process.env.GJC_FALLBACK_MAX_STAGED_EVENTS; + else process.env.GJC_FALLBACK_MAX_STAGED_EVENTS = previous; + } + }); + + it("falls back to the default cap for non-digit GJC_FALLBACK_MAX_STAGED_EVENTS values", async () => { + // Exponents and hex are not "positive integer (digits only)" input. A + // bare Number() parse would accept "3e0"/"0x3" as the cap 3, so these + // five staged events would overflow; digits-only parsing must reject + // them and keep the default 10_000-event cap instead. + for (const raw of ["3e0", "0x3"]) { + const previous = process.env.GJC_FALLBACK_MAX_STAGED_EVENTS; + process.env.GJC_FALLBACK_MAX_STAGED_EVENTS = raw; + try { + const mock = createMockModel(); + const streamFn = () => { + const stream = new AssistantMessageEventStream(); + queueMicrotask(() => { + let last: AssistantMessage | undefined; + for (let i = 0; i < 5; i += 1) { + const partial = assistantMessage(mock.model); + partial.content.push({ type: "text", text: `chunk-${i}` }); + last = partial; + stream.push({ type: "start", partial }); + } + stream.end(last); + }); + return stream; + }; + const agent = new Agent({ + initialState: { model: mock.model, systemPrompt: ["test"], tools: [], messages: [] }, + streamFn, + }); + await agent.prompt("run", { fallbackManaged: true }); + await agent.waitForIdle(); + // The invalid value fell back to the default cap, so the small run + // completes instead of tripping a misparsed 3-event limit. + expect(agent.state.error).toBeUndefined(); + } finally { + if (previous === undefined) delete process.env.GJC_FALLBACK_MAX_STAGED_EVENTS; + else process.env.GJC_FALLBACK_MAX_STAGED_EVENTS = previous; + } + } + }); + it("clamps above-ceiling GJC_FALLBACK_MAX_STAGED_* overrides and accepts exact-ceiling values", async () => { + // The caps exist to bound memory: an override near MAX_SAFE_INTEGER + // would trade the typed, bounded local_buffer_overflow for a process + // OOM. Above-ceiling values must clamp to the ceiling (with a warning) + // rather than being honored. Exactly-at-ceiling values are the accepted + // boundary (#wouldOverflow deliberately accepts equality), so they are + // honored verbatim — both sides of the boundary are pinned here. + const diagnostics = captureStagedCapClampWarnings(); + for (const [name, atCeiling, aboveCeiling, ceiling] of [ + ["GJC_FALLBACK_MAX_STAGED_EVENTS", "2000000", "2000001", 2_000_000], + ["GJC_FALLBACK_MAX_STAGED_BYTES", "1073741824", "1073741825", 1024 * 1024 * 1024], + ] as const) { + const previous = process.env[name]; + // Exactly at the ceiling: honored, no warning. + process.env[name] = atCeiling; + try { + expect( + name === "GJC_FALLBACK_MAX_STAGED_EVENTS" + ? managedAttemptMaxStagedEvents() + : managedAttemptMaxStagedBytes(), + ).toBe(ceiling); + } finally { + if (previous === undefined) delete process.env[name]; + else process.env[name] = previous; + } + // Above the ceiling: clamped back down, one warning. + process.env[name] = aboveCeiling; + try { + expect( + name === "GJC_FALLBACK_MAX_STAGED_EVENTS" + ? managedAttemptMaxStagedEvents() + : managedAttemptMaxStagedBytes(), + ).toBe(ceiling); + } finally { + if (previous === undefined) delete process.env[name]; + else process.env[name] = previous; + } + } + // Both clamp sites warn once each, naming the variable and the ceiling; + // the once-per-(knob, value) memoization still lets distinct values warn. + const clampWarnings = diagnostics.filter(entry => typeof entry.requested === "number"); + expect(clampWarnings).toHaveLength(2); + }); + + it("clamps beyond-safe-integer overrides through the lexical decimal path", () => { + // 2^53 and 2^53+1 fail Number.isSafeInteger, so a purely numeric parse + // would misclassify them as invalid and silently fall back to the + // DEFAULT instead of the documented clamp — an operator asking for a + // huge cap would get 10 000 events / 16 MiB. The lexical comparison + // must clamp both to the ceiling, and the warning payload must be a + // bounded digest (never the arbitrarily long raw string embedded in a + // log record). + const warnings = captureStagedCapClampWarnings(); + const previous = process.env.GJC_FALLBACK_MAX_STAGED_EVENTS; + for (const raw of ["9007199254740992", "9007199254740993", "9".repeat(64)]) { + process.env.GJC_FALLBACK_MAX_STAGED_EVENTS = raw; + expect(managedAttemptMaxStagedEvents()).toBe(2_000_000); + } + // The once-per-digest memoization collapses the two 2^53 values (same + // length, same 8-digit prefix) into one record and keeps the 64-digit + // one distinct — bounded logging without losing the clamp signal. + const digestWarnings = warnings.filter(entry => typeof entry.requested === "string"); + expect(digestWarnings).toHaveLength(2); + for (const entry of digestWarnings) { + expect(String(entry.requested).length).toBeLessThanOrEqual(64); + expect(String(entry.requested)).toContain("digits"); + } + expect(digestWarnings.every(entry => entry.ceiling === 2_000_000)).toBe(true); + if (previous === undefined) delete process.env.GJC_FALLBACK_MAX_STAGED_EVENTS; + else process.env.GJC_FALLBACK_MAX_STAGED_EVENTS = previous; + }); + + it("charges every retained batch item against the caps before retention", async () => { + // Adversarial multi-chunk growth: the provider streams assistant pairs + // (uncharged before this change) alongside measured message_update + // events. A byte cap smaller than the sum of the pair sizes must fail + // the attempt as a typed local overflow BEFORE the pair is retained, + // proving the assistant pair is charged against the same bound the + // measured events use — actual retention can no longer exceed the caps + // while the counters read under them. + const previous = process.env.GJC_FALLBACK_MAX_STAGED_BYTES; + process.env.GJC_FALLBACK_MAX_STAGED_BYTES = "2048"; + try { + 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: "x".repeat(4096) }); + // A provider that streams a large assistant pair after the + // lifecycle start: with the pair uncharged, the batch would + // retain it while the counters stayed low. + stream.push({ type: "text_start", contentIndex: 0, partial }); + stream.end(partial); + }); + return stream; + }; + const agent = new Agent({ + initialState: { model: mock.model, systemPrompt: ["test"], tools: [], messages: [] }, + streamFn, + onAssistantMessageEvent: () => {}, + }); + let outcomeCalls = 0; + await agent.prompt("run", { + fallbackManaged: true, + onManagedAttemptOutcome: () => { + outcomeCalls += 1; + return { type: "retry", continuation: () => {} }; + }, + }); + await agent.waitForIdle(); + // The retained batch as a whole exceeded the cap and was rejected + // as a typed local overflow without consuming the fallback chain. + expect(outcomeCalls).toBe(0); + expect(agent.state.error).toContain("provisional event buffer limit"); + expect((agent.state.messages.at(-1) as AssistantMessage).errorKind).toBe("local_buffer_overflow"); + } finally { + if (previous === undefined) delete process.env.GJC_FALLBACK_MAX_STAGED_BYTES; + else process.env.GJC_FALLBACK_MAX_STAGED_BYTES = previous; + } + }); + it("rejects a many-small-chunks payload without materializing the full budget", async () => { + // The discriminating pin the review asked for: a payload composed of + // many individually sub-limit strings must trip the typed overflow via + // the THROWING pre-allocation walk — which terminates serialization the + // moment the running byte count crosses the cap — rather than by + // materializing the whole JSON string first. On the pre-fix path the + // walk substituted "" and kept going, so the full value was built + // before any check ran; here the cap is crossed while the accumulated + // seen-count is far below the payload's total size. + const previous = process.env.GJC_FALLBACK_MAX_STAGED_BYTES; + // 256 KiB budget against a ~4 MiB payload of 4 KiB chunks: each chunk + // is well under the budget, only their accumulation crosses it. + process.env.GJC_FALLBACK_MAX_STAGED_BYTES = String(256 * 1024); + try { + const mock = createMockModel(); + const chunk = "x".repeat(4 * 1024); + const streamFn = () => { + const stream = new AssistantMessageEventStream(); + queueMicrotask(() => { + const partial = assistantMessage(mock.model); + stream.push({ type: "start", partial }); + partial.content.push({ type: "thinking", thinking: chunk.repeat(1024) }); + // A reasoning block of many accumulated sub-limit deltas: + // no single event is near the cap, and the pre-fix walk + // would have walked the entire ~4 MiB before checking. + stream.push({ type: "thinking_start", contentIndex: 0, partial }); + stream.end(partial); + }); + return stream; + }; + const agent = new Agent({ + initialState: { model: mock.model, systemPrompt: ["test"], tools: [], messages: [] }, + streamFn, + onAssistantMessageEvent: () => {}, + }); + let outcomeCalls = 0; + await agent.prompt("run", { + fallbackManaged: true, + onManagedAttemptOutcome: () => { + outcomeCalls += 1; + return { type: "retry", continuation: () => {} }; + }, + }); + await agent.waitForIdle(); + // The typed overflow fired without consuming the fallback chain and + // without ever building the full multi-megabyte serialization. + expect(outcomeCalls).toBe(0); + expect(agent.state.error).toContain("provisional event buffer limit"); + const terminal = agent.state.messages.at(-1) as AssistantMessage; + expect(terminal.errorKind).toBe("local_buffer_overflow"); + // The diagnostic names the cap that tripped, proving the guard — + // not a downstream crash — produced the failure. + const overflow = (terminal as unknown as { bufferOverflow?: { maxStagedBytes: number } }).bufferOverflow; + expect(overflow?.maxStagedBytes).toBe(256 * 1024); + } finally { + if (previous === undefined) delete process.env.GJC_FALLBACK_MAX_STAGED_BYTES; + else process.env.GJC_FALLBACK_MAX_STAGED_BYTES = previous; + } + }); + + it("falls back to the default cap for zero, negative, and non-numeric values", () => { + // Those never reach the clamp: the digits-only parse rejects them and + // keeps the documented default, so the guard can never be disabled by + // malformed input either. + for (const raw of ["0", "-5", "abc", ""]) { + const previous = process.env.GJC_FALLBACK_MAX_STAGED_EVENTS; + process.env.GJC_FALLBACK_MAX_STAGED_EVENTS = raw; + try { + expect(managedAttemptMaxStagedEvents()).toBe(MANAGED_ATTEMPT_MAX_STAGED_EVENTS); + } finally { + if (previous === undefined) delete process.env.GJC_FALLBACK_MAX_STAGED_EVENTS; + else process.env.GJC_FALLBACK_MAX_STAGED_EVENTS = previous; + } + } + }); + + it("accepts trusted staged-cap values with surrounding whitespace", () => { + const previous = process.env.GJC_FALLBACK_MAX_STAGED_EVENTS; + process.env.GJC_FALLBACK_MAX_STAGED_EVENTS = " 2 "; + try { + expect(managedAttemptMaxStagedEvents()).toBe(2); + } finally { + if (previous === undefined) delete process.env.GJC_FALLBACK_MAX_STAGED_EVENTS; + else process.env.GJC_FALLBACK_MAX_STAGED_EVENTS = previous; + } + }); it("retains queued follow-up input when its managed attempt is discarded for retry", async () => { const mock = createMockModel({ responses: [{ content: ["initial"] }, { content: ["retried"] }] });