fix(proxy): detect Responses failed fake-200 streams

src/app/v1/_lib/proxy/forwarder.ts

@@ -1502,7 +1502,8 @@ export class ProxyForwarder {
               detected.isError &&
               (detected.code === "FAKE_200_HTML_BODY" ||
                 detected.code === "FAKE_200_JSON_ERROR_NON_EMPTY" ||
-                detected.code === "FAKE_200_JSON_ERROR_MESSAGE_NON_EMPTY");
+                detected.code === "FAKE_200_JSON_ERROR_MESSAGE_NON_EMPTY" ||
+                detected.code === "FAKE_200_OPENAI_RESPONSE_FAILED");
 
             if (isStrongFake200) {
               const inferredStatus = inferUpstreamErrorStatusCodeFromText(inspectedText);

src/lib/utils/upstream-error-detection.ts

@@ -74,12 +74,15 @@ const FAKE_200_CODES = {
   JSON_ERROR_NON_EMPTY: "FAKE_200_JSON_ERROR_NON_EMPTY",
   JSON_ERROR_MESSAGE_NON_EMPTY: "FAKE_200_JSON_ERROR_MESSAGE_NON_EMPTY",
   JSON_MESSAGE_KEYWORD_MATCH: "FAKE_200_JSON_MESSAGE_KEYWORD_MATCH",
+  OPENAI_RESPONSE_FAILED: "FAKE_200_OPENAI_RESPONSE_FAILED",
 } as const;
 
 // SSE 快速过滤：仅当文本里“看起来存在 JSON key”时才进入 parseSSEData（避免无谓解析）。
 // 注意：这里必须是 `"key"\s*:` 形式，避免误命中 JSON 字符串内容里的 `\"key\"`。
 const MAY_HAVE_JSON_ERROR_KEY = /"error"\s*:/;
 const MAY_HAVE_JSON_MESSAGE_KEY = /"message"\s*:/;
+const MAY_HAVE_OPENAI_RESPONSES_FAILED_SIGNAL =
+  /(?:event:\s*response\.failed|"type"\s*:\s*"response\.failed"|"status"\s*:\s*"failed")/;
 
 const HTML_DOC_SNIFF_MAX_CHARS = 1024;
 const HTML_DOCTYPE_RE = /^<!doctype\s+html[\s>]/i;
@@ -236,6 +239,58 @@ function hasNonEmptyValue(value: unknown): boolean {
   return true;
 }
 
+type OpenAIResponsesFailedDetection = {
+  detail?: string;
+};
+
+function detectOpenAIResponsesFailed(
+  obj: Record<string, unknown>
+): OpenAIResponsesFailedDetection | null {
+  const eventType = typeof obj.type === "string" ? obj.type.trim() : "";
+  const sseEventType = typeof obj.__sseEvent === "string" ? obj.__sseEvent.trim() : "";
+  const response = isPlainRecord(obj.response) ? obj.response : obj;
+  const responseStatus = typeof response.status === "string" ? response.status.trim() : "";
+  const responseObject = typeof response.object === "string" ? response.object.trim() : "";
+  const responseId = typeof response.id === "string" ? response.id.trim() : "";
+
+  const looksLikeOpenAIResponse =
+    sseEventType.startsWith("response.") ||
+    eventType.startsWith("response.") ||
+    responseObject === "response" ||
+    responseId.startsWith("resp_");
+  const isFailedResponse =
+    sseEventType === "response.failed" ||
+    eventType === "response.failed" ||
+    responseStatus === "failed";
+  if (!looksLikeOpenAIResponse || !isFailedResponse) {
+    return null;
+  }
+
+  const responseError = response.error;
+  if (typeof responseError === "string" && responseError.trim()) {
+    return { detail: truncateForDetail(responseError) };
+  }
+
+  if (isPlainRecord(responseError)) {
+    const message = typeof responseError.message === "string" ? responseError.message : "";
+    if (message.trim()) {
+      return { detail: truncateForDetail(message) };
+    }
+
+    const code = typeof responseError.code === "string" ? responseError.code : "";
+    if (code.trim()) {
+      return { detail: truncateForDetail(code) };
+    }
+  }
+
+  const topLevelMessage = typeof obj.message === "string" ? obj.message : "";
+  if (topLevelMessage.trim()) {
+    return { detail: truncateForDetail(topLevelMessage) };
+  }
+
+  return {};
+}
+
 export function sanitizeErrorTextForDetail(text: string): string {
   // 注意：这里的目的不是“完美脱敏”，而是尽量降低上游错误信息中意外夹带敏感内容的风险。
   // 若后续发现更多敏感模式，可在不改变检测语义的前提下补充。
@@ -280,6 +335,15 @@ function detectFromJsonObject(
   rawJsonChars: number,
   options: Required<Pick<DetectionOptions, "maxJsonCharsForMessageCheck" | "messageKeyword">>
 ): UpstreamErrorDetectionResult {
+  const openAIResponsesFailed = detectOpenAIResponsesFailed(obj);
+  if (openAIResponsesFailed !== null) {
+    return {
+      isError: true,
+      code: FAKE_200_CODES.OPENAI_RESPONSE_FAILED,
+      ...(openAIResponsesFailed.detail ? { detail: openAIResponsesFailed.detail } : {}),
+    };
+  }
+
   // 判定优先级：
   // 1) `error` 非空：直接判定为错误（强信号）
   // 2) 小体积 JSON 下，`message` 命中关键字：判定为错误（弱信号，但能覆盖部分“错误只写在 message”场景）
@@ -387,35 +451,43 @@ export function detectUpstreamErrorFromSseOrJsonText(
     return { isError: false };
   }
 
-  // 情况 2：SSE 文本。快速过滤：既无 "error"/"message" key 时跳过解析
+  // 情况 2：SSE 文本。快速过滤：既无 "error"/"message" key，也无 Responses failed 信号时跳过解析
   // 注意：这里要求 key 命中 `"key"\s*:`，尽量避免误命中 JSON 字符串内容里的 `\"error\"`。
-  if (!MAY_HAVE_JSON_ERROR_KEY.test(text) && !MAY_HAVE_JSON_MESSAGE_KEY.test(text)) {
+  if (
+    !MAY_HAVE_JSON_ERROR_KEY.test(text) &&
+    !MAY_HAVE_JSON_MESSAGE_KEY.test(text) &&
+    !MAY_HAVE_OPENAI_RESPONSES_FAILED_SIGNAL.test(text)
+  ) {
     return { isError: false };
   }
 
   // parseSSEData 会把每个事件的 data 尝试解析成对象；我们只对 object data 做结构化判定。
   const events = parseSSEData(text);
   for (const evt of events) {
     if (!isPlainRecord(evt.data)) continue;
+    const eventData =
+      typeof evt.event === "string" && evt.event.trim().length > 0
+        ? { ...evt.data, __sseEvent: evt.event.trim() }
+        : evt.data;
     // 性能优化：只有在 message 是字符串、且“看起来足够小”时才需要精确计算 JSON 字符数。
     // 对大多数 SSE 事件（message 为对象、或没有 message），无需 JSON.stringify。
     let chars = 0;
-    const errorValue = evt.data.error;
-    const messageValue = evt.data.message;
+    const errorValue = eventData.error;
+    const messageValue = eventData.message;
     if (!hasNonEmptyValue(errorValue) && typeof messageValue === "string") {
       if (messageValue.length >= merged.maxJsonCharsForMessageCheck) {
         chars = merged.maxJsonCharsForMessageCheck; // >= 阈值即可跳过 message 关键字判定
       } else {
         try {
-          chars = JSON.stringify(evt.data).length;
+          chars = JSON.stringify(eventData).length;
         } catch {
           // stringify 失败时回退为近似值（仍保持“仅小体积 JSON 才做 message 检测”的意图）
           chars = messageValue.length;
         }
       }
     }
 
-    const res = detectFromJsonObject(evt.data, chars, merged);
+    const res = detectFromJsonObject(eventData, chars, merged);
     if (res.isError) return res;
   }
 

tests/unit/proxy/proxy-forwarder-fake-200-html.test.ts

@@ -371,6 +371,78 @@ describe("ProxyForwarder - fake 200 HTML body", () => {
     expect(mocks.recordSuccess).not.toHaveBeenCalledWith(1);
   });
 
+  test("200 + application/json 的非流式 Responses failed 应视为失败并切换供应商", async () => {
+    const provider1 = createProvider({ id: 1, name: "p1", key: "k1", maxRetryAttempts: 1 });
+    const provider2 = createProvider({ id: 2, name: "p2", key: "k2", maxRetryAttempts: 1 });
+
+    const session = createSession();
+    session.requestUrl = new URL("https://example.com/v1/responses");
+    session.originalUrlPathname = "/v1/responses";
+    session.endpointPolicy = resolveEndpointPolicy("/v1/responses");
+    session.setProvider(provider1);
+
+    mocks.pickRandomProviderWithExclusion.mockResolvedValueOnce(provider2);
+
+    const doForward = vi.spyOn(ProxyForwarder as any, "doForward");
+
+    const failedResponseBody = JSON.stringify({
+      id: "resp_failed",
+      object: "response",
+      status: "failed",
+      error: {
+        type: "rate_limit_error",
+        message: "Concurrency limit exceeded for user, please retry later",
+      },
+    });
+    const okJson = JSON.stringify({
+      id: "resp_ok",
+      object: "response",
+      status: "completed",
+      output: [],
+    });
+
+    doForward.mockResolvedValueOnce(
+      new Response(failedResponseBody, {
+        status: 200,
+        headers: {
+          "content-type": "application/json; charset=utf-8",
+          "content-length": String(failedResponseBody.length),
+        },
+      })
+    );
+
+    doForward.mockResolvedValueOnce(
+      new Response(okJson, {
+        status: 200,
+        headers: {
+          "content-type": "application/json; charset=utf-8",
+          "content-length": String(okJson.length),
+        },
+      })
+    );
+
+    const response = await ProxyForwarder.send(session);
+    expect(await response.text()).toContain("resp_ok");
+
+    expect(doForward).toHaveBeenCalledTimes(2);
+    expect(doForward.mock.calls[0][1].id).toBe(1);
+    expect(doForward.mock.calls[1][1].id).toBe(2);
+
+    expect(mocks.pickRandomProviderWithExclusion).toHaveBeenCalledWith(session, [1]);
+    expect(mocks.recordFailure).toHaveBeenCalledWith(
+      1,
+      expect.objectContaining({ message: "FAKE_200_OPENAI_RESPONSE_FAILED" })
+    );
+
+    const failure = mocks.recordFailure.mock.calls[0]?.[1];
+    expect(failure).toBeInstanceOf(ProxyError);
+    expect((failure as ProxyError).statusCode).toBe(502);
+    expect((failure as ProxyError).upstreamError?.rawBody).toBe(failedResponseBody);
+    expect((failure as ProxyError).upstreamError?.rawBodyTruncated).toBe(false);
+    expect(mocks.recordSuccess).toHaveBeenCalledWith(2);
+    expect(mocks.recordSuccess).not.toHaveBeenCalledWith(1);
+  });
+
   test("假200 JSON error 命中 rate limit 关键字时，应推断为 429 并在决策链中标记为推断", async () => {
     const provider1 = createProvider({ id: 1, name: "p1", key: "k1", maxRetryAttempts: 1 });
     const provider2 = createProvider({ id: 2, name: "p2", key: "k2", maxRetryAttempts: 1 });

tests/unit/proxy/response-handler-endpoint-circuit-isolation.test.ts

@@ -274,6 +274,37 @@ function createFake200StreamResponse(errorMessage: string = "invalid api key"):
   });
 }
 
+/** Create an OpenAI Responses SSE stream that reports a terminal response.failed event. */
+function createOpenAIResponsesFailedStreamResponse(): Response {
+  const body = [
+    "event: response.failed",
+    `data: ${JSON.stringify({
+      type: "response.failed",
+      response: {
+        id: "resp_123",
+        object: "response",
+        status: "failed",
+        error: {
+          code: "rate_limit_exceeded",
+          message: "Concurrency limit exceeded for user, please retry later",
+        },
+      },
+    })}`,
+    "",
+  ].join("\n");
+  const encoder = new TextEncoder();
+  const stream = new ReadableStream<Uint8Array>({
+    start(controller) {
+      controller.enqueue(encoder.encode(body));
+      controller.close();
+    },
+  });
+  return new Response(stream, {
+    status: 200,
+    headers: { "content-type": "text/event-stream" },
+  });
+}
+
 /** Create an SSE stream that returns non-200 HTTP status with error body. */
 function createNon200StreamResponse(statusCode: number): Response {
   const body = `data: ${JSON.stringify({ error: "rate limit exceeded" })}\n\n`;
@@ -327,6 +358,7 @@ function setupCommonMocks() {
   });
   vi.mocked(updateMessageRequestDetails).mockResolvedValue(undefined);
   vi.mocked(updateMessageRequestDuration).mockResolvedValue(undefined);
+  vi.mocked(SessionManager.updateSessionUsage).mockResolvedValue(undefined);
   vi.mocked(SessionManager.storeSessionResponse).mockResolvedValue(undefined);
   vi.mocked(SessionManager.clearSessionProvider).mockResolvedValue(undefined);
   vi.mocked(RateLimitService.trackCost).mockResolvedValue(undefined);
@@ -378,6 +410,33 @@ describe("Endpoint circuit breaker isolation", () => {
     ).toBe(true);
   });
 
+  it("OpenAI Responses response.failed with HTTP 200 should be treated as provider failure", async () => {
+    const session = createSession();
+    setDeferredMeta(session, 42);
+
+    const response = createOpenAIResponsesFailedStreamResponse();
+    await ProxyResponseHandler.dispatch(session, response);
+    await drainAsyncTasks();
+
+    expect(mockRecordFailure).toHaveBeenCalledWith(
+      1,
+      expect.objectContaining({ message: "FAKE_200_OPENAI_RESPONSE_FAILED" })
+    );
+    expect(mockRecordEndpointSuccess).not.toHaveBeenCalled();
+    expect(mockRecordEndpointFailure).not.toHaveBeenCalled();
+    expect(SessionManager.clearSessionProvider).toHaveBeenCalledWith("fake-session");
+    expect(updateMessageRequestDetails).toHaveBeenCalledWith(
+      1,
+      expect.objectContaining({
+        statusCode: 502,
+        errorMessage:
+          "FAKE_200_OPENAI_RESPONSE_FAILED: Concurrency limit exceeded for user, please retry later",
+        providerId: 1,
+      })
+    );
+    expect(RateLimitService.trackCost).not.toHaveBeenCalled();
+  });
+
   it("高并发模式下，fake-200 流式错误仍应记录核心失败，但跳过 session 观测写入", async () => {
     const session = createSession();
     session.setHighConcurrencyModeEnabled(true);