sscoble/pi-mono
1
0
Fork 0
mirror of https://github.com/badlogic/pi-mono.git synced 2026-11-05 23:37:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

fix(agent): bound shell execution output

Browse source
This commit is contained in:
Mario Zechner 2026-09-01 18:12:06 +02:00
parent 5009d0608c
commit 0095bce7db
16 changed files with 1297 additions and 994 deletions
Download patch file Download diff file Expand all files Collapse all files

445
packages/agent/docs/mobile-handoff/01-harness/03-execenv/execenv.md
View file

@ -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<Result<{ stdout: string; stderr: string; exitCode: number }, ExecutionError>>
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<TruncationResult, "content">;
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<string, string>;
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<Result<ShellExecResult, ExecutionError>>;
cleanup(context: Context): Promise<void>;
}
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.

28
packages/agent/docs/mobile-handoff/01-harness/04-tool-output/bash-worked-example.md
View file

@ -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<ExecutionToolContext, typeof bashSchema, BashToolDetails> {
return {
@ -23,21 +23,24 @@ export function createBashTool(): AgentHarnessTool<ExecutionToolContext, typeof
async execute(_id, { command, timeout }, signal, out, context) {
const env = context.env;
let view: ShellOutputView | undefined;
const result = getOrThrow(await env.exec(command, {
cwd: env.cwd,
inheritEnv: true,
timeout,
capture: { limits: this.output, intervalMs: 500, spill: true },
capture: { limits: this.output, spill: true },
onUpdate: (u) => {
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<WireOp[]>`, 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<WireOp[]>`, 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

33
packages/agent/docs/mobile-handoff/01-harness/04-tool-output/harness-tools.md
View file

@ -32,24 +32,10 @@ interface ToolOutput<TDetails extends JsonValue> {
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

285
packages/agent/docs/mobile-handoff/01-harness/04-tool-output/rate-limiting.md
View file

@ -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.

200
packages/agent/src/harness/env/nodejs.ts vendored
View file

@ -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<number | undefined, ExecutionError> {
if (timeout === undefined) return ok(undefined);
@ -282,7 +287,7 @@ function killProcessTree(pid: number): void {
}
}
function waitForChildProcess(child: ChildProcess): Promise<number | null> {
function waitForChildProcess(child: ChildProcess, spillIsDraining: () => boolean): Promise<number | null> {
return new Promise((resolvePromise, reject) => {
let settled = false;
let exited = false;
@ -314,7 +319,10 @@ function waitForChildProcess(child: ChildProcess): Promise<number | null> {
};
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<Result<{ stdout: string; stderr: string; exitCode: number }, ExecutionError>> {
): Promise<Result<ShellExecResult, ExecutionError>> {
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<typeof spawn> | undefined;
let timeoutId: ReturnType<typeof setTimeout> | undefined;
const spillPrefix: SpillChunk[] = [];
let spillPath: string | undefined;
const spillQueue: SpillChunk[] = [];
let spillStart: Promise<void> | 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<ShellExecResult, ExecutionError>) => {
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<void> => {
await spillStart;
const stream = spillStream;
if (stream === undefined || spillError !== undefined || stream.destroyed) return;
await new Promise<void>((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))),
);

173
packages/agent/src/harness/tools/bash.ts
View file

@ -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<typeof bashSchema>;
export interface BashToolDetails {
truncation?: TruncationResult;
truncation?: ShellOutputTruncation;
fullOutputPath?: string;
}
@ -56,7 +54,7 @@ export function createBashTool<TContext extends ExecutionToolContext = Execution
return {
name: "bash",
label: "bash",
description: `Execute a bash command in the current working directory. Returns stdout and stderr. Output is truncated to last ${DEFAULT_MAX_LINES} lines or ${DEFAULT_MAX_BYTES / 1024}KB (whichever is hit first). If truncated, full output is saved to a temp file. Optionally provide a timeout in seconds.`,
description: `Execute a bash command in the current working directory. Returns combined stdout and stderr. Output is truncated to last ${DEFAULT_MAX_LINES} lines or ${DEFAULT_MAX_BYTES / 1024}KB (whichever is hit first). If truncated, full output is saved to a temp file. Optionally provide a timeout in seconds.`,
parameters: bashSchema,
async execute(_toolCallId, { command, timeout }, onUpdate, toolContext, _invocation, context) {
validateTimeout(timeout);
@ -68,109 +66,80 @@ export function createBashTool<TContext extends ExecutionToolContext = Execution
inheritEnv: true,
};
await options?.prepare?.(execution, toolContext, context);
let getLatestProgress: (() => ShellCaptureProgress) | undefined;
let updateTimer: ReturnType<typeof setTimeout> | 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 };
},
};
}

56
packages/agent/src/harness/types.ts
View file

@ -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<TValue, TError> = { ok: true; value: TValue } | { ok: false; error: TError };
@ -313,6 +314,51 @@ export interface FileSystem {
cleanup(context: Context): Promise<void>;
}
/** 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<TruncationResult, "content">;
/** 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<Result<{ stdout: string; stderr: string; exitCode: number }, ExecutionError>>;
): Promise<Result<ShellExecResult, ExecutionError>>;
/** Release shell resources. Must be best-effort and must not throw or reject. */
cleanup(context: Context): Promise<void>;
}

87
packages/agent/src/harness/utils/adaptive-publisher.ts Normal file
View file

@ -0,0 +1,87 @@
export interface AdaptivePublisherOptions<TValue, TUpdate> {
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<TValue, TUpdate> {
readonly #options: AdaptivePublisherOptions<TValue, TUpdate>;
readonly #minIntervalMs: number;
readonly #targetBytesPerSecond: number;
#published: TValue | undefined;
#dirty = false;
#nextEmitAt = 0;
#timer: ReturnType<typeof setTimeout> | undefined;
#disposed = false;
constructor(options: AdaptivePublisherOptions<TValue, TUpdate>) {
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);
}
}

238
packages/agent/src/harness/utils/output-capture.ts Normal file
View file

@ -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<ShellOutputView, ShellOutputUpdate>;
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));
}

244
packages/agent/src/harness/utils/shell-output.ts
View file

@ -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<ShellExecOptions, "onStdout" | "onStderr"> {
export interface ShellCaptureOptions extends Omit<ShellExecOptions, "capture" | "onUpdate"> {
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<Result<ShellCaptureResult, ExecutionError>> {
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<Result<void, ExecutionError>> = 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 };

2
packages/agent/src/harness/utils/truncate.ts
View file

@ -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);

9
packages/agent/src/index.ts
View file

@ -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";

76
packages/agent/test/harness/adaptive-publisher.test.ts Normal file
View file

@ -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" });
});
});

213
packages/agent/test/harness/nodejs-env.test.ts
View file

@ -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<T>(promise: Promise<T>, ms: number, onTimeout?: () => void)
});
}
async function collectShellOutput(
env: NodeExecutionEnv,
command: string,
options: ShellExecOptions | undefined,
context: Parameters<NodeExecutionEnv["exec"]>[2],
): Promise<{ result: Awaited<ReturnType<NodeExecutionEnv["exec"]>>; 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<NodeExecutionEnv["createTempFile"]>[0],
context: Parameters<NodeExecutionEnv["createTempFile"]>[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 });

151
packages/agent/test/harness/output-capture.test.ts Normal file
View file

@ -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([]);
});
});

51
packages/agent/test/harness/tools.test.ts
View file

@ -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<Result<{ stdout: string; stderr: string; exitCode: number }, ExecutionError>> {
options?.onStdout?.("before\n", context);
setTimeout(() => options?.onStdout?.("late\n", context), 0);
return ok({ stdout: "before\n", stderr: "", exitCode: 0 });
): Promise<Result<ShellExecResult, ExecutionError>> {
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<Result<{ stdout: string; stderr: string; exitCode: number }, ExecutionError>> {
options?.onStdout?.("one\n", context);
_context: Context,
): Promise<Result<ShellExecResult, ExecutionError>> {
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<Result<{ stdout: string; stderr: string; exitCode: number }, ExecutionError>> {
): Promise<Result<ShellExecResult, ExecutionError>> {
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}`));
}
}