diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index d049e0eae..b6ff3992f 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -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 diff --git a/packages/durable/docs/spec.md b/packages/durable/docs/spec.md index d76dc09a9..362460b2c 100644 --- a/packages/durable/docs/spec.md +++ b/packages/durable/docs/spec.md @@ -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 diff --git a/packages/durable/src/env/index.ts b/packages/durable/src/env/index.ts index a9047c038..a410fa354 100644 --- a/packages/durable/src/env/index.ts +++ b/packages/durable/src/env/index.ts @@ -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 { diff --git a/packages/durable/src/harness/output.ts b/packages/durable/src/harness/output.ts index 6cf797544..429c5c931 100644 --- a/packages/durable/src/harness/output.ts +++ b/packages/durable/src/harness/output.ts @@ -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 diff --git a/packages/durable/src/harness/tool.ts b/packages/durable/src/harness/tool.ts index 0d6311f47..4c84ff062 100644 --- a/packages/durable/src/harness/tool.ts +++ b/packages/durable/src/harness/tool.ts @@ -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); diff --git a/packages/durable/src/harness/types.ts b/packages/durable/src/harness/types.ts index 9875f0918..941a3daa4 100644 --- a/packages/durable/src/harness/types.ts +++ b/packages/durable/src/harness/types.ts @@ -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 extend agent(context: Context): Promise; /** 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`. */ diff --git a/packages/durable/src/testing/env-conformance.ts b/packages/durable/src/testing/env-conformance.ts index ddee8ba2b..29be58738 100644 --- a/packages/durable/src/testing/env-conformance.ts +++ b/packages/durable/src/testing/env-conformance.ts @@ -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; @@ -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"); diff --git a/packages/durable/src/tools/bash.ts b/packages/durable/src/tools/bash.ts index b38a5f4fe..838213e4b 100644 --- a/packages/durable/src/tools/bash.ts +++ b/packages/durable/src/tools/bash.ts @@ -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 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, ); diff --git a/packages/durable/test/harness-output-skip.test.ts b/packages/durable/test/harness-output-skip.test.ts new file mode 100644 index 000000000..ef87504b3 --- /dev/null +++ b/packages/durable/test/harness-output-skip.test.ts @@ -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", + ); + }); +}); diff --git a/packages/durable/test/harness-tools.test.ts b/packages/durable/test/harness-tools.test.ts index d5b021e8f..a2a4c9226 100644 --- a/packages/durable/test/harness-tools.test.ts +++ b/packages/durable/test/harness-tools.test.ts @@ -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|\n[warn] Output truncated to its end: 3 lines, 18 bytes dropped\n", + ); + 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; diff --git a/packages/durable/test/tools.test.ts b/packages/durable/test/tools.test.ts index 48268d1f9..bdf72ea63 100644 --- a/packages/durable/test/tools.test.ts +++ b/packages/durable/test/tools.test.ts @@ -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> { + 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");