fix: emit stream errors in client protocol (#2355)

Integrated into release/v3.8.0

open-sse/handlers/chatCore.ts

@@ -3009,6 +3009,7 @@ export async function handleChatCore({
     provider,
     model,
     connectionId,
+    clientResponseFormat,
   });
 
   const dedupRequestBody = { ...translatedBody, model: `${provider}/${model}`, stream };

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<typeof createStreamController>;
 
+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();
       }

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;