fix(streams): release consumed completion payloads

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.

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.

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) => {

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`);
+});

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();

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" },
       ],