diff --git a/API.md b/API.md index b83de5e..d0e043c 100644 --- a/API.md +++ b/API.md @@ -310,6 +310,10 @@ Choose exactly one factory: `adapter` for a `defineAdapter` factory, or `integration` for a built-in SDK factory. Both return the same lifecycle update shape. The helper owns the integration's pool and finalizes only pending calls, without replaying SDK completion history, even when a call ID is reused. +Built-in integrations release their completion payload history after each +managed batch; the current batch and any values retained by the consumer remain +live. Call IDs remain recorded for duplicate suppression. Manual SDK streams +still retain all completed calls for `finish()`. Cleanup and backpressure rules above apply to both paths. Custom integration factories must use the supplied pool and return completion history with newly finalized calls appended, matching the built-in contract. diff --git a/CHANGELOG.md b/CHANGELOG.md index 3a1cb1c..de6017b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased +- Release consumed completion payloads in managed SDK readers while preserving + manual completion history and duplicate suppression. - Preserve Vercel AI SDK schema defaults and transforms in completed tool inputs, including `fullStream` calls without argument deltas, with replacement patches when the final input differs from streamed arguments. diff --git a/packages/core/src/internal/integration.js b/packages/core/src/internal/integration.js index bca1799..e6b3b1a 100644 --- a/packages/core/src/internal/integration.js +++ b/packages/core/src/internal/integration.js @@ -5,6 +5,13 @@ import { } from "./errors.js"; import { emitDiagnostic } from "./diagnostics.js"; +const completionHistories = new WeakMap(); + +export const clearCompletionHistory = (stream) => { + const history = completionHistories.get(stream); + if (history !== undefined) history.length = 0; +}; + export const createIntegration = ( pool, accept, @@ -104,7 +111,7 @@ export const createIntegration = ( } }; - return { + const stream = { push: (event) => push(event, false), pushAll: (event) => push(event, true), finish() { @@ -126,6 +133,8 @@ export const createIntegration = ( } }, }; + completionHistories.set(stream, completed); + return stream; }; export const append = (operations, id, delta) => { diff --git a/packages/core/src/managed-memory.test.js b/packages/core/src/managed-memory.test.js new file mode 100644 index 0000000..692bd59 --- /dev/null +++ b/packages/core/src/managed-memory.test.js @@ -0,0 +1,60 @@ +import assert from "node:assert/strict"; +import { execFileSync } from "node:child_process"; +import test from "node:test"; + +test("managed SDK readers release consumed payloads while manual streams retain history", () => { + const output = execFileSync(process.execPath, ["--expose-gc", "--input-type=module", "-e", ` + import assert from "node:assert/strict"; + import { setImmediate } from "node:timers/promises"; + import { defineAdapter, readStructured } from ${JSON.stringify(new URL("./index.js", import.meta.url).href)}; + import { vercelAI } from ${JSON.stringify(new URL("./vercel-ai.js", import.meta.url).href)}; + + const calls = 200; + function* events(custom = false) { + for (let id = 0; id < calls; id++) { + const text = JSON.stringify({ id, data: ("payload-" + id).repeat(3000) }); + if (custom) { + yield [{ type: "start", id }, { type: "delta", id, text }, { type: "end", id }]; + } else { + yield { type: "tool-input-start", id: String(id) }; + yield { type: "tool-input-delta", id: String(id), delta: text }; + yield { type: "tool-input-end", id: String(id) }; + } + } + } + async function consume(custom) { + const reader = readStructured(events(custom), { + ...(custom ? { adapter: defineAdapter(event => event) } : { integration: vercelAI }), + limits: { maxActiveStreams: 1 }, + }); + const references = []; + while (references.length < calls) { + const { value } = await reader.next(); + if (value.type === "complete") { + assert.equal(value.value.id, references.length); + references.push(new WeakRef(value.value)); + } + } + return { reader, references }; + } + const sdk = await consume(false); + const custom = await consume(true); + for (let attempt = 0; attempt < 5; attempt++) { + await setImmediate(); + global.gc(); + } + const retained = references => references.filter(ref => ref.deref() !== undefined).length; + const result = { sdk: retained(sdk.references), custom: retained(custom.references) }; + await sdk.reader.return(); + await custom.reader.return(); + + const manual = vercelAI(); + for (const event of events()) manual.pushAll(event); + assert.deepEqual(manual.finish().map(({ value }) => value.id), Array.from({ length: calls }, (_, id) => id)); + assert.equal(manual.finish().length, calls); + console.log(JSON.stringify(result)); + `], { encoding: "utf8", timeout: 20_000 }); + const retained = JSON.parse(output); + assert.ok(retained.custom <= 1, `custom reader retained ${retained.custom} payloads`); + assert.ok(retained.sdk <= 1, `SDK reader retained ${retained.sdk} payloads`); +}); diff --git a/packages/core/src/read-structured.js b/packages/core/src/read-structured.js index 964b4a0..3b5eeb9 100644 --- a/packages/core/src/read-structured.js +++ b/packages/core/src/read-structured.js @@ -1,6 +1,7 @@ import { createStructuredStreamPool } from "./index.js"; import { streamError } from "./internal/errors.js"; import { abortableSource } from "./internal/abortable-source.js"; +import { clearCompletionHistory } from "./internal/integration.js"; export async function* readStructured( events, @@ -21,6 +22,7 @@ export async function* readStructured( let source; let failed = false; const dispose = () => { + clearCompletionHistory(stream); if (pool === undefined) stream?.dispose(); else for (const id of pool.activeIds) pool.abort(id); }; @@ -40,7 +42,9 @@ export async function* readStructured( source = abortableSource(events, signal, dispose); for await (const event of source) { signal?.throwIfAborted(); - for (const update of stream.pushAll(event)) { + const updates = stream.pushAll(event); + clearCompletionHistory(stream); + for (const update of updates) { signal?.throwIfAborted(); yield update; } @@ -55,6 +59,7 @@ export async function* readStructured( // SDK finish() appends pending calls to its legacy completion history. const pending = pool.size; const completed = stream.finish(); + clearCompletionHistory(stream); if (pending > 0) { for (const result of completed.slice(-pending)) { signal?.throwIfAborted(); diff --git a/packages/core/src/read-structured.test.js b/packages/core/src/read-structured.test.js index 354a883..d87a3c0 100644 --- a/packages/core/src/read-structured.test.js +++ b/packages/core/src/read-structured.test.js @@ -230,6 +230,7 @@ test("SDK completion history does not duplicate reused IDs at EOF", async () => { type: "tool-input-start", id: "same" }, { type: "tool-input-delta", id: "same", delta: "1" }, { type: "tool-input-end", id: "same" }, + { type: "tool-input-end", id: "same" }, { type: "tool-input-start", id: "same" }, { type: "tool-input-delta", id: "same", delta: "2" }, ],