From 7e0e831dafd0ab4ee0aa8a9aee9ea4c557c90269 Mon Sep 17 00:00:00 2001 From: zverev Date: Tue, 6 Oct 2026 00:24:26 +0300 Subject: [PATCH] fix: restore incremental Responses reasoning --- CHANGELOG.md | 4 + index.ts | 25 +++- src/responses-stream.ts | 223 +++++++++++++++++++++++++++++++++ tests/test-pi-local.mjs | 169 ++++++++++++++++++++++--- tests/test-transport.ts | 271 ++++++++++++++++++++++++++++++++++++++++ 5 files changed, 672 insertions(+), 20 deletions(-) create mode 100644 src/responses-stream.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index eb7d567..2140be7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,9 @@ # Changelog +## Unreleased + +- Restore incremental reasoning on the Provider API OpenAI Responses wire by translating Command Code's non-standard `response.reasoning.delta` events into the canonical `response.reasoning_text.delta` before they reach pi-ai. Reasoning now streams while the reasoning item is still open instead of appearing only at `response.output_item.done`. Reproduced with `deepseek/deepseek-v4.1-flash-fast`; standard Responses events, Chat Completions, Anthropic Messages, and the generate fallback are unchanged. + ## 0.7.4 - 2026-10-05 - Use the Provider API's OpenAI Responses wire (`/provider/v1/responses`) for models whose `/provider/v1/models` entry advertises `/responses`, keeping OpenAI Chat Completions for chat-only models and Anthropic Messages for Claude. The wire is resolved per model from `supported_endpoints`, falling back to Chat Completions when the catalog omits the field. This also enables prompt caching (`prompt_cache_key` is forwarded on `/responses` but ignored on `/chat/completions`): repeated prefixes bill at cache-read instead of full input price ([#132](https://github.com/patlux/pi-commandcode-provider/issues/132), [#133](https://github.com/patlux/pi-commandcode-provider/pull/133)). diff --git a/index.ts b/index.ts index 8e39886..98489b4 100644 --- a/index.ts +++ b/index.ts @@ -38,6 +38,7 @@ import { getApiKey as getOAuthApiKey, login, refreshToken } from "./src/oauth.ts import { normalizeCommandCodeMessage } from "./src/overflow.ts" import { MODEL_COSTS, ZERO_MODEL_COST } from "./src/pricing.ts" import { registerCommandCodeQuota } from "./src/quota-command.ts" +import { normalizeCommandCodeResponsesResponse } from "./src/responses-stream.ts" import { createCommandCodeRuntime } from "./src/runtime.ts" import { createCommandCodeUsageProvider, type UsageProvider } from "./src/usage.ts" import { transcriptReadersFrom, withTranscriptPromptAndTools } from "./src/transcript.ts" @@ -222,17 +223,31 @@ export default async function (pi: ExtensionAPI) { modelApis.get(modelId) ?? apiForModelId(modelId) const transport = createCommandCodeTransportRouter({ createStream: () => new AssistantMessageEventStream(), - streamProvider: (model, context, options) => - streamNativeProvider( + streamProvider: (model, context, options) => { + const wire = resolveModelApi(model.id) + const resolvedOptions: Parameters[2] = + resolveStreamOptions(options) + const nativeOptions = + wire === "openai-responses" + ? { + ...resolvedOptions, + fetch: async (...args: Parameters) => + normalizeCommandCodeResponsesResponse( + await (resolvedOptions?.fetch ?? fetch)(...args), + ), + } + : resolvedOptions + return streamNativeProvider( { ...model, - api: resolveModelApi(model.id), + api: wire, cost: commandCodeCostRatesAt(model.id, model.cost), compat: model.compatConfig ?? model.compat, }, context, - resolveStreamOptions(options), - ), + nativeOptions, + ) + }, streamGenerate: (model, context, options) => streamGenerate( model, diff --git a/src/responses-stream.ts b/src/responses-stream.ts new file mode 100644 index 0000000..b9526cb --- /dev/null +++ b/src/responses-stream.ts @@ -0,0 +1,223 @@ +/** + * Normalizes Command Code's non-standard Responses reasoning deltas. + * + * Command Code emits `response.reasoning.delta` while a reasoning item is still + * streaming, but pi-ai 1.0.3 only understands the canonical + * `response.reasoning_text.delta`. Without translation the deltas are dropped + * and the reasoning text only appears at `response.output_item.done`. Rewriting + * the frame in flight restores incremental thinking without touching any other + * Responses event. + */ + +const SSE_MIME_TYPE = "text/event-stream" +const REASONING_DELTA_EVENT = "response.reasoning.delta" +const REASONING_TEXT_DELTA_EVENT = "response.reasoning_text.delta" + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value) +} + +interface SseLine { + content: string + ending: string +} + +/** Splits a complete SSE frame into lines, preserving each original ending. */ +function splitSseLines(frame: string): SseLine[] { + const lines: SseLine[] = [] + let start = 0 + let index = 0 + while (index < frame.length) { + const code = frame.charCodeAt(index) + if (code === 0x0a) { + lines.push({ content: frame.slice(start, index), ending: "\n" }) + index += 1 + start = index + } else if (code === 0x0d) { + if (index + 1 < frame.length && frame.charCodeAt(index + 1) === 0x0a) { + lines.push({ content: frame.slice(start, index), ending: "\r\n" }) + index += 2 + } else { + lines.push({ content: frame.slice(start, index), ending: "\r" }) + index += 1 + } + start = index + } else { + index += 1 + } + } + if (start < frame.length) { + lines.push({ content: frame.slice(start), ending: "" }) + } + return lines +} + +interface SseField { + field: string + value: string +} + +/** Parses one SSE line. Returns undefined for comments and blank lines. */ +function parseSseField(content: string): SseField | undefined { + if (content.startsWith(":")) return undefined + const colon = content.indexOf(":") + if (colon === -1) return { field: content, value: "" } + const field = content.slice(0, colon) + let value = content.slice(colon + 1) + if (value.startsWith(" ")) value = value.slice(1) + return { field, value } +} + +/** + * Rewrites `response.reasoning.delta` frames as `response.reasoning_text.delta` + * frames. Everything else, including malformed JSON, is returned unchanged. + */ +function normalizeResponsesSseFrame(frame: string): string { + const lines = splitSseLines(frame) + const dataValues: string[] = [] + for (let index = 0; index < lines.length; index += 1) { + const content = index === 0 ? stripBom(lines[index].content) : lines[index].content + const parsed = parseSseField(content) + if (parsed?.field === "data") dataValues.push(parsed.value) + } + if (dataValues.length === 0) return frame + + let payload: unknown + try { + payload = JSON.parse(dataValues.join("\n")) + } catch { + return frame + } + if (!isRecord(payload)) return frame + if (payload.type !== REASONING_DELTA_EVENT) return frame + if (typeof payload.delta !== "string") return frame + if (!Number.isInteger(payload.output_index) || (payload.output_index as number) < 0) return frame + + const replacement = { ...payload, type: REASONING_TEXT_DELTA_EVENT } + const replacementLine = `data: ${JSON.stringify(replacement)}` + const output: string[] = [] + let replacedData = false + for (let index = 0; index < lines.length; index += 1) { + const line = lines[index] + const content = index === 0 ? stripBom(line.content) : line.content + const bom = index === 0 && line.content.startsWith("\uFEFF") ? "\uFEFF" : "" + const parsed = parseSseField(content) + if (parsed?.field === "data") { + if (!replacedData) { + output.push(`${bom}${replacementLine}${line.ending}`) + replacedData = true + } + continue + } + if (parsed?.field === "event" && parsed.value === REASONING_DELTA_EVENT) { + output.push(`${bom}event: ${REASONING_TEXT_DELTA_EVENT}${line.ending}`) + continue + } + output.push(`${line.content}${line.ending}`) + } + return output.join("") +} + +/** Drops a leading UTF-8 BOM for field recognition only, never from output. */ +function stripBom(content: string): string { + return content.startsWith("\uFEFF") ? content.slice(1) : content +} + +/** Stateful splitter that emits one normalized frame per SSE event. */ +class ResponsesSseNormalizer { + private readonly decoder = new TextDecoder("utf-8", { ignoreBOM: true }) + private readonly encoder = new TextEncoder() + private buffer = "" + private scanIndex = 0 + private lineStart = 0 + + push(text: string, emit: (frame: string) => void, flush: boolean): void { + this.buffer += text + this.process(emit, flush) + } + + finish(emit: (frame: string) => void): void { + if (this.buffer.length === 0) return + emit(this.buffer) + this.buffer = "" + this.scanIndex = 0 + this.lineStart = 0 + } + + private process(emit: (frame: string) => void, flush: boolean): void { + while (this.scanIndex < this.buffer.length) { + const code = this.buffer.charCodeAt(this.scanIndex) + if (code === 0x0a) { + this.consumeLine(this.scanIndex + 1, this.scanIndex, emit) + } else if (code === 0x0d) { + const next = this.scanIndex + 1 + if (next >= this.buffer.length) { + if (!flush) return + this.consumeLine(next, this.scanIndex, emit) + } else if (this.buffer.charCodeAt(next) === 0x0a) { + this.consumeLine(next + 1, this.scanIndex, emit) + } else { + this.consumeLine(next, this.scanIndex, emit) + } + } else { + this.scanIndex += 1 + } + } + } + + private consumeLine(end: number, lineEnd: number, emit: (frame: string) => void): void { + if (lineEnd === this.lineStart) { + const frame = this.buffer.slice(0, end) + emit(normalizeResponsesSseFrame(frame)) + this.buffer = this.buffer.slice(end) + this.scanIndex = 0 + this.lineStart = 0 + return + } + this.lineStart = end + this.scanIndex = end + } + + transform(chunk: Uint8Array, controller: TransformStreamDefaultController): void { + const text = this.decoder.decode(chunk, { stream: true }) + this.push(text, (frame) => controller.enqueue(this.encoder.encode(frame)), false) + } + + flush(controller: TransformStreamDefaultController): void { + this.push( + this.decoder.decode(), + (frame) => controller.enqueue(this.encoder.encode(frame)), + true, + ) + this.finish((frame) => controller.enqueue(this.encoder.encode(frame))) + } +} + +/** + * Wraps a Command Code Responses HTTP response so streaming reasoning deltas + * reach pi-ai. Returns the original response when it cannot be an SSE stream. + */ +export function normalizeCommandCodeResponsesResponse(response: Response): Response { + if (!response.ok) return response + if (response.body === null) return response + const mimeType = response.headers.get("content-type")?.split(";")[0]?.trim().toLowerCase() + if (mimeType !== SSE_MIME_TYPE) return response + + const normalizer = new ResponsesSseNormalizer() + const body = response.body.pipeThrough( + new TransformStream({ + transform: (chunk, controller) => normalizer.transform(chunk, controller), + flush: (controller) => normalizer.flush(controller), + }), + ) + + const headers = new Headers(response.headers) + headers.delete("content-length") + headers.delete("content-encoding") + + return new Response(body, { + status: response.status, + statusText: response.statusText, + headers, + }) +} diff --git a/tests/test-pi-local.mjs b/tests/test-pi-local.mjs index ae95d48..93b5775 100644 --- a/tests/test-pi-local.mjs +++ b/tests/test-pi-local.mjs @@ -19,7 +19,7 @@ const EXT_PATH = resolve(PROJECT_DIR, "index.ts") const COMPAT_CALLER_EXT_PATH = resolve(__dirname, "fixtures", "compat-caller-extension.ts") const TEST_MODEL = "gpt-5.4" const CLAUDE_TEST_MODEL = "claude-sonnet-4-6" -const RESPONSES_TEST_MODEL = "gpt-5.6-sol" +const RESPONSES_TEST_MODEL = "deepseek/deepseek-v4.1-flash-fast" function findPiBinary() { if (process.env.PI_BIN) return process.env.PI_BIN @@ -72,6 +72,9 @@ let overflowRequestCount = 0 let modelsDelayMs = 0 let includeRefreshedModel = false let includeResponsesModel = false +// When set, the Responses mock holds the reasoning item open until the test +// resolves this promise, proving deltas arrive before the item completes. +let responsesReasoningGate function modelCatalog() { const data = [ @@ -111,7 +114,7 @@ function modelCatalog() { object: "model", created: 1779824324, owned_by: "command-code", - name: "GPT 5.6 Sol", + name: "DeepSeek V4.1 Flash Fast", context_length: 1_000_000, supported_endpoints: ["/chat/completions", "/responses"], }) @@ -130,7 +133,7 @@ function modelCatalog() { } /** Minimal OpenAI Responses stream: one assistant text message, then a terminal event. */ -function writeResponsesStream(res, text) { +async function writeResponsesStream(res, text) { const item = { type: "message", id: "msg_mock", @@ -138,14 +141,58 @@ function writeResponsesStream(res, text) { status: "completed", content: [{ type: "output_text", text, annotations: [] }], } - const send = (event) => res.write(`data: ${JSON.stringify(event)}\n\n`) + const send = (event, name) => + res.write(`${name ? `event: ${name}\n` : ""}data: ${JSON.stringify(event)}\n\n`) send({ type: "response.created", response: { id: "resp_mock", object: "response", status: "in_progress", output: [] }, }) + const reasoningItem = { + type: "reasoning", + id: "rs_mock", + summary: [], + content: [{ type: "reasoning_text", text: "We think" }], + } + const textIndex = responsesReasoningGate ? 1 : 0 + if (responsesReasoningGate) { + send({ + type: "response.output_item.added", + output_index: 0, + item: { type: "reasoning", id: "rs_mock", summary: [], content: [] }, + }) + send( + { + type: "response.reasoning.delta", + output_index: 0, + content_index: 0, + item_id: "rs_mock", + delta: "We", + }, + "response.reasoning.delta", + ) + send( + { + type: "response.reasoning.delta", + output_index: 0, + content_index: 0, + item_id: "rs_mock", + delta: " think", + }, + "response.reasoning.delta", + ) + await responsesReasoningGate + send({ + type: "response.reasoning.done", + output_index: 0, + content_index: 0, + item_id: "rs_mock", + text: "We think", + }) + send({ type: "response.output_item.done", output_index: 0, item: reasoningItem }) + } send({ type: "response.output_item.added", - output_index: 0, + output_index: textIndex, item: { type: "message", id: "msg_mock", @@ -156,26 +203,26 @@ function writeResponsesStream(res, text) { }) send({ type: "response.output_text.delta", - output_index: 0, + output_index: textIndex, content_index: 0, item_id: "msg_mock", delta: text, }) send({ type: "response.output_text.done", - output_index: 0, + output_index: textIndex, content_index: 0, item_id: "msg_mock", text, }) - send({ type: "response.output_item.done", output_index: 0, item }) + send({ type: "response.output_item.done", output_index: textIndex, item }) send({ type: "response.completed", response: { id: "resp_mock", object: "response", status: "completed", - output: [item], + output: responsesReasoningGate ? [reasoningItem, item] : [item], usage: { input_tokens: 1, output_tokens: 1, @@ -278,7 +325,7 @@ const server = createServer((req, res) => { : "overflow-recovered" : "mock-pi-ok" if (isResponsesRequest) { - writeResponsesStream(res, text) + writeResponsesStream(res, text).catch((error) => res.destroy(error)) return } if (isAnthropicRequest) { @@ -376,11 +423,25 @@ function runPi(args, timeoutOrOptions = 30_000) { }) } +/** Observe close before shutdown so cleanup never races session-file writes. */ +function rpcChildCloseWaiter(child) { + const closed = new Promise((resolve) => child.once("close", resolve)) + return async () => { + const timer = setTimeout(() => child.kill("SIGKILL"), 5_000) + try { + await closed + } finally { + clearTimeout(timer) + } + } +} + async function runRpcQuery( timeoutMs = 30_000, promptMessage = "say mock token", extraArgs = [], promptFields = {}, + options = {}, ) { const child = spawn( PI_BIN, @@ -393,7 +454,7 @@ async function runRpcQuery( "--provider", "commandcode", "--model", - TEST_MODEL, + options.model ?? TEST_MODEL, ...extraArgs, ], { @@ -403,6 +464,7 @@ async function runRpcQuery( }, ) + const waitForClose = rpcChildCloseWaiter(child) let stdout = "" let stderr = "" let buffer = "" @@ -449,6 +511,7 @@ async function runRpcQuery( try { const event = JSON.parse(trimmed) events.push(event) + options.onEvent?.(event) if (event.type === "response" && event.id === "prompt-1" && event.success === true) { sawPromptAccepted = true } @@ -476,6 +539,7 @@ async function runRpcQuery( }) const ok = await done + await waitForClose() return { ok, stdout, @@ -508,6 +572,7 @@ async function runRpcExtensionCommands(timeoutMs = 30_000) { }, ) + const waitForClose = rpcChildCloseWaiter(child) let buffer = "" let stderr = "" const events = [] @@ -619,6 +684,7 @@ async function runRpcExtensionCommands(timeoutMs = 30_000) { } } finally { child.kill() + await waitForClose() } } @@ -645,6 +711,7 @@ async function runRpcCompatCall(timeoutMs = 30_000) { }, ) + const waitForClose = rpcChildCloseWaiter(child) let buffer = "" let stderr = "" @@ -688,6 +755,7 @@ async function runRpcCompatCall(timeoutMs = 30_000) { return { message: await notification, stderr } } finally { child.kill() + await waitForClose() } } @@ -712,6 +780,7 @@ async function runRpcOverflowRecovery(timeoutMs = 60_000) { }, ) + const waitForClose = rpcChildCloseWaiter(child) let buffer = "" let stderr = "" const events = [] @@ -768,6 +837,7 @@ async function runRpcOverflowRecovery(timeoutMs = 60_000) { }) const outcome = await result + await waitForClose() return { ...outcome, requests: overflowRequestCount, @@ -1092,6 +1162,77 @@ try { assert.equal(lastRequestBody?.model, RESPONSES_TEST_MODEL) assert.equal(lastRequestBody?.stream, true) assert.ok(Array.isArray(lastRequestBody?.input), "Responses request should use the input array") + + console.log("[pi-local] Responses reasoning deltas stream before the item completes") + requestCount = 0 + let releaseResponsesReasoningGate + responsesReasoningGate = new Promise((resolve) => { + releaseResponsesReasoningGate = resolve + }) + const thinkingDeltas = [] + let sawThinkingEndBeforeGate = false + let thinkingEndCount = 0 + try { + const reasoningRpc = await runRpcQuery( + 30_000, + "say mock token", + ["--thinking", "high"], + {}, + { + model: RESPONSES_TEST_MODEL, + onEvent: (event) => { + if (event.type !== "message_update") return + const assistantEvent = event.assistantMessageEvent + if (assistantEvent?.type === "thinking_delta") { + thinkingDeltas.push(assistantEvent.delta) + if (thinkingDeltas.length === 2) releaseResponsesReasoningGate() + } else if (assistantEvent?.type === "thinking_end") { + thinkingEndCount += 1 + if (thinkingDeltas.length < 2) sawThinkingEndBeforeGate = true + } + }, + }, + ) + assert.equal( + reasoningRpc.ok, + true, + JSON.stringify( + { + stderr: reasoningRpc.stderr, + stdout: reasoningRpc.stdout, + events: reasoningRpc.events.slice(-10), + }, + null, + 2, + ), + ) + assert.equal(sawThinkingEndBeforeGate, false) + assert.deepEqual(thinkingDeltas, ["We", " think"]) + assert.equal(thinkingEndCount, 1) + const thinkingEnd = reasoningRpc.events.find( + (event) => + event.type === "message_update" && event.assistantMessageEvent?.type === "thinking_end", + ) + assert.equal(thinkingEnd?.assistantMessageEvent?.content, "We think") + const finalMessage = [...reasoningRpc.events] + .reverse() + .find((event) => event.type === "message_end" && event.message?.role === "assistant") + const thinkingBlock = finalMessage?.message?.content?.find((block) => block.type === "thinking") + assert.equal(thinkingBlock?.thinking, "We think") + assert.equal( + finalMessage?.message?.content?.some( + (block) => block.type === "text" && block.text.includes("mock-pi-ok"), + ), + true, + ) + assert.equal(lastRequestPath, "/provider/v1/responses") + assert.equal(requestCount, 1) + assert.notEqual(finalMessage?.message?.stopReason, "error") + } finally { + releaseResponsesReasoningGate() + responsesReasoningGate = undefined + } + includeResponsesModel = false console.log("[pi-local] Claude request through Anthropic Messages endpoint") @@ -1196,11 +1337,9 @@ try { assert.equal(overflowRpc.sawCompactionRetry, true, JSON.stringify(overflowRpc)) assert.equal(overflowRpc.stderrHasSecrets, false) overflowMode = false - - console.log("[pi-local] PASS") } finally { await new Promise((resolve) => server.close(resolve)) - // RPC children can still flush session files while exiting after SIGTERM. - // Retry transient ENOTEMPTY, but keep a persistent cleanup failure fatal. + // RPC children have closed; retain retries for transient filesystem errors. rmSync(tempHome, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 }) } +console.log("[pi-local] PASS") diff --git a/tests/test-transport.ts b/tests/test-transport.ts index 82e9771..274888e 100644 --- a/tests/test-transport.ts +++ b/tests/test-transport.ts @@ -1,6 +1,7 @@ import assert from "node:assert/strict" import { describe, it } from "node:test" +import { normalizeCommandCodeResponsesResponse } from "../src/responses-stream.ts" import { createCommandCodeTransportRouter } from "../src/transport.ts" import type { AssistantMessageEvent, @@ -245,3 +246,273 @@ describe("Command Code transport router", () => { assert.equal(generateCalls, 0) }) }) + +const encoder = new TextEncoder() + +function sseHeaders(extra: Record = {}): Record { + return { "content-type": "text/event-stream; charset=utf-8", ...extra } +} + +async function readAll(response: Response): Promise { + const reader = response.body!.getReader() + const decoder = new TextDecoder("utf-8", { ignoreBOM: true }) + let output = "" + for (;;) { + const { done, value } = await reader.read() + if (done) break + output += decoder.decode(value, { stream: true }) + } + output += decoder.decode() + return output +} + +async function readWithTimeout( + reader: ReadableStreamDefaultReader, + timeoutMs = 2_000, +): Promise> { + return await Promise.race([ + reader.read(), + new Promise((_, reject) => + setTimeout(() => reject(new Error(`read timed out after ${timeoutMs}ms`)), timeoutMs), + ), + ]) +} + +function findBytes(haystack: Uint8Array, needle: readonly number[]): number { + outer: for (let i = 0; i + needle.length <= haystack.length; i += 1) { + for (let j = 0; j < needle.length; j += 1) { + if (haystack[i + j] !== needle[j]) continue outer + } + return i + } + return -1 +} + +describe("Command Code Responses stream normalizer", () => { + it("emits the canonical event incrementally across split chunks", async () => { + const event = { + type: "response.reasoning.delta", + output_index: 0, + content_index: 0, + item_id: "rs_1", + delta: "наб", + } + const frame = `event: response.reasoning.delta\ndata: ${JSON.stringify(event)}\n\n` + const bytes = encoder.encode(frame) + const typeSplit = findBytes(bytes, [0x72, 0x65, 0x61, 0x73, 0x6f, 0x6e]) + 4 + // Land inside the first Cyrillic character so the decoder must buffer bytes. + const charSplit = findBytes(bytes, [0xd0, 0xbd]) + 1 + + let sourceController!: ReadableStreamDefaultController + const source = new ReadableStream({ + start(controller) { + sourceController = controller + }, + }) + const response = new Response(source, { status: 200, headers: sseHeaders() }) + const normalized = normalizeCommandCodeResponsesResponse(response) + assert.notEqual(normalized, response) + const reader = normalized.body!.getReader() + + sourceController.enqueue(bytes.slice(0, typeSplit)) + sourceController.enqueue(bytes.slice(typeSplit, charSplit)) + sourceController.enqueue(bytes.slice(charSplit)) + + const first = await readWithTimeout(reader) + assert.equal(first.done, false) + const text = new TextDecoder().decode(first.value) + assert.match(text, /^event: response\.reasoning_text\.delta\n/) + const dataLine = text.split("\n").find((line) => line.startsWith("data: ")) + assert.ok(dataLine, `missing data line in ${JSON.stringify(text)}`) + const parsed = JSON.parse(dataLine.slice("data: ".length)) + assert.equal(parsed.type, "response.reasoning_text.delta") + assert.equal(parsed.delta, "наб") + assert.equal(parsed.item_id, "rs_1") + assert.equal(parsed.output_index, 0) + assert.equal(parsed.content_index, 0) + + // The upstream response is still open, yet the delta was already emitted. + sourceController.close() + const end = await readWithTimeout(reader) + assert.equal(end.done, true) + }) + + it("preserves a leading BOM when rewriting event or data lines", async () => { + const alias = { type: "response.reasoning.delta", output_index: 0, delta: "hi" } + const canonical = { ...alias, type: "response.reasoning_text.delta" } + for (const includeEvent of [false, true]) { + const input = + "\uFEFF" + + (includeEvent ? "event: response.reasoning.delta\n" : "") + + `data: ${JSON.stringify(alias)}\n\n` + const expected = + "\uFEFF" + + (includeEvent ? "event: response.reasoning_text.delta\n" : "") + + `data: ${JSON.stringify(canonical)}\n\n` + const response = new Response(input, { headers: sseHeaders() }) + assert.equal(await readAll(normalizeCommandCodeResponsesResponse(response)), expected) + } + }) + + it("recognizes LF, CRLF, and CR boundaries and merges split data", async () => { + const split = (json: string): [string, string] => [json.slice(0, 1), json.slice(1)] + for (const ending of ["\n", "\r\n", "\r"]) { + const alias = { type: "response.reasoning.delta", output_index: 0, delta: "hi" } + const canonical = { + type: "response.reasoning_text.delta", + output_index: 1, + delta: "yo", + item_id: "rs_2", + } + const [head, tail] = split(JSON.stringify(alias)) + const frame = + `event: response.reasoning.delta${ending}` + + `data: ${head}${ending}` + + `data: ${tail}${ending}${ending}` + + `data: ${JSON.stringify(canonical)}${ending}${ending}` + + // Split the first CRLF pair across two chunks so the scanner must wait. + const firstCrlf = frame.indexOf("\r\n") + const chunks = + ending === "\r\n" && firstCrlf !== -1 + ? [frame.slice(0, firstCrlf + 1), frame.slice(firstCrlf + 1)] + : [frame] + + const source = new ReadableStream({ + start(controller) { + for (const chunk of chunks) controller.enqueue(encoder.encode(chunk)) + controller.close() + }, + }) + const response = new Response(source, { status: 200, headers: sseHeaders() }) + const output = await readAll(normalizeCommandCodeResponsesResponse(response)) + + assert.equal(output.split("response.reasoning.delta").length - 1, 0, output) + assert.equal(output.split("response.reasoning_text.delta").length - 1, 3, output) + assert.ok(output.includes('"delta":"hi"'), output) + assert.ok(output.includes(JSON.stringify(canonical)), output) + } + }) + + it("leaves standard and malformed frames untouched", async () => { + const aliasInDelta = { + type: "response.output_text.delta", + output_index: 0, + delta: "response.reasoning.delta", + } + const input = + `data: ${JSON.stringify({ type: "response.output_text.delta", output_index: 0, delta: "x" })}\n\n` + + `data: ${JSON.stringify({ type: "response.reasoning.done", output_index: 0, text: "t" })}\n\n` + + `data: ${JSON.stringify({ type: "response.output_item.done", output_index: 0, item: { type: "reasoning", id: "rs", content: [] } })}\n\n` + + `data: ${JSON.stringify(aliasInDelta)}\n\n` + + `: keep this comment\n\n` + + `data: [DONE]\n\n` + + `data: {not json\n\n` + + `data: ${JSON.stringify({ type: "response.reasoning.delta", output_index: 0 })}\n\n` + + `data: ${JSON.stringify({ type: "response.reasoning.delta", output_index: -1, delta: "x" })}\n\n` + + const source = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode(input)) + controller.close() + }, + }) + const response = new Response(source, { status: 200, headers: sseHeaders() }) + const output = await readAll(normalizeCommandCodeResponsesResponse(response)) + assert.equal(output, input) + }) + + it("returns non-SSE responses unchanged without reading the body", async () => { + const errorResponse = new Response("nope", { status: 500 }) + assert.equal(normalizeCommandCodeResponsesResponse(errorResponse), errorResponse) + + const noBody = new Response(null, { status: 200 }) + assert.equal(noBody.body, null) + assert.equal(normalizeCommandCodeResponsesResponse(noBody), noBody) + + let pulled = 0 + const body = new ReadableStream({ + pull(controller) { + pulled += 1 + controller.enqueue(encoder.encode("{}")) + controller.close() + }, + }) + const jsonResponse = new Response(body, { + status: 200, + headers: { "content-type": "application/json" }, + }) + assert.equal(normalizeCommandCodeResponsesResponse(jsonResponse), jsonResponse) + assert.equal(pulled, 0) + }) + + it("preserves status and request metadata while dropping body-encoding headers", async () => { + const source = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode("data: [DONE]\n\n")) + controller.close() + }, + }) + const response = new Response(source, { + status: 200, + headers: sseHeaders({ + "request-id": "req_123", + "content-length": "999", + "content-encoding": "gzip", + }), + }) + const normalized = normalizeCommandCodeResponsesResponse(response) + assert.equal(normalized.status, 200) + assert.equal(normalized.statusText, response.statusText) + assert.equal(normalized.headers.get("request-id"), "req_123") + assert.equal(normalized.headers.get("content-length"), null) + assert.equal(normalized.headers.get("content-encoding"), null) + }) + + it("propagates cancellation to the source stream", async () => { + let resolveCancel!: (reason: unknown) => void + const cancelled = new Promise((resolve) => { + resolveCancel = resolve + }) + const source = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode("data: {}\n\n")) + }, + cancel(reason) { + resolveCancel(reason) + }, + }) + const response = new Response(source, { status: 200, headers: sseHeaders() }) + const reader = normalizeCommandCodeResponsesResponse(response).body!.getReader() + await readWithTimeout(reader) + await reader.cancel("stop") + + const reason = await Promise.race([ + cancelled, + new Promise((_, reject) => + setTimeout(() => reject(new Error("source was not cancelled")), 2_000), + ), + ]) + assert.equal(reason, "stop") + }) + + it("surfaces an input stream error instead of a clean end", async () => { + const failure = new Error("boom") + let first = true + const source = new ReadableStream({ + pull(controller) { + if (first) { + first = false + controller.enqueue(encoder.encode("data: {}\n\n")) + return + } + controller.error(failure) + }, + }) + const response = new Response(source, { status: 200, headers: sseHeaders() }) + const reader = normalizeCommandCodeResponsesResponse(response).body!.getReader() + const chunk = await readWithTimeout(reader) + assert.equal(chunk.done, false) + await assert.rejects(readWithTimeout(reader), /boom/) + }) +})