From 427c8c15eccd639999fed8de75d4e5e4473c8da8 Mon Sep 17 00:00:00 2001 From: jcs130 Date: Wed, 11 Nov 2026 23:27:24 +0800 Subject: [PATCH] fix(providers): preserve upstream rejection status in failed responses Failed Responses carried HTTP 200 and were reduced to status 0, so MainLoop resubmitted explicit 400 rejections three times per batch. Preserve the embedded rejection status while retaining actual attempt HTTP status, partial output and usage. Transport and MainLoop integration tests verify one request per rejected batch. --- docs/providers.md | 3 ++- src/providers/transport/errors.ts | 26 ++++++++++++++++++- src/providers/transport/response-http.ts | 7 +++-- tests/core/loop.test.ts | 21 +++++++++++++++ tests/providers/transport/errors.test.ts | 13 +++++++++- .../providers/transport/response-http.test.ts | 19 +++++++++++++- 6 files changed, 83 insertions(+), 6 deletions(-) diff --git a/docs/providers.md b/docs/providers.md index dfc259bb2..3a7a9c135 100644 --- a/docs/providers.md +++ b/docs/providers.md @@ -1,4 +1,4 @@ - + # Provider @@ -123,6 +123,7 @@ router 自动加载。 一次生成可包含多次请求尝试,重试间隔为 1s / 4s / 10s;状态 0、408、429 与 5xx 重试,401/403 先刷新一次凭证。 流里的 `error` 事件是上游自己报的失败,不重试。非 2xx 的错误消息是 `LLM API <状态码>: <上游的错误说明>`, 说明取 JSON 正文的 `error.message` 或 `message`,没有就取正文，压成一行;完整正文在 `GenerationError.body`。 +HTTP 200 内的失败终态仍可能是请求拒绝。`GenerationError.status` 保留错误对象的 4xx/5xx 状态、已知请求拒绝/限流代码，或兼容网关的 `<400>` 等消息前缀；未知错误保持 0。`ProviderAttempt.status` 始终保留真实 HTTP 状态。明确的 400 拒绝因此不会被主循环当作网络失败在本批重提，原错误正文、失败输出及用量仍可审计。 响应带 `Retry-After` 时按它退避,等待与生成中的其他等待一样可被调用方中断。 已提交不可逆增量之后不再重试。终态只接受 `completed` 与 `incomplete`:截断落成 `incomplete_details.reason`,被截断的工具调用由 Core 标成未执行。输出字符数超过 diff --git a/src/providers/transport/errors.ts b/src/providers/transport/errors.ts index d7068a69d..bc4404c19 100644 --- a/src/providers/transport/errors.ts +++ b/src/providers/transport/errors.ts @@ -12,11 +12,35 @@ export class LLMError extends Error { /** An `error` event in a stream: the upstream's own failure report, which the transport does not retry. */ export class StreamErrorEvent extends LLMError { constructor(body: string) { - super(`Stream error event: ${upstreamText(body)}`, 0, body); + super(`Stream error event: ${upstreamText(body)}`, upstreamStatus(body), body); this.name = 'StreamErrorEvent'; } } +/** A stream can carry a request rejection after its HTTP 200 headers were sent. */ +export function upstreamStatus(body: string): number { + try { + const parsed: unknown = JSON.parse(body); + if (!parsed || typeof parsed !== 'object') return 0; + const error: unknown = 'error' in parsed ? parsed.error : parsed; + if (!error || typeof error !== 'object') return 0; + for (const key of ['status', 'status_code'] as const) { + const value = (error as Record)[key]; + if (typeof value === 'number' && Number.isInteger(value) && value >= 400 && value <= 599) return value; + } + // Compatibility gateways sometimes wrap the upstream status in a message prefix. + const message = 'message' in error && typeof error.message === 'string' ? error.message : ''; + const status = /^<([45]\d{2})>\s/.exec(message); + if (status) return Number(status[1]); + const code = 'code' in error ? error.code : null; + const type = 'type' in error ? error.type : null; + if (['invalid_request_error', 'context_length_exceeded', 'content_filter', 'data_inspection_failed', 'DataInspectionFailed'].includes(String(code)) + || type === 'invalid_request_error') return 400; + if (code === 'rate_limit_exceeded' || type === 'rate_limit_error') return 429; + } catch { /* Unstructured failures keep the unknown status. */ } + return 0; +} + /** The upstream's error text: `error.message` or `message` of a JSON body, else the body, on one line. */ export function upstreamText(body: string): string { let text = body; diff --git a/src/providers/transport/response-http.ts b/src/providers/transport/response-http.ts index c867a3982..de72c707c 100644 --- a/src/providers/transport/response-http.ts +++ b/src/providers/transport/response-http.ts @@ -3,7 +3,7 @@ import type { Request, Response, StreamEvent } from '../../protocol/open-respons import type { ItemOrigin } from '../../protocol/open-responses/context.ts'; import { GenerationError, priceUsage, unknownMeters, type GenerateOptions, type Generation, type ProviderAttempt, type PriceSnapshot, type TokenMeters } from '../../core/generation.ts'; import { ResponseProtocolError } from '../../protocol/open-responses/stream.ts'; -import { LLMError, StreamErrorEvent, abortError, parseRetryAfter, retryDelay, reqIdSuffix, upstreamText } from './errors.ts'; +import { LLMError, StreamErrorEvent, abortError, parseRetryAfter, retryDelay, reqIdSuffix, upstreamText, upstreamStatus } from './errors.ts'; import type { ResponseAssembly } from './response-assembly.ts'; export interface ResponseTransport { @@ -197,7 +197,10 @@ export async function generate(request: Request, options: GenerateOptions, origi attempt.serviceTier = parsed.serviceTier; } attempt.responseId = partial.id; - if (partial.status === 'failed') throw new LLMError(partial.error?.message ?? partial.error?.code ?? 'Response failed', 0, JSON.stringify(partial.error)); + if (partial.status === 'failed') { + const body = JSON.stringify(partial.error); + throw new LLMError(partial.error?.message ?? partial.error?.code ?? 'Response failed', upstreamStatus(body), body); + } if (partial.status !== 'completed' && partial.status !== 'incomplete') throw new LLMError(`Response has no terminal status: ${partial.status}`, 0, ''); attempt.outcome = options.signal?.aborted ? 'discarded' : partial.status; transport.succeeded(); diff --git a/tests/core/loop.test.ts b/tests/core/loop.test.ts index 8d75b29e4..3ba097177 100644 --- a/tests/core/loop.test.ts +++ b/tests/core/loop.test.ts @@ -3657,6 +3657,27 @@ describe("MainLoop 重新请求、轮次边界与统计", () => { assertPairing(rig.session.messages); }); + it('does not resubmit a terminal provider rejection carried by HTTP 200', async () => { + let requests = 0; + const response = { id: 'rejected-request', model: 'test', status: 'failed', output: [], + error: { code: 'server_error', message: '<400> InternalError.Algo.DataInspectionFailed: Input text data may contain inappropriate content.' }, usage: null }; + vi.stubGlobal('fetch', async () => { + requests++; + return new Response('data: ' + JSON.stringify({ type: 'response.created', sequence_number: 0, response: { ...response, status: 'in_progress', error: null } }) + '\n\n' + + 'data: ' + JSON.stringify({ type: 'response.failed', sequence_number: 1, response }) + '\n\n', { status: 200, headers: { 'content-type': 'text/event-stream' } }); + }); + try { + let ended = false; + const provider = new ResponsesProvider({ baseUrl: 'https://fixture.test/v1', apiKey: 'fixture' }); + rig = makeRig({ llm: { respond: (request, options) => provider.respond(request, { ...options, diagnostic: true }) }, + hooks: { onTurnEnded: () => { ended = true; } } }); + rig.start(); + await until(() => ended); + expect(requests).toBe(1); + expect(rig.session.messages.some(m => m.role === 'tool')).toBe(false); + } finally { vi.unstubAllGlobals(); } + }); + it("重新请求前投递退避期间已就绪的事件", async () => { rig = makeRig({ resubmit: { maxConsecutive: 2, maxPerBatch: 4, backoffMs: [120] } }); rig.start(); diff --git a/tests/providers/transport/errors.test.ts b/tests/providers/transport/errors.test.ts index 3bec09a7f..42b340433 100644 --- a/tests/providers/transport/errors.test.ts +++ b/tests/providers/transport/errors.test.ts @@ -1,8 +1,19 @@ import { afterEach, describe, expect, it, vi } from 'vitest'; -import { parseRetryAfter } from '../../../src/providers/transport/errors.ts'; +import { parseRetryAfter, upstreamStatus } from '../../../src/providers/transport/errors.ts'; afterEach(() => vi.useRealTimers()); +it('uses explicit upstream statuses while leaving ambiguous failures unknown', () => { + expect(upstreamStatus(JSON.stringify({ error: { status: 422, message: 'bad input' } }))).toBe(422); + expect(upstreamStatus(JSON.stringify({ code: 'server_error', message: '<429> overloaded' }))).toBe(429); + expect(upstreamStatus(JSON.stringify({ code: 'server_error', message: '<503> unavailable' }))).toBe(503); + expect(upstreamStatus(JSON.stringify({ code: 'invalid_request_error', message: 'invalid argument' }))).toBe(400); + expect(upstreamStatus(JSON.stringify({ code: 'rate_limit_exceeded' }))).toBe(429); + expect(upstreamStatus(JSON.stringify({ code: 'server_error', message: '400 characters received' }))).toBe(0); + expect(upstreamStatus(JSON.stringify({ status: 200, message: 'unknown failure' }))).toBe(0); + expect(upstreamStatus('network failure')).toBe(0); +}); + describe('parseRetryAfter', () => { it('数字按秒,HTTP 日期按距今,缺失与不成形为 null', () => { vi.useFakeTimers(); diff --git a/tests/providers/transport/response-http.test.ts b/tests/providers/transport/response-http.test.ts index a4ccc6fd3..66c6d3f51 100644 --- a/tests/providers/transport/response-http.test.ts +++ b/tests/providers/transport/response-http.test.ts @@ -13,6 +13,23 @@ import { ResponsesProvider } from '../../../src/providers/openai-responses-compa afterEach(() => { vi.unstubAllGlobals(); vi.useRealTimers(); }); +it.each([true, false])('preserves a wrapped rejection without rewriting HTTP metering (stream=%s)', async streaming => { + const response = { id: 'rejection', model: 'test', status: 'failed', output: [], + error: { code: 'server_error', message: '<400> InternalError.Algo.DataInspectionFailed: Input text data may contain inappropriate content.' }, usage: null }; + const fetcher = vi.fn(async () => new Response(streaming + ? 'data: ' + JSON.stringify({ type: 'response.created', sequence_number: 0, response: { ...response, status: 'in_progress', error: null } }) + '\n\n' + + 'data: ' + JSON.stringify({ type: 'response.failed', sequence_number: 1, response }) + '\n\n' + : JSON.stringify(response), { status: 200, headers: { 'content-type': streaming ? 'text/event-stream' : 'application/json' } })); + vi.stubGlobal('fetch', fetcher); + const provider = new ResponsesProvider({ baseUrl: 'https://fixture.test/v1', apiKey: 'fixture' }); + const error = await provider.respond({ model: 'test', input: 'hi' }, streaming ? { onEvent: () => {} } : {}).catch(e => e); + expect(error).toBeInstanceOf(GenerationError); + expect(error).toMatchObject({ status: 400, body: JSON.stringify(response.error), partial: response }); + expect(error.attempts).toHaveLength(1); + expect(error.attempts[0]).toMatchObject({ status: 200, outcome: 'failed', meters: { input: null, output: null } }); + expect(fetcher).toHaveBeenCalledTimes(1); +}); + class Fixture extends OpenAIHttpClient { constructor(log: Logger = nullLogger()) { super('https://fixture.test', log); } protected buildBody(spec: ModelSpec, messages: NativeChatMessage[], tools?: ToolSchema[]): Record { @@ -122,7 +139,7 @@ data: ${JSON.stringify(event)} try { const failure = await stream(await server.port); expect(failure).toBeInstanceOf(GenerationError); - expect(failure).toMatchObject({ status: 0, body: JSON.stringify(event) }); + expect(failure).toMatchObject({ status: 400, body: JSON.stringify(event) }); expect((failure as Error).message).toContain(said); expect(server.requests()).toBe(1); } finally { server.close(); }