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(); }