From 0095bce7db4fe6a63524b37535315f41b086656d Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Tue, 1 Sep 2026 18:12:06 +0200 Subject: [PATCH] fix(agent): bound shell execution output --- .../01-harness/03-execenv/execenv.md | 445 ++++-------------- .../04-tool-output/bash-worked-example.md | 28 +- .../04-tool-output/harness-tools.md | 33 +- .../04-tool-output/rate-limiting.md | 285 ++++------- packages/agent/src/harness/env/nodejs.ts | 200 ++++++-- packages/agent/src/harness/tools/bash.ts | 173 +++---- packages/agent/src/harness/types.ts | 56 ++- .../src/harness/utils/adaptive-publisher.ts | 87 ++++ .../agent/src/harness/utils/output-capture.ts | 238 ++++++++++ .../agent/src/harness/utils/shell-output.ts | 244 +++------- packages/agent/src/harness/utils/truncate.ts | 2 +- packages/agent/src/index.ts | 9 + .../test/harness/adaptive-publisher.test.ts | 76 +++ .../agent/test/harness/nodejs-env.test.ts | 213 ++++++--- .../agent/test/harness/output-capture.test.ts | 151 ++++++ packages/agent/test/harness/tools.test.ts | 51 +- 16 files changed, 1297 insertions(+), 994 deletions(-) create mode 100644 packages/agent/src/harness/utils/adaptive-publisher.ts create mode 100644 packages/agent/src/harness/utils/output-capture.ts create mode 100644 packages/agent/test/harness/adaptive-publisher.test.ts create mode 100644 packages/agent/test/harness/output-capture.test.ts diff --git a/packages/agent/docs/mobile-handoff/01-harness/03-execenv/execenv.md b/packages/agent/docs/mobile-handoff/01-harness/03-execenv/execenv.md index decc617dd..ec4d05a57 100644 --- a/packages/agent/docs/mobile-handoff/01-harness/03-execenv/execenv.md +++ b/packages/agent/docs/mobile-handoff/01-harness/03-execenv/execenv.md @@ -1,423 +1,148 @@ -# ExecutionEnv: Bounded Shell Output +# ExecutionEnv: bounded shell output -Moves output capping, spilling and coalescing out of individual tools and into -`Shell`, so they happen where the bytes originate. +**Status:** implemented in production source. The prototype files beside this document are historical evidence; production lives in: -> **Scope:** `packages/agent/src/harness/env/`, `harness/utils/`, -> `harness/tools/bash.ts`, and the `Shell` contract in `harness/types.ts`. -> -> **Independent of `01-delta` and `02-scopes`.** `Shell`'s three update kinds are -> its own vocabulary and stop at the `ToolOutput` sink; `Shell` never emits ops -> and knows nothing about `delta.md`. This unit can land in any order. -> -> **This one has working code**, not just a specification: -> -> | file | lands as | -> | --- | --- | -> | `output-capture.ts` | `harness/utils/output-capture.ts` — new | -> | `bash.after.ts` | `harness/tools/bash.ts` — the rewritten tool | -> | `nodejs-env.diff` | the `NodeExecutionEnv.exec` change | -> | `nodejs-env.test.diff` | the test migration, 29 passing | -> -> All of it typechecked and ran: `cd packages/agent && npx vitest run --config -> vitest.harness.config.ts nodejs-env` gives 29 passed, 1 skipped. +- `packages/agent/src/harness/utils/adaptive-publisher.ts` +- `packages/agent/src/harness/utils/output-capture.ts` +- `packages/agent/src/harness/env/nodejs.ts` +- `packages/agent/src/harness/tools/bash.ts` ---- +The generic `ToolOutput` use of the same publisher remains in `04-tool-output`. -## 1. The problem +## 1. Problem -Three separate faults, one root cause. - -**`exec` is unbounded at its root.** +The old `Shell.exec()` accumulated complete strings inside `NodeExecutionEnv`: ```ts -// packages/agent/src/harness/types.ts -exec(command, options, context): Promise> +stdout += chunk; +stderr += chunk; ``` -```ts -// packages/agent/src/harness/env/nodejs.ts:463,474 -child.stdout?.on("data", (chunk: string) => { stdout += chunk; /* … */ }); -child.stderr?.on("data", (chunk: string) => { stderr += chunk; /* … */ }); -``` +Bash truncated only after those strings had already been built. `cat 1gb.txt` therefore materialized a gigabyte in the worker before any tool-level bound could help. -`cat 1gb.txt` materialises 1 GB of string in the agent process. No cap exists -below the tool layer, so no tool-level truncation can prevent it — by the time -`bash.ts` sees the output it has already been accumulated in full. +Spilling also belongs where bytes originate. If execution is remote, a spill on the worker is inaccessible to the model's `read` and `grep` tools, and creating it after transport would first send the complete gigabyte over the connection. -**Truncation is tool-specific.** `bash.ts` owns a rolling buffer, `truncateTail`, -and a spill file. `read.ts` has its own truncation. Every future tool that -produces bulk output re-implements the same logic, differently. +## 2. Boundary -**Spill has no owner.** bash writes to `createTempFile({ prefix: "bash-" })` — -`/tmp`, cleaned up whenever the OS decides. The path is handed to the model so it -can `grep` the full output, which means the file must live on the machine where -the model's other tools run. +`ExecutionEnv` now owns: -That last point is the constraint that shapes everything: **the spill must be in -the execution environment.** Not in session storage, not on the agent machine. -If the exec env is a remote sandbox, a spill on the agent host is unreachable to -the model's own `read` and `grep`. +- a bounded head or tail view; +- complete byte and line totals; +- lazy source-local spill after the view first crosses its limits; +- persistent source-local spill with bounded write-stream backpressure; +- adaptive publication of the latest bounded state; +- a forced final publication before settlement. -Which means the capping has to happen there too. Otherwise reading a 1 GB file on -a sandbox host ships 1 GB to the agent machine, caps it, and ships the spill back. +It does not return or retain separate `stdout` and `stderr` values. Both pipes feed one arrival-ordered model-visible text view, matching bash and `ToolOutput`; preserving stream styling through tail eviction would require a segmented retained state that the Harness does not expose. Text is folded from updates, and `ShellExecResult` contains only exit and truncation/spill metadata. -## 2. Scope +Bash owns only command semantics and its model-visible footer. Its old rolling buffer, spill creation, 100 ms throttle, and full-output accumulation are gone. The existing two-second durable checkpoint request remains temporarily until `ToolOutput` owns durable cadence. -**In scope.** Bounded capture in `Shell`, spill inside the exec env, coalesced -live updates, capability reporting. - -**Relationship to the rest.** The three update kinds here are `Shell`'s own -vocabulary and stop at the `ToolOutput` sink; they do not travel further. -`harness-tools.md` §4 converts them into ops against `ToolOutputState`. `Shell` -never emits ops and knows nothing about `delta.md`. - -**Out of scope, deliberately.** Making a *remote* exec env work well with a local -worker. The answer there is to run the worker where the exec env is — the server -stays local, the worker relocates — and that is a separate work package. This -document assumes worker and exec env are colocated, which is the case for -`NodeExecutionEnv` today. - -## 3. Types +## 3. Contract ```ts -// ─── Policy ────────────────────────────────────────────────────── - -/** Which retained portion survives when output exceeds the limits. */ -export type OutputRetention = "head" | "tail"; - -/** How much output the caller can accept, and what to keep on overflow. */ -export interface OutputLimits { +interface ShellOutputLimits { maxBytes: number; maxLines: number; - /** Defaults to "tail". */ - retain?: OutputRetention; + retain?: "head" | "tail"; } -/** Requests bounded capture. Absent means unbounded. */ -export interface CaptureOptions { - limits: OutputLimits; - /** - * Minimum interval between onUpdate calls, in ms. Output arriving inside a - * window is coalesced, never dropped. 0 emits per chunk. Defaults to 0. - */ - intervalMs?: number; - /** - * Archive the full output to a file inside this environment, created lazily - * when the limits are first exceeded. The path is meaningful on the machine - * the command ran on. - */ +interface ShellOutputCaptureOptions { + limits: ShellOutputLimits; spill?: boolean; } -// ─── Captured output ───────────────────────────────────────────── +type ShellOutputTruncation = Omit; -export type ShellStream = "stdout" | "stderr"; - -/** Bounded view of a command's output, aggregated in arrival order. */ -export interface CapturedOutput { - /** The retained portion, already truncated per OutputLimits. */ - text: string; - /** Totals over everything produced, plus what was kept. */ - truncation: TruncationResult; - /** Present only if spill was requested and the limits were exceeded. */ +interface ShellOutputMetadata { + truncation: ShellOutputTruncation; spillPath?: string; - /** - * Size of the final line when it alone exceeded the byte cap - * (`truncation.lastLinePartial`). Only the env holds the raw buffer, so only - * the env can measure it — a tool cannot recover it from the retained view. - */ lastLineBytes?: number; } -export type ShellOutputUpdate = - | { kind: "chunk"; stream: ShellStream; text: string } - | { kind: "snapshot"; output: CapturedOutput } - | { kind: "counters"; truncation: TruncationResult }; - -// ─── Exec ──────────────────────────────────────────────────────── - -export interface ShellExecOptions { - cwd?: string; - env?: Record; - inheritEnv?: boolean; - timeout?: number; - - /** Bounded capture. Absent means unbounded. */ - capture?: CaptureOptions; - /** Live output. Replaces onStdout / onStderr. */ - onUpdate?: (update: ShellOutputUpdate, context: Context) => void; +interface ShellOutputView extends ShellOutputMetadata { + text: string; } -export interface ShellExecResult { - output: CapturedOutput; +type ShellOutputUpdate = + | { kind: "replace"; output: ShellOutputView } + | { kind: "append"; text: string; metadata: ShellOutputMetadata } + | { kind: "slide"; drop: number; text: string; metadata: ShellOutputMetadata } + | { kind: "metadata"; metadata: ShellOutputMetadata }; + +interface ShellExecResult extends ShellOutputMetadata { exitCode: number; } - -/** What this implementation can actually do. Tiers differ in fidelity. */ -export interface ShellCapabilities { - /** false for a backend that can only stream raw bytes. */ - capture: boolean; - spill: boolean; - /** false when the backend cannot emit incrementally. */ - liveUpdates: boolean; -} - -export interface Shell { - capabilities(): ShellCapabilities; - exec( - command: string, - options: ShellExecOptions | undefined, - context: Context, - ): Promise>; - cleanup(context: Context): Promise; -} - -export interface ExecutionEnv extends FileSystem, Shell {} ``` -`TruncationResult` is the existing type from -`packages/agent/src/harness/utils/truncate.ts`. It already carries `totalBytes`, -`totalLines`, `truncated`, `truncatedBy`, `outputBytes`, `outputLines`, which is -exactly what a UI needs to render "showing last 50 KB of 1.2 GB". No new type. +`drop` counts JavaScript string code units, matching `slice()`. Updates are ordered. A consumer applies them with `applyShellOutputUpdate()`. -`Shell` deliberately has **no tool vocabulary**. It emits text. Images, details, -usage, added tool names and terminate are `ToolOutput`'s concern and are produced -on the agent side. That is why `ShellOutputUpdate` has three variants and not six. +A complete replacement establishes the initial state or recovers when no verified overlap exists. An append carries only a growing suffix. A slide drops a prefix and appends the new suffix. Metadata updates move totals or the spill path without resending text. -## 4. Why three update kinds +The compatibility `executeShellWithCapture()` helper still returns one bounded final view. Its `onChunk` callback receives initial, append, and slide text only; metadata and post-turnover replacements are not mislabeled as new bytes. -They are the three states a bounded buffer can be in: text grows, text slides, -text stops. +## 4. Adaptive publication -| kind | when | why nothing else works | -|---|---|---| -| `chunk` | below the cap | append-only and complete; the common case, since most commands never reach the cap | -| `snapshot` | retained view changed by eviction | an append cannot express "and 20 bytes fell off the front" | -| `counters` | `head` mode past the cap | text is frozen forever; without this you resend `maxBytes` to move a number | +`AdaptivePublisher` keeps only the latest dirty state. Intermediate process writes never become a queue of output updates. -Reader: +Policy is harness-global rather than per tool: ```ts -switch (u.kind) { - case "chunk": view.text += u.text; break; - case "snapshot": view = u.output; break; - case "counters": view.truncation = u.truncation; break; -} +minIntervalMs = 100; +targetBytesPerSecond = 100 * 1024; +nextDelayMs = max(minIntervalMs, encodedUpdateBytes * 1000 / targetBytesPerSecond); ``` -### 4.1 Ordering invariant +The first dirty state after idle is immediate. Writes received before the deadline collapse into the latest bounded state. One trailing timer guarantees eventual publication. Finalization bypasses the deadline once, still bounded by the retained cap. -> Once the cap is crossed, a `chunk` must never follow without an intervening -> `snapshot`. +The publisher commits its baseline before invoking the consumer. If a consumer applies an update and then throws, finalization cannot emit the same delta twice. The command fails with `callback_error`. -The first flush that evicts is a `snapshot` regardless of size. Violating this -leaves the reader appending to a view whose front has been discarded — it grows -without bound and silently disagrees with `truncation`. +### Workload behavior -A `head` producer never evicts, so it emits chunks up to the cap and only -`counters` afterwards, and never needs a snapshot. +| workload | result | +| --- | --- | +| Finishes below cap | immediate initial state, small appends, forced final state | +| Below-cap trickle | isolated writes are immediate; sustained writes are at most 100 ms apart | +| Full-force output after cap | complete turnovers are cap-sized replacements spaced by their encoded size | +| Post-cap trickle | small verified `slide` updates remain responsive; the complete window is not resent | +| Burst then silence | one leading update and one trailing update | +| Error, timeout, or abort | latest bounded state is forced before the error settles | -### 4.2 Per-mode behaviour +At a 50 KB cap and 100 KB/s target, repeated complete turnovers settle near two updates per second. Small post-cap slides still use the 100 ms floor. -Worked example: `maxBytes = 100`, `intervalMs = 500`. +The rate bound is amortized. An immediate-after-idle update and forced terminal update may each create one cap-bounded burst. -**`retain: "tail"`** +## 5. Capture and spill -| flush | produced | total | evicted | emitted | -|---|---|---|---|---| -| 1 | 30 B | 30 | — | `chunk` 30 B | -| 2 | 40 B | 70 | — | `chunk` 40 B | -| 3 | 50 B | 120 | 20 B | **`snapshot`** 100 B | -| 4 | 10 B | 130 | 10 B | `snapshot` 100 B | -| 5 | 0 B | 130 | — | *(nothing)* | -| 6 | 5 B | 135 | 5 B | `snapshot` 100 B | +`OutputCapture` trims its decoded line-aware working buffer back to twice the byte cap when it exceeds four times the cap. Tail mode drops old text; head mode preserves the original prefix. This amortizes UTF-8 trimming instead of rescanning the retained window for every process chunk. -Flush 3 is the crossing. Note flushes 4 and 6: **once the buffer is full, every -byte in evicts a byte out**, so there is no cheap-trickle regime past the cap. -Tail collapses to: chunks until full, snapshots at the interval thereafter. +Spill creation is lazy. Before crossing, the source keeps at most the bounded prefix needed to create a complete archive. On crossing it: -**`retain: "head"`** +1. pauses stdout and stderr while it creates the file inside the execution environment; +2. opens one persistent append stream with a bounded 8 MB high-water mark; +3. writes the preserved raw prefix and subsequent raw chunks in arrival order; +4. resumes immediately while the writer accepts data; +5. pauses again only when `write()` reports backpressure, then resumes on `drain`. -| flush | produced | total | emitted | -|---|---|---|---| -| 1 | 30 B | 30 | `chunk` 30 B | -| 2 | 40 B | 70 | `chunk` 40 B | -| 3 | 50 B | 120 | `chunk` 30 B *(the part that fit)* | -| 4 | 10 B | 130 | `counters` | +This avoids both an unbounded promise chain and one async file-open/append cycle per process chunk. A spill create or stream-write failure kills the child and fails execution rather than publishing lossy success. -Head never evicts. After the cap only totals move. +The spill path is force-published as metadata when it becomes available. Spill writes are awaited before the final output flush. -**Head+tail is not offered.** `harness-tools.md` §3.1 rejected it and that stands: -`truncate.ts` exports `truncateHead` and `truncateTail`, neither combines them, and -none is being added. OpenCode's `preview` shape — half the budget from each end — -is a reasonable future addition for failed builds, where you want the invocation -and the error but not the middle, but it is out of scope here. +Node streams remain raw for spill throughput and exact archival bytes. `OutputCapture` uses one streaming `TextDecoder`, so a read boundary cannot split a code point; invalid display control characters are removed only from bounded snapshots, not by scanning the complete raw stream. Line totals count a final unterminated line, and `lastLineBytes` remains exact even when one line exceeds the working buffer. -### 4.3 Idle +## 6. Remote execution -No output in a window emits nothing. `counters` should use a slower cadence than -`intervalMs` (2 s is reasonable) since it exists only to keep a total live. +When worker and execution environment are colocated, updates are in-process and this publisher primarily bounds capture work. `ToolOutput` remains the downstream event/durability limiter. -### 4.3.1 Coalescing is a size lever, not only a rate lever +When an execution environment is physically remote from its worker, the same bounded updates cross that transport. Full-force output cannot transfer the complete stream: intermediate writes collapse at the source, while the complete stream remains in the source-local spill. -`intervalMs` sets how large a chunk the sink receives, and chunk size decides both -throughput and wire size downstream (`delta.md` §4.2). +Each real costly boundary gets its own publisher instance. A colocated deployment may bypass transport serialization; a custom tool that bypasses `ExecutionEnv` still passes through the future `ToolOutput` publisher. -Measured against a 50 KB window: at 100-byte chunks the op encoding puts **135% of -the input on the wire** — past the cap each write emits `truncate` + `append`, and -two op envelopes exceed a 100-byte payload. At 8 KB chunks it is 31%. +## 7. Remaining work -So an `intervalMs` of 0 is correct only for a tool that produces little output. For -anything streaming, coalescing is what keeps the encoding on the right side of that -curve, and the interval should be chosen with size in mind rather than only -"how often should the UI redraw". - -### 4.4 Traffic bound - -Worst case is `maxBytes / intervalMs`, independent of how much the command -produces. At 50 KB and 500 ms that is 100 KB/s whether the command emits 1 MB or -1 GB. This is the property that makes the pathological case safe, and it is why -snapshots are the safe path rather than a fallback. - -## 5. Implementation tiers - -**Tier 1 — Node (`NodeExecutionEnv`).** Mostly a move, not new code. The logic in -`packages/agent/src/harness/utils/shell-output.ts` already implements all of it: -a rolling buffer held at `DEFAULT_MAX_BYTES * 2` (`trimToLastUtf8Bytes`), -`truncateTail` applied at snapshot time, and lazy spill via `ensureFullOutputFile` -/ `appendFullOutput`. Relocate it into the env, generalise `truncateTail` to the -three retention modes, add coalescing. - -Note the existing two-stage structure and keep it: the raw buffer is trimmed to -`maxBytes * 2` as a memory guard, and the retained view is computed from it per -snapshot. `truncateTail` is line-aware — it walks backwards line by line and takes -a partial only when a single line alone exceeds `maxBytes` — so eviction is not -byte arithmetic. - -**Tier 2 — pi helper.** Same TypeScript on the far side of the connection. -Nothing to design. - -**Tier 3 — dumb SSH, shell script.** Implementable: - -```sh -cmd 2>&1 | tee "$SPILL" & -pid=$! -while kill -0 $pid 2>/dev/null; do - printf '\0SNAP\0'; tail -c "$MAXB" "$SPILL" - sleep "$INTERVAL" -done -``` - -Spill via `tee`, bounded snapshots via `tail -c` against the spill file. -`retain: "head"` is `head -c` and can stop reading early. Needs a framing -sentinel and a writable remote path. - -This tier can emit `snapshot` and `counters` but the chunk path is degraded, -because a `tail -c` pipeline buffers. That is acceptable: `snapshot` is -self-contained, so a snapshot-only stream is correct, just less live. It reports -`liveUpdates: false`. - -`capabilities()` exists so this degradation is declared rather than silent. A -backend reporting `capture: false` forces the caller to apply limits locally, -accepting that the bytes crossed the wire. - -## 6. Migration - -**Breaking.** `exec` no longer returns `{ stdout, stderr }`. Verified that no -caller consumes them separately — only `nodejs.ts` internals touch -`child.stdout` / `child.stderr` — so aggregating in arrival order costs nothing. -`chunk` carries `stream` so a renderer can still colour stderr. - -**`onStdout` / `onStderr` → `onUpdate`.** The callback-error path -(`callback_error`, abort on throw) carries over unchanged. - -**`shell-output.ts` collapses.** `executeShellWithCapture`, `createProgress`, -`tailOutput`, `ensureFullOutputFile` all move into the env. - -**`bash.ts` loses its truncation entirely.** No rolling buffer, no `truncateTail`, -no `createTempFile`, no `BASH_UPDATE_THROTTLE_MS`. `fullOutputPath` becomes -`spillPath` returned in `CapturedOutput`. Its `details` reduce to whatever remains -tool-specific. - -**`BASH_CHECKPOINT_INTERVAL_MS` goes, and nothing here replaces it.** The durable -cadence is not a `Shell` concern. - -The shipped `bash.after.ts` checkpoints when the update kind is `snapshot`, on the -reasoning that a snapshot is the only kind that can express eviction. **That is -wrong and should not be copied.** `nowCrossed` is sticky: once total output -exceeds the cap it never goes back, so every subsequent update is a snapshot. At -`intervalMs = 100` that is a durable write every 100 ms — the same cadence, not a -lower one. - -Two rates genuinely exist. The UI wants roughly 100 ms; durability wants roughly -two seconds. But a *tool* has no basis for choosing the second: it does not know -what a durable write costs, whether there is a sidecar, or what recovery is worth. -That belongs to the sink, and it is settled in `04-tool-output` — where the -`checkpoint` flag comes off the tool-facing API entirely. - -**`read.ts`** keeps its own truncation for now — it reads files rather than -running commands, so it does not go through `Shell`. Unifying that is a separate -question (§8). - -### 6.1 Three things that only surface when you run it - -Each of these passes typecheck and fails at runtime. - -**Await the spill chain before the final snapshot.** The spill file is created -lazily, on the first chunk, so its path does not exist yet when `exec` settles. -Snapshotting before that promise resolves reports `spillPath: undefined` for -output that *was* archived — the file is on disk, the caller just never learns -where. Await the chain, then `capture.flush()`, then snapshot. - -**Counting newlines is not counting lines.** A command emitting one 60 KB line -with no trailing newline has produced **one** line, not zero. The off-by-one -surfaces as a user-visible message reading `line 0`, and only for the -single-oversized-line case. Count newlines, plus one if the buffer does not end -with one. - -**`lastLineBytes` can only come from the env.** When one line alone exceeds the -byte cap, the footer reports that line's own size — and only the env holds the -raw buffer, so a tool cannot recover it from the retained view. -`CapturedOutput.lastLineBytes` carries it. - -There is also a portability trap in the shipped code: **no TypeScript parameter -properties.** `constructor(readonly x: T)` is rejected outright by Node's -`--experimental-strip-types`, which is the mode these files are meant to run -under. - -## 7. Spill lifetime - -The spill exists so the **model** can read it. That fixes its location: inside the -exec env, on the machine where the model's other tools run. - -It must **not** be deleted when the tool settles — the model may `grep` a spill -from an earlier turn in the same session. So spill lifetime is session-scoped, not -invocation-scoped, and something has to sweep it. - -Current behaviour is `createTempFile` into `/tmp` with OS-dependent cleanup, which -is the complaint that started this. Proposed: a session-scoped directory the env -owns, swept on session end, with a retention floor for crash cases. OpenCode uses -7-day retention on its `tool-output/` directory as precedent. - -## 8. Open questions - -- **Session-scoped spill directory.** Who creates it, who sweeps it, and what the - path looks like to the model. Needs an env-level concept that does not exist yet. -- **`read.ts` and other non-shell bulk output.** Should `FileSystem` grow the same - capture policy, or does `ToolOutput` handle those on the agent side because the - bytes already crossed? Leaning to the latter, since a file read is a single call - rather than a stream. -- **UTF-8 chunk boundaries.** A chunk boundary landing mid-codepoint bakes in a - replacement character. `setEncoding("utf8")` on the Node streams handles this - today; a tier-3 shell script does not. Codex spends real effort here - (`bytes_to_string_smart`, `chardetng`) — we should at minimum not corrupt UTF-8. -- **`counters` cadence.** Fixed 2 s, or derived from `intervalMs`. -- Whether head+tail retention is worth adding later, and if so whether the two - halves need separate budgets. -- **Serving spill to a UI.** When the worker is remote, the server cannot read the - spill file, so a "view full output" affordance must go through the worker. Out of - scope here, but it constrains the UI design. -- **Binary output.** A command emitting binary to stdout currently becomes lossy - UTF-8. Whether `Shell` should detect and refuse, or spill raw bytes and retain a - marker, is undecided. +- Move generic custom-tool composition, text retention, event publication, and durable cadence into `ToolOutput`. +- Replace whole `AgentToolResult` progress with Chord operations at that downstream boundary. +- Make memo and output checkpoint persistence one atomic transaction. +- Seed replay from durable output rather than deleting it. +- Put spills in an environment-owned session directory and sweep them after session lifetime plus a crash-retention floor. +- Decide explicit limits for images and structured details; text is bounded, those values are not yet. +- Define raw binary-output behavior. Current shell output remains lossy UTF-8 text. diff --git a/packages/agent/docs/mobile-handoff/01-harness/04-tool-output/bash-worked-example.md b/packages/agent/docs/mobile-handoff/01-harness/04-tool-output/bash-worked-example.md index 0cf268866..dc81efc99 100644 --- a/packages/agent/docs/mobile-handoff/01-harness/04-tool-output/bash-worked-example.md +++ b/packages/agent/docs/mobile-handoff/01-harness/04-tool-output/bash-worked-example.md @@ -13,7 +13,7 @@ state → facet → wire → consumer. ```ts // packages/agent/src/harness/tools/bash.ts -export interface BashToolDetails { spillPath?: string } +export interface BashToolDetails { spillPath?: string; truncation?: ShellOutputTruncation } export function createBashTool(): AgentHarnessTool { return { @@ -23,21 +23,24 @@ export function createBashTool(): AgentHarnessTool { - if (u.kind === "chunk") out.write(u.text); - if (u.kind === "snapshot") out.replace(u.output.text); - // counters: nothing to do, truncation totals ride the snapshot + view = applyShellOutputUpdate(view, u); + if (u.kind === "append") out.write(u.text); + else out.replace(view.text); + out.details.truncation = view.truncation; + if (view.spillPath) out.details.spillPath = view.spillPath; }, }, context)); - if (result.output.spillPath) out.details.spillPath = result.output.spillPath; - if (result.output.truncation.truncated) out.write(`\n\n[${describe(result.output.truncation)}]`); + if (result.spillPath) out.details.spillPath = result.spillPath; + if (result.truncation.truncated) out.write(`\n\n[${describe(result.truncation)}]`); if (result.exitCode) throw new Error(`Command exited with code ${result.exitCode}`); }, }; @@ -57,14 +60,11 @@ longer knows what was dropped. ## 2. The exec env caps at the source -`env.exec` applies `OutputLimits` where the bytes originate. For a sandbox host +`env.exec` applies `ShellOutputLimits` where the bytes originate. For a sandbox host that means `cat 1gb.txt` never ships 1 GB to the agent machine, and the spill -lands where the model's own `read` and `grep` run. See `executionenv.md`. +lands where the model's own `read` and `grep` run. See [`execenv.md`](../03-execenv/execenv.md). -Under the cap it emits `chunk`. Once the window starts evicting it emits -`snapshot`, because an append cannot express "and 20 bytes fell off the front". -For `retain: "tail"` there is no cheap-trickle regime past the cap — every byte -in evicts a byte out — so it is chunks until full, snapshots thereafter. +The initial state is a bounded `replace`. Growth emits `append`; a moving tail emits `slide { drop, text }`; complete turnover falls back to a bounded `replace`; metadata moves totals and spill paths without resending text. The adaptive publisher keeps small slides responsive and spaces cap-sized turnovers according to their encoded size. ## 3. The sink @@ -135,7 +135,7 @@ persisting in the main log forever. Retirement is a main-log `retireScope` recor so it commits atomically with the settle writes; the unlink is a consequence of replaying that record, not part of the transaction ([scopes.md §5](../02-scopes/scopes.md)). -A `list`, one encoded batch appended per flush, base batches tagged `"base"` so recovery reads backwards with `stopAtTag` and stops there ([scopes.md §11](../02-scopes/scopes.md#11-list-tags-and-stop-conditions)). The tracker emits structural ops; the producer calls `rebase()` periodically to write a capped root replacement and bound recovery replay. The write interval is the crash-loss knob and lives in the capture policy. +A `list`, one encoded batch appended per flush, base batches tagged `"base"` so recovery reads backwards with `stopAtTag` and stops there ([scopes.md §11](../02-scopes/scopes.md#11-list-tags-and-stop-conditions)). The tracker emits structural ops; the producer calls `rebase()` periodically to write a capped root replacement and bound recovery replay. The durable interval is a sink policy. Shell capture controls only source-state and transport publication; memo, terminal, and recovery-base flushes are forced by `ToolOutput`. Recovery seeds a fresh `ToolOutput` from that state — it does **not** delete it, which is what `clearReplayCheckpoint` does today and is a bug diff --git a/packages/agent/docs/mobile-handoff/01-harness/04-tool-output/harness-tools.md b/packages/agent/docs/mobile-handoff/01-harness/04-tool-output/harness-tools.md index f58432e84..2919a4414 100644 --- a/packages/agent/docs/mobile-handoff/01-harness/04-tool-output/harness-tools.md +++ b/packages/agent/docs/mobile-handoff/01-harness/04-tool-output/harness-tools.md @@ -32,24 +32,10 @@ interface ToolOutput { write(text: string): void; /** Append an image block. Images are never windowed. */ image(image: ImageContent): void; - /** - * Replace the retained text wholesale. - * - * For output originating in the exec env, which emits `chunk` below the cap and - * `snapshot` once the window starts evicting (`executionenv.md` §4). An append - * cannot express eviction, so a snapshot has to assign. The tracker still - * recovers `t` + `a` from the assignment by verified overlap, so this is not a - * whole-value replacement on the wire. - */ + /** Replace the retained text wholesale. Chord still recovers a verified `t` + `a` slide when possible. */ replace(text: string): void; - /** - * Record totals when the text itself did not change — `head` mode past the cap, - * where output keeps arriving but the retained view is frozen - * (`executionenv.md` §4). Without this a renderer cannot report how much was - * suppressed, and resending the capped text to move a counter is what - * `counters` exists to avoid. - */ - counters(truncation: TruncationResult): void; + /** Apply source-owned truncation totals and spill metadata without resending text. */ + capture(metadata: ShellOutputMetadata): void; /** The tool's own details object. Mutate it directly. */ readonly details: TDetails; @@ -175,8 +161,8 @@ Head+tail is **not** offered. `truncate.ts` exports `truncateHead` and ### 3.2 The tool does not truncate — the exec env does For output originating in the execution environment, capping, coalescing and -spilling happen **at the source**. See `executionenv.md`. The tool passes its -`OutputLimits` into `env.exec` and pipes the resulting updates into the sink. +spilling happen **at the source**. See [`execenv.md`](../03-execenv/execenv.md). The tool passes its +`ShellOutputLimits` into `env.exec` and pipes the resulting updates into the sink. This is not a convenience. Reading a 1 GB file on a sandbox host must not ship 1 GB to the agent machine to be capped there, and the spill file must land where @@ -201,7 +187,7 @@ interface ToolOutputState { usage?: Usage; addedTools?: string[]; terminate: boolean; - truncation: TruncationResult; // totals over everything ever written + truncation: ShellOutputTruncation; // totals over everything ever written, without duplicate text } ``` @@ -353,10 +339,7 @@ writes: [appendList(address, wire, isBase(ops) ? "base" : undefined)]; The landed tracker emits structural ops unconditionally. There is no serialized-size comparison or adaptive replacement heuristic. A producer requests a base batch explicitly with `rebase()`; the output sink's cap bounds that replacement, while periodic rebasing bounds recovery work. Production remeasurement rejects a text-specific append/truncate API: the generic path measured 2.43–2.46 µs per 50 KB rolling-window flush locally, below surrounding costs. Keep ordinary tracked string mutation; see the [decision record](../01-delta/append-decision.md). -**Checkpointing is not deleted.** An earlier draft called for removing -`BASH_CHECKPOINT_INTERVAL_MS`; that was wrong. Checkpoint frequency is the knob -trading durable write volume against how much a crash loses, and it belongs in -the exec env's capture policy now rather than in bash. +**Checkpointing is not deleted.** `BASH_CHECKPOINT_INTERVAL_MS` remains only as an interim compatibility mechanism. The generic sink owns durable frequency because Shell cannot price a storage write or enforce memo/output atomicity. Forced memo, terminal, and recovery-base writes bypass ordinary pacing while remaining cap-bounded. ### 7.4 Replay must seed, not discard @@ -467,7 +450,7 @@ no facet present. them in a loop grows `content` without limit. Currently treated as a tool bug. - Whether details need a bound for the same reason (§2.1). - Whether `retain: "head"` should keep emitting counter-only updates once capped - so a renderer can report how much was suppressed. `executionenv.md` answers this + so a renderer can report how much was suppressed. [`execenv.md`](../03-execenv/execenv.md) answers this for exec-originated output; agent-side output needs the same answer. - Whether a failing tool *should* be able to terminate, or whether the current inability is deliberate — there is a reasonable argument the model should diff --git a/packages/agent/docs/mobile-handoff/01-harness/04-tool-output/rate-limiting.md b/packages/agent/docs/mobile-handoff/01-harness/04-tool-output/rate-limiting.md index 564c167a7..eef04ebd5 100644 --- a/packages/agent/docs/mobile-handoff/01-harness/04-tool-output/rate-limiting.md +++ b/packages/agent/docs/mobile-handoff/01-harness/04-tool-output/rate-limiting.md @@ -1,250 +1,125 @@ -# 04-tool-output: update cadence and rate limiting +# Bounded output publication -**Status: design, not implementation.** This unit is not built. What follows is -the problem, what three other agents do about it, and a candidate solution with -its measured behaviour and its known holes. +**Status:** the shared adaptive publisher and its `ExecutionEnv` use are implemented. Generic `ToolOutput` integration remains design work. -Depends on `01-delta` (the op vocabulary), `02-scopes` (the durable list), and -`03-execenv` (where bytes originate). +Depends on landed Chord delta tracking, source-bounded execution output, and scoped storage for durable batches. ---- +## 1. Invariant -## 1. The invariant this unit exists to hold +> The durable record, what the model sees, and what the UI shows are the same bounded view. -> **The durable record, what the model sees, and what the UI shows are always the -> same view.** +A spill file is not a second view. It is a file inside the execution environment that the model reaches through ordinary file tools. -Not a simplification — a commitment. If the UI shows output the model never -received, a user reads it, forms a belief about what the model knows, and is -wrong. Showing less is better than showing something different. +Every uncontrolled producer boundary needs two independent bounds: -Two consequences that look like limitations and are not: +- **state size:** the latest retained text is capped; +- **publication:** both encoded bytes and event count are paced. -- **One byte budget, not two.** A separate "UI scrollback" budget would break the - invariant by construction. Codex has two (see §3) and their UI can scroll - through output the model never saw. -- **Two knobs on the tool, and only two:** how much to keep, and which end. Head - vs tail is genuinely per-command — `grep` and `find` want the first matches, - `npm test` wants the failure at the end — and nothing in the harness can infer - it. +Delta encoding is complementary. It compresses a published change; it does not cap state or decide when publication occurs. -The spill file does not violate this. It is not a view; it is a file on disk the -model reaches through `read`/`grep` like any other file. +## 2. Boundaries ---- - -## 2. The problem - -A tool produces output continuously. Every update must reach three consumers. How -often? - -**`intervalMs` is the wrong knob**, and it is what `03-execenv` currently ships. -It was chosen for an in-process TUI where 100 ms feels live. The cost of an emit -varies by ~1000x depending on throughput, so a fixed interval is wrong at both -ends: - -| | bytes per emit | at 100 emits/s | -| --- | --- | --- | -| slow build, 2 KB/s | 40 | 4 KB/s — fine | -| `cat 1gb.txt` | 51 000 | **5.1 MB/s** — untenable | - -**And no purely data-intrinsic policy exists.** "Emit every N bytes changed" gives -constant bytes per emit, so the emit *rate* scales with throughput. "Emit once per -window turnover" is 20 000 emits for 1 GB at a 50 KB window. A clock is -unavoidable; the question is whose, and what it measures. - -### 2.1 Where the delta encoding stops helping - -Measured, 50 KB tail window, one flush per arrival: - -| bytes arriving between flushes | ops | wire | vs a full replacement | -| --- | --- | --- | --- | -| 200 B | `t,a` | 232 B | 0% | -| 2 KB | `t,a` | 2 067 B | 4% | -| 25 KB | `t,a` | 25 504 B | 51% | -| 49 KB | `t,a` | 49 959 B | 100% | -| **50 KB and above** | **`s`** | **50 967 B** | **102%** | - -The delta helps in exact proportion to how little the window moved, and stops -entirely once arrival >= cap. At `cat` rates every update is a full `s`. That is -correct behaviour, not a failure — but it means **the pathological case is bounded -by the cap and the cadence, never by the encoding.** - -Related: at full turnover, `s`-on-the-field and `r`-on-the-root are within 0.3% -(51 833 vs 51 970 bytes). At low rates the field-level `s` is 224x smaller. So the -tracker naturally emits the narrowest op it can, exactly where narrowness pays. - ---- - -## 3. What other agents do - -Read before designing. All three differ from us and from each other. - -### tmux — backpressure at the source - -```c -#define READ_SIZE 1024 // max data held from a pty -#define READ_BACKOFF 512 // bytes waiting for the tty before backing off -#define READ_TIME 100 // µs to wait before the next read +```text +remote process + -> optional ExecutionEnv publisher + -> worker ToolOutput publisher + -> events + durability + replication ``` -tmux stops **reading the pty** when the client is behind, and discards output plus -a full redraw when the backlog gets large — which is our full-turnover `s`, -arrived at independently. +A publisher instance is required only before a real costly boundary: -The symptom that justifies it: *"when the user tries to ^C the command they have -to wait for this backlog to clear"*. **The real cost of a queue is interactivity, -not bandwidth.** Their failure mode is permanent data loss — users reported -scrollback silently truncated — which our spill avoids. +- a physically remote execution environment limits output before transport; +- `ToolOutput` limits custom tools and downstream event/storage traffic; +- a colocated environment can feed its bounded updates in process without another serialized transport; +- custom tools bypass `ExecutionEnv` but cannot bypass `ToolOutput`. -### Codex — no backpressure, deliberately, and two budgets +The control algorithm is shared. Payload vocabularies differ: Shell uses replace/append/slide/metadata; `ToolOutput` uses Chord operations. -```rust -// Continue reading to EOF to avoid back-pressure -pub const DEFAULT_OUTPUT_BYTES_CAP: usize = 1024 * 1024; // head, silent drop -pub(crate) const MAX_EXEC_OUTPUT_DELTAS_PER_CALL: usize = 10_000; -const READ_CHUNK_SIZE: usize = 8192; -``` +## 3. Why size or cadence alone is insufficient -Model gets the first 1 MB, head-retained, silently discarded past that. The UI -gets raw 8 KB chunks it appends itself, capped at 10 000 **events** — so up to -80 MB, bounded by count rather than rate. Crude, needs no timer and no consumer -signal, and breaks our invariant: their UI can scroll through output the model -never saw. +A fixed 50 KB snapshot is safe per event but not over time. At 100 ms it permits ten complete snapshots per second, approximately 500 KB/s plus envelopes. -### OpenCode — no live output at all +A byte budget alone also permits excessive tiny events and durable transactions. A minimum interval bounds count; encoded-size debt bounds bandwidth. + +The original handoff incorrectly treated `intervalMs = 100` as 100 emits/s. It is ten emits/s. The fixed interval was still non-adaptive: it delayed small trickles while allowing complete windows at the same frequency. + +## 4. Landed adaptive algorithm + +`packages/agent/src/harness/utils/adaptive-publisher.ts` implements: ```ts -// TODO: Add durable/live progress metadata streaming for long-running commands -// once V2 tool invocation progress context is wired. +nextDelayMs = max(globalMinEmitInterval, encodedUpdateBytes * 1000 / globalTargetBytesPerSecond); ``` -`collectStream` folds the whole stream into a 1 MB head-retained buffer and -returns once. Their **v2 spec** states the separation we should copy: - -> *"Model-output bounding is not producer memory management. Processes and -> streaming sources may need separate capture or spooling limits before a tool -> result exists. Those limits must be modeled at the producer boundary and must -> not masquerade as model-output truncation. A producer cannot claim a complete -> retained output after it has already discarded bytes."* - -Also worth taking: *"if complete retention fails, settlement fails operationally -rather than publishing lossy success"* — better to fail the tool than hand the -model a silently truncated result. Both Codex and OpenCode-v1 do the lossy thing. - -**We are the only one of the three that keeps the tail.** For `npm test`, where -the failure is at the end, head retention shows the successful setup and drops the -error. - ---- - -## 4. Candidate: a byte-rate limiter owned by the sink - -The constraint is bandwidth, so measure bandwidth. **Unit: bytes per second.** One -value in the sink, not per tool. +Current harness-global policy: ```ts -#maybeEmit(): void { - this.#credit = min(RATE, this.#credit + RATE * secondsSince(this.#lastRefill)); - this.#lastRefill = now(); - if (this.#credit <= 0) return; // accumulate instead - - const ops = this.#tracker.flush(); - if (ops.length === 0) return; - const wire = this.#encoder.encode(ops); - this.#credit -= sizeOf(wire); - this.#publish(wire); // ONE emit, all three consumers -} +minEmitInterval = 100 ms; +targetBytesPerSecond = 100 KB/s; ``` -`flush()` drains, so it can only be called once per batch — which is *why* durable -and UI share it, and why they cannot diverge. +Behavior: -### Measured, 50 KB budget, RATE = 100 KB/s, 10 s +1. The first dirty state after idle publishes immediately. +2. Writes before the next deadline collapse into the latest state. +3. One trailing timer publishes held state after the deadline. +4. Completion and correctness boundaries force one bounded publication. +5. The publisher commits its baseline before consumer delivery, preventing duplicate deltas if a consumer applies and then throws. -| producer | emits/s | wire | -| --- | --- | --- | -| slow build, 2 KB/s | 100/s | 40 KB | -| busy build, 50 KB/s | 100/s | 529 KB | -| **`cat`, 100 MB/s** | **2.2/s** | **1.1 MB** | -| `cat`, no limit (`intervalMs=100` today) | 100/s | **50.6 MB** | +This is an amortized token bucket with a cap-sized burst. A leading or forced terminal update may exceed the target over a short interval, but sustained encoded bytes converge to the target and sustained event count cannot exceed the minimum-interval floor except for explicit forced correctness writes. -**45x less wire for `cat`, and the slow case is more responsive than a 100 ms -interval, not less.** Same rule, opposite behaviour, because it measures the thing -that is actually scarce. Nobody configures a cadence. +## 5. Scenario traces -### 4.1 This depends on D1 being fixed first +Assume a 50 KB cap, 100 KB/s target, and 100 ms floor. -Holding back is only free if held-back writes collapse. **They currently do not** -— see `../01-delta/FINDINGS.md` D1: 1000 held-back interleaved writes produce 2001 -ops and 264 KB. Until that is fixed the rate limiter converts a bandwidth problem -into a memory problem. +### Below cap, completes -### 4.2 Known hole: the trailing emit +The initial state publishes immediately. Small appends publish no faster than the floor, and final dirty state is forced. Total encoded text is approximately the produced text. -`#maybeEmit` runs only when the tool writes. A command that bursts and then goes -quiet on a held write never sends it: +### Below cap, trickling -``` -t=0ms sent 319B (credit -19) -t=10ms sent 329B (credit -248) -t=20ms HELD - ...quiet for 30s... - tracker still dirty — the UI shows state from t=10ms, forever -``` +Writes arriving more than 100 ms apart publish immediately because the preceding deadline has already passed. Faster writes collapse into one append per floor interval. -Needs a **one-shot** timer armed only while something is held, with the delay -computed as `deficit / RATE`. That is not the knob we deleted: it is derived from -`RATE`, not configured, and it does not poll. +### Above cap, full force -### 4.3 Undecided +The retained state never exceeds 50 KB. Complete window turnovers encode as bounded replacements. A roughly 50 KB update buys about 500 ms of silence, yielding approximately two updates and 100 KB per second regardless of raw producer throughput. -- **Durability cadence.** Does the durable write ride the same emit, or run - slower? Riding it is simpler and preserves the invariant trivially. A separate - slower clock reduces writes but needs its own trailing timer and its own - argument for why divergence is acceptable. -- **Backpressure.** tmux stops reading the pipe; Codex explicitly does not. We - currently do not. tmux is the only one with a human pressing Ctrl-C in the - pane — which suggests backpressure buys *interactivity*, and we may be closer to - Codex here. Unmeasured: whether our abort path has the Ctrl-C latency problem. -- **Pull-based instead of a rate limit.** The consumer says "ready" and gets the - current state; no estimation at all. Measured: **1000 producer writes cost the - same one op as 1**, so a slow consumer sees a bigger jump rather than a queue. - Cleaner in principle, needs consumer plumbing and still needs a durability floor. +For shell execution, the complete stream goes to a source-local spill under backpressure rather than over the output-update channel. ---- +### Above cap, trickling -## 5. Consequences for `03-execenv` +A small tail movement encodes as truncate plus append (Chord) or `slide` (Shell). Its encoded size is small, so the 100 ms floor dominates and output remains responsive. Resending a complete 50 KB snapshot for each tiny slide would be both slower and larger. -**`BASH_CHECKPOINT_INTERVAL_MS` goes and nothing in `03-execenv` replaces it.** -The durable cadence is not a `Shell` concern. +### Burst then silence -`bash.after.ts` as shipped checkpoints when the update kind is `snapshot`, on the -reasoning that a snapshot is the only kind that can express eviction. **That is -wrong and should not be copied.** `nowCrossed` is sticky: once total output -exceeds the cap it never resets, so every subsequent update is a snapshot — a -durable write every `intervalMs`, the same cadence, not a lower one. +The leading state is immediate. Held writes collapse, and one trailing timer publishes the latest residue. There is no polling timer. -**Two knobs can also go:** +### Huge single write -- **`maxLines`** is redundant with `maxBytes`. 2000 lines x 25 bytes is the same - bound; 100 000 tiny lines still yields ~50 KB. Keep truncation *line-aware* — - cut at a line boundary at or below the budget — which is what `truncateTail` - already does, so removing the parameter does not change behaviour. -- **A byte limit does not break UTF-8.** Only slicing by *code units* does. - Binary-searching on encoded length lands on a codepoint boundary every time, and - `trimToLastBytes` already does this. Verified across mixed ASCII / accented / - CJK / emoji at 10, 20 and 30-byte budgets. +The producer may already have allocated its input, but the boundary retains and publishes only the configured cap. Shell spill keeps the complete source stream. Arbitrary custom-tool images and structured details still require separate limits. -**`counters` as an update kind can go too.** It exists so head mode can move a -number without resending the frozen text — but nothing needs that number *live*. -The final `ShellExecResult` carries the totals and the visible content was -complete when the cap filled. Head mode emits chunks until full, then silence. -That reduces three update kinds to two. +## 6. Forced writes -**`retain: "head" | "tail"` stays.** Head is correct for search-shaped output — -`grep`, `find`, `ls -R`, where results are unordered so the first N are as -informative as any N and arrive immediately. `truncateHead` is already used by -`read.ts`. +These bypass pacing once, while remaining state-size bounded: -Resulting `CaptureOptions`: `{ maxBytes, retain?, spill? }` — three, from five. +- command/tool completion; +- error or abort; +- a memo and output checkpoint committed atomically; +- an explicit recovery base/rebase. + +A terminal flush cancels the trailing timer before `tool_end`, preventing a late update for a settled invocation. + +## 7. `ToolOutput` application + +The sink will own one Chord tracker and encoder per invocation. Mutations remain local while publication is blocked; flush-time dirty tracking means held writes collapse without retaining one op per write. + +One sink flush feeds the live event and current model-visible state. Durable batches use the same logical flush, encoded as per-stream `WireOp[]`. Periodic `rebase()` bounds recovery replay; a memo correctness flush writes a tagged base batch in the same transaction as the memo. + +Durable cadence belongs to the sink, not to Shell or bash. Bash's existing two-second checkpoint request remains only as an interim compatibility mechanism until the sink migration. + +## 8. Remaining decisions + +- Explicit byte/count rejection policy for images. +- Bounds for structured details; arbitrary JSON cannot be meaningfully tail-windowed. +- Whether ordinary durable writes ride every live sink publication initially or use a slower measured cadence. Memo and terminal correctness flushes are not optional. +- Exact global production values after profiling real remote transport and storage. diff --git a/packages/agent/src/harness/env/nodejs.ts b/packages/agent/src/harness/env/nodejs.ts index b8c8fd6dd..cfd4690a5 100644 --- a/packages/agent/src/harness/env/nodejs.ts +++ b/packages/agent/src/harness/env/nodejs.ts @@ -1,6 +1,6 @@ import { type ChildProcess, spawn } from "node:child_process"; import { randomUUID } from "node:crypto"; -import { constants, createReadStream } from "node:fs"; +import { constants, createReadStream, createWriteStream, type WriteStream } from "node:fs"; import { access, appendFile, @@ -29,12 +29,17 @@ import { ok, type Result, type ShellExecOptions, + type ShellExecResult, toError, } from "../types.ts"; +import { OutputCapture } from "../utils/output-capture.ts"; const MAX_TIMEOUT_MS = 2_147_483_647; const MAX_TIMEOUT_SECONDS = MAX_TIMEOUT_MS / 1000; const EXIT_STDIO_GRACE_MS = 100; +const SPILL_HIGH_WATER_MARK = 8 * 1024 * 1024; + +type SpillChunk = string | Uint8Array; function resolveTimeoutMs(timeout: number | undefined): Result { if (timeout === undefined) return ok(undefined); @@ -282,7 +287,7 @@ function killProcessTree(pid: number): void { } } -function waitForChildProcess(child: ChildProcess): Promise { +function waitForChildProcess(child: ChildProcess, spillIsDraining: () => boolean): Promise { return new Promise((resolvePromise, reject) => { let settled = false; let exited = false; @@ -314,7 +319,10 @@ function waitForChildProcess(child: ChildProcess): Promise { }; const armIdleTimer = (): void => { if (postExitTimer) clearTimeout(postExitTimer); - postExitTimer = setTimeout(() => finalize(exitCode), EXIT_STDIO_GRACE_MS); + postExitTimer = setTimeout(() => { + if (spillIsDraining()) armIdleTimer(); + else finalize(exitCode); + }, EXIT_STDIO_GRACE_MS); }; const onData = (): void => { if (exited && !settled) armIdleTimer(); @@ -375,7 +383,7 @@ export class NodeExecutionEnv implements ExecutionEnv { command: string, options: ShellExecOptions | undefined, context: Context, - ): Promise> { + ): Promise> { const signal = context.abortSignal; if (signal?.aborted) return err(new ExecutionError("aborted", "aborted")); const timeoutMsResult = resolveTimeoutMs(options?.timeout); @@ -399,28 +407,110 @@ export class NodeExecutionEnv implements ExecutionEnv { } return await new Promise((resolvePromise) => { - let stdout = ""; - let stderr = ""; let settled = false; let timedOut = false; let callbackError: ExecutionError | undefined; + let spillError: ExecutionError | undefined; let child: ReturnType | undefined; let timeoutId: ReturnType | undefined; + const spillPrefix: SpillChunk[] = []; + let spillPath: string | undefined; + const spillQueue: SpillChunk[] = []; + let spillStart: Promise | undefined; + let spillStream: WriteStream | undefined; + let spillBackpressured = false; const onAbort = () => { - if (child?.pid) { - killProcessTree(child.pid); - } + if (child?.pid) killProcessTree(child.pid); }; + const failCallback = (error: unknown) => { + if (callbackError !== undefined) return; + const cause = toError(error); + callbackError = new ExecutionError("callback_error", cause.message, cause); + onAbort(); + }; + let capture: OutputCapture; + try { + capture = new OutputCapture(options?.capture, context, { + onUpdate: options?.onUpdate, + onError: failCallback, + }); + } catch (error) { + const cause = toError(error); + resolvePromise(err(new ExecutionError("unknown", cause.message, cause))); + return; + } - const settle = (result: Result<{ stdout: string; stderr: string; exitCode: number }, ExecutionError>) => { + const settle = (result: Result) => { + if (settled) return; + settled = true; if (timeoutId) clearTimeout(timeoutId); if (signal) signal.removeEventListener("abort", onAbort); if (child?.pid) this.activeChildPids.delete(child.pid); - if (settled) return; - settled = true; + capture.dispose(); resolvePromise(result); }; + const pauseOutput = () => { + child?.stdout?.pause(); + child?.stderr?.pause(); + }; + const resumeOutput = () => { + if (callbackError || spillError || timedOut || signal?.aborted || spillBackpressured) return; + child?.stdout?.resume(); + child?.stderr?.resume(); + }; + const failSpill = (error: unknown) => { + if (spillError !== undefined) return; + const cause = toError(error); + spillError = new ExecutionError( + "unknown", + `Failed to preserve complete shell output: ${cause.message}`, + cause, + ); + spillBackpressured = false; + onAbort(); + }; + const writeSpill = (chunk: SpillChunk): void => { + if (spillStream === undefined || chunk.length === 0) return; + if (spillStream.write(chunk) || spillBackpressured) return; + spillBackpressured = true; + pauseOutput(); + spillStream.once("drain", () => { + spillBackpressured = false; + resumeOutput(); + }); + }; + const startSpill = (chunk: SpillChunk): void => { + if (spillStream !== undefined) { + writeSpill(chunk); + return; + } + spillQueue.push(chunk); + if (spillStart !== undefined) return; + pauseOutput(); + spillStart = (async () => { + const created = await this.createTempFile({ prefix: "pi-output-", suffix: ".log" }, context); + if (!created.ok) throw created.error; + spillPath = created.value; + capture.setSpillPath(spillPath); + spillStream = createWriteStream(spillPath, { flags: "a", highWaterMark: SPILL_HIGH_WATER_MARK }); + spillStream.on("error", failSpill); + for (const queued of spillQueue) writeSpill(queued); + spillQueue.length = 0; + })() + .catch(failSpill) + .finally(resumeOutput); + }; + const finishSpill = async (): Promise => { + await spillStart; + const stream = spillStream; + if (stream === undefined || spillError !== undefined || stream.destroyed) return; + await new Promise((resolveFinish) => { + stream.once("error", () => resolveFinish()); + stream.once("finish", resolveFinish); + stream.end(); + }); + }; try { const commandFromStdin = shellConfig.value.commandTransport === "stdin"; @@ -447,48 +537,54 @@ export class NodeExecutionEnv implements ExecutionEnv { } timeoutId = - timeoutMs !== undefined - ? setTimeout(() => { + timeoutMs === undefined + ? undefined + : setTimeout(() => { timedOut = true; - if (child?.pid) { - killProcessTree(child.pid); - } - }, timeoutMs) - : undefined; + onAbort(); + }, timeoutMs); if (signal) { - if (signal.aborted) { - onAbort(); - } else { - signal.addEventListener("abort", onAbort, { once: true }); - } + if (signal.aborted) onAbort(); + else signal.addEventListener("abort", onAbort, { once: true }); } - child.stdout?.setEncoding("utf8"); - child.stderr?.setEncoding("utf8"); - child.stdout?.on("data", (chunk: string) => { - stdout += chunk; + const feed = (chunk: Uint8Array) => { try { - options?.onStdout?.(chunk, context); + const wasTruncated = capture.truncated; + capture.push(chunk); + if (!options?.capture?.spill || chunk.length === 0) return; + if (spillPath !== undefined || wasTruncated) { + startSpill(chunk); + } else if (capture.truncated) { + for (const prefix of spillPrefix) startSpill(prefix); + spillPrefix.length = 0; + startSpill(chunk); + } else { + spillPrefix.push(chunk); + } } catch (error) { - const cause = toError(error); - callbackError = new ExecutionError("callback_error", cause.message, cause); - onAbort(); + failCallback(error); } - }); - child.stderr?.on("data", (chunk: string) => { - stderr += chunk; - try { - options?.onStderr?.(chunk, context); - } catch (error) { - const cause = toError(error); - callbackError = new ExecutionError("callback_error", cause.message, cause); - onAbort(); - } - }); + }; + child.stdout?.on("data", feed); + child.stderr?.on("data", feed); - void waitForChildProcess(child).then( - (code) => { + void waitForChildProcess( + child, + () => + spillError === undefined && + spillStart !== undefined && + (spillStream === undefined || spillBackpressured), + ).then( + async (code) => { + await finishSpill(); + try { + capture.finish(); + capture.flush(); + } catch (error) { + failCallback(error); + } if (callbackError) { settle(err(callbackError)); return; @@ -501,7 +597,19 @@ export class NodeExecutionEnv implements ExecutionEnv { settle(err(new ExecutionError("aborted", "aborted"))); return; } - settle(ok({ stdout, stderr, exitCode: code ?? 0 })); + if (spillError) { + settle(err(spillError)); + return; + } + const output = capture.snapshot(); + settle( + ok({ + exitCode: code ?? 0, + truncation: output.truncation, + ...(output.spillPath === undefined ? {} : { spillPath: output.spillPath }), + ...(output.lastLineBytes === undefined ? {} : { lastLineBytes: output.lastLineBytes }), + }), + ); }, (error: Error) => settle(err(new ExecutionError("spawn_error", error.message, error))), ); diff --git a/packages/agent/src/harness/tools/bash.ts b/packages/agent/src/harness/tools/bash.ts index af04de9a0..eff1e18a4 100644 --- a/packages/agent/src/harness/tools/bash.ts +++ b/packages/agent/src/harness/tools/bash.ts @@ -1,13 +1,11 @@ import { type Static, Type } from "typebox"; import type { Context } from "../context.ts"; -import type { AgentHarnessTool } from "../types.ts"; -import { getOrThrow } from "../types.ts"; -import { executeShellWithCapture, type ShellCaptureProgress } from "../utils/shell-output.ts"; -import { DEFAULT_MAX_BYTES, DEFAULT_MAX_LINES, formatSize, type TruncationResult } from "../utils/truncate.ts"; +import type { AgentHarnessTool, ShellOutputTruncation, ShellOutputView } from "../types.ts"; +import { applyShellOutputUpdate } from "../utils/output-capture.ts"; +import { DEFAULT_MAX_BYTES, DEFAULT_MAX_LINES, formatSize } from "../utils/truncate.ts"; import type { ExecutionToolContext } from "./tool-context.ts"; const MAX_TIMEOUT_SECONDS = 2_147_483_647 / 1000; -const BASH_UPDATE_THROTTLE_MS = 100; const BASH_CHECKPOINT_INTERVAL_MS = 2_000; const bashSchema = Type.Object({ @@ -18,7 +16,7 @@ const bashSchema = Type.Object({ export type BashToolInput = Static; export interface BashToolDetails { - truncation?: TruncationResult; + truncation?: ShellOutputTruncation; fullOutputPath?: string; } @@ -56,7 +54,7 @@ export function createBashTool ShellCaptureProgress) | undefined; - let updateTimer: ReturnType | undefined; - let updateDirty = false; - let lastUpdateAt = 0; + let view: ShellOutputView | undefined; let lastCheckpointAt = Date.now(); let lastCheckpoint: string | undefined; - - const emitOutputUpdate = (): void => { - if (!updateDirty || !getLatestProgress) return; - updateDirty = false; - const now = Date.now(); - lastUpdateAt = now; - const progress = getLatestProgress(); - const snapshot = { - content: [{ type: "text" as const, text: progress.output }], - details: { - truncation: progress.truncation.truncated ? progress.truncation : undefined, - fullOutputPath: progress.fullOutputPath, - }, - }; - const encoded = JSON.stringify(snapshot); - const checkpoint = now - lastCheckpointAt >= BASH_CHECKPOINT_INTERVAL_MS && encoded !== lastCheckpoint; - onUpdate(snapshot, checkpoint ? { checkpoint: true } : undefined); - if (checkpoint) { - lastCheckpointAt = now; - lastCheckpoint = encoded; - } - }; - const clearUpdateTimer = (): void => { - if (!updateTimer) return; - clearTimeout(updateTimer); - updateTimer = undefined; - }; - const scheduleOutputUpdate = (): void => { - updateDirty = true; - const delay = BASH_UPDATE_THROTTLE_MS - (Date.now() - lastUpdateAt); - if (delay <= 0) { - clearUpdateTimer(); - emitOutputUpdate(); - return; - } - updateTimer ??= setTimeout(() => { - updateTimer = undefined; - emitOutputUpdate(); - }, delay); - }; + let acceptingUpdates = true; onUpdate({ content: [], details: undefined }); - try { - const capture = getOrThrow( - await executeShellWithCapture( - env, - execution.command, - { - cwd: execution.cwd, - env: execution.env, - inheritEnv: execution.inheritEnv, - timeout, - returnExecutionErrors: true, - onChunk: (_chunk, getProgress) => { - getLatestProgress = getProgress; - scheduleOutputUpdate(); + const result = await env.exec( + execution.command, + { + cwd: execution.cwd, + env: execution.env, + inheritEnv: execution.inheritEnv, + timeout, + capture: { + limits: { maxBytes: DEFAULT_MAX_BYTES, maxLines: DEFAULT_MAX_LINES, retain: "tail" }, + spill: true, + }, + onUpdate: (update) => { + if (!acceptingUpdates) return; + view = applyShellOutputUpdate(view, update); + const snapshot = { + content: [{ type: "text" as const, text: view.text }], + details: { + truncation: view.truncation.truncated ? view.truncation : undefined, + fullOutputPath: view.spillPath, }, - }, - context, - ), - ); - clearUpdateTimer(); - getLatestProgress = () => capture; - updateDirty = true; - emitOutputUpdate(); + }; + const now = Date.now(); + const encoded = JSON.stringify(snapshot); + const checkpoint = + now - lastCheckpointAt >= BASH_CHECKPOINT_INTERVAL_MS && encoded !== lastCheckpoint; + onUpdate(snapshot, checkpoint ? { checkpoint: true } : undefined); + if (checkpoint) { + lastCheckpointAt = now; + lastCheckpoint = encoded; + } + }, + }, + context, + ); + acceptingUpdates = false; - let outputText = capture.output; - let details: BashToolDetails | undefined; - if (capture.truncation.truncated) { - details = { truncation: capture.truncation, fullOutputPath: capture.fullOutputPath }; - const startLine = capture.truncation.totalLines - capture.truncation.outputLines + 1; - const endLine = capture.truncation.totalLines; - if (capture.truncation.lastLinePartial) { - const lastLineSize = formatSize(capture.lastLineBytes); - outputText += `\n\n[Showing last ${formatSize(capture.truncation.outputBytes)} of line ${endLine} (line is ${lastLineSize}). Full output: ${capture.fullOutputPath}]`; - } else if (capture.truncation.truncatedBy === "lines") { - outputText += `\n\n[Showing lines ${startLine}-${endLine} of ${capture.truncation.totalLines}. Full output: ${capture.fullOutputPath}]`; - } else { - outputText += `\n\n[Showing lines ${startLine}-${endLine} of ${capture.truncation.totalLines} (${formatSize(DEFAULT_MAX_BYTES)} limit). Full output: ${capture.fullOutputPath}]`; - } + let outputText = view?.text ?? ""; + const capture = result.ok ? { text: outputText, ...result.value } : view; + let details: BashToolDetails | undefined; + if (capture?.truncation.truncated) { + details = { truncation: capture.truncation, fullOutputPath: capture.spillPath }; + const startLine = capture.truncation.totalLines - capture.truncation.outputLines + 1; + const endLine = capture.truncation.totalLines; + if (capture.truncation.lastLinePartial) { + const lastLineSize = formatSize(capture.lastLineBytes ?? capture.truncation.outputBytes); + outputText += `\n\n[Showing last ${formatSize(capture.truncation.outputBytes)} of line ${endLine} (line is ${lastLineSize}). Full output: ${capture.spillPath}]`; + } else if (capture.truncation.truncatedBy === "lines") { + outputText += `\n\n[Showing lines ${startLine}-${endLine} of ${capture.truncation.totalLines}. Full output: ${capture.spillPath}]`; + } else { + outputText += `\n\n[Showing lines ${startLine}-${endLine} of ${capture.truncation.totalLines} (${formatSize(DEFAULT_MAX_BYTES)} limit). Full output: ${capture.spillPath}]`; } - - const appendStatus = (status: string): string => `${outputText ? `${outputText}\n\n` : ""}${status}`; - if (capture.cancelled) throw new Error(appendStatus("Command aborted")); - if (capture.executionError?.code === "timeout") { - throw new Error(appendStatus(`Command timed out after ${timeout} seconds`), { - cause: capture.executionError, - }); - } - if (capture.executionError) throw capture.executionError; - if (capture.exitCode !== 0 && capture.exitCode !== undefined) { - throw new Error(appendStatus(`Command exited with code ${capture.exitCode}`)); - } - return { content: [{ type: "text", text: outputText || "(no output)" }], details }; - } finally { - clearUpdateTimer(); } + + if (!result.ok) { + const status = + result.error.code === "timeout" + ? `Command timed out after ${timeout} seconds` + : result.error.code === "aborted" + ? "Command aborted" + : result.error.message; + throw new Error(outputText ? `${outputText}\n\n${status}` : status, { cause: result.error }); + } + if (result.value.exitCode !== 0) { + throw new Error( + `${outputText ? `${outputText}\n\n` : ""}Command exited with code ${result.value.exitCode}`, + ); + } + return { content: [{ type: "text", text: outputText || "(no output)" }], details }; }, }; } diff --git a/packages/agent/src/harness/types.ts b/packages/agent/src/harness/types.ts index baae97607..23ff7604b 100644 --- a/packages/agent/src/harness/types.ts +++ b/packages/agent/src/harness/types.ts @@ -3,6 +3,7 @@ import type { Static, TSchema } from "typebox"; import type { AgentTool, AgentToolResult } from "../types.ts"; import type { Context } from "./context.ts"; import type { JsonValue } from "./session/types.ts"; +import type { TruncationResult } from "./utils/truncate.ts"; /** Result of a fallible operation. Expected failures are returned as `ok: false` instead of thrown. */ export type Result = { ok: true; value: TValue } | { ok: false; error: TError }; @@ -313,6 +314,51 @@ export interface FileSystem { cleanup(context: Context): Promise; } +/** Which portion of bounded output survives after the limit is crossed. */ +export type ShellOutputRetention = "head" | "tail"; + +/** Source-side limits for one combined shell output view. */ +export interface ShellOutputLimits { + maxBytes: number; + maxLines: number; + /** Defaults to `"tail"`. */ + retain?: ShellOutputRetention; +} + +/** Bounded shell capture requested by the caller. */ +export interface ShellOutputCaptureOptions { + limits: ShellOutputLimits; + /** Preserve complete output in an execution-environment-local file after the limits are crossed. */ + spill?: boolean; +} + +/** Truncation metadata without a duplicate copy of the retained text. */ +export type ShellOutputTruncation = Omit; + +/** Metadata accompanying a bounded shell output view. */ +export interface ShellOutputMetadata { + truncation: ShellOutputTruncation; + spillPath?: string; + lastLineBytes?: number; +} + +/** Complete bounded shell output view. */ +export interface ShellOutputView extends ShellOutputMetadata { + text: string; +} + +/** Incremental source-side change to one bounded shell output view. */ +export type ShellOutputUpdate = + | { kind: "replace"; output: ShellOutputView } + | { kind: "append"; text: string; metadata: ShellOutputMetadata } + | { kind: "slide"; drop: number; text: string; metadata: ShellOutputMetadata } + | { kind: "metadata"; metadata: ShellOutputMetadata }; + +/** Bounded shell completion. Output text is delivered through {@link ShellExecOptions.onUpdate}. */ +export interface ShellExecResult extends ShellOutputMetadata { + exitCode: number; +} + /** Options for {@link Shell.exec}. */ export interface ShellExecOptions { /** Working directory for the command. Relative paths are resolved against {@link ExecutionEnv.cwd}. Defaults to {@link ExecutionEnv.cwd}. */ @@ -323,10 +369,10 @@ export interface ShellExecOptions { inheritEnv?: boolean; /** Timeout in seconds. Implementations should return a timeout error when the command exceeds this duration. Defaults to no timeout. */ timeout?: number; - /** Called with stdout chunks as they are produced. */ - onStdout?: (chunk: string, context: Context) => void; - /** Called with stderr chunks as they are produced. */ - onStderr?: (chunk: string, context: Context) => void; + /** Source-side bounded capture. Output is discarded when this and `onUpdate` are both absent. */ + capture?: ShellOutputCaptureOptions; + /** Called with bounded output changes. */ + onUpdate?: (update: ShellOutputUpdate, context: Context) => void; } /** Shell execution capability used by the harness. */ @@ -336,7 +382,7 @@ export interface Shell { command: string, options: ShellExecOptions | undefined, context: Context, - ): Promise>; + ): Promise>; /** Release shell resources. Must be best-effort and must not throw or reject. */ cleanup(context: Context): Promise; } diff --git a/packages/agent/src/harness/utils/adaptive-publisher.ts b/packages/agent/src/harness/utils/adaptive-publisher.ts new file mode 100644 index 000000000..d971ee969 --- /dev/null +++ b/packages/agent/src/harness/utils/adaptive-publisher.ts @@ -0,0 +1,87 @@ +export interface AdaptivePublisherOptions { + snapshot(): TValue; + update(previous: TValue | undefined, current: TValue): TUpdate | undefined; + measure(update: TUpdate): number; + publish(update: TUpdate): void; + onError(error: unknown): void; + minIntervalMs?: number; + targetBytesPerSecond?: number; +} + +/** + * Publishes the latest state without queuing intermediate mutations. + * + * The first dirty state after idle is immediate. Each publication then buys a + * delay proportional to its encoded size, with a minimum interval that also + * bounds event count. A single trailing timer guarantees eventual publication. + */ +export class AdaptivePublisher { + readonly #options: AdaptivePublisherOptions; + readonly #minIntervalMs: number; + readonly #targetBytesPerSecond: number; + #published: TValue | undefined; + #dirty = false; + #nextEmitAt = 0; + #timer: ReturnType | undefined; + #disposed = false; + + constructor(options: AdaptivePublisherOptions) { + this.#options = options; + this.#minIntervalMs = options.minIntervalMs ?? 100; + this.#targetBytesPerSecond = options.targetBytesPerSecond ?? 100 * 1024; + } + + markDirty(): void { + if (this.#disposed) return; + this.#dirty = true; + const wait = this.#nextEmitAt - Date.now(); + if (wait <= 0) { + this.flush(); + return; + } + this.#armTimer(wait); + } + + flush(force = false): void { + if (this.#disposed || !this.#dirty) return; + const now = Date.now(); + if (!force && now < this.#nextEmitAt) { + this.#armTimer(this.#nextEmitAt - now); + return; + } + if (this.#timer !== undefined) clearTimeout(this.#timer); + this.#timer = undefined; + const current = this.#options.snapshot(); + const update = this.#options.update(this.#published, current); + if (update === undefined) { + this.#published = current; + this.#dirty = false; + return; + } + const encodedBytes = this.#options.measure(update); + this.#published = current; + this.#dirty = false; + this.#nextEmitAt = now + Math.max(this.#minIntervalMs, (encodedBytes * 1000) / this.#targetBytesPerSecond); + // Commit before delivery. A consumer may apply the update and then throw or + // reenter the producer; retaining the old baseline would duplicate that delta. + this.#options.publish(update); + } + + dispose(): void { + if (this.#timer !== undefined) clearTimeout(this.#timer); + this.#timer = undefined; + this.#disposed = true; + } + + #armTimer(wait: number): void { + if (this.#timer !== undefined) return; + this.#timer = setTimeout(() => { + this.#timer = undefined; + try { + this.flush(); + } catch (error) { + this.#options.onError(error); + } + }, wait); + } +} diff --git a/packages/agent/src/harness/utils/output-capture.ts b/packages/agent/src/harness/utils/output-capture.ts new file mode 100644 index 000000000..b640a6a83 --- /dev/null +++ b/packages/agent/src/harness/utils/output-capture.ts @@ -0,0 +1,238 @@ +import type { Context } from "../context.ts"; +import type { ShellOutputCaptureOptions, ShellOutputMetadata, ShellOutputUpdate, ShellOutputView } from "../types.ts"; +import { AdaptivePublisher } from "./adaptive-publisher.ts"; +import { DEFAULT_MAX_BYTES, DEFAULT_MAX_LINES, truncateHead, truncateTail, utf8ByteLength } from "./truncate.ts"; + +export const OUTPUT_MIN_EMIT_INTERVAL_MS = 100; +export const OUTPUT_TARGET_BYTES_PER_SECOND = 100 * 1024; + +const INVALID_SHELL_OUTPUT = /[\x00-\x08\x0b-\x1f\ufff9-\ufffb]/g; +const textEncoder = new TextEncoder(); +const textDecoder = new TextDecoder(); + +interface OutputCaptureHandlers { + onUpdate?: (update: ShellOutputUpdate, context: Context) => void; + onError(error: unknown): void; +} + +/** + * Maintains and publishes one bounded shell-output view. + * + * Writes received while publication is rate-limited collapse into the latest + * view. Small changes remain responsive; complete window turnovers purchase a + * proportionally longer delay. The first update after idle and an explicit + * final flush are immediate. + */ +export class OutputCapture { + readonly #maxBytes: number; + readonly #maxLines: number; + readonly #retain: "head" | "tail"; + readonly #context: Context; + readonly #onUpdate: OutputCaptureHandlers["onUpdate"]; + + readonly #decoder = new TextDecoder(); + #buffer = ""; + #bufferBytes = 0; + #totalBytes = 0; + #newlines = 0; + #endsWithNewline = true; + #currentLineBytes = 0; + #spillPath: string | undefined; + #disposed = false; + readonly #publisher: AdaptivePublisher; + + constructor(options: ShellOutputCaptureOptions | undefined, context: Context, handlers: OutputCaptureHandlers) { + this.#maxBytes = options?.limits.maxBytes ?? DEFAULT_MAX_BYTES; + this.#maxLines = options?.limits.maxLines ?? DEFAULT_MAX_LINES; + this.#retain = options?.limits.retain ?? "tail"; + this.#context = context; + this.#onUpdate = handlers.onUpdate; + if (!Number.isFinite(this.#maxBytes) || this.#maxBytes <= 0) { + throw new TypeError("Output maxBytes must be a positive finite number"); + } + if (!Number.isInteger(this.#maxLines) || this.#maxLines <= 0) { + throw new TypeError("Output maxLines must be a positive integer"); + } + this.#publisher = new AdaptivePublisher({ + snapshot: () => this.snapshot(), + update: updateFrom, + measure: (update) => utf8ByteLength(JSON.stringify(update)), + publish: (update) => this.#onUpdate?.(update, this.#context), + onError: handlers.onError, + minIntervalMs: OUTPUT_MIN_EMIT_INTERVAL_MS, + targetBytesPerSecond: OUTPUT_TARGET_BYTES_PER_SECOND, + }); + } + + get truncated(): boolean { + return this.#totalBytes > this.#maxBytes || this.#totalLines() > this.#maxLines; + } + + push(chunk: string | Uint8Array): void { + if (this.#disposed) return; + if (typeof chunk === "string") { + this.#appendText(this.#decoder.decode()); + this.#appendText(chunk); + return; + } + this.#appendText(this.#decoder.decode(chunk, { stream: true })); + } + + finish(): void { + if (this.#disposed) return; + this.#appendText(this.#decoder.decode()); + } + + setSpillPath(path: string): void { + if (this.#disposed || this.#spillPath === path) return; + this.#spillPath = path; + this.#publisher.markDirty(); + this.flush(); + } + + snapshot(): ShellOutputView { + const retained = + this.#retain === "head" + ? truncateHead(this.#buffer, { maxBytes: this.#maxBytes, maxLines: this.#maxLines }) + : truncateTail(this.#buffer, { maxBytes: this.#maxBytes, maxLines: this.#maxLines }); + const totalLines = this.#totalLines(); + const truncated = this.truncated; + const { content, ...truncation } = retained; + return { + text: sanitizeShellOutput(content), + truncation: { + ...truncation, + truncated, + truncatedBy: truncated ? (totalLines > this.#maxLines ? "lines" : "bytes") : null, + totalBytes: this.#totalBytes, + totalLines, + }, + ...(this.#spillPath === undefined ? {} : { spillPath: this.#spillPath }), + ...(retained.lastLinePartial ? { lastLineBytes: this.#currentLineBytes } : {}), + }; + } + + flush(): void { + if (this.#disposed) return; + this.#publisher.flush(true); + } + + dispose(): void { + this.#publisher.dispose(); + this.#disposed = true; + } + + #appendText(text: string): void { + if (text === "") return; + const textBytes = utf8ByteLength(text); + this.#totalBytes += textBytes; + this.#newlines += countNewlines(text); + this.#endsWithNewline = text.endsWith("\n"); + const lastNewline = text.lastIndexOf("\n"); + this.#currentLineBytes = + lastNewline === -1 ? this.#currentLineBytes + textBytes : utf8ByteLength(text.slice(lastNewline + 1)); + this.#buffer += text; + this.#bufferBytes += textBytes; + + const guard = this.#maxBytes * 2; + if (this.#bufferBytes > guard * 2) { + this.#buffer = + this.#retain === "tail" + ? trimToLastUtf8Bytes(this.#buffer, guard) + : trimToFirstUtf8Bytes(this.#buffer, guard); + this.#bufferBytes = utf8ByteLength(this.#buffer); + } + this.#publisher.markDirty(); + } + + #totalLines(): number { + return this.#newlines + (this.#endsWithNewline || this.#totalBytes === 0 ? 0 : 1); + } +} + +export function applyShellOutputUpdate( + current: ShellOutputView | undefined, + update: ShellOutputUpdate, +): ShellOutputView { + switch (update.kind) { + case "replace": + return update.output; + case "append": + return { text: `${current?.text ?? ""}${update.text}`, ...update.metadata }; + case "slide": + return { text: `${current?.text.slice(update.drop) ?? ""}${update.text}`, ...update.metadata }; + case "metadata": + return { text: current?.text ?? "", ...update.metadata }; + } +} + +function updateFrom(previous: ShellOutputView | undefined, current: ShellOutputView): ShellOutputUpdate { + if (previous === undefined) return { kind: "replace", output: current }; + const metadata: ShellOutputMetadata = { + truncation: current.truncation, + ...(current.spillPath === undefined ? {} : { spillPath: current.spillPath }), + ...(current.lastLineBytes === undefined ? {} : { lastLineBytes: current.lastLineBytes }), + }; + if (current.text === previous.text) return { kind: "metadata", metadata }; + if (current.text.length > previous.text.length && current.text.slice(0, previous.text.length) === previous.text) { + return { kind: "append", text: current.text.slice(previous.text.length), metadata }; + } + const shared = suffixPrefixOverlap( + previous.text, + current.text, + Math.min(previous.text.length, current.text.length, current.truncation.maxBytes * 2), + ); + if (shared > 0) { + return { + kind: "slide", + drop: previous.text.length - shared, + text: current.text.slice(shared), + metadata, + }; + } + return { kind: "replace", output: current }; +} + +function suffixPrefixOverlap(before: string, after: string, scan: number): number { + if (before.length === 0 || after.length === 0 || scan === 0) return 0; + const tail = before.length > scan ? before.slice(before.length - scan) : before; + for (const probeLength of [Math.min(64, after.length), 1]) { + const probe = after.slice(0, probeLength); + let candidates = 0; + for (let index = tail.indexOf(probe); index !== -1; index = tail.indexOf(probe, index + 1)) { + if (++candidates > 8) break; + const overlapLength = tail.length - index; + if (overlapLength <= after.length && tail.slice(index) === after.slice(0, overlapLength)) { + return overlapLength; + } + } + if (probeLength === 1) break; + } + return 0; +} + +export function sanitizeShellOutput(text: string): string { + return text.replace(INVALID_SHELL_OUTPUT, ""); +} + +function countNewlines(text: string): number { + let count = 0; + for (let index = text.indexOf("\n"); index !== -1; index = text.indexOf("\n", index + 1)) count++; + return count; +} + +function trimToLastUtf8Bytes(text: string, maxBytes: number): string { + const bytes = textEncoder.encode(text); + if (bytes.length <= maxBytes) return text; + let start = bytes.length - maxBytes; + while (start < bytes.length && ((bytes[start] ?? 0) & 0xc0) === 0x80) start++; + return textDecoder.decode(bytes.subarray(start)); +} + +function trimToFirstUtf8Bytes(text: string, maxBytes: number): string { + const bytes = textEncoder.encode(text); + if (bytes.length <= maxBytes) return text; + let end = maxBytes; + while (end > 0 && ((bytes[end] ?? 0) & 0xc0) === 0x80) end--; + return textDecoder.decode(bytes.subarray(0, end)); +} diff --git a/packages/agent/src/harness/utils/shell-output.ts b/packages/agent/src/harness/utils/shell-output.ts index 754d35a16..fc0e4c847 100644 --- a/packages/agent/src/harness/utils/shell-output.ts +++ b/packages/agent/src/harness/utils/shell-output.ts @@ -1,5 +1,14 @@ import type { Context } from "../context.ts"; -import { type ExecutionEnv, ExecutionError, err, ok, type Result, type ShellExecOptions, toError } from "../types.ts"; +import { + type ExecutionEnv, + type ExecutionError, + err, + ok, + type Result, + type ShellExecOptions, + type ShellOutputView, +} from "../types.ts"; +import { applyShellOutputUpdate, sanitizeShellOutput } from "./output-capture.ts"; import { DEFAULT_MAX_BYTES, DEFAULT_MAX_LINES, type TruncationResult, truncateTail } from "./truncate.ts"; export interface ShellCaptureProgress { @@ -9,7 +18,7 @@ export interface ShellCaptureProgress { lastLineBytes: number; } -export interface ShellCaptureOptions extends Omit { +export interface ShellCaptureOptions extends Omit { onChunk?: (chunk: string, getProgress: () => ShellCaptureProgress, context: Context) => void; /** Return shell execution failures with captured output instead of as a failed Result. */ returnExecutionErrors?: boolean; @@ -22,179 +31,82 @@ export interface ShellCaptureResult extends ShellCaptureProgress { executionError?: ExecutionError; } -function toExecutionError(error: unknown): ExecutionError { - if (error instanceof ExecutionError) return error; - const cause = toError(error); - return new ExecutionError("unknown", cause.message, cause); -} - -export function sanitizeBinaryOutput(str: string): string { - return Array.from(str) - .filter((char) => { - const code = char.codePointAt(0); - if (code === undefined) return false; - if (code === 0x09 || code === 0x0a || code === 0x0d) return true; - if (code <= 0x1f) return false; - if (code >= 0xfff9 && code <= 0xfffb) return false; - return true; - }) - .join(""); -} - -function trimToLastUtf8Bytes(text: string, maxBytes: number, encoder: { encode(input?: string): Uint8Array }): string { - const bytes = encoder.encode(text); - if (bytes.byteLength <= maxBytes) return text; - let start = bytes.byteLength - maxBytes; - while (start < bytes.byteLength && ((bytes[start] ?? 0) & 0xc0) === 0x80) start++; - return new TextDecoder().decode(bytes.subarray(start)); +function progressFrom(output: ShellOutputView): ShellCaptureProgress { + return { + output: output.text, + truncation: { content: output.text, ...output.truncation }, + ...(output.spillPath === undefined ? {} : { fullOutputPath: output.spillPath }), + lastLineBytes: output.lastLineBytes ?? 0, + }; } +/** + * Compatibility collector for callers that need one bounded final view. + * Source-side capture, adaptive publication, and spilling remain owned by the + * execution environment. + */ export async function executeShellWithCapture( env: ExecutionEnv, command: string, options: ShellCaptureOptions | undefined, context: Context, ): Promise> { - let tailOutput = ""; - const maxOutputBytes = DEFAULT_MAX_BYTES * 2; - const encoder = new TextEncoder(); - - let totalBytes = 0; - let completedLines = 0; - let hasOpenLine = false; - let currentLineBytes = 0; - let fullOutputPath: string | undefined; - let fullOutputRequested = false; - let acceptingOutput = true; - let writeChain: Promise> = Promise.resolve(ok(undefined)); - let captureError: ExecutionError | undefined; - - const appendFullOutput = (text: string): void => { - if (!fullOutputRequested || captureError) return; - writeChain = writeChain.then(async (previous) => { - if (!previous.ok) return previous; - if (!fullOutputPath) return err(new ExecutionError("unknown", "Full output path was not created")); - const appendResult = await env.appendFile(fullOutputPath, text, context); - return appendResult.ok ? ok(undefined) : err(toExecutionError(appendResult.error)); - }); - }; - - const ensureFullOutputFile = (initialContent: string): void => { - if (fullOutputRequested || captureError) return; - fullOutputRequested = true; - writeChain = writeChain.then(async (previous) => { - if (!previous.ok) return previous; - const tempFile = await env.createTempFile({ prefix: "bash-", suffix: ".log" }, context); - if (!tempFile.ok) return err(toExecutionError(tempFile.error)); - fullOutputPath = tempFile.value; - const appendResult = await env.appendFile(tempFile.value, initialContent, context); - return appendResult.ok ? ok(undefined) : err(toExecutionError(appendResult.error)); - }); - }; - - const createProgress = (): ShellCaptureProgress => { - const tailTruncation = truncateTail(tailOutput); - const totalLines = completedLines + (hasOpenLine ? 1 : 0); - const truncated = totalLines > DEFAULT_MAX_LINES || totalBytes > DEFAULT_MAX_BYTES; - const truncation: TruncationResult = { - ...tailTruncation, - truncated, - truncatedBy: truncated - ? (tailTruncation.truncatedBy ?? (totalBytes > DEFAULT_MAX_BYTES ? "bytes" : "lines")) - : null, - totalLines, - totalBytes, - }; - return { - output: truncated ? truncation.content : tailOutput, - truncation, - fullOutputPath, - lastLineBytes: currentLineBytes, - }; - }; - - const onChunk = (chunk: string): void => { - if (!acceptingOutput) return; - try { - const text = sanitizeBinaryOutput(chunk).replace(/\r/g, ""); - const textBytes = encoder.encode(text).byteLength; - totalBytes += textBytes; - const newlineCount = text.split("\n").length - 1; - completedLines += newlineCount; - const lastNewline = text.lastIndexOf("\n"); - if (lastNewline >= 0) { - const trailingText = text.slice(lastNewline + 1); - currentLineBytes = encoder.encode(trailingText).byteLength; - hasOpenLine = trailingText.length > 0; - } else if (text.length > 0) { - currentLineBytes += textBytes; - hasOpenLine = true; - } - - tailOutput += text; - const totalLines = completedLines + (hasOpenLine ? 1 : 0); - if ((totalBytes > DEFAULT_MAX_BYTES || totalLines > DEFAULT_MAX_LINES) && !fullOutputRequested) { - ensureFullOutputFile(tailOutput); - } else if (fullOutputRequested) { - appendFullOutput(text); - } - tailOutput = trimToLastUtf8Bytes(tailOutput, maxOutputBytes, encoder); - options?.onChunk?.(text, createProgress, context); - } catch (error) { - captureError = toExecutionError(error); - } - }; - - try { - const result = await env.exec( - command, - { - cwd: options?.cwd, - env: options?.env, - inheritEnv: options?.inheritEnv, - timeout: options?.timeout, - onStdout: onChunk, - onStderr: onChunk, + let output: ShellOutputView | undefined; + const result = await env.exec( + command, + { + cwd: options?.cwd, + env: options?.env, + inheritEnv: options?.inheritEnv, + timeout: options?.timeout, + capture: { + limits: { maxBytes: DEFAULT_MAX_BYTES, maxLines: DEFAULT_MAX_LINES, retain: "tail" }, + spill: true, }, - context, - ); - acceptingOutput = false; - let progress = createProgress(); - if (progress.truncation.truncated && !fullOutputRequested) ensureFullOutputFile(tailOutput); - const writeResult = await writeChain; - if (!writeResult.ok) return err(writeResult.error); - if (captureError) return err(captureError); - progress = createProgress(); + onUpdate: (update, updateContext) => { + const previous = output; + output = applyShellOutputUpdate(output, update); + const chunk = + update.kind === "append" || update.kind === "slide" + ? update.text + : update.kind === "replace" && previous === undefined + ? output.text + : undefined; + // A metadata-only update and a post-cap replacement contain no new + // incremental chunk. Reporting their complete view would duplicate bytes + // for callers that accumulate this compatibility callback. + if (chunk) options?.onChunk?.(chunk, () => progressFrom(output!), updateContext); + }, + }, + context, + ); - if (!result.ok) { - if (result.error.code === "aborted" || context.abortSignal?.aborted) { - return ok({ - ...progress, - exitCode: undefined, - cancelled: true, - truncated: progress.truncation.truncated, - }); - } - if (options?.returnExecutionErrors) { - return ok({ - ...progress, - exitCode: undefined, - cancelled: false, - truncated: progress.truncation.truncated, - executionError: result.error, - }); - } - return err(result.error); - } - const cancelled = context.abortSignal?.aborted ?? false; - return ok({ - ...progress, - exitCode: cancelled ? undefined : result.value.exitCode, - cancelled, - truncated: progress.truncation.truncated, - }); - } catch (error) { - acceptingOutput = false; - return err(toExecutionError(error)); + if (output === undefined) { + const { content, ...truncation } = truncateTail(""); + output = { text: content, truncation }; } + const progress = progressFrom(output); + if (!result.ok) { + if (result.error.code === "aborted" || context.abortSignal?.aborted) { + return ok({ ...progress, exitCode: undefined, cancelled: true, truncated: progress.truncation.truncated }); + } + if (options?.returnExecutionErrors) { + return ok({ + ...progress, + exitCode: undefined, + cancelled: false, + truncated: progress.truncation.truncated, + executionError: result.error, + }); + } + return err(result.error); + } + return ok({ + ...progress, + exitCode: result.value.exitCode, + cancelled: false, + truncated: result.value.truncation.truncated, + }); } + +export { sanitizeShellOutput as sanitizeBinaryOutput }; diff --git a/packages/agent/src/harness/utils/truncate.ts b/packages/agent/src/harness/utils/truncate.ts index 8150458cb..07a81ebf3 100644 --- a/packages/agent/src/harness/utils/truncate.ts +++ b/packages/agent/src/harness/utils/truncate.ts @@ -51,7 +51,7 @@ interface RuntimeBuffer { const runtimeBuffer = (globalThis as { Buffer?: RuntimeBuffer }).Buffer; const nonAsciiPattern = /[^\x00-\x7f]/; -function utf8ByteLength(content: string): number { +export function utf8ByteLength(content: string): number { if (runtimeBuffer) return runtimeBuffer.byteLength(content, "utf8"); const firstNonAscii = content.search(nonAsciiPattern); diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index d07613be7..4dcf3e307 100644 --- a/packages/agent/src/index.ts +++ b/packages/agent/src/index.ts @@ -132,9 +132,18 @@ export { type PromptTemplate, type Shell, type ShellExecOptions, + type ShellExecResult, + type ShellOutputCaptureOptions, + type ShellOutputLimits, + type ShellOutputMetadata, + type ShellOutputRetention, + type ShellOutputTruncation, + type ShellOutputUpdate, + type ShellOutputView, type Skill, toError, } from "./harness/types.ts"; +export { applyShellOutputUpdate } from "./harness/utils/output-capture.ts"; export * from "./harness/utils/shell-output.ts"; export * from "./harness/utils/truncate.ts"; export * from "./proxy.ts"; diff --git a/packages/agent/test/harness/adaptive-publisher.test.ts b/packages/agent/test/harness/adaptive-publisher.test.ts new file mode 100644 index 000000000..b6f13efc8 --- /dev/null +++ b/packages/agent/test/harness/adaptive-publisher.test.ts @@ -0,0 +1,76 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { AdaptivePublisher } from "../../src/harness/utils/adaptive-publisher.ts"; + +afterEach(() => { + vi.useRealTimers(); +}); + +describe("AdaptivePublisher", () => { + it("bounds event count and spaces large publications by encoded size", () => { + vi.useFakeTimers(); + vi.setSystemTime(0); + let value = "a"; + const updates: string[] = []; + const publisher = new AdaptivePublisher({ + snapshot: () => value, + update: (_previous, current) => current, + measure: (update) => update.length, + publish: (update) => updates.push(update), + onError: (error) => { + throw error; + }, + minIntervalMs: 100, + targetBytesPerSecond: 100, + }); + + publisher.markDirty(); + value = "x".repeat(100); + publisher.markDirty(); + vi.advanceTimersByTime(100); + expect(updates).toEqual(["a", "x".repeat(100)]); + + value = "held"; + publisher.markDirty(); + vi.advanceTimersByTime(999); + expect(updates).toHaveLength(2); + vi.advanceTimersByTime(1); + expect(updates).toEqual(["a", "x".repeat(100), "held"]); + }); + + it("commits its baseline before a consumer throws", () => { + vi.useFakeTimers(); + vi.setSystemTime(0); + let value = "a"; + const updates: Array<{ previous: string | undefined; current: string }> = []; + let throwAfterApply = false; + const publisher = new AdaptivePublisher({ + snapshot: () => value, + update: (previous, current) => ({ previous, current }), + measure: () => 1, + publish: (update) => { + updates.push(update); + if (throwAfterApply) throw new Error("consumer failed after apply"); + }, + onError: () => {}, + minIntervalMs: 100, + targetBytesPerSecond: 100, + }); + + publisher.markDirty(); + vi.advanceTimersByTime(100); + value = "ab"; + throwAfterApply = true; + expect(() => publisher.markDirty()).toThrow("consumer failed after apply"); + publisher.flush(true); + expect(updates).toEqual([ + { previous: undefined, current: "a" }, + { previous: "a", current: "ab" }, + ]); + + throwAfterApply = false; + vi.advanceTimersByTime(100); + value = "abc"; + publisher.markDirty(); + expect(updates.at(-1)).toEqual({ previous: "ab", current: "abc" }); + }); +}); diff --git a/packages/agent/test/harness/nodejs-env.test.ts b/packages/agent/test/harness/nodejs-env.test.ts index 1833ccc51..f7446f46e 100644 --- a/packages/agent/test/harness/nodejs-env.test.ts +++ b/packages/agent/test/harness/nodejs-env.test.ts @@ -7,7 +7,8 @@ import { pathToFileURL } from "node:url"; import { afterEach, describe, expect, it } from "vitest"; import { BACKGROUND_CONTEXT, withAbortSignal } from "../../src/harness/context.ts"; import { NodeExecutionEnv } from "../../src/harness/env/nodejs.ts"; -import { FileError, getOrThrow } from "../../src/harness/types.ts"; +import { FileError, getOrThrow, type ShellExecOptions, type ShellOutputView } from "../../src/harness/types.ts"; +import { applyShellOutputUpdate } from "../../src/harness/utils/output-capture.ts"; import { executeShellWithCapture } from "../../src/harness/utils/shell-output.ts"; import { createTempDir } from "./session-test-utils.ts"; @@ -32,6 +33,26 @@ function withTimeout(promise: Promise, ms: number, onTimeout?: () => void) }); } +async function collectShellOutput( + env: NodeExecutionEnv, + command: string, + options: ShellExecOptions | undefined, + context: Parameters[2], +): Promise<{ result: Awaited>; output: ShellOutputView | undefined }> { + let output: ShellOutputView | undefined; + const result = await env.exec( + command, + { + ...options, + onUpdate: (update) => { + output = applyShellOutputUpdate(output, update); + }, + }, + context, + ); + return { result, output }; +} + function toBashSingleQuotedArg(value: string): string { return `'${value.replace(/\\/g, "/").replace(/'/g, `'"'"'`)}'`; } @@ -59,6 +80,18 @@ function cleanupDetachedChild(pidFile: string): void { } catch {} } +class FailingSpillExecutionEnv extends NodeExecutionEnv { + override async createTempFile( + options: Parameters[0], + context: Parameters[1], + ) { + if (options?.prefix === "pi-output-") { + return { ok: true as const, value: join(this.cwd, "missing", "spill.log") }; + } + return super.createTempFile(options, context); + } +} + afterEach(async () => { for (const path of chmodRestorePaths.splice(0)) { try { @@ -294,16 +327,15 @@ describe("NodeExecutionEnv", () => { it("executes commands in cwd with env overrides", async () => { const root = createTempDir(); const env = new NodeExecutionEnv({ cwd: root }); - const result = getOrThrow( - await env.exec( - 'printf \'%s:%s\' "$PWD" "$NODE_ENV_TEST"', - { - env: { NODE_ENV_TEST: "ok" }, - }, - BACKGROUND_CONTEXT, - ), + const collected = await collectShellOutput( + env, + 'printf \'%s:%s\' "$PWD" "$NODE_ENV_TEST"', + { env: { NODE_ENV_TEST: "ok" } }, + BACKGROUND_CONTEXT, ); - expect(result).toEqual({ stdout: `${await realpath(root)}:ok`, stderr: "", exitCode: 0 }); + const result = getOrThrow(collected.result); + expect(collected.output?.text).toBe(`${await realpath(root)}:ok`); + expect(result.exitCode).toBe(0); }); it.each([ @@ -326,15 +358,14 @@ describe("NodeExecutionEnv", () => { PI_NODE_ENV_PRESERVED_TEST: "preserved", }, }); - const result = getOrThrow( - await env.exec( - `printf '%s:%s|%s|%s' "\${PI_SESSION_FILE+x}" "\${PI_SESSION_FILE-}" "$PI_CODING_AGENT" "$PI_NODE_ENV_PRESERVED_TEST"`, - { env: overrides }, - BACKGROUND_CONTEXT, - ), + const collected = await collectShellOutput( + env, + `printf '%s:%s|%s|%s' "\${PI_SESSION_FILE+x}" "\${PI_SESSION_FILE-}" "$PI_CODING_AGENT" "$PI_NODE_ENV_PRESERVED_TEST"`, + { env: overrides }, + BACKGROUND_CONTEXT, ); - - expect(result.stdout).toBe(`${expectedSessionFile}|true|preserved`); + getOrThrow(collected.result); + expect(collected.output?.text).toBe(`${expectedSessionFile}|true|preserved`); }, ); @@ -347,18 +378,14 @@ describe("NodeExecutionEnv", () => { process.env[inheritedKey] = "host"; try { const env = new NodeExecutionEnv({ cwd: root, shellEnv: { [configuredKey]: "configured" } }); - const result = getOrThrow( - await env.exec( - `printf '%s:%s:%s' "\${${inheritedKey}-}" "\${${configuredKey}-}" "\${${explicitKey}-}"`, - { - inheritEnv: false, - env: { [explicitKey]: "explicit" }, - }, - BACKGROUND_CONTEXT, - ), + const collected = await collectShellOutput( + env, + `printf '%s:%s:%s' "\${${inheritedKey}-}" "\${${configuredKey}-}" "\${${explicitKey}-}"`, + { inheritEnv: false, env: { [explicitKey]: "explicit" } }, + BACKGROUND_CONTEXT, ); - - expect(result.stdout).toBe("::explicit"); + getOrThrow(collected.result); + expect(collected.output?.text).toBe("::explicit"); } finally { if (previousInherited === undefined) delete process.env[inheritedKey]; else process.env[inheritedKey] = previousInherited; @@ -392,11 +419,16 @@ describe("NodeExecutionEnv", () => { const wslEnv = new NodeExecutionEnv({ cwd: root, shellPath }); const nameExpansion = "$" + "{name}"; - const result = getOrThrow( - await wslEnv.exec(`name='World'; echo "Hello, ${nameExpansion}!"`, undefined, BACKGROUND_CONTEXT), + const collected = await collectShellOutput( + wslEnv, + `name='World'; echo "Hello, ${nameExpansion}!"`, + undefined, + BACKGROUND_CONTEXT, ); - - expect(result).toEqual({ stdout: "Hello, World!\n", stderr: "args:-s\n", exitCode: 0 }); + const result = getOrThrow(collected.result); + expect(collected.output?.text).toContain("Hello, World!"); + expect(collected.output?.text).toContain("args:-s"); + expect(result.exitCode).toBe(0); } finally { process.chdir(originalCwd); process.env.PATH = originalPath; @@ -414,18 +446,18 @@ describe("NodeExecutionEnv", () => { const env = new NodeExecutionEnv({ cwd: root }); const controller = new AbortController(); try { - const result = getOrThrow( - await withTimeout( - env.exec( - createInheritedStdioCommand(pidFile), - undefined, - withAbortSignal(controller.signal, BACKGROUND_CONTEXT), - ), - 3000, - () => controller.abort(), + const collected = await withTimeout( + collectShellOutput( + env, + createInheritedStdioCommand(pidFile), + undefined, + withAbortSignal(controller.signal, BACKGROUND_CONTEXT), ), + 3000, + () => controller.abort(), ); - expect(result.stdout).toContain("child-exiting"); + getOrThrow(collected.result); + expect(collected.output?.text).toContain("child-exiting"); } finally { controller.abort(); cleanupDetachedChild(pidFile); @@ -445,28 +477,27 @@ describe("NodeExecutionEnv", () => { await expect(withTimeout(execution, 3000)).resolves.toMatchObject({ ok: true }); }); - it("streams stdout and stderr chunks", async () => { + it("combines stdout and stderr into one bounded view", async () => { const root = createTempDir(); const env = new NodeExecutionEnv({ cwd: root }); - let stdout = ""; - let stderr = ""; + const updates: string[] = []; + let output: ShellOutputView | undefined; const result = getOrThrow( await env.exec( "printf out; printf err >&2", { - onStdout: (chunk) => { - stdout += chunk; - }, - onStderr: (chunk) => { - stderr += chunk; + onUpdate: (update) => { + updates.push(update.kind); + output = applyShellOutputUpdate(output, update); }, }, BACKGROUND_CONTEXT, ), ); - expect(result).toEqual({ stdout: "out", stderr: "err", exitCode: 0 }); - expect(stdout).toBe("out"); - expect(stderr).toBe("err"); + expect(result.exitCode).toBe(0); + expect(output?.text).toContain("out"); + expect(output?.text).toContain("err"); + expect(updates[0]).toBe("replace"); }); it("reports a missing working directory before spawning", async () => { @@ -484,7 +515,8 @@ describe("NodeExecutionEnv", () => { const root = createTempDir(); const env = new NodeExecutionEnv({ cwd: root }); const result = getOrThrow(await env.exec("exit 7", undefined, BACKGROUND_CONTEXT)); - expect(result).toEqual({ stdout: "", stderr: "", exitCode: 7 }); + expect(result.exitCode).toBe(7); + expect(result.truncation.totalBytes).toBe(0); }); it("returns timeout errors for commands exceeding the timeout", async () => { @@ -501,7 +533,7 @@ describe("NodeExecutionEnv", () => { const result = await env.exec( "printf out", { - onStdout: () => { + onUpdate: () => { throw new Error("callback failed"); }, }, @@ -582,6 +614,75 @@ describe("NodeExecutionEnv", () => { } }); + it("does not create a spill before bounded output crosses its limits", async () => { + const root = createTempDir(); + const env = new NodeExecutionEnv({ cwd: root }); + const result = getOrThrow( + await env.exec( + "printf short", + { + capture: { limits: { maxBytes: 100, maxLines: 10, retain: "tail" }, spill: true }, + onUpdate: () => {}, + }, + BACKGROUND_CONTEXT, + ), + ); + expect(result.spillPath).toBeUndefined(); + }); + + it("preserves exact raw bytes in the spill while decoding a bounded text view", async () => { + const root = createTempDir(); + const env = new NodeExecutionEnv({ cwd: root }); + const expected = [0x66, 0x80, 0x00, 0x6f]; + const result = getOrThrow( + await env.exec( + `${JSON.stringify(process.execPath)} -e "process.stdout.write(Buffer.from([${expected.join(",")}]))"`, + { + capture: { limits: { maxBytes: 1, maxLines: 10, retain: "tail" }, spill: true }, + onUpdate: () => {}, + }, + BACKGROUND_CONTEXT, + ), + ); + expect(result.spillPath).toBeDefined(); + expect([...getOrThrow(await env.readBinaryFile(result.spillPath!, BACKGROUND_CONTEXT))]).toEqual(expected); + }); + + it("fails rather than silently losing a requested spill", async () => { + const root = createTempDir(); + const env = new FailingSpillExecutionEnv({ cwd: root }); + const result = await env.exec( + "printf 12345678901234567890", + { + capture: { limits: { maxBytes: 10, maxLines: 10, retain: "tail" }, spill: true }, + onUpdate: () => {}, + }, + BACKGROUND_CONTEXT, + ); + expect(result).toMatchObject({ + ok: false, + error: { code: "unknown", message: expect.stringContaining("Failed to preserve complete shell output") }, + }); + }); + + it("preserves complete output when spill-stream backpressure pauses a process that exits quickly", async () => { + const root = createTempDir(); + const env = new NodeExecutionEnv({ cwd: root }); + const size = 500_000; + const result = getOrThrow( + await env.exec( + `${JSON.stringify(process.execPath)} -e "process.stdout.write('x'.repeat(${size}))"`, + { + capture: { limits: { maxBytes: 10, maxLines: 10, retain: "tail" }, spill: true }, + onUpdate: () => {}, + }, + BACKGROUND_CONTEXT, + ), + ); + expect(result.spillPath).toBeDefined(); + expect(getOrThrow(await env.readTextFile(result.spillPath!, BACKGROUND_CONTEXT))).toHaveLength(size); + }); + it("captures large shell output to a full output file through the execution env", async () => { const root = createTempDir(); const env = new NodeExecutionEnv({ cwd: root }); diff --git a/packages/agent/test/harness/output-capture.test.ts b/packages/agent/test/harness/output-capture.test.ts new file mode 100644 index 000000000..4a0834cb4 --- /dev/null +++ b/packages/agent/test/harness/output-capture.test.ts @@ -0,0 +1,151 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { BACKGROUND_CONTEXT } from "../../src/harness/context.ts"; +import type { ShellOutputUpdate, ShellOutputView } from "../../src/harness/types.ts"; +import { applyShellOutputUpdate, OutputCapture, sanitizeShellOutput } from "../../src/harness/utils/output-capture.ts"; + +function createCapture(options?: { maxBytes?: number; maxLines?: number; retain?: "head" | "tail" }) { + const updates: ShellOutputUpdate[] = []; + const errors: unknown[] = []; + const capture = new OutputCapture( + { + limits: { + maxBytes: options?.maxBytes ?? 50, + maxLines: options?.maxLines ?? 100, + retain: options?.retain ?? "tail", + }, + }, + BACKGROUND_CONTEXT, + { + onUpdate: (update) => updates.push(update), + onError: (error) => errors.push(error), + }, + ); + return { capture, updates, errors }; +} + +function fold(updates: ShellOutputUpdate[]): ShellOutputView | undefined { + let output: ShellOutputView | undefined; + for (const update of updates) output = applyShellOutputUpdate(output, update); + return output; +} + +afterEach(() => { + vi.useRealTimers(); +}); + +describe("OutputCapture", () => { + it("removes invalid control characters without changing text or line boundaries", () => { + const input = "a\0b\tc\nd\re\u0007f\ufff9g\ufffbh😀"; + expect(sanitizeShellOutput(input)).toBe("ab\tc\ndefgh😀"); + const { capture } = createCapture(); + capture.push(input); + expect(capture.snapshot().text).toBe("ab\tc\ndefgh😀"); + }); + it("decodes UTF-8 split across raw process chunks", () => { + const { capture } = createCapture(); + const bytes = new TextEncoder().encode("😀"); + capture.push(bytes.subarray(0, 2)); + expect(capture.snapshot().text).toBe(""); + capture.push(bytes.subarray(2)); + capture.finish(); + expect(capture.snapshot().text).toBe("😀"); + }); + + it("publishes the first bounded view immediately and trickling appends responsively", () => { + vi.useFakeTimers(); + vi.setSystemTime(0); + const { capture, updates } = createCapture(); + + capture.push("one"); + expect(updates).toHaveLength(1); + expect(updates[0]?.kind).toBe("replace"); + + vi.advanceTimersByTime(150); + capture.push(" two"); + expect(updates).toHaveLength(2); + expect(updates[1]).toMatchObject({ kind: "append", text: " two" }); + expect(fold(updates)?.text).toBe("one two"); + }); + + it("collapses a burst into one trailing update", () => { + vi.useFakeTimers(); + vi.setSystemTime(0); + const { capture, updates } = createCapture(); + capture.push("a"); + capture.push("b"); + capture.push("c"); + expect(updates).toHaveLength(1); + + vi.advanceTimersByTime(100); + expect(updates).toHaveLength(2); + expect(updates[1]).toMatchObject({ kind: "append", text: "bc" }); + expect(fold(updates)?.text).toBe("abc"); + }); + + it("publishes a small slide for post-cap trickle", () => { + vi.useFakeTimers(); + vi.setSystemTime(0); + const { capture, updates } = createCapture({ maxBytes: 10 }); + capture.push("abcdefghij"); + vi.advanceTimersByTime(150); + capture.push("k"); + + expect(updates[1]).toMatchObject({ kind: "slide", drop: 1, text: "k" }); + expect(fold(updates)?.text).toBe("bcdefghijk"); + expect(fold(updates)?.truncation.totalBytes).toBe(11); + }); + + it("keeps the exact byte count for a single line larger than its working buffer", () => { + vi.useFakeTimers(); + const { capture } = createCapture({ maxBytes: 10 }); + capture.push("x".repeat(100)); + expect(capture.snapshot()).toMatchObject({ + text: "x".repeat(10), + lastLineBytes: 100, + truncation: { lastLinePartial: true }, + }); + }); + + it("uses a cap-bounded replacement after complete turnover", () => { + vi.useFakeTimers(); + vi.setSystemTime(0); + const { capture, updates } = createCapture({ maxBytes: 10 }); + capture.push("abcdefghij"); + capture.push("x".repeat(100)); + vi.advanceTimersByTime(100); + + expect(updates[1]?.kind).toBe("replace"); + expect(fold(updates)?.text).toHaveLength(10); + expect(fold(updates)?.truncation.totalBytes).toBe(110); + }); + + it("forces held state and cancels its trailing timer on dispose", () => { + vi.useFakeTimers(); + vi.setSystemTime(0); + const { capture, updates } = createCapture(); + capture.push("a"); + capture.push("b"); + capture.flush(); + expect(fold(updates)?.text).toBe("ab"); + capture.dispose(); + vi.advanceTimersByTime(1_000); + expect(updates).toHaveLength(2); + }); + + it("preserves the original head after its raw guard is crossed", () => { + vi.useFakeTimers(); + const { capture } = createCapture({ maxBytes: 100, maxLines: 2, retain: "head" }); + capture.push(`first\nsecond\n${"tail".repeat(100)}`); + expect(capture.snapshot().text).toBe("first\nsecond"); + }); + + it("publishes spill metadata without resending text", () => { + vi.useFakeTimers(); + const { capture, updates, errors } = createCapture(); + capture.push("output"); + capture.setSpillPath("/tmp/output.log"); + expect(updates.at(-1)).toMatchObject({ kind: "metadata", metadata: { spillPath: "/tmp/output.log" } }); + expect(fold(updates)?.spillPath).toBe("/tmp/output.log"); + expect(errors).toEqual([]); + }); +}); diff --git a/packages/agent/test/harness/tools.test.ts b/packages/agent/test/harness/tools.test.ts index 1c099b95e..7bd73075d 100644 --- a/packages/agent/test/harness/tools.test.ts +++ b/packages/agent/test/harness/tools.test.ts @@ -15,8 +15,9 @@ import { ok, type Result, type ShellExecOptions, + type ShellExecResult, } from "../../src/harness/types.ts"; -import { DEFAULT_MAX_LINES } from "../../src/harness/utils/truncate.ts"; +import { DEFAULT_MAX_BYTES, DEFAULT_MAX_LINES, truncateTail } from "../../src/harness/utils/truncate.ts"; import { createTempDir } from "./session-test-utils.ts"; const noUpdate = () => {}; @@ -94,15 +95,36 @@ class BlockingEditExecutionEnv extends NodeExecutionEnv { } } +function fakeShellOutput(text: string, options: ShellExecOptions | undefined, spillPath?: string): ShellExecResult { + const limits = options?.capture?.limits ?? { + maxBytes: DEFAULT_MAX_BYTES, + maxLines: DEFAULT_MAX_LINES, + retain: "tail" as const, + }; + const truncated = truncateTail(text, limits); + const { content, ...truncation } = truncated; + options?.onUpdate?.( + { + kind: "replace", + output: { text: content, truncation, ...(spillPath === undefined ? {} : { spillPath }) }, + }, + BACKGROUND_CONTEXT, + ); + return { exitCode: 0, truncation, ...(spillPath === undefined ? {} : { spillPath }) }; +} + class LateOutputExecutionEnv extends NodeExecutionEnv { override async exec( _command: string, options: ShellExecOptions | undefined, context: Context, - ): Promise> { - options?.onStdout?.("before\n", context); - setTimeout(() => options?.onStdout?.("late\n", context), 0); - return ok({ stdout: "before\n", stderr: "", exitCode: 0 }); + ): Promise> { + const result = fakeShellOutput("before\n", options); + setTimeout(() => { + const truncation = truncateTail("before\nlate\n", options?.capture?.limits); + options?.onUpdate?.({ kind: "replace", output: { text: truncation.content, truncation } }, context); + }, 0); + return ok(result); } } @@ -119,16 +141,15 @@ class CheckpointOutputExecutionEnv extends NodeExecutionEnv { override async exec( _command: string, options: ShellExecOptions | undefined, - context: Context, - ): Promise> { - options?.onStdout?.("one\n", context); + _context: Context, + ): Promise> { + fakeShellOutput("one\n", options); await delay(2_100); - options?.onStdout?.("two\n", context); + fakeShellOutput("one\ntwo\n", options); await delay(100); - options?.onStdout?.("three\n", context); + fakeShellOutput("one\ntwo\nthree\n", options); await delay(2_000); - options?.onStdout?.("four\n", context); - return ok({ stdout: "one\ntwo\nthree\nfour\n", stderr: "", exitCode: 0 }); + return ok(fakeShellOutput("one\ntwo\nthree\nfour\n", options)); } } @@ -137,9 +158,11 @@ class TimeoutOutputExecutionEnv extends NodeExecutionEnv { _command: string, options: ShellExecOptions | undefined, context: Context, - ): Promise> { + ): Promise> { const output = `${Array.from({ length: TRUNCATED_OUTPUT_LINES }, (_, index) => `line-${index + 1}`).join("\n")}\n`; - options?.onStdout?.(output, context); + const spillPath = getOrThrow(await this.createTempFile({ prefix: "timeout-", suffix: ".log" }, context)); + getOrThrow(await this.writeFile(spillPath, output, context)); + fakeShellOutput(output, options, spillPath); return err(new ExecutionError("timeout", `timeout:${options?.timeout}`)); } }