From 6ffe9fe5bf6fc1f457b54fe7118c8b8500bcf6a0 Mon Sep 17 00:00:00 2001 From: Thyrst Date: Tue, 27 Oct 2026 11:30:18 +0000 Subject: [PATCH 1/5] Fix SDK stream completion at DONE (LAN-873) --- package.json | 2 +- pnpm-lock.yaml | 50 ++++++- src/openai.spec.ts | 54 +++---- src/vercel-ai/close-on-sse-done.spec.ts | 177 +++++++++++++++++++++++ src/vercel-ai/close-on-sse-done.ts | 98 +++++++++++++ src/vercel-ai/langtail-language-model.ts | 19 ++- src/vercel-ai/stream-completion.spec.ts | 174 ++++++++++++++++++++++ 7 files changed, 539 insertions(+), 35 deletions(-) create mode 100644 src/vercel-ai/close-on-sse-done.spec.ts create mode 100644 src/vercel-ai/close-on-sse-done.ts create mode 100644 src/vercel-ai/stream-completion.spec.ts diff --git a/package.json b/package.json index 7f4069b..cd4ec81 100644 --- a/package.json +++ b/package.json @@ -41,7 +41,7 @@ "copyfiles": "^2.4.1", "fs-extra": "^11.2.0", "jsdom": "^24.1.0", - "nock": "14.0.0-beta.5", + "nock": "14.0.17", "prettier": "^3.2.5", "tsup": "^8.0.2", "typescript": "^5.7.3", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 9585c9f..5df22c2 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -53,8 +53,8 @@ devDependencies: specifier: ^24.1.0 version: 24.1.0 nock: - specifier: 14.0.0-beta.5 - version: 14.0.0-beta.5 + specifier: 14.0.17 + version: 14.0.17 prettier: specifier: ^3.2.5 version: 3.2.5 @@ -381,6 +381,18 @@ packages: typescript: 5.8.2 dev: false + /@mswjs/interceptors@0.41.9: + resolution: {integrity: sha512-VVPPgHyQ6ShqnrmDWuxjmUIsO9gWyOZFmuOfLd9LfBGQJwZfy0gvv9pbHSJuoFNIYC7ZDX9aoFwowjcdSC4E8w==} + engines: {node: '>=18'} + dependencies: + '@open-draft/deferred-promise': 2.2.0 + '@open-draft/logger': 0.3.0 + '@open-draft/until': 2.1.0 + is-node-process: 1.2.0 + outvariant: 1.4.3 + strict-event-emitter: 0.5.1 + dev: true + /@nodelib/fs.scandir@2.1.5: resolution: {integrity: sha512-vq24Bq3ym5HEQm2NKCr3yXDwjc7vTsEThRDnkp2DK9p1uqLR+DHurm/NOTo0KG7HYHU7eppKZj3MyqYuMBf62g==} engines: {node: '>= 8'} @@ -402,6 +414,21 @@ packages: fastq: 1.17.1 dev: true + /@open-draft/deferred-promise@2.2.0: + resolution: {integrity: sha512-CecwLWx3rhxVQF6V4bAgPS5t+So2sTbPgAzafKkVizyi7tlwpcFpdFqq+wqF2OwNBmqFuu6tOyouTuxgpMfzmA==} + dev: true + + /@open-draft/logger@0.3.0: + resolution: {integrity: sha512-X2g45fzhxH238HKO4xbSr7+wBS8Fvw6ixhTDuvLd5mqh6bJJCFAPwU9mPDxbcrRtfxv4u5IHCEH77BmxvXmmxQ==} + dependencies: + is-node-process: 1.2.0 + outvariant: 1.4.3 + dev: true + + /@open-draft/until@2.1.0: + resolution: {integrity: sha512-U69T3ItWHvLwGg5eJ0n3I62nWuE6ilHlmz7zM0npLBRvPRd7e6NYmg54vvRtP5mZG7kZqZCFVdsTWo7BPtBujg==} + dev: true + /@opentelemetry/api@1.9.0: resolution: {integrity: sha512-3giAOQvZiH5F9bMlMiv8+GSPMeqg0dbaeo58/0SlA9sxSqZhnUtxzX9/2FzyhS9sWQf5S0GJE0AKBrFqjpeYcg==} engines: {node: '>=8.0.0'} @@ -1265,6 +1292,10 @@ packages: is-extglob: 2.1.1 dev: true + /is-node-process@1.2.0: + resolution: {integrity: sha512-Vg4o6/fqPxIjtxgUH5QLJhwZ7gW5diGCVlXpuUfELC62CuxM1iHcRe51f2W1FDy04Ai4KJkagKjx3XaqyfRKXw==} + dev: true + /is-number@7.0.0: resolution: {integrity: sha512-41Cifkg6e8TylSpdtTpeLVMqvSBEVzTttHvERD741+pnZ8ANv0004MRL43QKPDlK9cGvNp6NZWZUBlbGXYxxng==} engines: {node: '>=0.12.0'} @@ -1522,10 +1553,11 @@ packages: resolution: {integrity: sha512-Yd3UES5mWCSqR+qNT93S3UoYUkqAZ9lLg8a7g9rimsWmYGK8cVToA4/sF3RrshdyV3sAGMXVUmpMYOw+dLpOuw==} dev: false - /nock@14.0.0-beta.5: - resolution: {integrity: sha512-u255tf4DYvyErTlPZA9uTfXghiZZy+NflUOFONPVKZ5tP0yaHwKig28zyFOLhu8y5YcCRC+V5vDk4HHileh2iw==} - engines: {node: '>= 18'} + /nock@14.0.17: + resolution: {integrity: sha512-EjRr1weMa4ALQX35AgZTEnP+weJJjlW1KGDiNM2IQC2069YDHas4f4B4UUYR+TTLyKWxJvOz2wObDKQs/LNreA==} + engines: {node: '>=18.20.0 <20 || >=20.12.1'} dependencies: + '@mswjs/interceptors': 0.41.9 json-stringify-safe: 5.0.1 propagate: 2.0.1 dev: true @@ -1626,6 +1658,10 @@ packages: - encoding dev: false + /outvariant@1.4.3: + resolution: {integrity: sha512-+Sl2UErvtsoajRDKCE5/dBz4DIvHXQQnAxtQTF04OJxY0+DyZXSo5P5Bb7XYWOh81syohlYL24hbDwxedPUJCA==} + dev: true + /p-limit@5.0.0: resolution: {integrity: sha512-/Eaoq+QyLSiXQ4lyYV23f14mZRQcXnxfHrN0vCai+ak9G0pp9iEQukIIZq5NccEvwRB8PUnZT0KsOoDCINS1qQ==} engines: {node: '>=18'} @@ -1938,6 +1974,10 @@ packages: resolution: {integrity: sha512-JPbdCEQLj1w5GilpiHAx3qJvFndqybBysA3qUOnznweH4QbNYUsW/ea8QzSrnh0vNsezMMw5bcVool8lM0gwzg==} dev: true + /strict-event-emitter@0.5.1: + resolution: {integrity: sha512-vMgjE/GGEPEFnhFub6pa4FmJBRBVOLpIII2hvCZ8Kzb7K0hlHo7mQv6xYrBvCL2LtAIBwFUK8wvuJgTVSQ5MFQ==} + dev: true + /strict-uri-encode@2.0.0: resolution: {integrity: sha512-QwiXZgpRcKkhTj2Scnn++4PKtWsH0kpzZ62L2R6c/LUVYv7hVnZqcg2+sMuT6R7Jusu1vviK/MFsu6kNJfWlEQ==} engines: {node: '>=4'} diff --git a/src/openai.spec.ts b/src/openai.spec.ts index c669b56..0b78393 100644 --- a/src/openai.spec.ts +++ b/src/openai.spec.ts @@ -5,41 +5,43 @@ import nock from "nock" import { ChatCompletionChunkSchema } from "./dataSchema" import OpenAI from "openai" - -const lt = createOpenAIProxy(new OpenAI({ apiKey: process.env.LANGTAIL_API_KEY })) +const lt = createOpenAIProxy( + new OpenAI({ apiKey: process.env.LANGTAIL_API_KEY || "local-test" }), +) describe("Langtail Proxy", () => { - it("should support streaming", async () => { - const proxyCompletion = await lt.chat.completions.create({ - messages: [ - { role: "system", content: "You are a helpful assistant." }, - { role: "user", content: "What is the weather like?" }, - ], - stream: true, - model: "gpt-3.5-turbo", - doNotRecord: false, - metadata: { - "custom-field": "1", - }, - }) - let partCount = 0 - for await (const part of proxyCompletion) { - partCount++ + it.skipIf(process.env.TEST_LIVE !== "true")( + "should support streaming", + async () => { + const proxyCompletion = await lt.chat.completions.create({ + messages: [ + { role: "system", content: "You are a helpful assistant." }, + { role: "user", content: "What is the weather like?" }, + ], + stream: true, + model: "gpt-3.5-turbo", + doNotRecord: false, + metadata: { + "custom-field": "1", + }, + }) + let partCount = 0 + for await (const part of proxyCompletion) { + partCount++ - ChatCompletionChunkSchema.parse(part) - } + ChatCompletionChunkSchema.parse(part) + } - expect(partCount > 1).toBe(true) - }) + expect(partCount > 1).toBe(true) + }, + ) it("should not record", async (t) => { nock(baseURL) // nock works by intercepting requests at the network level, if open AI switches to undici we will need to intercept differently .post("/chat/completions") .reply(200, function (uri, req) { - expect(this.req.headers["x-langtail-do-not-record"][0]).toBe("true") - expect(this.req.headers["x-langtail-metadata-custom-field"][0]).toBe( - "1", - ) + expect(this.req.headers["x-langtail-do-not-record"]).toBe("true") + expect(this.req.headers["x-langtail-metadata-custom-field"]).toBe("1") // check that body has no metadata or doNotRecord expect(req).toMatchInlineSnapshot(` diff --git a/src/vercel-ai/close-on-sse-done.spec.ts b/src/vercel-ai/close-on-sse-done.spec.ts new file mode 100644 index 0000000..4d182e0 --- /dev/null +++ b/src/vercel-ai/close-on-sse-done.spec.ts @@ -0,0 +1,177 @@ +// @vitest-environment node +import { describe, expect, it, vi } from "vitest" +import { closeOnSseDone } from "./close-on-sse-done" + +const encoder = new TextEncoder() + +describe("SSE protocol completion", () => { + it.each(["\n", "\r", "\r\n"])( + "handles %j framing at every byte boundary", + async (newline) => { + const payload = 'data: {"content":"hello 💓 [DONE]"}' + newline + newline + const done = "data: [DONE]" + newline + newline + const bytes = encoder.encode(payload + done) + for (let split = 0; split <= bytes.length; split++) { + const cancel = vi.fn() + const source = new ReadableStream({ + start(controller) { + controller.enqueue(bytes.slice(0, split)) + controller.enqueue(bytes.slice(split)) + controller.enqueue( + encoder.encode("data: must-not-be-forwarded\n\n"), + ) + }, + cancel, + }) + const result = await new Response(closeOnSseDone(source)).text() + // A CR terminates the final record without waiting for an optional LF. + expect(result).toBe( + payload + (newline === "\r\n" ? done.slice(0, -1) : done), + ) + expect(cancel).toHaveBeenCalledOnce() + } + }, + ) + + it("does not mistake comments, event names, multiline data, or JSON strings for DONE", async () => { + const payload = + ': [DONE]\n\nevent: [DONE]\n\ndata: [DO\ndata: NE]\n\ndata: {"text":"[DONE]"}\n\n' + const cancel = vi.fn() + const source = new ReadableStream({ + start(controller) { + controller.enqueue(payload) + controller.close() + }, + cancel, + }) + expect(await new Response(closeOnSseDone(source)).text()).toBe(payload) + expect(cancel).not.toHaveBeenCalled() + }) + + it("preserves usage after finish_reason and before DONE", async () => { + const payload = + 'data: {"finish_reason":"tool_calls"}\n\ndata: {"usage":{"total_tokens":42}}\n\ndata:[DONE]\n\n' + const source = new ReadableStream({ + start(controller) { + controller.enqueue(payload) + }, + }) + expect(await new Response(closeOnSseDone(source)).text()).toBe(payload) + }) + + it("does not wait for a hung upstream cancellation promise", async () => { + const source = new ReadableStream({ + start(controller) { + controller.enqueue("data: [DONE]\n\n") + }, + cancel() { + return new Promise(() => {}) + }, + }) + expect(await new Response(closeOnSseDone(source)).text()).toBe( + "data: [DONE]\n\n", + ) + }) + + it("forwards caller cancellation and source errors", async () => { + const cancel = vi.fn() + const source = new ReadableStream({ cancel }) + const reader = closeOnSseDone(source).getReader() + const pending = reader.read() + await reader.cancel("stop") + await expect(pending).resolves.toMatchObject({ done: true }) + expect(cancel).toHaveBeenCalledWith("stop") + const error = new Error("reset") + const failed = new ReadableStream({ + start(controller) { + controller.error(error) + }, + }) + await expect(new Response(closeOnSseDone(failed)).text()).rejects.toBe( + error, + ) + }) + + it("forwards large records and unterminated EOF without retaining payload text", async () => { + const payload = "data: " + "x".repeat(256 * 1024) + "\n\ndata: unfinished" + const source = new ReadableStream({ + start(controller) { + controller.enqueue(payload) + controller.close() + }, + }) + expect(await new Response(closeOnSseDone(source)).text()).toBe(payload) + }) +}) + +it("does not finish on an unterminated DONE record", async () => { + let upstream!: ReadableStreamDefaultController + const cancel = vi.fn() + const source = new ReadableStream({ + start(c) { + upstream = c + }, + cancel, + }) + const reader = closeOnSseDone(source).getReader() + upstream.enqueue("data: [DONE]\n") + expect((await reader.read()).done).toBe(false) + expect(cancel).not.toHaveBeenCalled() + upstream.enqueue("\n") + expect((await reader.read()).done).toBe(false) + expect((await reader.read()).done).toBe(true) + expect(cancel).toHaveBeenCalledOnce() +}) + +it("does not treat extra data fields or trailing characters as DONE", async () => { + const payload = + "data: [DONE]\ndata:\n\ndata: [DONE] suffix\n\ndata: [DONE] \n\n" + const source = new ReadableStream({ + start(c) { + c.enqueue(payload) + c.close() + }, + }) + expect(await new Response(closeOnSseDone(source)).text()).toBe(payload) +}) + +it("handles a stream delivered one byte at a time", async () => { + const payload = + ': heartbeat\r\ndata: {"content":"💓"}\r\n\r\ndata: [DONE]\n\n' + const bytes = encoder.encode(payload) + let index = 0 + const cancel = vi.fn() + const source = new ReadableStream({ + pull(c) { + if (index < bytes.length) c.enqueue(bytes.slice(index, ++index)) + }, + cancel, + }) + expect(await new Response(closeOnSseDone(source)).text()).toBe(payload) + expect(cancel).toHaveBeenCalledOnce() +}) + +it("observes rejected upstream cancellation without failing completed output", async () => { + const source = new ReadableStream({ + start(c) { + c.enqueue("data: [DONE]\n\n") + }, + cancel() { + return Promise.reject(new Error("cancel rejected")) + }, + }) + expect(await new Response(closeOnSseDone(source)).text()).toBe( + "data: [DONE]\n\n", + ) +}) + +it("does not wait for hung cancellation when the caller stops a silent read", async () => { + const cancel = vi.fn(() => new Promise(() => {})) + const reader = closeOnSseDone( + new ReadableStream({ cancel }), + ).getReader() + const pending = reader.read() + await reader.cancel("caller stopped") + await expect(pending).resolves.toEqual({ done: true, value: undefined }) + expect(cancel).toHaveBeenCalledWith("caller stopped") +}) diff --git a/src/vercel-ai/close-on-sse-done.ts b/src/vercel-ai/close-on-sse-done.ts new file mode 100644 index 0000000..b3c040f --- /dev/null +++ b/src/vercel-ai/close-on-sse-done.ts @@ -0,0 +1,98 @@ +/** + * End an SSE response at the protocol's [DONE] event, even if the HTTP + * connection stays open. Only SSE framing is inspected; payloads are neither + * parsed as JSON nor retained. Cancellation must not delay downstream EOF. + */ +export function closeOnSseDone( + source: ReadableStream, +): ReadableStream { + const reader = source.getReader() + const encoder = new TextEncoder() + let line = "" + let longLine = false + let data: string | undefined + let afterCr = false + let finished = false + + function endLine() { + if (!line && !longLine) { + const done = data === "[DONE]" + data = undefined + return done + } + if (line === "data" || line.startsWith("data:")) { + // Multiple data fields are joined with newlines in SSE, so they cannot + // masquerade as the single-line terminal event. + const value = line.slice(5).replace(/^ /, "") + data = data === undefined && !longLine ? value : "not-done" + } + line = "" + longLine = false + return false + } + + function terminalOffset(bytes: Uint8Array) { + for (let i = 0; i < bytes.length; i++) { + const byte = bytes[i]! + if (afterCr && byte === 10) { + afterCr = false + continue + } + afterCr = byte === 13 + if (byte === 10 || byte === 13) { + if (endLine()) return i + 1 + } else if (line.length < 12) { + line += String.fromCharCode(byte) + } else { + longLine = true + } + } + return undefined + } + + function cancelUpstream(reason?: unknown) { + // Some sources do not settle their cancellation promise promptly. Closing + // the consumer must not wait for them, but always observe cancellation errors. + void reader + .cancel(reason) + .catch(() => {}) + .finally(() => reader.releaseLock()) + } + + return new ReadableStream({ + async pull(controller) { + if (finished) return + try { + const result = await reader.read() + if (finished) return + if (result.done) { + finished = true + reader.releaseLock() + controller.close() + return + } + const bytes = + typeof result.value === "string" + ? encoder.encode(result.value) + : result.value + const end = terminalOffset(bytes) + controller.enqueue(end === undefined ? bytes : bytes.slice(0, end)) + if (end !== undefined) { + finished = true + controller.close() + cancelUpstream() + } + } catch (error) { + if (!finished) { + finished = true + reader.releaseLock() + controller.error(error) + } + } + }, + cancel(reason) { + finished = true + cancelUpstream(reason) + }, + }) +} diff --git a/src/vercel-ai/langtail-language-model.ts b/src/vercel-ai/langtail-language-model.ts index 45b465f..00ad081 100644 --- a/src/vercel-ai/langtail-language-model.ts +++ b/src/vercel-ai/langtail-language-model.ts @@ -17,6 +17,7 @@ import { postJsonToApi, } from "@ai-sdk/provider-utils" import { z } from "zod" +import { closeOnSseDone } from "./close-on-sse-done" import { convertToOpenAIChatMessages } from "./convert-to-openai-chat-messages" import { mapLangtailFinishReason } from "./map-langtail-finish-reason" import { LangtailChatSettings } from "./langtail-chat-settings" @@ -535,9 +536,21 @@ export class LangtailChatLanguageModel< headers: this.headers, body: body, failedResponseHandler: openaiFailedResponseHandler, - successfulResponseHandler: createEventSourceResponseHandler( - langtailChatChunksSchema, - ), + successfulResponseHandler: async ({ response, ...context }) => { + const handler = createEventSourceResponseHandler( + langtailChatChunksSchema, + ) + return handler({ + ...context, + response: response.body + ? new Response(closeOnSseDone(response.body), { + status: response.status, + statusText: response.statusText, + headers: response.headers, + }) + : response, + }) + }, abortSignal: options.abortSignal, }) diff --git a/src/vercel-ai/stream-completion.spec.ts b/src/vercel-ai/stream-completion.spec.ts new file mode 100644 index 0000000..c9e36be --- /dev/null +++ b/src/vercel-ai/stream-completion.spec.ts @@ -0,0 +1,174 @@ +// @vitest-environment node +import { createServer } from "node:http" +import type { AddressInfo } from "node:net" +import { streamText, tool } from "ai" +import { z } from "zod" +import { expect, it } from "vitest" +import { createLangtail } from "./langtail-provider" + +it.each(["\n", "\r", "\r\n"])( + "finishes tools and usage at DONE with %j framing without HTTP EOF", + async (newline) => { + let onDisconnect!: () => void + const disconnected = new Promise((resolve) => { + onDisconnect = resolve + }) + const event = (choices: unknown[], usage?: object) => + "data: " + + JSON.stringify({ + id: "test", + model: "test", + created: 1, + choices, + ...(usage ? { usage } : {}), + }) + + newline + + newline + const server = createServer((request, response) => { + request.resume() + response.on("close", () => { + onDisconnect() + }) + response.writeHead(200, { "content-type": "text/event-stream" }) + response.write( + event([ + { + index: 0, + delta: { + tool_calls: [ + { + index: 0, + id: "call", + type: "function", + function: { name: "Read", arguments: '{"path":"a.txt"}' }, + }, + ], + }, + finish_reason: null, + }, + ]), + ) + response.write( + event([{ index: 0, delta: {}, finish_reason: "tool_calls" }]), + ) + response.write( + event([], { prompt_tokens: 7, completion_tokens: 3, total_tokens: 10 }), + ) + response.write("data: [DONE]" + newline + newline) + // Deliberately leave the HTTP response open after protocol completion. + }) + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)) + const abort = new AbortController() + let deadline: ReturnType | undefined + try { + const provider = createLangtail({ + baseURL: `http://127.0.0.1:${(server.address() as AddressInfo).port}`, + apiKey: "local-test", + }) + const result = streamText({ + model: provider("test"), + prompt: "test", + maxRetries: 0, + abortSignal: abort.signal, + toolCallStreaming: true, + tools: { Read: tool({ parameters: z.object({ path: z.string() }) }) }, + }) + const drain = (async () => { + for await (const _ of result.fullStream) { + } + return "finished" + })() + void drain.catch(() => {}) + const outcome = await Promise.race([ + drain, + new Promise((resolve) => { + deadline = setTimeout( + () => resolve("still waiting for HTTP EOF"), + 2000, + ) + }), + ]) + expect(outcome).toBe("finished") + expect(await result.finishReason).toBe("tool-calls") + expect(await result.toolCalls).toMatchObject([ + { toolName: "Read", args: { path: "a.txt" } }, + ]) + expect(await result.usage).toMatchObject({ + promptTokens: 7, + completionTokens: 3, + }) + await disconnected + } finally { + clearTimeout(deadline) + abort.abort() + server.closeAllConnections() + await new Promise((resolve) => server.close(() => resolve())) + } + }, +) + +it.each(["provider error", "caller abort"])( + "preserves partial text on %s", + async (mode) => { + const abort = new AbortController() + let onDisconnect!: () => void + const disconnected = new Promise((resolve) => { + onDisconnect = resolve + }) + const server = createServer((request, response) => { + request.resume() + response.on("close", onDisconnect) + response.writeHead(200, { "content-type": "text/event-stream" }) + response.write( + 'data: {"id":"partial","choices":[{"index":0,"delta":{"content":"Partial 💓"},"finish_reason":null}]}\n\n', + ) + if (mode === "provider error") { + response.end( + 'data: {"error":{"message":"Upstream LLM stream interrupted","code":"UPSTREAM_STREAM_INTERRUPTED"}}\n\n', + ) + } + // Caller-abort case keeps the connection open until the client cancels. + }) + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)) + try { + const result = streamText({ + model: createLangtail({ + apiKey: "local-test", + baseURL: `http://127.0.0.1:${(server.address() as AddressInfo).port}`, + })("test"), + prompt: "test", + maxRetries: 0, + abortSignal: abort.signal, + onError() {}, + }) + let text = "" + const errors: unknown[] = [] + const drain = (async () => { + for await (const part of result.fullStream) { + if (part.type === "text-delta") { + text += part.textDelta + if (mode === "caller abort") abort.abort() + } + if (part.type === "error") errors.push(part.error) + } + })() + if (mode === "caller abort") { + await expect(drain).rejects.toMatchObject({ name: "AbortError" }) + } else { + await drain + } + expect(text).toBe("Partial 💓") + if (mode === "provider error") { + expect(errors).toEqual([ + expect.objectContaining({ code: "UPSTREAM_STREAM_INTERRUPTED" }), + ]) + expect(await result.finishReason).toBe("error") + } + await disconnected + } finally { + abort.abort() + server.closeAllConnections() + await new Promise((resolve) => server.close(() => resolve())) + } + }, +) From 71ba306f73344b964c904b61bfa29837da8ddf1b Mon Sep 17 00:00:00 2001 From: Thyrst Date: Tue, 27 Oct 2026 12:12:00 +0000 Subject: [PATCH 2/5] Handle leading SSE BOM and expand SDK lifecycle regressions --- src/vercel-ai/close-on-sse-done.spec.ts | 59 ++++ src/vercel-ai/close-on-sse-done.ts | 10 + src/vercel-ai/stream-cancellation.spec.ts | 113 +++++++ src/vercel-ai/stream-lifecycle.spec.ts | 348 ++++++++++++++++++++++ 4 files changed, 530 insertions(+) create mode 100644 src/vercel-ai/stream-cancellation.spec.ts create mode 100644 src/vercel-ai/stream-lifecycle.spec.ts diff --git a/src/vercel-ai/close-on-sse-done.spec.ts b/src/vercel-ai/close-on-sse-done.spec.ts index 4d182e0..ec50104 100644 --- a/src/vercel-ai/close-on-sse-done.spec.ts +++ b/src/vercel-ai/close-on-sse-done.spec.ts @@ -175,3 +175,62 @@ it("does not wait for hung cancellation when the caller stops a silent read", as await expect(pending).resolves.toEqual({ done: true, value: undefined }) expect(cancel).toHaveBeenCalledWith("caller stopped") }) + +it("recognizes a leading UTF-8 BOM even when it is split across chunks", async () => { + const bytes = encoder.encode("\uFEFFdata: [DONE]\n\n") + for (let split = 0; split <= bytes.length; split++) { + const cancel = vi.fn() + const source = new ReadableStream({ + start(c) { + c.enqueue(bytes.slice(0, split)) + c.enqueue(bytes.slice(split)) + }, + cancel, + }) + const reader = closeOnSseDone(source).getReader() + try { + let received = 0 + while (received < bytes.length) { + const { value } = await reader.read() + received += value?.length ?? 0 + } + expect(cancel).toHaveBeenCalledOnce() + expect((await reader.read()).done).toBe(true) + } finally { + await reader.cancel() + } + } +}) + +it("preserves framing across 200 reproducible random chunk layouts", async () => { + let seed = 873 + const next = () => (seed = (Math.imul(seed, 1664525) + 1013904223) >>> 0) + for (let run = 0; run < 200; run++) { + const newline = ["\n", "\r", "\r\n"][next() % 3]! + const prefix = + (run % 2 ? "\uFEFF" : "") + + ': [DONE]\r\ndata: {"text":"💓 [DONE]"}\n\ndata: [DONE]\ndata: extra\n\n' + const expected = + prefix + "data:[DONE]" + newline + (newline === "\r\n" ? "\r" : newline) + const bytes = encoder.encode( + prefix + "data:[DONE]" + newline + newline + "data: ignored\n\n", + ) + let offset = 0 + const cancel = vi.fn() + const source = new ReadableStream({ + pull(c) { + if (offset < bytes.length) { + const end = offset + 1 + (next() % 23) + c.enqueue(bytes.slice(offset, end)) + offset = end + } + }, + cancel, + }) + const actual = new Uint8Array( + await new Response(closeOnSseDone(source)).arrayBuffer(), + ) + expect(actual).toEqual(encoder.encode(expected)) + expect(cancel).toHaveBeenCalledOnce() + } +}) diff --git a/src/vercel-ai/close-on-sse-done.ts b/src/vercel-ai/close-on-sse-done.ts index b3c040f..1a1e681 100644 --- a/src/vercel-ai/close-on-sse-done.ts +++ b/src/vercel-ai/close-on-sse-done.ts @@ -13,6 +13,7 @@ export function closeOnSseDone( let data: string | undefined let afterCr = false let finished = false + let initialBytes = 0 function endLine() { if (!line && !longLine) { @@ -34,6 +35,15 @@ export function closeOnSseDone( function terminalOffset(bytes: Uint8Array) { for (let i = 0; i < bytes.length; i++) { const byte = bytes[i]! + // TextDecoder strips one leading UTF-8 BOM. Match that behavior in the + // framing scanner too, even when the three bytes arrive separately. + if (initialBytes < 3) { + initialBytes++ + if (initialBytes === 3 && line === "\xEF\xBB" && byte === 0xbf) { + line = "" + continue + } + } if (afterCr && byte === 10) { afterCr = false continue diff --git a/src/vercel-ai/stream-cancellation.spec.ts b/src/vercel-ai/stream-cancellation.spec.ts new file mode 100644 index 0000000..ce11ef6 --- /dev/null +++ b/src/vercel-ai/stream-cancellation.spec.ts @@ -0,0 +1,113 @@ +// @vitest-environment node +import { createServer } from "node:http" +import type { ServerResponse } from "node:http" +import type { AddressInfo } from "node:net" +import { expect, it } from "vitest" +import { createLangtail } from "./langtail-provider" + +const delta = + 'data: {"id":"test","choices":[{"index":0,"delta":{"content":"hello"},"finish_reason":null}]}\n\n' +const callOptions = { + prompt: [ + { + role: "user" as const, + content: [{ type: "text" as const, text: "test" }], + }, + ], + inputFormat: "prompt" as const, + mode: { type: "regular" as const }, +} + +it.each([ + "before headers", + "silent body", + "partial event", + ...Array.from({ length: 10 }, (_, i) => `active ${i}`), +])("cancels real HTTP transport: %s", async (mode) => { + let requested!: () => void + const requestSeen = new Promise((resolve) => { + requested = resolve + }) + let disconnected!: () => void + const closed = new Promise((resolve) => { + disconnected = resolve + }) + let timer: ReturnType | undefined + const server = createServer((request, response) => { + request.resume() + response.on("close", () => { + clearInterval(timer) + disconnected() + }) + requested() + if (mode === "before headers") return + response.writeHead(200, { "content-type": "text/event-stream" }) + response.flushHeaders() + if (mode === "partial event") response.write('data: {"choices":') + if (mode.startsWith("active")) + timer = setInterval(() => response.write(delta), 1) + }) + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)) + const abort = new AbortController() + try { + const model = createLangtail({ + apiKey: "local-test", + baseURL: `http://127.0.0.1:${(server.address() as AddressInfo).port}`, + })("test") + const pending = model.doStream({ + ...callOptions, + abortSignal: abort.signal, + }) + await requestSeen + if (mode === "before headers") { + abort.abort() + await expect(pending).rejects.toMatchObject({ name: "AbortError" }) + } else { + const { stream } = await pending + const reader = stream.getReader() + if (mode.startsWith("active")) { + let part: any + do { + part = (await reader.read()).value + } while (part.type !== "text-delta") + } + await reader.cancel("client stopped") + } + await closed + } finally { + abort.abort() + clearInterval(timer) + server.closeAllConnections() + await new Promise((resolve) => server.close(() => resolve())) + } +}) + +it("propagates a real socket reset after delivered text", async () => { + let response!: ServerResponse + const server = createServer((request, res) => { + request.resume() + response = res + res.writeHead(200, { "content-type": "text/event-stream" }) + res.write(delta) + }) + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)) + try { + const model = createLangtail({ + apiKey: "local-test", + baseURL: `http://127.0.0.1:${(server.address() as AddressInfo).port}`, + })("test") + const { stream } = await model.doStream(callOptions) + const reader = stream.getReader() + let part: any + do { + part = (await reader.read()).value + } while (part.type !== "text-delta") + expect(part.textDelta).toBe("hello") + const next = reader.read() + response.destroy() + await expect(next).rejects.toBeInstanceOf(Error) + } finally { + server.closeAllConnections() + await new Promise((resolve) => server.close(() => resolve())) + } +}) diff --git a/src/vercel-ai/stream-lifecycle.spec.ts b/src/vercel-ai/stream-lifecycle.spec.ts new file mode 100644 index 0000000..acb0c9f --- /dev/null +++ b/src/vercel-ai/stream-lifecycle.spec.ts @@ -0,0 +1,348 @@ +// @vitest-environment node +import { afterEach, expect, it, vi } from "vitest" +import { createLangtail } from "./langtail-provider" + +const encode = (s: string) => new TextEncoder().encode(s) +const event = (choices: unknown[], usage?: object) => + `data: ${JSON.stringify({ id: "completion-test", model: "test-model", created: 123, choices, ...(usage ? { usage } : {}) })}\n\n` +const chunk = (delta: object, reason: string | null = null) => + event([{ index: 0, delta, finish_reason: reason }]) +const text = chunk({ content: "Hello 💓 [DONE]" }) +const finish = chunk({}, "stop") +const usage = event([], { + prompt_tokens: 17, + completion_tokens: 9, + total_tokens: 26, +}) +const done = "data: [DONE]\n\n" +const provider = createLangtail({ + baseURL: "https://local.invalid", + apiKey: "test", +}) +const callOptions = { + prompt: [ + { + role: "user" as const, + content: [{ type: "text" as const, text: "test" }], + }, + ], + inputFormat: "prompt" as const, + mode: { type: "regular" as const }, +} + +afterEach(() => { + vi.unstubAllGlobals() +}) + +function setup(body: string, split: "one" | "bytes" | "random", eof: boolean) { + const bytes = encode(body) + const cancel = vi.fn() + let offset = 0 + let seed = 12345 + const source = new ReadableStream({ + pull(c) { + if (offset >= bytes.length) { + if (eof) c.close() + return + } + seed = (Math.imul(seed, 1664525) + 1013904223) >>> 0 + const length = + split === "one" ? bytes.length : split === "bytes" ? 1 : 1 + (seed % 37) + c.enqueue(bytes.slice(offset, (offset += length))) + }, + cancel, + }) + vi.stubGlobal( + "fetch", + vi.fn( + async () => + new Response(source, { + headers: { + "content-type": "text/event-stream", + "x-langtail-request-id": "request-test", + }, + }), + ), + ) + return { cancel, source } +} + +async function collect() { + const result = await provider("test").doStream(callOptions) + const reader = result.stream.getReader() + const parts: any[] = [] + try { + while (true) { + const { value, done } = await reader.read() + if (done) break + parts.push(value) + } + } finally { + reader.releaseLock() + } + return { result, parts, finish: parts.find((p) => p.type === "finish") } +} + +const scenarios = [ + ["text", text + finish + usage], + [ + "zero usage", + text + + finish + + event([], { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0 }), + ], + ["length", text + chunk({}, "length") + usage], + ["content filter", text + chunk({}, "content_filter") + usage], + [ + "reasoning", + chunk({ reasoning_content: "Thinking 💭" }) + text + finish + usage, + ], + [ + "signed reasoning", + chunk({ + reasoning_details: [ + { + type: "reasoning.text", + text: "Thinking", + signature: "sig-test", + format: "anthropic-claude-v1", + index: 0, + }, + ], + }) + + text + + finish + + usage, + ], + [ + "parallel tools", + chunk({ + tool_calls: [ + { + index: 0, + id: "call-0", + type: "function", + function: { name: "Read", arguments: '{"path":' }, + }, + { + index: 1, + id: "call-1", + type: "function", + function: { name: "Read", arguments: '{"path":' }, + }, + ], + }) + + chunk({ + tool_calls: [ + { index: 1, function: { arguments: '"b.txt"}' } }, + { index: 0, function: { arguments: '"a.txt"}' } }, + ], + }) + + chunk({}, "tool_calls") + + usage, + ], + ["heartbeats", ": ping\n\n" + text + ": ping\n\n" + finish + usage], + ["BOM", "\uFEFF" + text + finish + usage], + ["large event", chunk({ content: "💓".repeat(16384) }) + finish + usage], +] as const + +it.each(scenarios)( + "preserves %s results across fragmentation and HTTP lifetime", + async (name, body) => { + setup(body + done, "one", true) + const reference = await collect() + expect(reference.parts.filter((p) => p.type === "error")).toEqual([]) + expect(reference.parts.filter((p) => p.type === "finish")).toHaveLength(1) + if (name === "parallel tools") { + expect( + reference.parts.filter((p) => p.type === "tool-call"), + ).toMatchObject([ + { toolCallId: "call-1", toolName: "Read", args: '{"path":"b.txt"}' }, + { toolCallId: "call-0", toolName: "Read", args: '{"path":"a.txt"}' }, + ]) + } else { + expect( + reference.parts + .filter((p) => p.type === "text-delta") + .map((p) => p.textDelta) + .join(""), + ).toBe(name === "large event" ? "💓".repeat(16384) : "Hello 💓 [DONE]") + } + expect(reference.finish.usage).toMatchObject( + name === "zero usage" + ? { promptTokens: 0, completionTokens: 0 } + : { promptTokens: 17, completionTokens: 9 }, + ) + for (const split of name === "large event" + ? (["random"] as const) + : (["bytes", "random"] as const)) { + const { cancel } = setup( + body + done + "data: garbage-after-done\n\n", + split, + false, + ) + const actual = await collect() + // V2 generates text/reasoning block IDs; those identifiers are not payload. + const stable = (parts: any[]) => parts.map(({ id, ...part }) => part) + expect(stable(actual.parts)).toEqual(stable(reference.parts)) + expect(cancel).toHaveBeenCalledOnce() + } + }, +) + +it("does not end at finish_reason before delayed usage arrives", async () => { + let upstream!: ReadableStreamDefaultController + const cancel = vi.fn() + vi.stubGlobal( + "fetch", + vi.fn( + async () => + new Response( + new ReadableStream({ + start(c) { + upstream = c + }, + cancel, + }), + ), + ), + ) + const result = await provider("test").doStream(callOptions) + const reader = result.stream.getReader() + upstream.enqueue(encode(text + finish)) + let part: any + do { + part = (await reader.read()).value + } while (part.type !== "text-delta") + expect(cancel).not.toHaveBeenCalled() + upstream.enqueue(encode(usage + done)) + const parts: any[] = [] + while (true) { + const next = await reader.read() + if (next.done) break + parts.push(next.value) + } + expect(parts.find((p) => p.type === "finish").usage).toMatchObject({ + promptTokens: 17, + completionTokens: 9, + }) + expect(cancel).toHaveBeenCalledOnce() +}) + +it.each([ + ["empty EOF", ""], + ["finish without DONE", text + finish + usage], + ["text without finish", text], + ["unterminated JSON", text + 'data: {"choices":'], +])("preserves existing clean EOF behavior: %s", async (_name, body) => { + setup(body, "random", true) + const { parts } = await collect() + expect(parts.filter((p) => p.type === "finish")).toHaveLength(1) + expect(parts.filter((p) => p.type === "error")).toHaveLength( + _name === "unterminated JSON" ? 1 : 0, + ) +}) + +it.each([ + [ + "error envelope", + 'data: {"error":{"message":"Upstream LLM stream interrupted","code":"UPSTREAM_STREAM_INTERRUPTED"}}\n\n', + ], + ["malformed JSON", 'data: {"choices":\n\n'], + ["invalid schema", 'data: {"choices":42}\n\n'], +])("surfaces %s after partial text", async (_name, tail) => { + setup(text + tail + done, "bytes", false) + const { parts, finish } = await collect() + expect( + parts + .filter((p) => p.type === "text-delta") + .map((p) => p.textDelta) + .join(""), + ).toBe("Hello 💓 [DONE]") + expect(parts.filter((p) => p.type === "error")).toHaveLength(1) + expect(finish.finishReason).toBe("error") +}) + +it.each([0, 1, 20])( + "propagates provider read failures after %i bytes of a partial event", + async (tailLength) => { + const failure = new Error("provider connection reset") + let upstream!: ReadableStreamDefaultController + vi.stubGlobal( + "fetch", + vi.fn( + async () => + new Response( + new ReadableStream({ + start(c) { + upstream = c + }, + }), + ), + ), + ) + const result = await provider("test").doStream(callOptions) + const reader = result.stream.getReader() + upstream.enqueue(encode(text)) + let part: any + do { + part = (await reader.read()).value + } while (part.type !== "text-delta") + if (tailLength) + upstream.enqueue(encode('data: {"choices": [{}]}').slice(0, tailLength)) + const pending = reader.read() + upstream.error(failure) + await expect(pending).rejects.toBe(failure) + }, +) + +it.each([false, true])( + "cancels silent response, partial event=%s", + async (partial) => { + const cancel = vi.fn() + const source = new ReadableStream({ + start(c) { + if (partial) c.enqueue(encode('data: {"choices":')) + }, + cancel, + }) + vi.stubGlobal( + "fetch", + vi.fn(async () => new Response(source)), + ) + const { stream } = await provider("test").doStream(callOptions) + const reader = stream.getReader() + // V2's start event is synchronous; start draining until cancellation. + const drain = (async () => { + while (!(await reader.read()).done) {} + })() + await reader.cancel("stop") + await drain + await vi.waitFor(() => expect(cancel).toHaveBeenCalledWith("stop")) + }, +) + +it.each([401, 429, 503])( + "preserves pre-stream HTTP %i errors", + async (status) => { + vi.stubGlobal( + "fetch", + vi.fn( + async () => + new Response( + JSON.stringify({ + error: { + message: "provider rejected request", + type: "api_error", + code: "test", + }, + }), + { status }, + ), + ), + ) + await expect(provider("test").doStream(callOptions)).rejects.toMatchObject({ + statusCode: status, + }) + }, +) From e448b3c7a08dd7383c65c4ed28cb7d205ddae069 Mon Sep 17 00:00:00 2001 From: Thyrst Date: Tue, 27 Oct 2026 12:17:12 +0000 Subject: [PATCH 3/5] Use GPT 6.1 Luna for live CI tests on both SDK branches --- .github/workflows/main.yml | 2 +- src/LangtailPrompts.spec.ts | 15 +++++++++++++++ src/openai.spec.ts | 4 +++- 3 files changed, 19 insertions(+), 2 deletions(-) diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index 044e594..73c20db 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -5,7 +5,7 @@ on: tags: - "*" pull_request: - branches: [main] + branches: [main, "v0.16.x"] jobs: build: diff --git a/src/LangtailPrompts.spec.ts b/src/LangtailPrompts.spec.ts index 74db6cb..89f8d71 100644 --- a/src/LangtailPrompts.spec.ts +++ b/src/LangtailPrompts.spec.ts @@ -75,6 +75,11 @@ describe.skipIf(!liveTesting)( const completion = await lt.invoke({ prompt, environment: "staging", + // Override the deployed fixture's retired model for live inference. + model: "gpt-6.1-luna", + temperature: 1, + reasoning_effort: "low", + max_tokens: 512, variables: { about: "cowboy Bebop", }, @@ -91,6 +96,11 @@ describe.skipIf(!liveTesting)( const completion = await lt.invoke({ prompt: "short-story-teller", environment: "staging", + // Override the deployed fixture's retired model for live inference. + model: "gpt-6.1-luna", + temperature: 1, + reasoning_effort: "low", + max_tokens: 512, variables: { about: "Aragorn", }, @@ -107,6 +117,11 @@ describe.skipIf(!liveTesting)( const proxyCompletion = await lt.invoke({ prompt, environment: "staging", + // Override the deployed fixture's retired model for live inference. + model: "gpt-6.1-luna", + temperature: 1, + reasoning_effort: "low", + max_tokens: 512, variables: { about: "napoleon", }, diff --git a/src/openai.spec.ts b/src/openai.spec.ts index 0b78393..65f3b3f 100644 --- a/src/openai.spec.ts +++ b/src/openai.spec.ts @@ -19,7 +19,9 @@ describe("Langtail Proxy", () => { { role: "user", content: "What is the weather like?" }, ], stream: true, - model: "gpt-3.5-turbo", + model: "gpt-6.1-luna", + reasoning_effort: "low", + max_tokens: 512, doNotRecord: false, metadata: { "custom-field": "1", From a252604cb2bc20ac459d2ac0810f94f86e3cf381 Mon Sep 17 00:00:00 2001 From: Thyrst Date: Tue, 27 Oct 2026 12:28:13 +0000 Subject: [PATCH 4/5] Use available GPT 6 Luna model in live CI tests --- src/LangtailPrompts.spec.ts | 6 +++--- src/openai.spec.ts | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/LangtailPrompts.spec.ts b/src/LangtailPrompts.spec.ts index 89f8d71..07b43ae 100644 --- a/src/LangtailPrompts.spec.ts +++ b/src/LangtailPrompts.spec.ts @@ -76,7 +76,7 @@ describe.skipIf(!liveTesting)( prompt, environment: "staging", // Override the deployed fixture's retired model for live inference. - model: "gpt-6.1-luna", + model: "gpt-6-luna", temperature: 1, reasoning_effort: "low", max_tokens: 512, @@ -97,7 +97,7 @@ describe.skipIf(!liveTesting)( prompt: "short-story-teller", environment: "staging", // Override the deployed fixture's retired model for live inference. - model: "gpt-6.1-luna", + model: "gpt-6-luna", temperature: 1, reasoning_effort: "low", max_tokens: 512, @@ -118,7 +118,7 @@ describe.skipIf(!liveTesting)( prompt, environment: "staging", // Override the deployed fixture's retired model for live inference. - model: "gpt-6.1-luna", + model: "gpt-6-luna", temperature: 1, reasoning_effort: "low", max_tokens: 512, diff --git a/src/openai.spec.ts b/src/openai.spec.ts index 65f3b3f..9d02eb2 100644 --- a/src/openai.spec.ts +++ b/src/openai.spec.ts @@ -19,7 +19,7 @@ describe("Langtail Proxy", () => { { role: "user", content: "What is the weather like?" }, ], stream: true, - model: "gpt-6.1-luna", + model: "gpt-6-luna", reasoning_effort: "low", max_tokens: 512, doNotRecord: false, From c8114a36a49e0a559d0c7f469e72ed915ad79901 Mon Sep 17 00:00:00 2001 From: Thyrst Date: Tue, 27 Oct 2026 12:53:03 +0000 Subject: [PATCH 5/5] Bump SDK version to 0.16.22 --- CHANGELOG.md | 5 +++++ package.json | 2 +- 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 67bb1da..26842ee 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,10 @@ # Changelog +## 0.16.22 + +- Finish Vercel AI SDK streams when a complete `[DONE]` event arrives, preserving tool calls and usage without waiting for the HTTP connection to close. +- Handle split SSE records and a leading UTF-8 BOM; cancel the response body without delaying completion when cancellation stalls or fails. + ## 0.16.21 - Add Meta's `meta-responses-v1` to the `reasoning_details` format union. It was missing, and since `ReasoningDetailArraySchema` drops entries it cannot parse, Meta reasoning details (e.g. from Muse Spark 1.2) were silently discarded instead of erroring — losing reasoning continuity across turns. diff --git a/package.json b/package.json index cd4ec81..1819124 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "langtail", - "version": "0.16.21", + "version": "0.16.22", "description": "", "main": "./Langtail.js", "packageManager": "pnpm@8.15.6",