feat(durable): windowed shell output with counted skips

- ShellExecOptions.window and ShellOutputInfo.skipped: environments may omit
  output outside the retained tail and report exact counts
- ToolExecutionApi.outputWindow (tail limits and progress pace) and
  output(chunk, skipped); the bash tool passes both through
- OutputBuffer counts skips; property tests against the unskipped stream
- fix: progress snapshots no longer change which tail is kept
This commit is contained in:
Mario Zechner 2026-10-04 22:43:16 +02:00
parent acfc60198e
commit cdf79797b3
11 changed files with 391 additions and 17 deletions

View file

@ -14,10 +14,12 @@
- Argv form of `exec()`, which runs a program without a shell.
- `createEnvConformance()` and `registerEnvConformance()` in `@earendil-works/pi-durable/testing` for checking custom `ExecutionEnv` implementations.
- `settings.progress` with `partialIntervalMs` and `outputIntervalMs` configures how often generation partials and running tool output are committed; defaults stay 100 ms ([#10357](https://github.com/earendil-works/pi/issues/10357))
- `ShellExecOptions.window` and `ShellOutputInfo.skipped`: an environment may omit shell output outside the caller's retained tail and report the omission, so remote environments need not transfer output the caller drops. `ToolExecutionApi.outputWindow` provides the window and `output(chunk, skipped)` counts omissions; the `bash` tool passes them through.
### Fixed
- `NodeExecutionEnv.flushFile()` on a directory fails with `is_directory` on Windows, as on POSIX.
- Tail-retained tool output no longer depends on when progress commits happened: a progress snapshot compacted the stored output to the kept window, which could move where a later window's first line started.
## [1.0.2] - 2026-10-04

View file

@ -2820,6 +2820,23 @@ sanitized, and throttled. The `bash` tool pipes those chunks into `output()`,
reports the spill path as a diagnostic, and throws on a nonzero exit or timeout;
the error result still carries the retained output and diagnostics.
An environment that moves output over a slow link, such as one on another host,
need not move all of it. `api.outputWindow` names the tail a tail-retaining
call keeps and the pace of its progress commits; the `bash` tool passes it as
`ShellExecOptions.window`. The environment may then omit output and report
the omission as `info.skipped` on the chunk that follows: the decoded byte
count, the newline count, and whether the omitted text ended with a newline.
It may omit only output followed by more than the window, by at least one byte
or one line, and delivers all of that following output in the same chunk, so
the omitted text can never be part of the kept tail and no progress commit sees
a gap. `output(chunk, skipped)` adds the omission to the dropped counts, so
the retained tail, the dropped counts, and the `truncated` diagnostic are the
same as if every byte had arrived; only the moments at which progress is
sampled differ. The environment should deliver no faster than the pace, since
progress commits sample no faster. A head-retaining call has no window. A
wrapper that replaces `output` to transform text sets `outputWindow` to
`undefined`, so no omitted text bypasses its transform.
`exec` takes a string or an argv array. A string runs through the
environment's shell. An array runs its first element directly with the rest as
arguments, without a shell, so a host that builds a command from data, such as

View file

@ -194,10 +194,51 @@ export interface ShellExecOptions {
*/
onOutput?: (text: string, context: Context, info: ShellOutputInfo) => void;
spill?: ShellSpillOptions;
/**
* The caller keeps only this tail of the output, so the environment may omit output outside it and report the omission
* as `info.skipped`. Without it, every chunk is delivered.
*/
window?: ShellOutputWindow;
}
/**
* The tail of the combined output a caller keeps, and how often it samples it. An environment that transfers output
* over a slow link uses it to omit what the caller would drop anyway and to send no faster than the caller commits.
*/
export interface ShellOutputWindow {
/** UTF-8 bytes of decoded text kept at the end of the output. */
maxBytes: number;
/** Lines kept at the end of the output. */
maxLines: number;
/** Minimum pause between the caller's samples of the output. */
minIntervalMs: number;
/** Each sample also pauses the caller in proportion to its size at this rate. */
bytesPerSecond: number;
}
/**
* Output an environment omitted, measured on the decoded text `onOutput` would have received: every U+FFFD counts as
* three bytes, and no sanitizing is applied.
*/
export interface ShellOutputSkip {
/** UTF-8 byte length of the omitted text. */
bytes: number;
/** Newlines (U+000A) in the omitted text. */
newlines: number;
/** Whether the omitted text ends with a newline. */
endsWithNewline: boolean;
}
export interface ShellOutputInfo {
stream: "stdout" | "stderr";
/**
* Output omitted immediately before this chunk, only with `window`. The chunk then holds all output after the
* omission up to its end, and that is more than the window by at least one byte or one line: more than
* `window.maxBytes` bytes or more than `window.maxLines` newlines. So the omitted text can never be in the kept tail.
* The omission and such a chunk may span both streams in arrival order; `stream` then names the chunk's last stream.
* Callers that need the streams apart do not pass `window`.
*/
skipped?: ShellOutputSkip;
}
export interface Shell {

View file

@ -1,3 +1,4 @@
import type { ShellOutputSkip } from "../env/index.ts";
import { utf8ByteLength } from "../truncate.ts";
/** Retention limits of one tool's output. */
@ -128,12 +129,32 @@ export class OutputBuffer {
return this.#storedBytes;
}
/** Accept a chunk; returns whether anything was accepted. */
push(chunk: string | Uint8Array): boolean {
/**
* Accept a chunk; returns whether anything was accepted. `skipped` is output omitted right before the chunk, which
* must be more than the tail window by at least one byte or line (`ShellOutputInfo.skipped`); only tail retention
* accepts it.
*/
push(chunk: string | Uint8Array, skipped?: ShellOutputSkip): boolean {
// Bytes of an incomplete character from an earlier byte chunk come first.
const text =
typeof chunk === "string" ? this.#decoder.decode() + chunk : this.#decoder.decode(chunk, { stream: true });
return this.#accept(text);
const pending = typeof chunk === "string" || skipped !== undefined ? this.#decoder.decode() : "";
const text = typeof chunk === "string" ? chunk : this.#decoder.decode(chunk, { stream: true });
if (skipped === undefined) return this.#accept(pending + text);
if (this.#limits.retain !== "tail") throw new Error("Skipped output requires tail retention");
this.#accept(pending);
this.#skip(skipped);
this.#accept(text);
return true;
}
/** Count omitted output; nothing stored before it can be in the window once the text after it arrives. */
#skip(skipped: ShellOutputSkip): void {
if (skipped.bytes === 0) return;
this.#totalBytes += skipped.bytes;
this.#totalNewlines += skipped.newlines;
this.#endsWithNewline = skipped.endsWithNewline;
this.#chunks = [];
this.#storedBytes = 0;
this.#storedNewlines = 0;
}
/** Flush an incomplete trailing character as a replacement character; call when the stream ends. */
@ -178,10 +199,11 @@ export class OutputBuffer {
const kept = boundOutput(stored, this.#limits);
const storedLines = lines(this.#storedNewlines, stored === "" || stored.endsWith("\n"));
const keptLines = storedLines - kept.droppedLines;
// Tail windows never reach back before this one, so only the kept slice needs storing.
// Tail windows never reach back before this one, but finding a later window's first line needs what precedes it:
// keep the shortest suffix longer than the window by a byte or a line, as `#accept` does.
if (this.#limits.retain === "tail" || this.#chunks.length > 1) {
const text = this.#limits.retain === "tail" ? kept.text : stored;
const bytes = this.#limits.retain === "tail" ? kept.bytes : this.#storedBytes;
const text = this.#limits.retain === "tail" ? tailMargin(stored, this.#limits) : stored;
const bytes = this.#limits.retain === "tail" ? utf8ByteLength(text) : this.#storedBytes;
this.#chunks = text === "" ? [] : [{ text, bytes, newlines: countNewlines(text) }];
this.#storedBytes = bytes;
this.#storedNewlines = this.#chunks[0]?.newlines ?? 0;
@ -194,6 +216,26 @@ export class OutputBuffer {
}
}
/**
* The shortest suffix of `text` with more than `maxBytes` bytes or more than `maxLines` newlines, or all of it. The
* tail window of any text that ends with this suffix, followed by anything, is the same as of `text` followed by it.
*/
function tailMargin(text: string, limits: OutputLimits): string {
const bytes = encoder.encode(text);
const byteStart = bytes.length > limits.maxBytes ? characterEnd(bytes, bytes.length - limits.maxBytes - 1) : 0;
let lineStart = 0;
let newlines = 0;
for (let index = bytes.lastIndexOf(NEWLINE); index !== -1; index = bytes.lastIndexOf(NEWLINE, index - 1)) {
if (++newlines > limits.maxLines) {
lineStart = index;
break;
}
if (index === 0) break;
}
const start = Math.max(byteStart, lineStart);
return start === 0 ? text : decoder.decode(bytes.subarray(start));
}
/** Lines of text with `newlines` newlines; a final unterminated line counts. */
function lines(newlines: number, terminated: boolean): number {
return newlines + (terminated ? 0 : 1);
@ -206,7 +248,7 @@ function countNewlines(text: string): number {
}
/** Each progress commit also buys a pause proportional to what it wrote. */
const PROGRESS_BYTES_PER_SECOND = 100 * 1024;
export const PROGRESS_BYTES_PER_SECOND = 100 * 1024;
/**
* Adaptive progress commits: the first change after an idle period commits at once; each commit then delays the next

View file

@ -19,7 +19,7 @@ import type {
} from "../types.ts";
import { assignJson } from "./json.ts";
import { clearProgress, finishSlot, LiveDoc, type ToolSlot, toolSlot } from "./live.ts";
import { boundOutput, OutputBuffer, type OutputLimits, Progress } from "./output.ts";
import { boundOutput, OutputBuffer, type OutputLimits, PROGRESS_BYTES_PER_SECOND, Progress } from "./output.ts";
import type {
ToolControl,
ToolDiagnostic,
@ -190,10 +190,19 @@ async function run(
callId: call.id,
registry: runtime.registry,
agent: runtime.agent,
output: (chunk) => {
output: (chunk, skipped) => {
assertLive();
if (reported.output.push(chunk)) progress.mark();
if (reported.output.push(chunk, skipped)) progress.mark();
},
outputWindow:
limits.retain === "tail"
? {
maxBytes: limits.maxBytes,
maxLines: limits.maxLines,
minIntervalMs: runtime.settings.progress.outputIntervalMs,
bytesPerSecond: PROGRESS_BYTES_PER_SECOND,
}
: undefined,
diagnostic: (diagnostic) => {
assertLive();
reported.diagnostics.push(copyJson(diagnostic, { omitUndefinedProperties: true }) as ToolDiagnostic);

View file

@ -14,7 +14,7 @@ import type {
Usage,
UserMessage,
} from "@earendil-works/pi-ai";
import type { ExecutionEnv } from "../env/index.ts";
import type { ExecutionEnv, ShellOutputSkip, ShellOutputWindow } from "../env/index.ts";
import type {
ConversationId,
ConversationOwnership,
@ -173,8 +173,17 @@ export interface ToolExecutionApi<TDetails extends JsonValue = JsonValue> extend
agent(context: Context): Promise<Agent>;
/** Built by `HarnessOptions.env` for this call; `undefined` without an environment. */
readonly env: ExecutionEnv | undefined;
/** Append running output; it becomes the result content when the result omits `content`. */
output(chunk: string | Uint8Array): void;
/**
* Append running output; it becomes the result content when the result omits `content`. `skipped` counts output
* omitted before the chunk, as reported by an environment given `outputWindow` (`ShellOutputInfo.skipped`).
*/
output(chunk: string | Uint8Array, skipped?: ShellOutputSkip): void;
/**
* The tail this call's output keeps and the pace of its progress commits, for `ShellExecOptions.window`; `undefined`
* when the tool keeps the head of its output, which cannot accept skips. A wrapper that replaces `output` and
* transforms text must also replace this with `undefined`, so skipped text cannot bypass its transform.
*/
readonly outputWindow: ShellOutputWindow | undefined;
/** Record a model-visible remark about this call. */
diagnostic(diagnostic: ToolDiagnostic): void;
/** Replace running details; the last value becomes the result details when the result omits `details`. */

View file

@ -5,6 +5,7 @@ import type { EnvConformanceCase, EnvConformanceOptions } from "./types.ts";
const context = BACKGROUND_CONTEXT;
const decoder = new TextDecoder();
const encoder = new TextEncoder();
type EnvTest = (env: ExecutionEnv) => Promise<void>;
@ -199,6 +200,42 @@ export function createEnvConformance(options: EnvConformanceOptions): readonly E
assert.strictEqual(errorCode(await env.exec([], undefined, context)), "spawn_error");
}),
createCase("windowed exec keeps the exact tail and counts what it skips", async (env) => {
const lines = 2000;
const window = { maxBytes: 200, maxLines: 5, minIntervalMs: 0, bytesPerSecond: 1_000_000_000 };
let bytes = 0;
let newlines = 0;
let tail = "";
const result = await env.exec(
[...shell, `i=0; while [ $i -lt ${lines} ]; do echo line-$i; i=$((i+1)); done`],
{
window,
onOutput: (text, _context, info) => {
if (info.skipped !== undefined) {
bytes += info.skipped.bytes;
newlines += info.skipped.newlines;
const after = encoder.encode(text).length;
const afterNewlines = text.split("\n").length - 1;
assert.ok(
after > window.maxBytes || afterNewlines > window.maxLines,
"a skip is followed by more than the window",
);
tail = "";
}
bytes += encoder.encode(text).length;
newlines += text.split("\n").length - 1;
tail += text;
},
},
context,
);
assert.strictEqual(getOrThrow(result).exitCode, 0);
const expected = Array.from({ length: lines }, (_, index) => `line-${index}\n`);
assert.strictEqual(bytes, expected.join("").length);
assert.strictEqual(newlines, lines);
assert.ok(tail.endsWith(expected.slice(-window.maxLines).join("")), "the delivered output ends with the tail");
}),
createCase("argv exec distinguishes timeout from abort", async (env) => {
const timedOut = await env.exec([...shell, "sleep 2"], { timeout: 0.1 }, context);
assert.strictEqual(errorCode(timedOut), "timeout");

View file

@ -40,7 +40,8 @@ function validateTimeout(timeout: number | undefined): void {
/**
* Runs a command through the environment's shell. Its output streams to `api.output()`, where the Harness keeps the
* tail within the default limits; the result content is that retained output. Output beyond the limits is spilled to a
* tail within the default limits; the result content is that retained output. The retained window goes to the
* environment, which may omit output outside it and report how much it omitted, so dropped counts stay exact. Output beyond the limits is spilled to a
* file whose path is reported as a diagnostic. A nonzero exit or timeout throws, which makes an error result that still
* carries the output and diagnostics.
*/
@ -68,8 +69,10 @@ export function createBashTool(options?: BashToolOptions): ToolRegistration<type
env: execution.env,
inheritEnv: execution.inheritEnv,
...(timeout === undefined ? {} : { timeout }),
onOutput: (text) => api.output(text),
onOutput: (text, _context, info) => api.output(text, info.skipped),
spill: { afterBytes: DEFAULT_MAX_BYTES, afterLines: DEFAULT_MAX_LINES },
// An environment may then omit output outside the retained tail and report the omission.
...(api.outputWindow === undefined ? {} : { window: api.outputWindow }),
},
context,
);

View file

@ -0,0 +1,144 @@
import { describe, expect, it } from "vitest";
import type { ShellOutputSkip } from "../src/env/index.ts";
import { OutputBuffer, type OutputLimits } from "../src/harness/output.ts";
/** Deterministic PRNG (mulberry32) so failures reproduce from the printed seed. */
function random(seed: number): () => number {
let state = seed >>> 0;
return () => {
state = (state + 0x6d2b79f5) >>> 0;
let t = state;
t = Math.imul(t ^ (t >>> 15), t | 1);
t ^= t + Math.imul(t ^ (t >>> 7), t | 61);
return ((t ^ (t >>> 14)) >>> 0) / 4294967296;
};
}
// Decoded shell output: ASCII, multi-byte and astral characters, replacement characters, CR/LF, tabs, and control
// characters that sanitizing removes after bounding.
const ALPHABET = ["a", "b", "z", " ", "\n", "\n", "\n", "\r\n", "\t", "é", "€", "😀", "\ufffd", "\x01", "\x1b"];
function randomChunk(next: () => number): string {
const length = Math.floor(next() * 12);
let text = "";
for (let index = 0; index < length; index++) text += ALPHABET[Math.floor(next() * ALPHABET.length)];
return text;
}
function measure(text: string): ShellOutputSkip {
let newlines = 0;
for (const character of text) if (character === "\n") newlines++;
return { bytes: Buffer.byteLength(text, "utf8"), newlines, endsWithNewline: text.endsWith("\n") };
}
/** Whether `text` proves that everything before it is outside the tail window. */
function exceedsWindow(text: string, limits: OutputLimits): boolean {
const { bytes, newlines } = measure(text);
return bytes > limits.maxBytes || newlines > limits.maxLines;
}
/** Code-point boundaries of `text`, so a cut never splits a surrogate pair. */
function boundaries(text: string): number[] {
const result = [0];
let offset = 0;
for (const character of text) {
offset += character.length;
result.push(offset);
}
return result;
}
/**
* Feed random output to one buffer in full and to another through a reference skipper that follows the contract of
* `ShellOutputInfo.skipped`: it holds undelivered text, and when it flushes, it may replace a prefix of it by counts if
* the rest exceeds the window. Snapshots must agree whenever both buffers have seen the same output.
*/
function checkSeed(seed: number): void {
const next = random(seed);
const limits: OutputLimits = {
maxBytes: 1 + Math.floor(next() * 40),
maxLines: 1 + Math.floor(next() * 5),
retain: "tail",
};
const full = new OutputBuffer(limits);
const skipping = new OutputBuffer(limits);
let pending = "";
let skips = 0;
const flush = (): void => {
if (pending === "") return;
const cuts = boundaries(pending).filter((cut) => cut > 0 && exceedsWindow(pending.slice(cut), limits));
if (cuts.length > 0 && next() < 0.7) {
const cut = cuts[Math.floor(next() * cuts.length)]!;
skipping.push(pending.slice(cut), measure(pending.slice(0, cut)));
skips++;
} else {
skipping.push(pending);
}
pending = "";
// Progress snapshots happen between deliveries and compact stored chunks.
if (next() < 0.5) skipping.snapshot();
expect(skipping.snapshot(), `seed ${seed}`).toEqual(full.snapshot());
};
const chunks = Math.floor(next() * 40);
for (let index = 0; index < chunks; index++) {
const chunk = randomChunk(next);
full.push(chunk);
if (next() < 0.3) full.snapshot();
pending += chunk;
if (next() < 0.3) flush();
}
flush();
full.end();
skipping.end();
expect(skipping.snapshot(), `seed ${seed}`).toEqual(full.snapshot());
expect(skips >= 0).toBe(true);
}
describe("OutputBuffer skipped output", () => {
// Progress commits snapshot at arbitrary moments; snapshots compact what is stored and must never change the window.
// A compaction to exactly the kept window lost the character that decides where a later window's first line starts.
it("keeps the same tail whenever progress snapshots happen", () => {
for (let seed = 1; seed <= 3000; seed++) {
const next = random(seed);
const limits: OutputLimits = {
maxBytes: 1 + Math.floor(next() * 40),
maxLines: 1 + Math.floor(next() * 5),
retain: "tail",
};
const plain = new OutputBuffer(limits);
const sampled = new OutputBuffer(limits);
const chunks = Math.floor(next() * 30);
for (let index = 0; index < chunks; index++) {
const chunk = randomChunk(next);
plain.push(chunk);
sampled.push(chunk);
if (next() < 0.4) sampled.snapshot();
}
expect(sampled.snapshot(), `seed ${seed}`).toEqual(plain.snapshot());
}
});
it("matches the full stream for every legal skip pattern", () => {
for (let seed = 1; seed <= 3000; seed++) checkSeed(seed);
});
it("counts skipped bytes, lines and the final newline exactly", () => {
const limits: OutputLimits = { maxBytes: 1000, maxLines: 2, retain: "tail" };
const full = new OutputBuffer(limits);
const skipping = new OutputBuffer(limits);
// The skipped text ends without a newline, so its last line continues in the delivered text.
const omitted = "one\ntwo\nthr";
const kept = "ee\nfour\nfive\nsix";
full.push(omitted + kept);
skipping.push(kept, measure(omitted));
expect(skipping.snapshot()).toEqual(full.snapshot());
expect(skipping.snapshot()).toEqual({ text: "five\nsix", droppedBytes: 19, droppedLines: 4 });
});
it("refuses skips for head retention", () => {
const buffer = new OutputBuffer({ maxBytes: 10, maxLines: 2, retain: "head" });
expect(() => buffer.push("x\ny\nz\n", { bytes: 3, newlines: 1, endsWithNewline: true })).toThrow(
"tail retention",
);
});
});

View file

@ -325,6 +325,48 @@ describe("tool results", () => {
await harness.close(context);
});
it("offers the tail window with the configured pace, not a head window, and accepts skipped output", async () => {
const setup = { ...chatSetup(), settings: { progress: { outputIntervalMs: 250 } } };
const windows: unknown[] = [];
addTool(
setup.registry,
tool(
"tailed",
async (_args, api) => {
windows.push(api.outputWindow);
api.output("dropped\n");
api.output("x\ny\n", { bytes: 8, newlines: 1, endsWithNewline: true });
return {};
},
{ outputLimits: { maxLines: 1, retain: "tail" } },
),
);
addTool(
setup.registry,
tool(
"headed",
async (_args, api) => {
windows.push(api.outputWindow);
return {};
},
{ outputLimits: { maxLines: 1 } },
),
);
const { harness, entries } = await run(setup, [calls(["tailed", {}, "c1"], ["headed", {}, "c2"]), DONE]);
expect(windows).toContainEqual({
maxBytes: 50 * 1024,
maxLines: 1,
minIntervalMs: 250,
bytesPerSecond: 100 * 1024,
});
expect(windows).toContainEqual(undefined);
// 8 bytes written, 8 skipped, then "x\n" dropped by the window: 3 lines, 18 bytes in all.
expect(resultText(results(entries)[0]!)).toBe(
"y\n|<harness>\n[warn] Output truncated to its end: 3 lines, 18 bytes dropped\n</harness>",
);
await harness.close(context);
});
it("bounds explicit text content and keeps other content", async () => {
const setup = chatSetup();
const image = { type: "image", data: "AAAA", mimeType: "image/png" } as const;

View file

@ -23,6 +23,8 @@ import {
type Result,
type ShellExecOptions,
type ShellExecResult,
type ShellOutputSkip,
type ShellOutputWindow,
} from "../src/env/index.ts";
import { NodeExecutionEnv } from "../src/env/node.ts";
import { withFileMutationQueue } from "../src/tools/file-mutation-queue.ts";
@ -518,6 +520,32 @@ describe("durable tools", () => {
});
describe("bash", () => {
it("passes the retained window to the environment and forwards what it skipped", async () => {
const window: ShellOutputWindow = { maxBytes: 4, maxLines: 1, minIntervalMs: 100, bytesPerSecond: 1024 };
const skipped: ShellOutputSkip = { bytes: 6, newlines: 2, endsWithNewline: true };
let received: ShellOutputWindow | undefined;
class SkippingEnv extends NodeExecutionEnv {
override async exec(
_command: string | readonly string[],
options: ShellExecOptions | undefined,
context: Context,
): Promise<Result<ShellExecResult, ExecutionError>> {
received = options?.window;
options?.onOutput?.("tail\n", context, { stream: "stdout", skipped });
return { ok: true, value: { exitCode: 0 } };
}
}
const calls: [string, ShellOutputSkip | undefined][] = [];
const api = {
...fakeApi(new SkippingEnv({ cwd: createTempDir() })).api,
outputWindow: window,
output: (chunk: string | Uint8Array, skip?: ShellOutputSkip) => calls.push([String(chunk), skip]),
} as ToolExecutionApi;
await createBashTool().execute({ command: "anything" }, api, BACKGROUND_CONTEXT);
expect(received).toEqual(window);
expect(calls).toEqual([["tail\n", skipped]]);
});
it("streams combined stdout and stderr and returns no content of its own", async () => {
const result = await run(createBashTool(), { command: "printf out; printf err >&2" }, createEnv());
expect(result.output.join("")).toContain("out");