diff --git a/.changeset/workers-ai-provider-streaming-no-double-text.md b/.changeset/workers-ai-provider-streaming-no-double-text.md new file mode 100644 index 000000000..69e99a8ce --- /dev/null +++ b/.changeset/workers-ai-provider-streaming-no-double-text.md @@ -0,0 +1,5 @@ +--- +"workers-ai-provider": patch +--- + +Fix doubled streaming text. Workers AI's OpenAI-compatible streaming emits both the native top-level `response` field and `choices[0].delta.content` in the same SSE chunk with identical content; the stream mapper emitted a `text-delta` for each, doubling every token (e.g. `"Hello world"` → `"HelloHello world world"`). The two are now treated as mutually exclusive — `choices[0].delta.content` is used only when the native `response` field is absent for that chunk. diff --git a/packages/workers-ai-provider/src/streaming.ts b/packages/workers-ai-provider/src/streaming.ts index f1937cf8e..79b43c599 100644 --- a/packages/workers-ai-provider/src/streaming.ts +++ b/packages/workers-ai-provider/src/streaming.ts @@ -221,7 +221,16 @@ export function getMappedStream( }); } - const textDelta = delta.content as string | undefined; + // Workers AI's OpenAI-compatible streaming emits BOTH the native + // top-level `response` field and `choices[0].delta.content` in the same + // chunk, with identical content. Emitting a text-delta for each doubles + // the output (e.g. "Hello" -> "HelloHello"). Treat them as mutually + // exclusive: only use `choices[].delta.content` when the native + // `response` field was absent for this chunk. + const textDelta = + nativeResponse != null && nativeResponse !== "" + ? undefined + : (delta.content as string | undefined); if (textDelta && textDelta.length > 0) { if (bufferContentForSalvage) { contentBuffer += textDelta; diff --git a/packages/workers-ai-provider/test/stream-text.test.ts b/packages/workers-ai-provider/test/stream-text.test.ts index 5eb32f600..47d4ccb40 100644 --- a/packages/workers-ai-provider/test/stream-text.test.ts +++ b/packages/workers-ai-provider/test/stream-text.test.ts @@ -57,6 +57,52 @@ describe("REST API - Streaming Text Tests", () => { expect(accumulatedText).toBe("Hello chunk1Hello chunk2"); }); + it("should not duplicate text when a chunk carries BOTH `response` and `choices[].delta.content`", async () => { + // Workers AI's OpenAI-compatible streaming emits both the native `response` + // field and `choices[0].delta.content` in the same chunk, with identical + // content. The mapper must treat them as mutually exclusive, otherwise every + // token is emitted twice ("Hello world" -> "HelloHello world world"). + server.use( + http.post( + `https://api.cloudflare.com/client/v4/accounts/${TEST_ACCOUNT_ID}/ai/run/${TEST_MODEL}`, + async () => { + return new Response( + [ + `data: {"response":"Hello ","choices":[{"delta":{"content":"Hello "},"finish_reason":null}]}\n\n`, + `data: {"response":"world","choices":[{"delta":{"content":"world"},"finish_reason":null}]}\n\n`, + `data: {"response":"","choices":[{"delta":{},"finish_reason":"stop"}]}\n\n`, + "data: [DONE]\n\n", + ].join(""), + { + headers: { + "Content-Type": "text/event-stream", + "Transfer-Encoding": "chunked", + }, + status: 200, + }, + ); + }, + ), + ); + + const workersai = createWorkersAI({ + accountId: TEST_ACCOUNT_ID, + apiKey: TEST_API_KEY, + }); + + const result = streamText({ + model: workersai(TEST_MODEL), + prompt: "Say hello world", + }); + + let accumulatedText = ""; + for await (const chunk of result.textStream) { + accumulatedText += chunk; + } + + expect(accumulatedText).toBe("Hello world"); + }); + it("should handle chunk without 'response' field gracefully", async () => { server.use( http.post(