diff --git a/API.md b/API.md index 528f10e..c36c612 100644 --- a/API.md +++ b/API.md @@ -372,6 +372,11 @@ If an event changes several calls and then fails, `pushAll` throws without returning a partial batch. It does not roll back previously returned live views. Use a separate pool per response/session. +The OpenAI Responses adapter uses the final `arguments` from +`response.function_call_arguments.done`, including when earlier deltas are +missing or incomplete. Repeated completions for an inactive call ID are ignored; +an explicit new start allows that ID to be reused. + The Vercel adapter accepts both `streamText(...).fullStream` and `toUIMessageStream()`. Within SDK step boundaries (`start-step` / `finish-step`), `tool-input-end` ends the deltas; the subsequent `tool-call` supplies the final diff --git a/CHANGELOG.md b/CHANGELOG.md index 33c1a76..4403758 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,9 @@ ## Unreleased +- Use OpenAI final arguments when deltas are missing or incomplete and suppress + repeated completion events while preserving explicit call ID reuse. + ## 0.1.15 - Add a Mistral Responses example that normalizes SDK frames and reuses the diff --git a/packages/core/src/openai.js b/packages/core/src/openai.js index 0919538..f5e3208 100644 --- a/packages/core/src/openai.js +++ b/packages/core/src/openai.js @@ -13,9 +13,7 @@ export const openAI = (pool = createStructuredStreamPool(), options) => return append(calls, event.item_id, event.delta); } if (event.type === "response.function_call_arguments.done") { - if (!calls.has(event.item_id)) - calls.start(event.item_id, event.arguments); - return calls.complete(event.item_id); + return calls.complete(event.item_id, () => event.arguments); } return undefined; }, diff --git a/packages/core/src/openai.test.js b/packages/core/src/openai.test.js new file mode 100644 index 0000000..e3e8442 --- /dev/null +++ b/packages/core/src/openai.test.js @@ -0,0 +1,112 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { createStructuredStreamPool, readStructured } from "./index.js"; +import { openAI } from "./openai.js"; + +const start = (id) => ({ + type: "response.output_item.added", + item: { type: "function_call", id }, +}); +const delta = (id, text) => ({ + type: "response.function_call_arguments.delta", item_id: id, delta: text, +}); +const done = (id, text) => ({ + type: "response.function_call_arguments.done", item_id: id, arguments: text, +}); + +test("OpenAI completion uses final arguments with missing, partial, or different deltas", () => { + const text = '{"city":"Paris"}'; + for (const snapshots of ["live", "immutable"]) { + for (const prefix of [undefined, "", '{"city":"Pa', '{"old":true}', text]) { + const pool = createStructuredStreamPool({ snapshots }); + const stream = openAI(pool); + let previous; + if (prefix !== undefined) { + stream.pushAll(start("a")); + previous = stream.pushAll(delta("a", prefix))[0].partialValue; + } + const saved = structuredClone(previous); + const updates = stream.pushAll(done("a", text)); + assert.deepEqual(updates.map(({ type }) => type), + prefix === undefined ? ["start", "complete"] : ["complete"]); + const result = updates.at(-1); + assert.deepEqual(result.value, { city: "Paris" }); + assert.deepEqual(result.partialValue, result.value); + assert.equal(result.text, text); + assert.equal(result.bytesSeen, Buffer.byteLength(text)); + assert.equal(result.complete, true); + assert.deepEqual(previous, saved); + if (prefix === text) { + assert.equal(result.partialValue, previous); + assert.deepEqual(result.changes, []); + } else if (prefix !== undefined) { + assert.deepEqual(result.changes[0], { op: "set", path: [], value: {} }); + } + if (snapshots === "immutable") assert.ok(Object.isFrozen(result.value)); + assert.equal(pool.size, 0); + assert.deepEqual(stream.finish().map(({ value }) => value), [{ city: "Paris" }]); + } + } +}); + +test("OpenAI ignores duplicate completions while preserving history and explicit ID reuse", () => { + for (const snapshots of ["live", "immutable"]) { + for (const method of ["push", "pushAll"]) { + const stream = openAI(createStructuredStreamPool({ snapshots })); + stream[method](done("a", "1")); + assert.deepEqual(stream[method](done("a", "{]")), method === "push" ? undefined : []); + assert.deepEqual(stream.finish().map(({ value }) => value), [1]); + stream[method](start("a")); + stream[method](delta("a", "2")); + stream[method](done("a", "2")); + assert.deepEqual(stream[method](done("a", "2")), method === "push" ? undefined : []); + assert.deepEqual(stream.finish().map(({ value }) => value), [1, 2]); + assert.deepEqual(stream.finish().map(({ value }) => value), [1, 2]); + } + } +}); + +test("invalid OpenAI final arguments remain terminal and release all active calls", () => { + for (const snapshots of ["live", "immutable"]) { + for (const [limits, text, code] of [ + [{}, "{]", "MISMATCHED_CLOSING"], + [{ maxBytes: 4 }, '"long"', "MAX_BYTES_EXCEEDED"], + [{ maxDepth: 1 }, "[[]]", "MAX_DEPTH_EXCEEDED"], + ]) { + const pool = createStructuredStreamPool({ snapshots, ...limits }); + const stream = openAI(pool); + stream.pushAll(start("a")); + stream.pushAll(delta("a", "{}")); + stream.pushAll(start("b")); + let failure; + assert.throws(() => stream.pushAll(done("a", text)), (error) => { + failure = error; + assert.equal(error.code, code); + assert.equal(error.id, "a"); + assert.equal(error.operation, "finish"); + assert.equal(error.adapter, "openai"); + assert.equal(error.eventType, "response.function_call_arguments.done"); + return true; + }); + assert.equal(pool.size, 0); + assert.throws(() => stream.finish(), (error) => error === failure); + } + } +}); + +test("managed OpenAI completion emits final arguments once and finalizes reused IDs at EOF", async () => { + for (const snapshots of ["live", "immutable"]) { + for (const prefix of [undefined, "", '{"city":"Pa', '{"city":"Paris"}']) { + const events = [ + ...(prefix === undefined ? [] : [start("a"), delta("a", prefix)]), + done("a", '{"city":"Paris"}'), + done("a", '{"city":"Paris"}'), + start("a"), delta("a", "42"), + ]; + const values = []; + for await (const update of readStructured(events, { integration: openAI, limits: { snapshots } })) + if (update.type === "complete") values.push(update.value); + assert.deepEqual(values, [{ city: "Paris" }, 42]); + } + } +});