diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 4bb5ad83e5..44c6e704f1 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -3009,6 +3009,7 @@ export async function handleChatCore({ provider, model, connectionId, + clientResponseFormat, }); const dedupRequestBody = { ...translatedBody, model: `${provider}/${model}`, stream }; diff --git a/open-sse/utils/streamHandler.ts b/open-sse/utils/streamHandler.ts index 2bfbcef68e..ea33f18e79 100644 --- a/open-sse/utils/streamHandler.ts +++ b/open-sse/utils/streamHandler.ts @@ -1,4 +1,5 @@ import { trackPendingRequest } from "@/lib/usageDb"; +import { FORMATS } from "../translator/formats.ts"; // Stream handler with disconnect detection - shared for all providers @@ -16,10 +17,41 @@ type StreamControllerOptions = { provider?: string; model?: string; connectionId?: string | null; + clientResponseFormat?: string | null; }; type StreamController = ReturnType; +function isResponsesClientFormat(clientResponseFormat?: string | null): boolean { + return ( + clientResponseFormat === FORMATS.OPENAI_RESPONSES || + clientResponseFormat === FORMATS.OPENAI_RESPONSE + ); +} + +function responsesErrorType(statusCode: number): string { + if (statusCode === 429) return "rate_limit_error"; + if (statusCode === 401 || statusCode === 403) return "authentication_error"; + if (statusCode >= 400 && statusCode < 500) return "invalid_request_error"; + return "server_error"; +} + +function responsesErrorCode(statusCode: number): string { + if (statusCode === 429) return "rate_limit_exceeded"; + if (statusCode === 401) return "invalid_authentication"; + if (statusCode === 403) return "permission_denied"; + if (statusCode >= 400 && statusCode < 500) return "bad_request"; + return "server_error"; +} + +function claudeErrorType(statusCode: number): string { + if (statusCode === 429) return "rate_limit_error"; + if (statusCode === 401) return "authentication_error"; + if (statusCode === 403) return "permission_error"; + if (statusCode >= 400 && statusCode < 500) return "invalid_request_error"; + return "api_error"; +} + // Get HH:MM:SS timestamp function getTimeString() { return new Date().toLocaleTimeString("en-US", { @@ -45,6 +77,7 @@ export function createStreamController({ provider, model, connectionId, + clientResponseFormat, }: StreamControllerOptions = {}) { const abortController = new AbortController(); const startTime = Date.now(); @@ -138,9 +171,67 @@ export function createStreamController({ }, abort: () => abortController.abort(), + clientResponseFormat, }; } +function buildStreamErrorChunks( + errorMsg: string, + statusCode: number, + clientResponseFormat?: string | null +) { + const encoder = new TextEncoder(); + if (isResponsesClientFormat(clientResponseFormat)) { + const errorEvent = { + type: "response.failed", + response: { + id: null, + status: "failed", + error: { + message: errorMsg, + type: responsesErrorType(statusCode), + code: responsesErrorCode(statusCode), + }, + }, + }; + + return [encoder.encode(`event: response.failed\ndata: ${JSON.stringify(errorEvent)}\n\n`)]; + } + + if (clientResponseFormat === FORMATS.CLAUDE) { + const errorEvent = { + type: "error", + error: { + type: claudeErrorType(statusCode), + message: errorMsg, + }, + }; + + return [encoder.encode(`event: error\ndata: ${JSON.stringify(errorEvent)}\n\n`)]; + } + + const errorEvent = { + object: "chat.completion.chunk", + choices: [ + { + index: 0, + delta: {}, + finish_reason: "error", + }, + ], + error: { + message: errorMsg, + type: "upstream_error", + code: statusCode, + }, + }; + + return [ + encoder.encode(`data: ${JSON.stringify(errorEvent)}\n\n`), + encoder.encode(`data: [DONE]\n\n`), + ]; +} + /** * Create transform stream with disconnect detection * Wraps existing transform stream and adds abort capability @@ -175,25 +266,13 @@ export function createDisconnectAwareStream(transformStream, streamController) { ? Number((error as { statusCode?: unknown }).statusCode) || 500 : 500; - const errorEvent = { - object: "chat.completion.chunk", - choices: [ - { - index: 0, - delta: {}, - finish_reason: "error", - }, - ], - error: { - message: errorMsg, - type: "upstream_error", - code: statusCode, - }, - }; - - const encoder = new TextEncoder(); - controller.enqueue(encoder.encode(`data: ${JSON.stringify(errorEvent)}\n\n`)); - controller.enqueue(encoder.encode(`data: [DONE]\n\n`)); + for (const chunk of buildStreamErrorChunks( + errorMsg, + statusCode, + streamController.clientResponseFormat + )) { + controller.enqueue(chunk); + } controller.close(); } diff --git a/tests/unit/stream-handler.test.ts b/tests/unit/stream-handler.test.ts index c74741dd9c..a960632e11 100644 --- a/tests/unit/stream-handler.test.ts +++ b/tests/unit/stream-handler.test.ts @@ -6,6 +6,7 @@ import { createStreamController, pipeWithDisconnect, } from "../../open-sse/utils/streamHandler.ts"; +import { FORMATS } from "../../open-sse/translator/formats.ts"; import { clearPendingRequests, getPendingRequests, @@ -57,6 +58,104 @@ test("createDisconnectAwareStream converts upstream errors into SSE error chunks assert.match(text, /\[DONE\]/); }); +test("createDisconnectAwareStream emits Responses API failure events for Responses clients", async () => { + const upstreamError = Object.assign(new Error("responses stream died"), { statusCode: 503 }); + const transformStream = { + readable: new ReadableStream({ + start(controller) { + controller.error(upstreamError); + }, + }), + writable: { + getWriter() { + return { + abort() {}, + }; + }, + }, + }; + + const stream = createDisconnectAwareStream( + transformStream, + createStreamController({ clientResponseFormat: FORMATS.OPENAI_RESPONSES }) + ); + const text = await readStreamText(stream); + + assert.match(text, /event: response\.failed/); + assert.match(text, /"type":"response\.failed"/); + assert.match(text, /"message":"responses stream died"/); + assert.match(text, /"type":"server_error"/); + assert.match(text, /"code":"server_error"/); + assert.doesNotMatch(text, /chat\.completion\.chunk/); + assert.doesNotMatch(text, /"finish_reason":"error"/); + assert.doesNotMatch(text, /\[DONE\]/); +}); + +test("createDisconnectAwareStream treats legacy OpenAI response format alias as Responses", async () => { + const upstreamError = Object.assign(new Error("legacy responses alias died"), { + statusCode: 429, + }); + const transformStream = { + readable: new ReadableStream({ + start(controller) { + controller.error(upstreamError); + }, + }), + writable: { + getWriter() { + return { + abort() {}, + }; + }, + }, + }; + + const stream = createDisconnectAwareStream( + transformStream, + createStreamController({ clientResponseFormat: FORMATS.OPENAI_RESPONSE }) + ); + const text = await readStreamText(stream); + + assert.match(text, /event: response\.failed/); + assert.match(text, /"type":"rate_limit_error"/); + assert.match(text, /"code":"rate_limit_exceeded"/); + assert.doesNotMatch(text, /chat\.completion\.chunk/); + assert.doesNotMatch(text, /\[DONE\]/); +}); + +test("createDisconnectAwareStream emits Claude SSE errors for Claude clients", async () => { + const upstreamError = Object.assign(new Error("claude stream died"), { statusCode: 502 }); + const transformStream = { + readable: new ReadableStream({ + start(controller) { + controller.error(upstreamError); + }, + }), + writable: { + getWriter() { + return { + abort() {}, + }; + }, + }, + }; + + const stream = createDisconnectAwareStream( + transformStream, + createStreamController({ clientResponseFormat: FORMATS.CLAUDE }) + ); + const text = await readStreamText(stream); + + assert.match(text, /event: error/); + assert.match(text, /"type":"error"/); + assert.match(text, /"type":"api_error"/); + assert.match(text, /"message":"claude stream died"/); + assert.doesNotMatch(text, /"code"/); + assert.doesNotMatch(text, /chat\.completion\.chunk/); + assert.doesNotMatch(text, /"finish_reason":"error"/); + assert.doesNotMatch(text, /\[DONE\]/); +}); + test("createDisconnectAwareStream cancel propagates disconnect reason and aborts the writer", async () => { let aborted = false; let disconnectEvent = null;