diff --git a/src/app/v1/_lib/proxy/forwarder.ts b/src/app/v1/_lib/proxy/forwarder.ts index 28bc7eb3f..fb1add656 100644 --- a/src/app/v1/_lib/proxy/forwarder.ts +++ b/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); diff --git a/src/lib/utils/upstream-error-detection.ts b/src/lib/utils/upstream-error-detection.ts index 49e2be3d0..e6bea0908 100644 --- a/src/lib/utils/upstream-error-detection.ts +++ b/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 = /^]/i; @@ -236,6 +239,58 @@ function hasNonEmptyValue(value: unknown): boolean { return true; } +type OpenAIResponsesFailedDetection = { + detail?: string; +}; + +function detectOpenAIResponsesFailed( + obj: Record +): 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> ): 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,9 +451,13 @@ 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 }; } @@ -397,17 +465,21 @@ export function detectUpstreamErrorFromSseOrJsonText( 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; @@ -415,7 +487,7 @@ export function detectUpstreamErrorFromSseOrJsonText( } } - const res = detectFromJsonObject(evt.data, chars, merged); + const res = detectFromJsonObject(eventData, chars, merged); if (res.isError) return res; } diff --git a/tests/unit/proxy/proxy-forwarder-fake-200-html.test.ts b/tests/unit/proxy/proxy-forwarder-fake-200-html.test.ts index 8b57d3915..14ff039cb 100644 --- a/tests/unit/proxy/proxy-forwarder-fake-200-html.test.ts +++ b/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 }); diff --git a/tests/unit/proxy/response-handler-endpoint-circuit-isolation.test.ts b/tests/unit/proxy/response-handler-endpoint-circuit-isolation.test.ts index 05b6bfa7a..1a6835302 100644 --- a/tests/unit/proxy/response-handler-endpoint-circuit-isolation.test.ts +++ b/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({ + 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); diff --git a/tests/unit/proxy/response-handler-non200.test.ts b/tests/unit/proxy/response-handler-non200.test.ts index 74e0bc909..8c59c113b 100644 --- a/tests/unit/proxy/response-handler-non200.test.ts +++ b/tests/unit/proxy/response-handler-non200.test.ts @@ -258,6 +258,178 @@ describe("Non-200 Status Code Handling", () => { expect(result.code).toBe("FAKE_200_EMPTY_BODY"); }); + it("should detect OpenAI Responses response.failed SSE event", () => { + const sse = [ + "event: response.failed", + 'data: {"type":"response.failed","response":{"id":"resp_123","status":"failed","error":{"code":"rate_limit_exceeded","message":"Concurrency limit exceeded for user, please retry later"}}}', + "", + ].join("\n"); + + const result = detectUpstreamErrorFromSseOrJsonText(sse); + + expect(result.isError).toBe(true); + if (result.isError) { + expect(result.code).toBe("FAKE_200_OPENAI_RESPONSE_FAILED"); + expect(result.detail).toBe("Concurrency limit exceeded for user, please retry later"); + } + }); + + it("should detect OpenAI Responses failure when only the SSE event name says response.failed", () => { + const sse = [ + "event: response.failed", + 'data: {"response":{"error":{"message":"Concurrency limit exceeded for user, please retry later"}}}', + "", + ].join("\n"); + + const result = detectUpstreamErrorFromSseOrJsonText(sse); + + expect(result.isError).toBe(true); + if (result.isError) { + expect(result.code).toBe("FAKE_200_OPENAI_RESPONSE_FAILED"); + expect(result.detail).toBe("Concurrency limit exceeded for user, please retry later"); + } + }); + + it("should detect OpenAI Responses failure when SSE event name is response.failed but data type is generic", () => { + const sse = [ + "event: response.failed", + 'data: {"type":"response","response":{"error":{"message":"Concurrency limit exceeded for user, please retry later"}}}', + "", + ].join("\n"); + + const result = detectUpstreamErrorFromSseOrJsonText(sse); + + expect(result.isError).toBe(true); + if (result.isError) { + expect(result.code).toBe("FAKE_200_OPENAI_RESPONSE_FAILED"); + expect(result.detail).toBe("Concurrency limit exceeded for user, please retry later"); + } + }); + + it("should detect wrapped OpenAI Responses failed JSON object", () => { + const body = JSON.stringify({ + type: "response.failed", + response: { + id: "resp_456", + status: "failed", + error: { + code: "server_error", + message: "Upstream worker failed", + }, + }, + }); + + const result = detectUpstreamErrorFromSseOrJsonText(body); + + expect(result.isError).toBe(true); + if (result.isError) { + expect(result.code).toBe("FAKE_200_OPENAI_RESPONSE_FAILED"); + expect(result.detail).toBe("Upstream worker failed"); + } + }); + + it("should detect flat OpenAI Responses failed JSON object", () => { + const body = JSON.stringify({ + id: "resp_789", + object: "response", + status: "failed", + error: { + code: "rate_limit_exceeded", + message: "Too many concurrent requests", + }, + }); + + const result = detectUpstreamErrorFromSseOrJsonText(body); + + expect(result.isError).toBe(true); + if (result.isError) { + expect(result.code).toBe("FAKE_200_OPENAI_RESPONSE_FAILED"); + expect(result.detail).toBe("Too many concurrent requests"); + } + }); + + it("should detect OpenAI Responses failed event when error detail is not extractable", () => { + const body = JSON.stringify({ + type: "response.failed", + response: { + id: "resp_numeric_code", + status: "failed", + error: { + code: 429, + type: "rate_limit_error", + }, + }, + }); + + const result = detectUpstreamErrorFromSseOrJsonText(body); + + expect(result.isError).toBe(true); + if (result.isError) { + expect(result.code).toBe("FAKE_200_OPENAI_RESPONSE_FAILED"); + expect(result.detail).toBeUndefined(); + } + }); + + it("should detect OpenAI Responses failed event without error payload", () => { + const body = JSON.stringify({ + type: "response.failed", + response: { + id: "resp_missing_error", + status: "failed", + }, + }); + + const result = detectUpstreamErrorFromSseOrJsonText(body); + + expect(result.isError).toBe(true); + if (result.isError) { + expect(result.code).toBe("FAKE_200_OPENAI_RESPONSE_FAILED"); + expect(result.detail).toBeUndefined(); + } + }); + + it("should detect OpenAI Responses failed SSE event without error payload", () => { + const sse = [ + "event: response.failed", + 'data: {"type":"response.failed","response":{"id":"resp_sse_missing_error","status":"failed"}}', + "", + ].join("\n"); + + const result = detectUpstreamErrorFromSseOrJsonText(sse); + + expect(result.isError).toBe(true); + if (result.isError) { + expect(result.code).toBe("FAKE_200_OPENAI_RESPONSE_FAILED"); + expect(result.detail).toBeUndefined(); + } + }); + + it("should not treat successful OpenAI Responses completed event as fake 200", () => { + const sse = [ + "event: response.completed", + 'data: {"type":"response.completed","response":{"id":"resp_ok","status":"completed","usage":{"input_tokens":1,"output_tokens":1}}}', + "", + ].join("\n"); + + const result = detectUpstreamErrorFromSseOrJsonText(sse); + + expect(result.isError).toBe(false); + }); + + it("should not classify normal OpenAI chat completion SSE chunks as Responses failures", () => { + const sse = [ + 'data: {"id":"chatcmpl_123","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"hello"}}]}', + "", + 'data: {"id":"chatcmpl_123","object":"chat.completion.chunk","choices":[{"index":0,"delta":{},"finish_reason":"stop"}],"usage":{"prompt_tokens":3,"completion_tokens":1,"total_tokens":4}}', + "", + "data: [DONE]", + ].join("\n"); + + const result = detectUpstreamErrorFromSseOrJsonText(sse); + + expect(result.isError).toBe(false); + }); + it("should return isError=false for successful JSON without error field", () => { const result = detectUpstreamErrorFromSseOrJsonText( '{"choices":[{"message":{"content":"hi"}}]}'