fix(ai,chord,durable,coding-agent): parse streamed tool-call arguments incrementally (fixes #9265)

- ai: StreamingJsonParser / ToolCallArgumentsStream read each argument delta once (tolerant,
  immutable snapshots, linear cost); every provider, pi-messages and the faux provider use it
  and keep stream state (partialJson, partialArgs, index, customInput, streamIndex, Bedrock
  reasoning bytes) off the content blocks; grammar tool input appends per delta
- agent: streamProxy uses the same parser
- chord: Delta append ops own their text (ownedSlice), so ops and replicas no longer keep the
  whole growing string alive
- durable, ai, mcp: bounded error excerpts and summaries own their characters
- coding-agent: tool calls render streamed arguments once per frame, not once per delta
This commit is contained in:
Mario Zechner 2026-10-24 08:00:07 +00:00
parent 136b87e8a7
commit f715a15c10
47 changed files with 3187 additions and 455 deletions

View file

@ -17,6 +17,12 @@
- Document, conversation view, task graph, and event watches use Chord's `Watch` implementation; behaviour is unchanged.
### Fixed
- Fixed a conversation watcher (`watchEvents()`, conversation views) holding a copy of a streamed text or tool call per partial commit, through Chord append ops that kept the whole string alive; a 140 KB answer followed by a 140 KB tool call held well over 100 MB ([#9265](https://github.com/earendil-works/pi/issues/9265))
- Fixed a nested call's error summary keeping the call's whole output alive through a slice of it ([#9265](https://github.com/earendil-works/pi/issues/9265))
- Fixed task span error messages and codemode call previews keeping the whole text they were cut from alive ([#9265](https://github.com/earendil-works/pi/issues/9265))
## [1.4.0] - 2026-10-21
### Breaking Changes

View file

@ -35,6 +35,7 @@
* call with an error result.
*/
import { type Context, copyJson, type JsonValue } from "@earendil-works/chord";
import { ownedSlice } from "@earendil-works/chord/delta";
import type { TextContent } from "@earendil-works/pi-ai";
import {
type CodemodeHostCall,
@ -198,7 +199,8 @@ const ARGS_PREVIEW_CHARS = 200;
const ERROR_PREVIEW_CHARS = 500;
function cut(text: string, max: number): string {
return text.length > max ? `${text.slice(0, max - 3)}...` : text;
// A copy: the row outlives the arguments or error text it previews.
return text.length > max ? `${ownedSlice(text, 0, max - 3)}...` : text;
}
/** A function scripts call as a top-level global; its result is journaled. */

View file

@ -1,5 +1,6 @@
import { type Context, copyJson, type JsonValue } from "@earendil-works/chord";
import { awaitWithContext, withAbortSignal, withoutAbortSignal, withTelemetry } from "@earendil-works/chord/context";
import { ownedSlice } from "@earendil-works/chord/delta";
import type { Models } from "@earendil-works/pi-ai";
import {
createTypedMetrics,
@ -1801,7 +1802,7 @@ export class TaskScheduler {
const message = terminal.status === "orphaned" ? terminal.reason : terminal.error.message;
span.setStatus({
status: "error",
error: { name: `task_${terminal.status}`, message: message.slice(0, 500) },
error: { name: `task_${terminal.status}`, message: ownedSlice(message, 0, 500) },
});
} else if (outcome === "aborted") {
span.setStatus({ status: "ok" });

View file

@ -1,6 +1,6 @@
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 { overlap, ownedSlice } from "@earendil-works/chord/delta";
import {
boundOutput,
OutputBuffer,
@ -864,9 +864,12 @@ function truncated(
/** Error text a nested call's summary keeps. */
const MAX_SUMMARY_ERROR_CHARS = 500;
/** How a nested call ended, for its slot: error text, from its diagnostics or else its `output`, bounded. */
/**
* How a nested call ended, for its slot: error text, from its diagnostics or else its `output`, bounded. The bound is a
* copy, so the slot never keeps a large output alive through a slice of it.
*/
function summaryOf(result: NestedToolExecutionResult, output: Content): NestedToolSummary {
const error = result.isError ? errorTextOf(result, output).slice(0, MAX_SUMMARY_ERROR_CHARS) : "";
const error = result.isError ? ownedSlice(errorTextOf(result, output), 0, MAX_SUMMARY_ERROR_CHARS) : "";
return {
isError: result.isError,
...(result.durationMs === undefined ? {} : { durationMs: result.durationMs }),

View file

@ -1,3 +1,5 @@
import { getHeapStatistics, setFlagsFromString } from "node:v8";
import { runInNewContext } from "node:vm";
import type { Draft } from "@earendil-works/chord";
import {
type AssistantMessage,
@ -620,3 +622,90 @@ describe("agent events", () => {
await harness.close(context);
});
});
const gc = (() => {
setFlagsFromString("--expose-gc");
return runInNewContext("gc") as () => void;
})();
/** Heap bytes in use after a full collection. */
async function heapInUse(): Promise<number> {
for (let round = 0; round < 3; round++) {
await drained();
gc();
}
return getHeapStatistics().used_heap_size;
}
// #9265
describe("streamed partials", () => {
it("streams a tool-call partial as toolcall_delta appends at its argument paths", async () => {
const setup = chatSetup({ tokensPerSecond: 500, tokenSize: { min: 1, max: 1 } });
setup.settings.progress = { partialIntervalMs: 20, outputIntervalMs: 100 };
const args = { path: "src/file.ts", content: "const a = 1;\n".repeat(150) };
setup.faux.setResponses([
fauxAssistantMessage([fauxToolCall("missing", args, { id: "c1" })], { stopReason: "toolUse" }),
fauxAssistantMessage([fauxText("done")]),
]);
const { harness, root } = await openChat(new MemoryStorage(), setup);
const partials = partialsOf(harness);
const { stream, events } = await listen(harness, root);
await (await root.submit({ type: "input", content: "go" }, context)).wait(context);
await drained();
const changes = events().flatMap((event) => (event.type === "message_update" ? event.changes : []));
const deltas = changes.filter((change) => change.type === "toolcall_delta");
expect(deltas.length).toBeGreaterThan(10);
expect(deltas.every((change) => ["path", "content"].includes(String(change.path)))).toBe(true);
// The call is resent whole only when a member appears (at most once per argument); growing strings are appends.
// With the raw argument text on the block, every commit resent the whole call.
expect(changes.filter((change) => change.type === "message")).toEqual([]);
expect(changes.filter((change) => change.type === "block").length).toBeLessThanOrEqual(2);
// The committed partials carry only the parsed arguments, never the raw argument text.
for (const partial of partials) {
for (const block of partial.content) {
expect(Object.keys(block).sort()).toEqual(["arguments", "id", "name", "type"]);
}
}
await stream.stop();
await harness.close(context);
});
it("keeps a watcher's memory proportional to a long streamed text and tool call, not to their commits", async () => {
const setup = chatSetup({ tokensPerSecond: 16_000, tokenSize: { min: 16, max: 16 } });
setup.settings.progress = { partialIntervalMs: 1, outputIntervalMs: 100 };
const line = (index: number) => `line ${index}: some streamed source text with "quotes" and \\ escapes\n`;
const text = Array.from({ length: 2_000 }, (_, index) => line(index)).join("");
const content = Array.from({ length: 2_000 }, (_, index) => line(index)).join("");
setup.faux.setResponses([
fauxAssistantMessage([fauxText(text), fauxToolCall("missing", { path: "a.ts", content }, { id: "c1" })], {
stopReason: "toolUse",
}),
fauxAssistantMessage([fauxText("done")]),
]);
const { harness, root } = await openChat(new MemoryStorage(), setup);
const { stream, batches } = await listen(harness, root);
const before = await heapInUse();
const submission = await root.submit({ type: "input", content: "go" }, context);
const partialContent = async () =>
(await harness.snapshot(LiveDoc, root.id, context))?.generation?.message?.content;
const shown = async (index: number): Promise<number> => {
const block = (await partialContent())?.[index];
if (block?.type === "text") return block.text.length;
return block?.type === "toolCall" ? String(block.arguments.content ?? "").length : 0;
};
const retained: number[] = [];
// Measure while the text, then the tool call, is almost complete: the watcher's view holds the most then.
await waitFor(async () => (await shown(0)) > text.length * 0.95, 30_000);
retained.push((await heapInUse()) - before);
await waitFor(async () => (await shown(1)) > content.length * 0.95, 30_000);
retained.push((await heapInUse()) - before);
await submission.wait(context);
const commits = batches.length;
expect(commits).toBeGreaterThan(500);
// The two 140 KB strings, the watcher's recorded batches and the rest of the stream: about 3 MB. Sliced append ops
// made the watcher's view hold a copy of the text per commit: well over 100 MB.
for (const bytes of retained) expect(bytes).toBeLessThan(20_000_000);
await stream.stop();
await harness.close(context);
}, 60_000);
});