From c8b96a61e799838582c779afb4ae4c80d0dfbdd7 Mon Sep 17 00:00:00 2001
From: luulam <luulam0911@gmail.com>
Date: Tue, 4 Aug 2026 23:34:00 +0700
Subject: [PATCH] =?UTF-8?q?feat(open-sse):=20non-streaming=20stream-error?=
 =?UTF-8?q?=20pattern=20match=20=E2=86=92=20502=20fallback?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../handlers/chatCore/sseToJsonHandler.js     | 14 +++-
 tests/unit/sse-error-patterns.test.js         | 75 +++++++++++++++++++
 2 files changed, 88 insertions(+), 1 deletion(-)
 create mode 100644 tests/unit/sse-error-patterns.test.js

diff --git a/open-sse/handlers/chatCore/sseToJsonHandler.js b/open-sse/handlers/chatCore/sseToJsonHandler.js
index 9638824b..10eeab60 100644
--- a/open-sse/handlers/chatCore/sseToJsonHandler.js
+++ b/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 || {};
diff --git a/tests/unit/sse-error-patterns.test.js b/tests/unit/sse-error-patterns.test.js
new file mode 100644
index 00000000..59cced66
--- /dev/null
+++ b/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);
+  });
+});