From 4a5d37291531ad48b8e536c892557877d219463c Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Thu, 15 Oct 2026 09:59:46 +0000 Subject: [PATCH] feat(chord,durable): move OutputBuffer and Progress to @earendil-works/chord/output OutputBuffer, Progress, boundOutput, sanitizeOutput, characterEnd and the output types move from Durable's internal harness/output.ts to the new Chord subpath, with OutputSkip (the shape of Durable's ShellOutputSkip) and utf8ByteLength. Durable imports them from Chord; its truncate.ts re-exports Chord's utf8ByteLength instead of keeping a second copy. --- packages/chord/CHANGELOG.md | 1 + packages/chord/README.md | 1 + packages/chord/package.json | 5 ++ .../src/harness => chord/src}/output.ts | 58 ++++++++++++++++--- .../test/output-skip.test.ts} | 5 +- .../test/output.test.ts} | 2 +- packages/durable/src/env/index.ts | 1 + packages/durable/src/harness/tool.ts | 8 ++- packages/durable/src/tools/read.ts | 2 +- packages/durable/src/truncate.ts | 39 ++----------- packages/durable/test/env-truncate.test.ts | 2 + .../test/tools-read-differential.test.ts | 2 +- tsconfig.json | 1 + 13 files changed, 77 insertions(+), 50 deletions(-) rename packages/{durable/src/harness => chord/src}/output.ts (86%) rename packages/{durable/test/harness-output-skip.test.ts => chord/test/output-skip.test.ts} (96%) rename packages/{durable/test/harness-output.test.ts => chord/test/output.test.ts} (99%) diff --git a/packages/chord/CHANGELOG.md b/packages/chord/CHANGELOG.md index 67de7cae6..ffab7f77e 100644 --- a/packages/chord/CHANGELOG.md +++ b/packages/chord/CHANGELOG.md @@ -24,6 +24,7 @@ ### Added +- `@earendil-works/chord/output`: `OutputBuffer`, which keeps a head or tail window of a stream at a cost proportional to each chunk, `Progress`, which paces commits of a changing value by their size, plus `boundOutput`, `sanitizeOutput`, `characterEnd`, `utf8ByteLength`, and the `OutputLimits`, `BoundedOutput`, `OutputSlice`, and `OutputSkip` types. Moved from `@earendil-works/pi-durable`, which now uses them from here. - Remote references to one object share one binding per host and key (type, normalized input identity, principal key, contract fingerprint): one hold, one stream per state member, calls over the same hold; every reference keeps its own status, observers and `release`. `HostOptions.linger` (`{ types: { [typeName]: ms }, cap }`, cap default 4) keeps a binding held and hydrated after its last reference is released, so a reference taken again meanwhile costs nothing on the link - Holds from clients carry an opaque `origin`: the id the accepting host gave the client connection, forwarded by relays and honored only from the parent link, so a host can tell client connections apart; a disconnect releases that client's holds at once - Flow control by units: the first unit of a new reference passes the window, so a `get` never waits behind other references' state; units for reloads and re-holds wait, built lazily from the current values; waiting ops are capped per link (`LinkOptions.pendingCap`, default four windows), beyond which the oldest streams become resets (a relay asks downstream with `resync`) diff --git a/packages/chord/README.md b/packages/chord/README.md index 7badaec07..6ba689043 100644 --- a/packages/chord/README.md +++ b/packages/chord/README.md @@ -18,6 +18,7 @@ Contracts are runtime values described with [TypeBox](https://github.com/sinclai | `@earendil-works/chord/node` | Child host processes: `createHostManager`, `forkDriver`, `childControl`, `processLock`; the facet module loader `nodeLoader`. | Node | | `@earendil-works/chord/context` | Explicit cancellation and invocation values: `BACKGROUND_CONTEXT`, `withCancel`, `withAbortSignal`, `withoutAbortSignal`, `withContextValue`, `awaitWithContext`. | Node and browsers | | `@earendil-works/chord/delta` | The transactional JSON tracker, structural diff, immutable apply, and the wire encoder/decoder for operation batches. | Node and browsers | +| `@earendil-works/chord/output` | Bounded output streams and paced progress: `OutputBuffer` (head or tail window of a stream, `OutputLimits`, `OutputSkip`), `boundOutput`, `sanitizeOutput`, `utf8ByteLength`, and `Progress` (adaptive commits of a changing value). | Node and browsers | ## Quick start diff --git a/packages/chord/package.json b/packages/chord/package.json index 48a38067f..cadc4bc20 100644 --- a/packages/chord/package.json +++ b/packages/chord/package.json @@ -16,6 +16,11 @@ "types": "./dist/context/index.d.ts", "import": "./dist/context/index.js" }, + "./output": { + "source": "./src/output.ts", + "types": "./dist/output.d.ts", + "import": "./dist/output.js" + }, "./delta": { "source": "./src/delta/index.ts", "types": "./dist/delta/index.d.ts", diff --git a/packages/durable/src/harness/output.ts b/packages/chord/src/output.ts similarity index 86% rename from packages/durable/src/harness/output.ts rename to packages/chord/src/output.ts index de9aa526a..54d1f2c71 100644 --- a/packages/durable/src/harness/output.ts +++ b/packages/chord/src/output.ts @@ -1,7 +1,9 @@ -import type { ShellOutputSkip } from "../env/index.ts"; -import { utf8ByteLength } from "../truncate.ts"; +/** + * Bounded output streams and adaptive progress commits: `OutputBuffer` keeps a head or tail window of a stream at a cost + * proportional to each chunk, and `Progress` paces commits of a changing value by their size. + */ -/** Retention limits of one tool's output. */ +/** Retention limits of one output stream. */ export type OutputLimits = { readonly maxBytes: number; readonly maxLines: number; readonly retain: "head" | "tail" }; /** Retained output and what the limits dropped. */ @@ -15,6 +17,19 @@ export type OutputSlice = { readonly droppedLines: number; }; +/** + * Output a producer omitted before a chunk, measured on the decoded text: every U+FFFD counts as three bytes, and no + * sanitizing is applied. + */ +export interface OutputSkip { + /** UTF-8 byte length of the omitted text. */ + readonly bytes: number; + /** Newlines (U+000A) in the omitted text. */ + readonly newlines: number; + /** Whether the omitted text ends with a newline. */ + readonly endsWithNewline: boolean; +} + const NEWLINE = 0x0a; const INVALID_OUTPUT = /[\x00-\x08\x0b-\x1f\ufff9-\ufffb]/g; const encoder = new TextEncoder(); @@ -105,7 +120,7 @@ function lineCount(bytes: Uint8Array): number { } /** - * Bounded running output of one tool call. Accepting a chunk costs time proportional to the chunk: head retention stops + * Bounded running output of one stream, such as a tool call's or a process's. Accepting a chunk costs time proportional to the chunk: head retention stops * storing once the window is full, and tail retention drops stored text the window no longer needs when it snapshots. * Counts of the whole stream are kept so the dropped totals stay exact. */ @@ -138,10 +153,9 @@ export class OutputBuffer { /** * 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. + * must be more than the tail window by at least one byte or line; only tail retention accepts it. */ - push(chunk: string | Uint8Array, skipped?: ShellOutputSkip): boolean { + push(chunk: string | Uint8Array, skipped?: OutputSkip): boolean { // Bytes of an incomplete character from an earlier byte chunk come first. const pending = typeof chunk === "string" || skipped !== undefined ? this.#decoder.decode() : ""; let text = typeof chunk === "string" ? chunk : this.#decoder.decode(chunk, { stream: true }); @@ -157,7 +171,7 @@ export class OutputBuffer { } /** Count omitted output; nothing stored before it can be in the window once the text after it arrives. */ - #skip(skipped: ShellOutputSkip): void { + #skip(skipped: OutputSkip): void { if (skipped.bytes === 0) return; this.#totalBytes += skipped.bytes; this.#totalNewlines += skipped.newlines; @@ -251,6 +265,34 @@ function lines(newlines: number, terminated: boolean): number { return newlines + (terminated ? 0 : 1); } +interface RuntimeBuffer { + byteLength(content: string, encoding: "utf8"): number; +} + +const runtimeBuffer = (globalThis as { Buffer?: RuntimeBuffer }).Buffer; +const nonAsciiPattern = /[^\x00-\x7f]/; + +/** UTF-8 byte length of `content`; a lone surrogate counts as U+FFFD, three bytes. */ +export function utf8ByteLength(content: string): number { + if (runtimeBuffer) return runtimeBuffer.byteLength(content, "utf8"); + const firstNonAscii = content.search(nonAsciiPattern); + if (firstNonAscii === -1) return content.length; + let bytes = firstNonAscii; + for (let i = firstNonAscii; i < content.length; i++) { + const code = content.charCodeAt(i); + if (code <= 0x7f) bytes += 1; + else if (code <= 0x7ff) bytes += 2; + else if (code >= 0xd800 && code <= 0xdbff && i + 1 < content.length) { + const next = content.charCodeAt(i + 1); + if (next >= 0xdc00 && next <= 0xdfff) { + bytes += 4; + i++; + } else bytes += 3; + } else bytes += 3; + } + return bytes; +} + function countNewlines(text: string): number { let count = 0; for (let index = text.indexOf("\n"); index !== -1; index = text.indexOf("\n", index + 1)) count++; diff --git a/packages/durable/test/harness-output-skip.test.ts b/packages/chord/test/output-skip.test.ts similarity index 96% rename from packages/durable/test/harness-output-skip.test.ts rename to packages/chord/test/output-skip.test.ts index ef87504b3..43d2edf1d 100644 --- a/packages/durable/test/harness-output-skip.test.ts +++ b/packages/chord/test/output-skip.test.ts @@ -1,6 +1,5 @@ import { describe, expect, it } from "vitest"; -import type { ShellOutputSkip } from "../src/env/index.ts"; -import { OutputBuffer, type OutputLimits } from "../src/harness/output.ts"; +import { OutputBuffer, type OutputLimits, type OutputSkip } from "../src/output.ts"; /** Deterministic PRNG (mulberry32) so failures reproduce from the printed seed. */ function random(seed: number): () => number { @@ -25,7 +24,7 @@ function randomChunk(next: () => number): string { return text; } -function measure(text: string): ShellOutputSkip { +function measure(text: string): OutputSkip { let newlines = 0; for (const character of text) if (character === "\n") newlines++; return { bytes: Buffer.byteLength(text, "utf8"), newlines, endsWithNewline: text.endsWith("\n") }; diff --git a/packages/durable/test/harness-output.test.ts b/packages/chord/test/output.test.ts similarity index 99% rename from packages/durable/test/harness-output.test.ts rename to packages/chord/test/output.test.ts index a64b456ac..3109164fd 100644 --- a/packages/durable/test/harness-output.test.ts +++ b/packages/chord/test/output.test.ts @@ -1,5 +1,5 @@ import { afterEach, describe, expect, it, vi } from "vitest"; -import { boundOutput, OutputBuffer, type OutputLimits, Progress, sanitizeOutput } from "../src/harness/output.ts"; +import { boundOutput, OutputBuffer, type OutputLimits, Progress, sanitizeOutput } from "../src/output.ts"; const head = (maxLines: number, maxBytes = 1000): OutputLimits => ({ maxBytes, maxLines, retain: "head" }); const tail = (maxLines: number, maxBytes = 1000): OutputLimits => ({ maxBytes, maxLines, retain: "tail" }); diff --git a/packages/durable/src/env/index.ts b/packages/durable/src/env/index.ts index 89437d3c2..0d8da86f7 100644 --- a/packages/durable/src/env/index.ts +++ b/packages/durable/src/env/index.ts @@ -334,6 +334,7 @@ export interface ShellOutputWindow { * 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. */ +/** The shape of `OutputSkip` from `@earendil-works/chord/output`, which `OutputBuffer.push()` accepts. */ export interface ShellOutputSkip { /** UTF-8 byte length of the omitted text. */ bytes: number; diff --git a/packages/durable/src/harness/tool.ts b/packages/durable/src/harness/tool.ts index ca779139e..50e13203f 100644 --- a/packages/durable/src/harness/tool.ts +++ b/packages/durable/src/harness/tool.ts @@ -1,6 +1,13 @@ import { type Context, copyJson, type Draft, type JsonRepresentation, type JsonValue } from "@earendil-works/chord"; import { awaitWithContext } from "@earendil-works/chord/context"; import { overlap } from "@earendil-works/chord/delta"; +import { + boundOutput, + OutputBuffer, + type OutputLimits, + PROGRESS_BYTES_PER_SECOND, + Progress, +} from "@earendil-works/chord/output"; import type { ImageContent, TextContent, ToolCall, ToolResultMessage } from "@earendil-works/pi-ai"; import { validateToolArguments } from "@earendil-works/pi-ai/utils/validation"; import { Compile } from "typebox/compile"; @@ -32,7 +39,6 @@ import { type ToolSlot, toolSlot, } from "./live.ts"; -import { boundOutput, OutputBuffer, type OutputLimits, PROGRESS_BYTES_PER_SECOND, Progress } from "./output.ts"; import type { NestedToolExecutionResult, ToolControl, diff --git a/packages/durable/src/tools/read.ts b/packages/durable/src/tools/read.ts index bea5c812c..cede96ef3 100644 --- a/packages/durable/src/tools/read.ts +++ b/packages/durable/src/tools/read.ts @@ -1,4 +1,5 @@ import type { Context } from "@earendil-works/chord"; +import { characterEnd } from "@earendil-works/chord/output"; import { type Static, Type } from "typebox"; import { type BinaryReader, @@ -9,7 +10,6 @@ import { startsWithBom, } from "../env/index.ts"; import { defineTool } from "../harness/define.ts"; -import { characterEnd } from "../harness/output.ts"; import type { ToolDiagnostic, ToolExecutionApi, ToolExecutionResult, ToolRegistration } from "../harness/types.ts"; import { DEFAULT_MAX_BYTES, diff --git a/packages/durable/src/truncate.ts b/packages/durable/src/truncate.ts index ed2041f78..54d9e84ce 100644 --- a/packages/durable/src/truncate.ts +++ b/packages/durable/src/truncate.ts @@ -1,3 +1,4 @@ +import { utf8ByteLength } from "@earendil-works/chord/output"; /** * Shared truncation utilities for tool outputs. * @@ -5,7 +6,7 @@ * - Line limit (default: 2000 lines) * - Byte limit (default: 50KB) * - * Never returns partial lines. Tool output streams are bounded by `src/harness/output.ts` instead. + * Never returns partial lines. Tool output streams are bounded by `OutputBuffer` from `@earendil-works/chord/output` instead. */ export const DEFAULT_MAX_LINES = 2000; @@ -43,40 +44,8 @@ export interface TruncationOptions { maxBytes?: number; } -interface RuntimeBuffer { - byteLength(content: string, encoding: "utf8"): number; -} - -const runtimeBuffer = (globalThis as { Buffer?: RuntimeBuffer }).Buffer; -const nonAsciiPattern = /[^\x00-\x7f]/; - -export function utf8ByteLength(content: string): number { - if (runtimeBuffer) return runtimeBuffer.byteLength(content, "utf8"); - - const firstNonAscii = content.search(nonAsciiPattern); - if (firstNonAscii === -1) return content.length; - - let bytes = firstNonAscii; - for (let i = firstNonAscii; i < content.length; i++) { - const code = content.charCodeAt(i); - if (code <= 0x7f) { - bytes += 1; - } else if (code <= 0x7ff) { - bytes += 2; - } else if (code >= 0xd800 && code <= 0xdbff && i + 1 < content.length) { - const next = content.charCodeAt(i + 1); - if (next >= 0xdc00 && next <= 0xdfff) { - bytes += 4; - i++; - } else { - bytes += 3; - } - } else { - bytes += 3; - } - } - return bytes; -} +// One implementation, shared with the output buffer. +export { utf8ByteLength }; function splitLinesForCounting(content: string): string[] { if (content.length === 0) return []; diff --git a/packages/durable/test/env-truncate.test.ts b/packages/durable/test/env-truncate.test.ts index 7a2f3f5ef..93665252f 100644 --- a/packages/durable/test/env-truncate.test.ts +++ b/packages/durable/test/env-truncate.test.ts @@ -25,6 +25,8 @@ describe("truncate utilities", () => { const output = execFileSync( process.execPath, [ + // Workspace packages from source, as the test runner resolves them, not from a possibly stale build. + "--conditions=source", "--import", new URL("./fixtures/delete-buffer.ts", import.meta.url).href, fileURLToPath(new URL("./fixtures/utf8-byte-length-without-buffer.ts", import.meta.url)), diff --git a/packages/durable/test/tools-read-differential.test.ts b/packages/durable/test/tools-read-differential.test.ts index f105a171e..e8486be10 100644 --- a/packages/durable/test/tools-read-differential.test.ts +++ b/packages/durable/test/tools-read-differential.test.ts @@ -2,9 +2,9 @@ import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; +import { characterEnd } from "@earendil-works/chord/output"; import { afterAll, describe, expect, it } from "vitest"; import { NodeExecutionEnv } from "../src/env/node.ts"; -import { characterEnd } from "../src/harness/output.ts"; import type { ToolDiagnostic, ToolExecutionApi } from "../src/harness/types.ts"; import { detectSupportedImageMimeType } from "../src/tools/image.ts"; import { createReadTool } from "../src/tools/read.ts"; diff --git a/tsconfig.json b/tsconfig.json index 14f1d161d..39b7cfd3b 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -9,6 +9,7 @@ "@earendil-works/chord/context": ["./packages/chord/src/context/index.ts"], "@earendil-works/chord/delta": ["./packages/chord/src/delta/index.ts"], "@earendil-works/chord/node": ["./packages/chord/src/node.ts"], + "@earendil-works/chord/output": ["./packages/chord/src/output.ts"], "@earendil-works/pi-telemetry": ["./packages/telemetry/src/index.ts"], "@earendil-works/pi-telemetry/testing": ["./packages/telemetry/src/testing/index.ts"], "@earendil-works/pi-codemode": ["./packages/codemode/src/index.ts"],