feat(open-sse): non-streaming stream-error pattern match → 502 fallback

open-sse/handlers/chatCore/sseToJsonHandler.js

@@ -1,4 +1,5 @@
 import { convertResponsesStreamToJson } from "../../transformer/streamToJsonConverter.js";
+import { matchStreamErrorPatterns } from "../../utils/streamErrorPatterns.js";
 import { createErrorResult } from "../../utils/error.js";
 import { HTTP_STATUS } from "../../config/runtimeConfig.js";
 import { FORMATS } from "../../translator/formats.js";
@@ -108,7 +109,7 @@ export function parseSSEToOpenAIResponse(rawSSE, fallbackModel) {
  * Handle case: provider forced streaming but client wants JSON.
  * Supports both Codex/Responses API SSE and standard Chat Completions SSE.
  */
-export async function handleForcedSSEToJson({ providerResponse, sourceFormat, provider, model, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, trackDone, appendLog, reqTag, log }) {
+export async function handleForcedSSEToJson({ providerResponse, sourceFormat, provider, model, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, trackDone, appendLog, reqTag, log, streamErrorPatterns }) {
   const contentType = providerResponse.headers.get("content-type") || "";
   const isSSE = contentType.includes("text/event-stream") || (contentType === "" && isResponsesProvider(provider));
   if (!isSSE) return null; // not handled here
@@ -209,6 +210,17 @@ export async function handleForcedSSEToJson({ providerResponse, sourceFormat, pr
       );
     }
 
+    // Config-driven in-stream error detection: the request "succeeded" at the
+    // HTTP level, but the content signals an upstream failure — treat it as an
+    // error so account/combo fallback and FAILED logging kick in.
+    const matchedPattern = matchStreamErrorPatterns(
+      streamErrorPatterns?.[provider],
+      parsed.choices?.[0]?.message?.content || ""
+    );
+    if (matchedPattern) {
+      return createErrorResult(HTTP_STATUS.BAD_GATEWAY, `Stream error pattern matched: ${matchedPattern}`);
+    }
+
     if (onRequestSuccess) await onRequestSuccess();
 
     const usage = parsed.usage || {};

tests/unit/sse-error-patterns.test.js

@@ -0,0 +1,75 @@
+import { describe, it, expect } from "vitest";
+import { handleForcedSSEToJson } from "../../open-sse/handlers/chatCore/sseToJsonHandler.js";
+
+const encoder = new TextEncoder();
+const sseResponse = (chunks) => {
+  const body = new ReadableStream({
+    start(c) {
+      for (const ch of chunks) c.enqueue(encoder.encode(`data: ${JSON.stringify(ch)}\n\n`));
+      c.enqueue(encoder.encode("data: [DONE]\n\n"));
+      c.close();
+    },
+  });
+  return new Response(body, { status: 200, headers: { "content-type": "text/event-stream" } });
+};
+
+const mkChunk = (content, finish = null) => ({
+  id: "x",
+  object: "chat.completion.chunk",
+  created: 1,
+  model: "m",
+  choices: [{ index: 0, delta: content ? { content } : {}, finish_reason: finish }],
+});
+
+const baseCtx = {
+  provider: "fakeprovider",
+  model: "m",
+  body: { stream: false },
+  stream: true,
+  translatedBody: null,
+  finalBody: null,
+  requestStartTime: Date.now(),
+  connectionId: "c1",
+  apiKey: null,
+  clientRawRequest: null,
+  onRequestSuccess: null,
+  pxpipe: null,
+  reqTag: "",
+  log: null,
+  trackDone: () => {},
+  appendLog: () => {},
+  reqLogger: null,
+  toolNameMap: null,
+  sourceFormat: "openai",
+};
+
+describe("Layer 1 — non-streaming stream error patterns", () => {
+  it("returns a 502 error result when content matches a configured pattern", async () => {
+    const result = await handleForcedSSEToJson({
+      ...baseCtx,
+      streamErrorPatterns: { fakeprovider: ["Network connection lost"] },
+      providerResponse: sseResponse([mkChunk("Network connection lost."), mkChunk(null, "stop")]),
+    });
+    expect(result.success).toBe(false);
+    expect(result.status).toBe(502);
+    expect(result.error).toContain("Network connection lost");
+  });
+
+  it("succeeds when content does not match", async () => {
+    const result = await handleForcedSSEToJson({
+      ...baseCtx,
+      streamErrorPatterns: { fakeprovider: ["Network connection lost"] },
+      providerResponse: sseResponse([mkChunk("hello world"), mkChunk(null, "stop")]),
+    });
+    expect(result.success).toBe(true);
+  });
+
+  it("ignores patterns for other providers", async () => {
+    const result = await handleForcedSSEToJson({
+      ...baseCtx,
+      streamErrorPatterns: { otherprovider: ["hello"] },
+      providerResponse: sseResponse([mkChunk("hello world"), mkChunk(null, "stop")]),
+    });
+    expect(result.success).toBe(true);
+  });
+});