From 96b629165b500464949c5ebce9c8d091a99821dc Mon Sep 17 00:00:00 2001 From: RainbowBird Date: Mon, 7 Sep 2026 00:08:47 +0800 Subject: [PATCH] fix(server): pass through final upstream errors (#2333) Signed-off-by: RainbowBird --- .agents/skills/create-pr/SKILL.md | 1 + AGENTS.md | 1 + server/AGENTS.md | 10 + .../routes/openai/v1/http/response.test.ts | 22 ++ .../api/src/routes/openai/v1/http/response.ts | 20 ++ .../v1/operations/chat-completions/index.ts | 15 +- .../adapters/tts/dashscope-cosyvoice.test.ts | 20 +- .../src/services/adapters/tts/index.test.ts | 67 +++++-- .../api/src/services/adapters/tts/types.ts | 18 +- .../api/src/services/adapters/tts/unspeech.ts | 8 +- .../src/services/domain/llm-router/router.ts | 189 ++++++++++++------ .../domain/llm-router/tests/router.test.ts | 166 +++++++++++---- .../services/domain/openai-speech/index.ts | 16 +- 13 files changed, 417 insertions(+), 136 deletions(-) create mode 100644 server/apps/api/src/routes/openai/v1/http/response.test.ts diff --git a/.agents/skills/create-pr/SKILL.md b/.agents/skills/create-pr/SKILL.md index 068264d21..ac0e73a5e 100644 --- a/.agents/skills/create-pr/SKILL.md +++ b/.agents/skills/create-pr/SKILL.md @@ -15,6 +15,7 @@ Create a reviewable PR from the exact commits intended for publication. 4. Publish the intended commits through the available GitHub/`gh` workflow. 5. Compose the PR body with a concise summary, exact verification commands, and the required visual table. 6. Create the PR, then open it and verify its title, base/head branches, body, and embedded images. +7. Get the PR review threads, comments, and check status. If a comment identifies a confirmed error, fix it, run focused checks, push the update, reply with evidence, and resolve the thread. ## Visual Evidence Workflow diff --git a/AGENTS.md b/AGENTS.md index 7b140e33c..a318617e0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -256,6 +256,7 @@ as a first language. ## PR / Workflow Tips - When asked to create, open, publish, or prepare a pull request, always use the repo-local `create-pr` skill. For user-visible changes it orchestrates `use-vishot` and the matching runtime variant, then uploads before/after screenshots as GitHub user assets in the PR body. +- After you create a pull request, get its review threads, comments, and check status. If a review identifies a confirmed error, fix it, run focused checks, push the update, reply with evidence, and resolve the thread. - Rebase pulls; branch naming `username/feat/short-name`; clear commit messages (gitmoji is prohibited). - Summarize changes, how tested (commands), and follow-ups. - Improve legacy you touch; avoid one-off patterns. diff --git a/server/AGENTS.md b/server/AGENTS.md index 4e7ea17e5..3f3cd1984 100644 --- a/server/AGENTS.md +++ b/server/AGENTS.md @@ -9,3 +9,13 @@ - Use `safeParse` if the caller branches on valid and invalid data. - Do not use `typeof`, `Record`, or type casts as runtime input validation. - Infer TypeScript types from Valibot schemas. Do not duplicate the contract in an interface. + +## Architecture decisions + +- Put server ADRs in `../airi-docs/adr/`. +- Create or update the ADR before you change a server boundary. +- A server boundary includes an HTTP contract, provider contract, persistence model, or cross-module lifecycle. +- Add a module dependency graph, an affected-file tree, and a sequence diagram to every implementation ADR. +- State the decision, scope, non-goals, and test plan in the ADR. +- Keep the ADR status `accepted` only after the decision is confirmed. +- Update the ADR when implementation changes the accepted design. diff --git a/server/apps/api/src/routes/openai/v1/http/response.test.ts b/server/apps/api/src/routes/openai/v1/http/response.test.ts new file mode 100644 index 000000000..34c0f932e --- /dev/null +++ b/server/apps/api/src/routes/openai/v1/http/response.test.ts @@ -0,0 +1,22 @@ +import { describe, expect, it } from 'vitest' + +import { buildSafeErrorResponseHeaders } from './response' + +describe('buildSafeErrorResponseHeaders', () => { + it('keeps the provider error content type and retry delay only', () => { + const headers = buildSafeErrorResponseHeaders(new Response('provider error', { + status: 429, + headers: { + 'content-type': 'application/problem+json', + 'retry-after': '30', + 'set-cookie': 'provider-session=secret', + 'x-request-id': 'provider-request-id', + }, + })) + + expect(headers.get('content-type')).toBe('application/problem+json') + expect(headers.get('retry-after')).toBe('30') + expect(headers.get('set-cookie')).toBeNull() + expect(headers.get('x-request-id')).toBeNull() + }) +}) diff --git a/server/apps/api/src/routes/openai/v1/http/response.ts b/server/apps/api/src/routes/openai/v1/http/response.ts index 668a39f54..ffd357b3b 100644 --- a/server/apps/api/src/routes/openai/v1/http/response.ts +++ b/server/apps/api/src/routes/openai/v1/http/response.ts @@ -5,6 +5,11 @@ const SAFE_RESPONSE_HEADERS = new Set([ 'cache-control', ]) +const SAFE_ERROR_RESPONSE_HEADERS = new Set([ + 'content-type', + 'retry-after', +]) + export function buildSafeResponseHeaders(response: Response): Headers { const headers = new Headers() response.headers.forEach((value, key) => { @@ -13,3 +18,18 @@ export function buildSafeResponseHeaders(response: Response): Headers { }) return headers } + +/** + * Builds the headers for a final upstream error response. + * + * Upstream errors can expose useful body types and retry delays. Other + * upstream headers can contain credentials, cookies, or provider internals. + */ +export function buildSafeErrorResponseHeaders(response: Response): Headers { + const headers = new Headers() + response.headers.forEach((value, key) => { + if (SAFE_ERROR_RESPONSE_HEADERS.has(key.toLowerCase())) + headers.set(key, value) + }) + return headers +} diff --git a/server/apps/api/src/routes/openai/v1/operations/chat-completions/index.ts b/server/apps/api/src/routes/openai/v1/operations/chat-completions/index.ts index 38dffcce7..559df1dfa 100644 --- a/server/apps/api/src/routes/openai/v1/operations/chat-completions/index.ts +++ b/server/apps/api/src/routes/openai/v1/operations/chat-completions/index.ts @@ -9,7 +9,7 @@ import { useLogger } from '@guiiai/logg' import { extractUsageFromBody } from '../../../../../services/domain/billing/billing' import { createBadRequestError } from '../../../../../utils/error' import { nanoid } from '../../../../../utils/id' -import { buildSafeResponseHeaders } from '../../http/response' +import { buildSafeErrorResponseHeaders, buildSafeResponseHeaders } from '../../http/response' import { createOpenAiRouteBilling } from '../../middlewares/billing' import { createRouteTelemetry, newRouteContext } from '../../middlewares/telemetry' @@ -143,7 +143,7 @@ export function chatCompletions(deps: V1RouteDeps): GatewayCallback<'chat.comple return new Response(response.body, { status: response.status, - headers: buildSafeResponseHeaders(response), + headers: buildSafeErrorResponseHeaders(response), }) } @@ -259,7 +259,8 @@ async function routeChatAliasCandidates(input: { routeCtx: ReturnType }> { let lastError: unknown - for (const modelId of input.modelIds) { + for (let index = 0; index < input.modelIds.length; index += 1) { + const modelId = input.modelIds[index] const routeCtx = newRouteContext() try { const response = await input.deps.llmRouter.route({ @@ -268,7 +269,13 @@ async function routeChatAliasCandidates(input: { headers: {}, abortSignal: input.abortSignal, }, routeCtx) - return { modelId, response, routeCtx } + if (response.ok || index === input.modelIds.length - 1) + return { modelId, response, routeCtx } + + // The alias owns the next configured model candidate. Its non-2xx body + // cannot reach the client while a later candidate can still serve the + // request, so release it before the next route attempt. + await response.body?.cancel().catch(() => {}) } catch (err) { if (input.abortSignal?.aborted) diff --git a/server/apps/api/src/services/adapters/tts/dashscope-cosyvoice.test.ts b/server/apps/api/src/services/adapters/tts/dashscope-cosyvoice.test.ts index 64e6929c9..c31529e8e 100644 --- a/server/apps/api/src/services/adapters/tts/dashscope-cosyvoice.test.ts +++ b/server/apps/api/src/services/adapters/tts/dashscope-cosyvoice.test.ts @@ -3,6 +3,7 @@ import { Buffer } from 'node:buffer' import { describe, expect, it, vi } from 'vitest' import { dashscopeCosyvoiceAdapter } from './dashscope-cosyvoice' +import { TtsUpstreamResponseError } from './types' const UNSPEECH = 'http://unspeech.local:5933' const SPEECH_URL = `${UNSPEECH}/v1/audio/speech` @@ -52,11 +53,12 @@ describe('dashscopeCosyvoiceAdapter', () => { expect(Array.from(out)).toEqual(Array.from(audioBytes)) }) - it('throws Error with .status when unspeech returns non-2xx (router walks to next key)', async () => { + it('throws TtsUpstreamResponseError when unspeech returns non-2xx', async () => { const fetchImpl = vi.fn().mockResolvedValueOnce(new Response('bad key', { status: 401 })) - await expect( - dashscopeCosyvoiceAdapter.send( + let caught: unknown + try { + await dashscopeCosyvoiceAdapter.send( { text: 'hi', voice: 'longxiaochun_v2' }, { keyPlaintext: Buffer.from('sk-test', 'utf8'), @@ -65,8 +67,16 @@ describe('dashscopeCosyvoiceAdapter', () => { adapterParams: {}, fetchImpl: fetchImpl as unknown as typeof fetch, }, - ), - ).rejects.toMatchObject({ status: 401, message: expect.stringContaining('401') }) + ) + } + catch (error) { + caught = error + } + + expect(caught).toBeInstanceOf(TtsUpstreamResponseError) + if (!(caught instanceof TtsUpstreamResponseError)) + throw caught + expect(caught.response.status).toBe(401) expect(fetchImpl).toHaveBeenCalledTimes(1) }) diff --git a/server/apps/api/src/services/adapters/tts/index.test.ts b/server/apps/api/src/services/adapters/tts/index.test.ts index 0d0aa59b0..3c4fbce37 100644 --- a/server/apps/api/src/services/adapters/tts/index.test.ts +++ b/server/apps/api/src/services/adapters/tts/index.test.ts @@ -4,6 +4,7 @@ import { describe, expect, it, vi } from 'vitest' import { ApiError } from '../../../utils/error' import { getAdapter } from './index' +import { TtsUpstreamResponseError } from './types' describe('getAdapter', () => { it('returns the azure adapter by id', () => { @@ -280,20 +281,31 @@ describe('azureAdapter.send', () => { expect(fetchImpl).not.toHaveBeenCalled() }) - it('throws Error with .status when unspeech non-2xx', async () => { + it('throws TtsUpstreamResponseError when unspeech returns non-2xx', async () => { const adapter = getAdapter('azure') const fetchImpl = vi.fn(async () => new Response('upstream rejected', { status: 401 })) as unknown as typeof fetch - await expect(adapter.send( - { text: 'hi', voice: 'en-US-AvaMultilingualNeural' }, - { - keyPlaintext: Buffer.from('k', 'utf8'), - baseURL: 'https://eastasia.tts.speech.microsoft.com/cognitiveservices/v1', - unspeechBaseURL: 'http://unspeech.local:5933', - adapterParams: { region: 'eastasia' }, - fetchImpl, - }, - )).rejects.toMatchObject({ status: 401 }) + let caught: unknown + try { + await adapter.send( + { text: 'hi', voice: 'en-US-AvaMultilingualNeural' }, + { + keyPlaintext: Buffer.from('k', 'utf8'), + baseURL: 'https://eastasia.tts.speech.microsoft.com/cognitiveservices/v1', + unspeechBaseURL: 'http://unspeech.local:5933', + adapterParams: { region: 'eastasia' }, + fetchImpl, + }, + ) + } + catch (error) { + caught = error + } + + expect(caught).toBeInstanceOf(TtsUpstreamResponseError) + if (!(caught instanceof TtsUpstreamResponseError)) + throw caught + expect(caught.response.status).toBe(401) }) }) @@ -459,20 +471,31 @@ describe('stepfunAdapter', () => { expect(body.extra_body.voice_label).toEqual({ emotion: '高兴' }) }) - it('throws Error with .status when unspeech returns non-2xx', async () => { + it('throws TtsUpstreamResponseError when unspeech returns non-2xx', async () => { const adapter = getAdapter('stepfun') const fetchImpl = vi.fn(async () => new Response('bad key', { status: 401 })) as unknown as typeof fetch - await expect(adapter.send( - { text: 'hi', voice: 'cixingnansheng' }, - { - keyPlaintext: Buffer.from('bad-key', 'utf8'), - baseURL: 'https://api.stepfun.com', - unspeechBaseURL: 'http://unspeech.local', - adapterParams: { model: 'stepaudio-2.5-tts' }, - fetchImpl, - }, - )).rejects.toMatchObject({ status: 401 }) + let caught: unknown + try { + await adapter.send( + { text: 'hi', voice: 'cixingnansheng' }, + { + keyPlaintext: Buffer.from('bad-key', 'utf8'), + baseURL: 'https://api.stepfun.com', + unspeechBaseURL: 'http://unspeech.local', + adapterParams: { model: 'stepaudio-2.5-tts' }, + fetchImpl, + }, + ) + } + catch (error) { + caught = error + } + + expect(caught).toBeInstanceOf(TtsUpstreamResponseError) + if (!(caught instanceof TtsUpstreamResponseError)) + throw caught + expect(caught.response.status).toBe(401) }) it('preserves an unspeech request abort for router timeout classification', async () => { diff --git a/server/apps/api/src/services/adapters/tts/types.ts b/server/apps/api/src/services/adapters/tts/types.ts index 1c52ef44c..81f36a902 100644 --- a/server/apps/api/src/services/adapters/tts/types.ts +++ b/server/apps/api/src/services/adapters/tts/types.ts @@ -70,6 +70,19 @@ export interface TtsResult { body: ArrayBuffer | ReadableStream } +/** + * A non-2xx response from the immediate TTS upstream. + * + * The router uses this response for configured fallback. It returns the final + * response to the client when every permitted fallback has failed. + */ +export class TtsUpstreamResponseError extends Error { + constructor(readonly response: Response) { + super(`TTS upstream responded with ${response.status}`) + this.name = 'TtsUpstreamResponseError' + } +} + /** * Stable provider identifier for the v1 adapter registry. * @@ -120,9 +133,8 @@ export interface TtsVoiceCatalogContext { * * Returns: * - A {@link TtsResult} on 2xx upstream responses. - * - Throws (Error subclass) on upstream non-2xx — the router maps the error to - * the next fallback key/upstream or to a 5xx for the caller. Adapters MUST - * NOT swallow upstream failures. + * - Throws {@link TtsUpstreamResponseError} on upstream non-2xx. The router + * can use the response for a fallback or return it to the caller. */ export interface TtsAdapter { /** Stable id used by the registry and config (`tts.upstreams[i].adapter`). */ diff --git a/server/apps/api/src/services/adapters/tts/unspeech.ts b/server/apps/api/src/services/adapters/tts/unspeech.ts index 9d8117262..f7f631ce6 100644 --- a/server/apps/api/src/services/adapters/tts/unspeech.ts +++ b/server/apps/api/src/services/adapters/tts/unspeech.ts @@ -6,6 +6,7 @@ import { errorMessageFrom } from '@moeru/std' import { generateSpeechResponse, listVoices, UnSpeechAPIError } from 'unspeech' import { createBadGatewayError, createInternalError } from '../../../utils/error' +import { TtsUpstreamResponseError } from './types' interface SendSpeechOptions { ctx: TtsAdapterContext @@ -72,9 +73,10 @@ export async function sendSpeechViaUnSpeech(options: SendSpeechOptions): Promise throw error if (error instanceof UnSpeechAPIError) { - const err = new Error(`${providerLabel} tts upstream ${error.status}: ${error.responseBody.slice(0, 256)}`) as Error & { status?: number } - err.status = error.status - throw err + throw new TtsUpstreamResponseError(new Response(error.responseBody, { + status: error.status, + headers: error.responseHeaders, + })) } throw createInternalError(`${providerLabel} tts fetch failed: ${errorMessageFrom(error) ?? 'unknown'}`) diff --git a/server/apps/api/src/services/domain/llm-router/router.ts b/server/apps/api/src/services/domain/llm-router/router.ts index 96461015b..df84ab99f 100644 --- a/server/apps/api/src/services/domain/llm-router/router.ts +++ b/server/apps/api/src/services/domain/llm-router/router.ts @@ -8,6 +8,7 @@ import type { EnvelopeCrypto } from '../../../utils/envelope-crypto' import type { ConfigKVService } from '../../adapters/config-kv' import type { TtsAdapterId, TtsInput } from '../../adapters/tts/types' import type { ConcurrencyLedger } from './concurrency-ledger' +import type { UpstreamAttempt } from './error-mapping' import type { LlmRouteContext, LlmRouteRequest, LlmRoutingGroup, LlmUpstream, RouteFailureTriggers, TtsRoutingGroup, TtsUpstream } from './types' import { Buffer as NodeBuffer } from 'node:buffer' @@ -24,12 +25,29 @@ import { AIRI_ATTR_GEN_AI_GATEWAY_UPSTREAM_URL, } from '../../../utils/observability' import { getAdapter } from '../../adapters/tts' +import { TtsUpstreamResponseError } from '../../adapters/tts/types' import { createConfigLoader } from './config-loader' import { mapUpstreamError } from './error-mapping' import { createKeyRotator } from './key-rotator' const UPSTREAM_BODY_SNIPPET_MAX = 256 +interface HttpAttemptFailure { + keyId: string + status: number | 'timeout' + bodySnippet?: string + errorMessage?: string + response?: Response +} + +function toDiagnosticAttempt({ provider, keyId, status, bodySnippet, errorMessage }: HttpAttemptFailure & { provider: string }): UpstreamAttempt { + return { provider, keyId, status, bodySnippet, errorMessage } +} + +async function discardUpstreamResponse(response: Response | undefined): Promise { + await response?.body?.cancel().catch(() => {}) +} + /** * Read at most `maxBytes` from an upstream non-2xx response body for * diagnostic logging, then cancel the rest so the socket can return to @@ -214,8 +232,8 @@ function ttsVoicesCacheKey(provider: string, modelName: string): string { * Returns: * - `route(req)` — picks an upstream + key, fetches the upstream, walks * fallback on non-2xx until one succeeds or every (upstream, key) has - * been tried. Returns a `Response` on the first 2xx; throws `ApiError` - * per KTD-1 mapping on full exhaustion. + * been tried. Returns a `Response` on the first 2xx or terminal upstream + * HTTP error. Throws `ApiError` when every attempt fails before a response. * * The router does NOT open its own OTel span — the route handler in U4 * owns the span. The router only enriches the *active* span with @@ -243,15 +261,15 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { req: LlmRouteRequest, perAttemptTimeoutMs: number, fallbackHttpCodes: number[], - onAttemptFailure: (failure: { keyId: string, status: number | 'timeout', bodySnippet?: string, errorMessage?: string }) => void, + onAttemptFailure: (failure: HttpAttemptFailure) => void, ): Promise< | { kind: 'ok', response: Response, attemptIndex: number, upstreamModel: string } - | { kind: 'exhausted', failures: Array<{ keyId: string, status: number | 'timeout', bodySnippet?: string, errorMessage?: string }> } + | { kind: 'exhausted', failures: HttpAttemptFailure[] } > { const provider = deriveProviderTag(upstream.baseURL) const rotator = createKeyRotator(upstream, options.envelopeCrypto, req.modelName, options.gatewayMetrics, provider) - const failures: Array<{ keyId: string, status: number | 'timeout', bodySnippet?: string, errorMessage?: string }> = [] + const failures: HttpAttemptFailure[] = [] let attemptIndex = 0 for (const key of rotator) { @@ -299,6 +317,7 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { } if (response.ok) { + await Promise.all(failures.map(failure => discardUpstreamResponse(failure.response))) // First 2xx wins. Enrich the active span and return. trace.getActiveSpan()?.setAttributes({ [AIRI_ATTR_GEN_AI_GATEWAY_UPSTREAM_URL]: upstream.baseURL, @@ -311,17 +330,15 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { const status = response.status // NOTICE: - // Drain at most UPSTREAM_BODY_SNIPPET_MAX bytes of the failed body - // for diagnostic logging (operators need to see the upstream's real - // error, not just the status code), then cancel the rest so the - // socket can return to the pool. Without the cancel, a 401/429/5xx - // fallback storm leaves half-read bodies in flight and exhausts the - // connection pool exactly when the upstream is sick. + // Read a cloned response for diagnostic logging. The original response + // stays unread because it can become the final client response. When a + // later fallback wins, this router cancels the discarded response. // Source: codex review 2026-05-15 HIGH #2 (cancel) + cause-propagation // follow-up 2026-05-16 (snippet). - const bodySnippet = await readUpstreamBodySnippet(response) - failures.push({ keyId: key.id, status, bodySnippet }) - onAttemptFailure({ keyId: key.id, status, bodySnippet }) + const bodySnippet = await readUpstreamBodySnippet(response.clone()) + const failure = { keyId: key.id, status, bodySnippet, response } + failures.push(failure) + onAttemptFailure(failure) options.gatewayMetrics?.fallbackCount.add(1, { provider, from_key: key.id, @@ -345,6 +362,7 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { // timeout. The router does NOT fall back on caller-abort: there is // no longer a client waiting for a response. if (req.abortSignal?.aborted) { + await Promise.all(failures.map(failure => discardUpstreamResponse(failure.response))) logger.withError(err).withFields({ keyId: key.id }).debug('Caller aborted upstream fetch; propagating without fallback') throw err } @@ -355,8 +373,9 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { // ApiError.cause so operators can tell apart "DNS failed" from // "attempt-timeout" without re-running the request. const errorMessage = errorMessageFromUnknown(err) - failures.push({ keyId: key.id, status: 'timeout', errorMessage }) - onAttemptFailure({ keyId: key.id, status: 'timeout', errorMessage }) + const failure = { keyId: key.id, status: 'timeout' as const, errorMessage } + failures.push(failure) + onAttemptFailure(failure) options.gatewayMetrics?.fallbackCount.add(1, { provider, from_key: key.id, @@ -372,6 +391,7 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { attemptIndex += 1 } + await Promise.all(failures.slice(0, -1).map(failure => discardUpstreamResponse(failure.response))) return { kind: 'exhausted', failures } } @@ -392,8 +412,9 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { const defaults = slice.defaults ?? { perAttemptTimeoutMs: 30000, fullChainTimeoutMs: 60000, fallbackHttpCodes: [401, 402, 403, 429, 500, 502, 503, 504] } const fallbackHttpCodes = llmModel.fallbackTriggers?.httpCodes ?? defaults.fallbackHttpCodes ?? [401, 402, 403, 429, 500, 502, 503, 504] - const allFailures: Array<{ provider: string, keyId: string, status: number | 'timeout', bodySnippet?: string, errorMessage?: string }> = [] + const allFailures: Array = [] let triedUpstreams = 0 + let terminalResponse: Response | undefined async function attemptUpstream(upstream: LlmUpstream, index: number) { const provider = deriveProviderTag(upstream.baseURL) @@ -424,12 +445,13 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { return { kind: 'exhausted' as const, statuses: result.failures.map(failure => failure.status), + response: result.failures.at(-1)?.response, } } async function routeGroup(group: LlmRoutingGroup): Promise< | { kind: 'ok', response: Response } - | { kind: 'exhausted', statuses: Array, transitionBlocked: boolean } + | { kind: 'exhausted', statuses: Array, transitionBlocked: boolean, response?: Response } > { const statuses: Array = [] for (let groupCandidateIndex = 0; groupCandidateIndex < group.upstreamIds.length; groupCandidateIndex += 1) { @@ -447,10 +469,12 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { statuses.push(...result.statuses) const hasNextCandidate = groupCandidateIndex < group.upstreamIds.length - 1 if (hasNextCandidate && !failuresMatch(result.statuses, group.retryOn)) - return { kind: 'exhausted', statuses, transitionBlocked: true } + return { kind: 'exhausted', statuses, transitionBlocked: true, response: result.response } + if (hasNextCandidate) + await discardUpstreamResponse(result.response) } - return { kind: 'exhausted', statuses, transitionBlocked: false } + return { kind: 'exhausted', statuses, transitionBlocked: false, response: allFailures.at(-1)?.response } } if (llmModel.routing != null) { @@ -466,8 +490,11 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { || !hasNextGroup || !failuresMatch(result.statuses, group.continueOn) ) { + if (result.response != null) + terminalResponse = result.response break } + await discardUpstreamResponse(result.response) } } else { @@ -475,6 +502,10 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { const result = await attemptUpstream(llmModel.upstreams[index], index) if (result.kind === 'ok') return result.response + if (index < llmModel.upstreams.length - 1) + await discardUpstreamResponse(result.response) + else if (result.response != null) + terminalResponse = result.response } } @@ -512,6 +543,9 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { } } + if (terminalResponse != null) + return terminalResponse + throw mapUpstreamError( lastFailure.status, { @@ -519,15 +553,15 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { triedUpstreams, lastStatusCode: lastFailure.status, }, - allFailures, + allFailures.map(toDiagnosticAttempt), ) } /** * Run one TTS upstream's key list in order, parallel to {@link dispatchOneUpstream} * but delegating actual HTTP to the provider adapter. Adapters surface - * upstream non-2xx as `Error & { status: number }`; network failures / - * timeouts arrive as plain `Error` with no status. + * upstream non-2xx as {@link TtsUpstreamResponseError}; network failures + * and timeouts arrive as errors without an upstream response. */ async function dispatchOneTtsUpstream( upstream: TtsUpstream, @@ -539,15 +573,15 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { perAttemptTimeoutMs: number, fallbackHttpCodes: number[], unspeechBaseURL: string, - onAttemptFailure: (failure: { keyId: string, status: number | 'timeout', errorMessage?: string }) => void, + onAttemptFailure: (failure: HttpAttemptFailure) => void, ): Promise< | { kind: 'ok', contentType: string, body: ArrayBuffer | ReadableStream, attemptIndex: number } - | { kind: 'exhausted', failures: Array<{ keyId: string, status: number | 'timeout', errorMessage?: string }> } + | { kind: 'exhausted', failures: HttpAttemptFailure[] } > { const providerTag = deriveProviderTag(upstream.baseURL) const rotator = createKeyRotator(upstream, options.envelopeCrypto, modelName, options.gatewayMetrics, providerTag) const adapter = getAdapter(providerId) - const failures: Array<{ keyId: string, status: number | 'timeout', errorMessage?: string }> = [] + const failures: HttpAttemptFailure[] = [] let attemptIndex = 0 for (const key of rotator) { @@ -587,14 +621,38 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { [AIRI_ATTR_GEN_AI_GATEWAY_KEY_ID]: key.id, [AIRI_ATTR_GEN_AI_GATEWAY_FALLBACK_DEPTH]: attemptIndex, }) + await Promise.all(failures.map(failure => discardUpstreamResponse(failure.response))) return { kind: 'ok', contentType: result.contentType, body: result.body, attemptIndex } } catch (err) { if (abortSignal?.aborted) { + await Promise.all(failures.map(failure => discardUpstreamResponse(failure.response))) logger.withError(err).withFields({ keyId: key.id }).debug('Caller aborted upstream tts fetch; propagating without fallback') throw err } + if (err instanceof TtsUpstreamResponseError) { + const failure = { keyId: key.id, status: err.response.status, response: err.response } + failures.push(failure) + onAttemptFailure(failure) + options.gatewayMetrics?.fallbackCount.add(1, { + provider: providerTag, + from_key: key.id, + reason: String(failure.status), + }) + options.gatewayMetrics?.upstreamErrors.add(1, { + provider: providerTag, + status_code: failure.status, + }) + if (!fallbackHttpCodes.includes(failure.status)) { + attemptIndex += 1 + break + } + logger.withError(err).withFields({ keyId: key.id, upstream: upstream.baseURL }).warn('Upstream TTS attempt failed') + attemptIndex += 1 + continue + } + // Adapter contract (see `server/apps/api/src/services/tts-adapters/types.ts` // and the three impls): // @@ -613,34 +671,20 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { if (err instanceof ApiError && err.statusCode < 500) throw err - const rawStatus - = (err as { status?: unknown }).status - ?? (err instanceof ApiError ? err.statusCode : undefined) - const failureStatus: number | 'timeout' = typeof rawStatus === 'number' ? rawStatus : 'timeout' - // TTS adapters bake the upstream body snippet into err.message - // (azure: `azure tts upstream 403: `, cosyvoice / volcengine - // analogous), so a single errorMessage carries both the status - // and the upstream payload diagnostics. + const rawStatus = err instanceof ApiError ? err.statusCode : undefined + const failureStatus: number | 'timeout' = rawStatus ?? 'timeout' const errorMessage = errorMessageFromUnknown(err) - failures.push({ keyId: key.id, status: failureStatus, errorMessage }) - onAttemptFailure({ keyId: key.id, status: failureStatus, errorMessage }) + const failure = { keyId: key.id, status: failureStatus, errorMessage } + failures.push(failure) + onAttemptFailure(failure) options.gatewayMetrics?.fallbackCount.add(1, { provider: providerTag, from_key: key.id, reason: String(failureStatus), }) - if (typeof rawStatus === 'number') { - options.gatewayMetrics?.upstreamErrors.add(1, { - provider: providerTag, - status_code: rawStatus, - }) - if (!fallbackHttpCodes.includes(rawStatus)) { - // Same key-level policy as chat: stop rotating credentials in this - // candidate. The enclosing group separately decides whether another - // candidate may be tried. - attemptIndex += 1 - break - } + if (typeof rawStatus === 'number' && !fallbackHttpCodes.includes(rawStatus)) { + attemptIndex += 1 + break } logger.withError(err).withFields({ keyId: key.id, upstream: upstream.baseURL }).warn('Upstream TTS attempt failed') } @@ -651,6 +695,7 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { attemptIndex += 1 } + await Promise.all(failures.slice(0, -1).map(failure => discardUpstreamResponse(failure.response))) return { kind: 'exhausted', failures } } @@ -672,13 +717,13 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { modelName: string, attemptUpstream: (upstream: TtsUpstream, index: number) => Promise< | { kind: 'ok', response: Response } - | { kind: 'exhausted', sawTooManyRequests: boolean, statuses: Array } + | { kind: 'exhausted', sawTooManyRequests: boolean, statuses: Array, response?: Response } >, retryOn?: RouteFailureTriggers, strategy: 'least-inflight' | 'ordered' = 'least-inflight', ): Promise< | { kind: 'ok', response: Response } - | { kind: 'exhausted', statuses: Array, transitionBlocked: boolean } + | { kind: 'exhausted', statuses: Array, transitionBlocked: boolean, response?: Response } > { async function markSaturated(upstream: TtsUpstream, poolId: string): Promise { await ledger.markSaturated(poolId, ttsPoolSaturationTtlSeconds) @@ -719,6 +764,13 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { let dispatchedAny = false let attemptedPools = 0 const statuses: Array = [] + let lastResponse: Response | undefined + + async function discardPreviousPoolResponse(): Promise { + await discardUpstreamResponse(lastResponse) + lastResponse = undefined + } + for (let rankedIndex = 0; rankedIndex < ranked.length; rankedIndex += 1) { const { upstream, index, poolId, maxConcurrency } = ranked[rankedIndex] const hasNextCandidate = rankedIndex < ranked.length - 1 @@ -726,14 +778,16 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { // Unlimited pool — dispatch without occupying a slot. dispatchedAny = true attemptedPools += 1 + await discardPreviousPoolResponse() const result = await attemptUpstream(upstream, index) if (result.kind === 'ok') return result statuses.push(...result.statuses) + lastResponse = result.response if (result.sawTooManyRequests) await markSaturated(upstream, poolId) if (hasNextCandidate && retryOn != null && !failuresMatch(result.statuses, retryOn)) - return { kind: 'exhausted', statuses, transitionBlocked: true } + return { kind: 'exhausted', statuses, transitionBlocked: true, response: result.response } continue } @@ -750,14 +804,16 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { dispatchedAny = true attemptedPools += 1 try { + await discardPreviousPoolResponse() const result = await attemptUpstream(upstream, index) if (result.kind === 'ok') return result statuses.push(...result.statuses) + lastResponse = result.response if (result.sawTooManyRequests) await markSaturated(upstream, poolId) if (hasNextCandidate && retryOn != null && !failuresMatch(result.statuses, retryOn)) - return { kind: 'exhausted', statuses, transitionBlocked: true } + return { kind: 'exhausted', statuses, transitionBlocked: true, response: result.response } } finally { await ledger.release(poolId) @@ -780,6 +836,7 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { kind: 'exhausted', statuses, transitionBlocked: attemptedPools !== upstreams.length, + response: lastResponse, } } @@ -804,8 +861,9 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { const unspeechBaseURL = (await options.configKV.getOrThrow('UNSPEECH_UPSTREAM')).restBaseURL - const allFailures: Array<{ provider: string, keyId: string, status: number | 'timeout', errorMessage?: string }> = [] + const allFailures: Array = [] let triedUpstreams = 0 + let terminalResponse: Response | undefined // tts upstream schema has no per-upstream timeoutMs (see ttsUpstreamSchema); // the defaults bucket alone governs per-attempt timeout. @@ -817,7 +875,7 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { // so the caller can circuit-break thatpool. async function attemptUpstream(upstream: TtsUpstream, index: number): Promise< | { kind: 'ok', response: Response } - | { kind: 'exhausted', sawTooManyRequests: boolean, statuses: Array } + | { kind: 'exhausted', sawTooManyRequests: boolean, statuses: Array, response?: Response } > { const providerTag = deriveProviderTag(upstream.baseURL) triedUpstreams += 1 @@ -849,12 +907,13 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { kind: 'exhausted', sawTooManyRequests: result.failures.some(f => f.status === 429), statuses: result.failures.map(failure => failure.status), + response: result.failures.at(-1)?.response, } } async function routeGroup(group: TtsRoutingGroup): Promise< | { kind: 'ok', response: Response } - | { kind: 'exhausted', statuses: Array, transitionBlocked: boolean } + | { kind: 'exhausted', statuses: Array, transitionBlocked: boolean, response?: Response } > { const indexedUpstreams = group.upstreamIds.map((upstreamId) => { const index = ttsModel.upstreams.findIndex(upstream => upstream.id === upstreamId) @@ -892,10 +951,12 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { statuses.push(...result.statuses) const hasNextCandidate = groupCandidateIndex < indexedUpstreams.length - 1 if (hasNextCandidate && !failuresMatch(result.statuses, group.retryOn)) - return { kind: 'exhausted', statuses, transitionBlocked: true } + return { kind: 'exhausted', statuses, transitionBlocked: true, response: result.response } + if (hasNextCandidate) + await discardUpstreamResponse(result.response) } - return { kind: 'exhausted', statuses, transitionBlocked: false } + return { kind: 'exhausted', statuses, transitionBlocked: false, response: allFailures.at(-1)?.response } } if (ttsModel.routing != null) { @@ -911,8 +972,11 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { || !hasNextGroup || !failuresMatch(result.statuses, group.continueOn) ) { + if (result.response != null) + terminalResponse = result.response break } + await discardUpstreamResponse(result.response) } } else { @@ -925,12 +989,18 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { const result = await attemptUpstream(ttsModel.upstreams[i], i) if (result.kind === 'ok') return result.response + if (i === ttsModel.upstreams.length - 1 && result.response != null) + terminalResponse = result.response + else + await discardUpstreamResponse(result.response) } } else { const result = await routeTtsAcrossPools(ttsModel.upstreams, req.modelName, attemptUpstream) if (result.kind === 'ok') return result.response + if (result.response != null) + terminalResponse = result.response } } @@ -957,6 +1027,9 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { } } + if (terminalResponse != null) + return terminalResponse + throw mapUpstreamError( lastFailure.status, { @@ -964,7 +1037,7 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { triedUpstreams, lastStatusCode: lastFailure.status, }, - allFailures, + allFailures.map(toDiagnosticAttempt), ) } diff --git a/server/apps/api/src/services/domain/llm-router/tests/router.test.ts b/server/apps/api/src/services/domain/llm-router/tests/router.test.ts index 68eca3909..d7ee78618 100644 --- a/server/apps/api/src/services/domain/llm-router/tests/router.test.ts +++ b/server/apps/api/src/services/domain/llm-router/tests/router.test.ts @@ -281,10 +281,12 @@ describe('createLlmRouterService', () => { expect(ctx.upstreamModel).toBe('real/upstream-id') }) - it('multi-key fallback: k1=401 then k2=200 → returns 200 and records fallbackCount once', async () => { + // https://github.com/moeru-ai/airi/pull/2333#discussion_r3820516102 + it('pr #2333: multi-key fallback releases the failed chat response before returning success', async () => { const { config, crypto } = makeConfig({ upstreams: [{ baseURL: 'https://up-a.example/v1', keyIds: ['k1', 'k2'] }] }) + const failedResponse = failResponse(401) const fetchImpl = vi.fn() - .mockResolvedValueOnce(failResponse(401)) + .mockResolvedValueOnce(failedResponse) .mockResolvedValueOnce(happyResponse({ ok: 1 })) const metrics = makeMetrics() @@ -299,6 +301,7 @@ describe('createLlmRouterService', () => { const res = await router.route({ modelName: 'openai/gpt-5-mini', body: {} }) expect(res.status).toBe(200) + expect(failedResponse.bodyUsed).toBe(true) expect(fetchImpl.mock.calls.length).toBe(2) expect((metrics.fallbackCount.add as ReturnType).mock.calls.length).toBe(1) @@ -310,6 +313,36 @@ describe('createLlmRouterService', () => { expect((metrics.keyExhaustedCount.add as ReturnType).mock.calls.length).toBe(0) }) + // https://github.com/moeru-ai/airi/pull/2333#discussion_r3820516102 + it('pr #2333: caller abort releases earlier failed chat responses', async () => { + const { config, crypto } = makeConfig({ upstreams: [{ baseURL: 'https://up-a.example/v1', keyIds: ['k1', 'k2'] }] }) + const failedResponse = failResponse(401) + const callerAbort = new AbortController() + const fetchImpl = vi.fn() + .mockResolvedValueOnce(failedResponse) + .mockImplementationOnce(async () => { + callerAbort.abort(new Error('client disconnected')) + throw callerAbort.signal.reason + }) + + const router = createLlmRouterService({ + configKV: makeConfigKV(config), + envelopeCrypto: crypto, + gatewayMetrics: null, + fetchImpl, + redis: makeRedisStub(), + concurrencyLedger: makeLedger(), + }) + + await expect(router.route({ + modelName: 'openai/gpt-5-mini', + body: {}, + abortSignal: callerAbort.signal, + })).rejects.toThrow('client disconnected') + + expect(failedResponse.bodyUsed).toBe(true) + }) + it('cross-upstream fallback: upstream A keys all 401, upstream B[0] = 200 → returns 200 without terminal exhaustion', async () => { const { config, crypto } = makeConfig({ upstreams: [ @@ -340,14 +373,20 @@ describe('createLlmRouterService', () => { expect((metrics.fallbackCount.add as ReturnType).mock.calls.length).toBe(2) }) - it('full exhaustion: every upstream + every key 401 → throws 502 BAD_GATEWAY (KTD-1 last-cause = 401 → 502)', async () => { + it('full exhaustion: every upstream + every key 401 → returns the final upstream response', async () => { const { config, crypto } = makeConfig({ upstreams: [ { baseURL: 'https://up-a.example/v1', keyIds: ['kA1'] }, { baseURL: 'https://up-b.example/v1', keyIds: ['kB1'] }, ], }) - const fetchImpl = vi.fn(async () => failResponse(401)) + const fetchImpl = vi.fn(async () => new Response('provider denied', { + status: 401, + headers: { + 'content-type': 'text/plain', + 'retry-after': '30', + }, + })) const metrics = makeMetrics() const router = createLlmRouterService({ @@ -359,16 +398,11 @@ describe('createLlmRouterService', () => { concurrencyLedger: makeLedger(), }) - try { - await router.route({ modelName: 'openai/gpt-5-mini', body: {} }) - throw new Error('expected throw') - } - catch (err) { - expect(err).toBeInstanceOf(ApiError) - expect((err as ApiError).statusCode).toBe(502) - expect((err as ApiError).errorCode).toBe('BAD_GATEWAY') - expect((err as ApiError).details).toMatchObject({ triedKeys: 2, triedUpstreams: 2, lastStatusCode: 401 }) - } + const response = await router.route({ modelName: 'openai/gpt-5-mini', body: {} }) + expect(response.status).toBe(401) + expect(response.headers.get('content-type')).toBe('text/plain') + expect(response.headers.get('retry-after')).toBe('30') + await expect(response.text()).resolves.toBe('provider denied') const exhaustionCalls = (metrics.keyExhaustedCount.add as ReturnType).mock.calls expect(exhaustionCalls.length).toBe(1) @@ -431,6 +465,10 @@ describe('createLlmRouterService', () => { expect(first).toMatchObject({ keyId: 'kA1', status: 401 }) expect(first.bodySnippet).toEqual(expect.stringContaining('key disabled')) expect(first.errorMessage).toBeUndefined() + // https://github.com/moeru-ai/airi/pull/2333#discussion_r3828016906 + // `Response` owns headers and a body stream. The diagnostic cause must + // keep only the documented serializable attempt fields. + expect(first.response).toBeUndefined() const second = (cause!.attempts as Array>)[1] expect(second).toMatchObject({ keyId: 'kB1', status: 'timeout' }) @@ -439,7 +477,7 @@ describe('createLlmRouterService', () => { } }) - it('same-status exhaustion: all keys 429 → throws 503 + sameStatusExhaustion incremented per provider', async () => { + it('same-status exhaustion: all keys 429 → returns 429 and increments sameStatusExhaustion per provider', async () => { const { config, crypto } = makeConfig({ upstreams: [ { baseURL: 'https://up-a.example/v1', keyIds: ['kA1', 'kA2'] }, @@ -458,7 +496,8 @@ describe('createLlmRouterService', () => { concurrencyLedger: makeLedger(), }) - await expect(router.route({ modelName: 'openai/gpt-5-mini', body: {} })).rejects.toMatchObject({ statusCode: 503, errorCode: 'SERVICE_UNAVAILABLE' }) + const response = await router.route({ modelName: 'openai/gpt-5-mini', body: {} }) + expect(response.status).toBe(429) const calls = (metrics.sameStatusExhaustion.add as ReturnType).mock.calls expect(calls.length).toBe(2) @@ -780,10 +819,11 @@ describe('createLlmRouterService', () => { const router = makeGroupedLlmRouter(fetchImpl) - await expect(router.route({ + const response = await router.route({ modelName: 'openai/gpt-5-mini', body: { messages: [] }, - })).rejects.toBeInstanceOf(ApiError) + }) + expect(response.status).toBe(429) expect(calledURLs).toEqual([ 'https://api.stepfun.com/step_plan/v1/chat/completions', @@ -801,10 +841,11 @@ describe('createLlmRouterService', () => { const router = makeGroupedLlmRouter(fetchImpl) - await expect(router.route({ + const response = await router.route({ modelName: 'openai/gpt-5-mini', body: { messages: [] }, - })).rejects.toBeInstanceOf(ApiError) + }) + expect(response.status).toBe(401) expect(calledURLs).toEqual([ 'https://api.stepfun.com/step_plan/v1/chat/completions', @@ -823,17 +864,19 @@ describe('createLlmRouterService', () => { // walked every key + upstream before surfacing — wasting upstream quota // and hiding the actual user-facing 400 behind a 502 mapping. // - // After patch: ApiError 4xx propagates immediately; ApiError 5xx folds - // into the network-failure fallback path using `statusCode`; `Error & - // { status }` stays on the existing fallback policy. + // After patch: ApiError 4xx propagates immediately. ApiError 5xx uses + // `statusCode` and obeys `fallbackHttpCodes`. TtsUpstreamResponseError + // stays on the existing fallback policy. describe('routeTts adapter error handling', () => { function makeTtsConfig(opts: { provider?: 'azure' upstreams?: Array<{ baseURL: string, keyIds: string[], adapterParams?: Record }> + fallbackHttpCodes?: number[] }): { config: RouterConfig, crypto: ReturnType } { const crypto = createEnvelopeCrypto({ masterKey: freshMasterKey() }) const modelName = 'tts-test' const upstreams = opts.upstreams ?? [{ baseURL: 'https://up-a.example', keyIds: ['kA1'] }] + const fallbackHttpCodes = opts.fallbackHttpCodes ?? [401, 429, 500, 502, 503, 504] const upstreamConfigs = upstreams.map(u => ({ baseURL: u.baseURL, keys: u.keyIds.map((id) => { @@ -850,14 +893,14 @@ describe('createLlmRouterService', () => { [modelName]: { provider: opts.provider ?? 'azure', upstreams: upstreamConfigs, - fallbackTriggers: { httpCodes: [401, 429, 500, 502, 503, 504], onTimeout: true }, + fallbackTriggers: { httpCodes: fallbackHttpCodes, onTimeout: true }, }, }, }, defaults: { perAttemptTimeoutMs: 5000, fullChainTimeoutMs: 10000, - fallbackHttpCodes: [401, 429, 500, 502, 503, 504], + fallbackHttpCodes, }, } as RouterConfig return { config, crypto } @@ -936,17 +979,52 @@ describe('createLlmRouterService', () => { expect((metrics.fallbackCount.add as ReturnType).mock.calls.length).toBe(1) }) - it('upstream `Error & { status: 401 }` folds into the existing fallback path', async () => { - // azure adapter throws `Error & { status: number }` on upstream non-2xx - // (see azure.ts:189-194). 401 is in fallbackHttpCodes so we must try - // the next key. + // https://github.com/moeru-ai/airi/pull/2333#discussion_r3820516115 + it('pr #2333: apiError 5xx stops key fallback when fallbackHttpCodes excludes the status', async () => { + const { config, crypto } = makeTtsConfig({ + upstreams: [{ baseURL: 'https://az.example', keyIds: ['kA1', 'kA2'], adapterParams: { region: 'eastasia' } }], + fallbackHttpCodes: [401, 429], + }) + const fetchImpl = vi.fn(async () => { + throw new TypeError('network unreachable') + }) + + const router = createLlmRouterService({ + configKV: makeConfigKV(config), + envelopeCrypto: crypto, + gatewayMetrics: null, + fetchImpl, + redis: makeRedisStub(), + concurrencyLedger: makeLedger(), + }) + + let caught: unknown + try { + await router.routeTts({ + modelName: 'tts-test', + input: { text: 'hi', voice: 'en-US-AvaMultilingualNeural' }, + }) + } + catch (error) { + caught = error + } + + expect(caught).toBeInstanceOf(ApiError) + expect((caught as ApiError).statusCode).toBe(502) + expect(fetchImpl).toHaveBeenCalledTimes(1) + }) + + it('upstream 401 folds into the existing fallback path', async () => { + // The Azure adapter throws TtsUpstreamResponseError on upstream non-2xx. + // 401 is in fallbackHttpCodes so we must try the next key. const { config, crypto } = makeTtsConfig({ upstreams: [{ baseURL: 'https://az.example', keyIds: ['kA1', 'kA2'], adapterParams: { region: 'eastasia' } }] }) let callIdx = 0 + const failedResponse = failResponse(401) const fetchImpl = vi.fn(async () => { callIdx += 1 if (callIdx === 1) - return failResponse(401) + return failedResponse return new Response(new Uint8Array([0x01]), { status: 200, headers: { 'content-type': 'audio/mpeg' } }) }) const metrics = makeMetrics() @@ -966,6 +1044,7 @@ describe('createLlmRouterService', () => { }) expect(res.status).toBe(200) + expect(failedResponse.bodyUsed).toBe(true) expect(fetchImpl).toHaveBeenCalledTimes(2) const fallbackCalls = (metrics.fallbackCount.add as ReturnType).mock.calls expect(fallbackCalls.length).toBe(1) @@ -995,10 +1074,12 @@ describe('createLlmRouterService', () => { concurrencyLedger: makeLedger(), }) - await expect(router.routeTts({ + const response = await router.routeTts({ modelName: 'tts-test', input: { text: 'hi', voice: 'en-US-AvaMultilingualNeural' }, - })).rejects.toMatchObject({ statusCode: 502 }) + }) + expect(response.status).toBe(451) + await expect(response.json()).resolves.toEqual({ error: 'bad' }) const exhaustionCalls = (metrics.keyExhaustedCount.add as ReturnType).mock.calls expect(exhaustionCalls.length).toBe(1) @@ -1323,10 +1404,11 @@ describe('createLlmRouterService', () => { const router = makeGroupedStepfunRouter(fetchImpl) - await expect(router.routeTts({ + const response = await router.routeTts({ modelName: 'stepfun/stepaudio-2.5-tts', input: { text: '你好' }, - })).rejects.toBeInstanceOf(ApiError) + }) + expect(response.status).toBe(429) expect(calledProfiles).toEqual(['step-plan', 'step-plan']) expect(calledProfiles).not.toContain('default') @@ -1342,10 +1424,11 @@ describe('createLlmRouterService', () => { const router = makeGroupedStepfunRouter(fetchImpl) - await expect(router.routeTts({ + const response = await router.routeTts({ modelName: 'stepfun/stepaudio-2.5-tts', input: { text: '你好' }, - })).rejects.toBeInstanceOf(ApiError) + }) + expect(response.status).toBe(401) expect(calledProfiles).toEqual(['step-plan']) }) @@ -1638,10 +1721,11 @@ describe('createLlmRouterService', () => { const router = makePoolRouter(config, crypto, ledger, fetchImpl) - await expect(router.routeTts({ + const response = await router.routeTts({ modelName: 'tts-pool', input: { text: 'hi' }, - })).rejects.toBeInstanceOf(ApiError) + }) + expect(response.status).toBe(402) expect(selectedAppIds).toEqual(['plan-b']) }) @@ -1728,7 +1812,8 @@ describe('createLlmRouterService', () => { const fetchImpl = vi.fn(async () => failResponse(429)) as unknown as typeof fetch const router = makePoolRouter(config, crypto, ledger, fetchImpl) - await expect(router.routeTts({ modelName: 'tts-pool', input: { text: 'hi' } })).rejects.toBeInstanceOf(ApiError) + const response = await router.routeTts({ modelName: 'tts-pool', input: { text: 'hi' } }) + expect(response.status).toBe(429) expect(markSaturated).toHaveBeenCalledWith('app-1', expect.any(Number)) }) @@ -1743,7 +1828,8 @@ describe('createLlmRouterService', () => { const fetchImpl = vi.fn(async () => failResponse(500)) as unknown as typeof fetch const router = makePoolRouter(config, crypto, ledger, fetchImpl) - await expect(router.routeTts({ modelName: 'tts-pool', input: { text: 'hi' } })).rejects.toBeInstanceOf(ApiError) + const response = await router.routeTts({ modelName: 'tts-pool', input: { text: 'hi' } }) + expect(response.status).toBe(500) expect(markSaturated).not.toHaveBeenCalled() }) diff --git a/server/apps/api/src/services/domain/openai-speech/index.ts b/server/apps/api/src/services/domain/openai-speech/index.ts index 2b223bc18..18692b3ce 100644 --- a/server/apps/api/src/services/domain/openai-speech/index.ts +++ b/server/apps/api/src/services/domain/openai-speech/index.ts @@ -28,6 +28,11 @@ const SAFE_RESPONSE_HEADERS = new Set([ 'cache-control', ]) +const SAFE_ERROR_RESPONSE_HEADERS = new Set([ + 'content-type', + 'retry-after', +]) + function asRecord(value: unknown): Record | undefined { if (typeof value !== 'object' || value == null || Array.isArray(value)) return undefined @@ -199,7 +204,7 @@ export function createOpenAiSpeechService(deps: OpenAiSpeechServiceDeps) { .warn('tts speech delivered with upstream error status') return new Response(response.body, { status: response.status, - headers: buildSafeResponseHeaders(response), + headers: buildSafeErrorResponseHeaders(response), }) } @@ -383,3 +388,12 @@ function buildSafeResponseHeaders(response: Response): Headers { }) return headers } + +function buildSafeErrorResponseHeaders(response: Response): Headers { + const headers = new Headers() + response.headers.forEach((value, key) => { + if (SAFE_ERROR_RESPONSE_HEADERS.has(key.toLowerCase())) + headers.set(key, value) + }) + return headers +}