diff --git a/apps/server/src/services/adapters/config-kv.test.ts b/apps/server/src/services/adapters/config-kv.test.ts index f10f9394b..c2643532d 100644 --- a/apps/server/src/services/adapters/config-kv.test.ts +++ b/apps/server/src/services/adapters/config-kv.test.ts @@ -151,6 +151,174 @@ describe('configKVService', () => { }) }) + it('llm router config should preserve explicit LLM and TTS provider groups', async () => { + await service.set('LLM_ROUTER_CONFIG', { + llm: { + models: { + 'step-3.5-flash': { + upstreams: [ + { + id: 'plan', + baseURL: 'https://api.stepfun.com/step_plan/v1', + keys: [{ id: 'plan-key', ciphertext: 'plan-ciphertext' }], + headerTemplate: 'Bearer {KEY}', + }, + { + id: 'paygo', + baseURL: 'https://api.stepfun.com/v1', + keys: [{ id: 'paygo-key', ciphertext: 'paygo-ciphertext' }], + headerTemplate: 'Bearer {KEY}', + }, + ], + routing: { + groups: [ + { + id: 'plan', + upstreamIds: ['plan'], + retryOn: { httpCodes: [402, 429, 500, 502, 503, 504], onTimeout: true }, + continueOn: { httpCodes: [402], onTimeout: false }, + }, + { + id: 'paygo', + upstreamIds: ['paygo'], + retryOn: { httpCodes: [429, 500, 502, 503, 504], onTimeout: true }, + }, + ], + }, + fallbackTriggers: { + httpCodes: [401, 402, 403, 429, 500, 502, 503, 504], + onTimeout: true, + }, + }, + }, + }, + tts: { + models: { + 'stepfun/stepaudio-2.5-tts': { + provider: 'stepfun', + upstreams: [ + { + id: 'plan', + baseURL: 'https://api.stepfun.com', + keys: [{ id: 'plan-key', ciphertext: 'plan-ciphertext' }], + adapterParams: { endpointProfile: 'step-plan' }, + maxConcurrency: 1, + }, + { + id: 'paygo', + baseURL: 'https://api.stepfun.com', + keys: [{ id: 'paygo-key', ciphertext: 'paygo-ciphertext' }], + adapterParams: { endpointProfile: 'default' }, + }, + ], + routing: { + groups: [ + { + id: 'plan', + upstreamIds: ['plan'], + strategy: 'least-inflight', + retryOn: { httpCodes: [402, 429, 500, 502, 503, 504], onTimeout: true }, + continueOn: { httpCodes: [402], onTimeout: false }, + }, + { + id: 'paygo', + upstreamIds: ['paygo'], + strategy: 'ordered', + retryOn: { httpCodes: [429, 500, 502, 503, 504], onTimeout: true }, + }, + ], + }, + fallbackTriggers: { + httpCodes: [401, 402, 429, 500, 502, 503, 504], + onTimeout: true, + }, + }, + }, + }, + defaults: { + perAttemptTimeoutMs: 30000, + fullChainTimeoutMs: 60000, + fallbackHttpCodes: [401, 402, 403, 429, 500, 502, 503, 504], + }, + }) + + const value = await service.getOrThrow('LLM_ROUTER_CONFIG') + const model = value.tts.models['stepfun/stepaudio-2.5-tts'] + + expect(value.llm.models['step-3.5-flash'].routing?.groups.map(group => group.id)).toEqual(['plan', 'paygo']) + expect(model.routing?.groups.map(group => group.id)).toEqual(['plan', 'paygo']) + expect(model.routing?.groups[0].continueOn).toEqual({ + httpCodes: [402], + onTimeout: false, + }) + }) + + it('rejects a TTS provider group that references an unknown upstream', async () => { + redis._store.set(configRedisKey('LLM_ROUTER_CONFIG'), JSON.stringify({ + llm: { models: {} }, + tts: { + models: { + tts: { + provider: 'stepfun', + upstreams: [{ + id: 'plan', + baseURL: 'https://api.stepfun.com', + keys: [{ id: 'plan-key', ciphertext: 'ciphertext' }], + }], + routing: { + groups: [{ + id: 'plan', + upstreamIds: ['missing'], + strategy: 'ordered', + retryOn: { httpCodes: [402], onTimeout: false }, + }], + }, + }, + }, + }, + })) + + await expect(service.getOptional('LLM_ROUTER_CONFIG')) + .rejects + .toMatchObject({ + statusCode: 503, + errorCode: 'CONFIG_INVALID', + }) + }) + + it('rejects least-inflight routing without an explicit concurrency cap', async () => { + redis._store.set(configRedisKey('LLM_ROUTER_CONFIG'), JSON.stringify({ + llm: { models: {} }, + tts: { + models: { + tts: { + provider: 'stepfun', + upstreams: [{ + id: 'plan', + baseURL: 'https://api.stepfun.com', + keys: [{ id: 'plan-key', ciphertext: 'ciphertext' }], + }], + routing: { + groups: [{ + id: 'plan', + upstreamIds: ['plan'], + strategy: 'least-inflight', + retryOn: { httpCodes: [402], onTimeout: false }, + }], + }, + }, + }, + }, + })) + + await expect(service.getOptional('LLM_ROUTER_CONFIG')) + .rejects + .toMatchObject({ + statusCode: 503, + errorCode: 'CONFIG_INVALID', + }) + }) + it('set should store string values as JSON strings', async () => { await service.set('STRIPE_FLUX_PRODUCT_ID', 'prod_abc123') diff --git a/apps/server/src/services/adapters/config-kv.ts b/apps/server/src/services/adapters/config-kv.ts index ca11562ff..c572c1a78 100644 --- a/apps/server/src/services/adapters/config-kv.ts +++ b/apps/server/src/services/adapters/config-kv.ts @@ -9,8 +9,9 @@ import { configRedisKey } from '../../utils/redis-keys' /** * LLM/TTS router config tree. Single composite entry under configKV holds the - * entire routing surface: per-model upstream list, per-upstream key array - * (envelope-encrypted ciphertexts), fallback triggers, default timeouts. + * entire routing surface: per-model upstream list, optional candidate groups, + * per-upstream key array (envelope-encrypted ciphertexts), transition policies, + * and default timeouts. * * Schema enforces: * - key entry id must not contain `|` — the envelope-crypto AAD uses `|` as @@ -30,6 +31,18 @@ export const fallbackTriggersSchema = optional( { httpCodes: [401, 402, 403, 429, 500, 502, 503, 504], onTimeout: true }, ) +/** + * Explicit allow-list for one routing transition. + * + * Unlike {@link fallbackTriggersSchema}, this contract has no permissive + * defaults: an omitted status or timeout never authorizes a transition across + * a configured routing boundary. + */ +export const routeFailureTriggersSchema = object({ + httpCodes: optional(array(number()), []), + onTimeout: optional(boolean(), false), +}) + export const keyEntrySchema = object({ id: pipe( string(), @@ -40,6 +53,11 @@ export const keyEntrySchema = object({ }) export const llmUpstreamSchema = object({ + id: optional(pipe( + string(), + nonEmpty('llm.upstreams[].id must not be empty'), + regex(/^[^|]+$/, 'llm.upstreams[].id must not contain "|"'), + )), baseURL: pipe(string(), nonEmpty('llm.upstreams[].baseURL must not be empty')), overrideModel: optional(string()), keys: pipe(array(keyEntrySchema), check(v => v.length >= 1, 'llm.upstreams[].keys must contain at least 1 entry')), @@ -47,15 +65,58 @@ export const llmUpstreamSchema = object({ timeoutMs: optional(number()), }) -export const llmModelSchema = object({ - upstreams: pipe(array(llmUpstreamSchema), check(v => v.length >= 1, 'llm.models[].upstreams must contain at least 1 entry')), - fallbackTriggers: fallbackTriggersSchema, +export const llmRoutingGroupSchema = object({ + id: pipe(string(), nonEmpty('llm.routing.groups[].id must not be empty')), + upstreamIds: pipe( + array(pipe(string(), nonEmpty('llm.routing.groups[].upstreamIds[] must not be empty'))), + check(v => v.length >= 1, 'llm.routing.groups[].upstreamIds must contain at least 1 entry'), + check(v => new Set(v).size === v.length, 'llm.routing.groups[].upstreamIds must be unique'), + ), + retryOn: routeFailureTriggersSchema, + continueOn: optional(routeFailureTriggersSchema), }) +export const llmRoutingSchema = object({ + groups: pipe( + array(llmRoutingGroupSchema), + check(v => v.length >= 1, 'llm.routing.groups must contain at least 1 entry'), + check(v => new Set(v.map(group => group.id)).size === v.length, 'llm.routing.groups[].id must be unique'), + ), +}) + +export const llmModelSchema = pipe( + object({ + upstreams: pipe(array(llmUpstreamSchema), check(v => v.length >= 1, 'llm.models[].upstreams must contain at least 1 entry')), + routing: optional(llmRoutingSchema), + fallbackTriggers: fallbackTriggersSchema, + }), + check((model) => { + if (model.routing == null) + return true + const upstreamIds = model.upstreams.map(upstream => upstream.id) + return upstreamIds.every(id => id != null) + && new Set(upstreamIds).size === upstreamIds.length + }, 'llm.models[].upstreams must have unique ids when routing is configured'), + check((model) => { + if (model.routing == null) + return true + const upstreamIds = new Set(model.upstreams.map(upstream => upstream.id)) + const referencedIds = model.routing.groups.flatMap(group => group.upstreamIds) + return referencedIds.length === upstreamIds.size + && new Set(referencedIds).size === referencedIds.length + && referencedIds.every(id => upstreamIds.has(id)) + }, 'llm.routing.groups must reference every upstream id exactly once'), +) + const ttsProviderSchema = picklist(['azure', 'dashscope-cosyvoice', 'stepfun', 'volcengine']) const asrProviderSchema = picklist(['aliyun-nls']) export const ttsUpstreamSchema = object({ + id: optional(pipe( + string(), + nonEmpty('tts.upstreams[].id must not be empty'), + regex(/^[^|]+$/, 'tts.upstreams[].id must not contain "|"'), + )), baseURL: pipe(string(), nonEmpty('tts.upstreams[].baseURL must not be empty')), keys: pipe(array(keyEntrySchema), check(v => v.length >= 1, 'tts.upstreams[].keys must contain at least 1 entry')), adapterParams: optional(record(string(), any()), {}), @@ -68,6 +129,26 @@ export const ttsUpstreamSchema = object({ maxConcurrency: optional(pipe(number(), check(v => v >= 1, 'tts.upstreams[].maxConcurrency must be >= 1 when set'))), }) +export const ttsRoutingGroupSchema = object({ + id: pipe(string(), nonEmpty('tts.routing.groups[].id must not be empty')), + upstreamIds: pipe( + array(pipe(string(), nonEmpty('tts.routing.groups[].upstreamIds[] must not be empty'))), + check(v => v.length >= 1, 'tts.routing.groups[].upstreamIds must contain at least 1 entry'), + check(v => new Set(v).size === v.length, 'tts.routing.groups[].upstreamIds must be unique'), + ), + strategy: optional(picklist(['ordered', 'least-inflight']), 'ordered'), + retryOn: routeFailureTriggersSchema, + continueOn: optional(routeFailureTriggersSchema), +}) + +export const ttsRoutingSchema = object({ + groups: pipe( + array(ttsRoutingGroupSchema), + check(v => v.length >= 1, 'tts.routing.groups must contain at least 1 entry'), + check(v => new Set(v.map(group => group.id)).size === v.length, 'tts.routing.groups[].id must be unique'), + ), +}) + export const streamingTtsUpstreamSchema = object({ baseURL: pipe(string(), nonEmpty('UNSPEECH_UPSTREAM.streaming.baseURL must not be empty')), keys: pipe(array(keyEntrySchema), check(v => v.length >= 1, 'UNSPEECH_UPSTREAM.streaming.keys must contain at least 1 entry')), @@ -88,11 +169,39 @@ export const unspeechUpstreamSchema = object({ streaming: optional(streamingTtsUpstreamSchema), }) -export const ttsModelSchema = object({ - provider: ttsProviderSchema, - upstreams: pipe(array(ttsUpstreamSchema), check(v => v.length >= 1, 'tts.models[].upstreams must contain at least 1 entry')), - fallbackTriggers: fallbackTriggersSchema, -}) +export const ttsModelSchema = pipe( + object({ + provider: ttsProviderSchema, + upstreams: pipe(array(ttsUpstreamSchema), check(v => v.length >= 1, 'tts.models[].upstreams must contain at least 1 entry')), + routing: optional(ttsRoutingSchema), + fallbackTriggers: fallbackTriggersSchema, + }), + check((model) => { + if (model.routing == null) + return true + const upstreamIds = model.upstreams.map(upstream => upstream.id) + return upstreamIds.every(id => id != null) + && new Set(upstreamIds).size === upstreamIds.length + }, 'tts.models[].upstreams must have unique ids when routing is configured'), + check((model) => { + if (model.routing == null) + return true + const upstreamIds = new Set(model.upstreams.map(upstream => upstream.id)) + const referencedIds = model.routing.groups.flatMap(group => group.upstreamIds) + return referencedIds.length === upstreamIds.size + && new Set(referencedIds).size === referencedIds.length + && referencedIds.every(id => upstreamIds.has(id)) + }, 'tts.routing.groups must reference every upstream id exactly once'), + check((model) => { + if (model.routing == null) + return true + const upstreamById = new Map(model.upstreams.map(upstream => [upstream.id, upstream])) + return model.routing.groups.every(group => + group.strategy !== 'least-inflight' + || group.upstreamIds.every(id => upstreamById.get(id)?.maxConcurrency != null), + ) + }, 'tts.routing least-inflight groups require maxConcurrency on every upstream'), +) export const asrUpstreamSchema = object({ keys: pipe(array(keyEntrySchema), check(v => v.length >= 1, 'asr.upstreams[].keys must contain at least 1 entry')), diff --git a/apps/server/src/services/adapters/tts/index.test.ts b/apps/server/src/services/adapters/tts/index.test.ts index 092dbd02d..0d0aa59b0 100644 --- a/apps/server/src/services/adapters/tts/index.test.ts +++ b/apps/server/src/services/adapters/tts/index.test.ts @@ -298,7 +298,7 @@ describe('azureAdapter.send', () => { }) describe('stepfunAdapter', () => { - it('lists StepFun voices through unspeech provider=stepfun', async () => { + it('uses unspeech as the StepFun voice-catalog source', async () => { const adapter = getAdapter('stepfun') const fetchImpl = vi.fn(async () => new Response(JSON.stringify({ voices: [{ @@ -323,6 +323,7 @@ describe('stepfunAdapter', () => { }), ]), ) + expect(fetchImpl).toHaveBeenCalledTimes(1) const [calledUrl] = (fetchImpl as unknown as { mock: { calls: [string, RequestInit][] } }).mock.calls[0] expect(calledUrl).toBe('http://unspeech.local/api/voices?provider=stepfun') }) @@ -348,7 +349,7 @@ describe('stepfunAdapter', () => { }, { keyPlaintext: Buffer.from('step-key', 'utf8'), - baseURL: 'https://api.stepfun.com/v1/audio/speech', + baseURL: 'https://api.stepfun.com', unspeechBaseURL: 'http://unspeech.local:5933', adapterParams: { model: 'stepaudio-2.5-tts' }, fetchImpl, @@ -379,6 +380,57 @@ describe('stepfunAdapter', () => { expect(result.body).toBeInstanceOf(ArrayBuffer) }) + it('passes the Step Plan endpoint profile to unspeech', async () => { + const adapter = getAdapter('stepfun') + const fetchImpl = vi.fn(async () => new Response(new Uint8Array([1, 2, 3]), { + status: 200, + headers: { 'content-type': 'audio/mpeg' }, + })) as unknown as typeof fetch + + const result = await adapter.send( + { + text: '你好', + voice: 'cixingnansheng', + responseFormat: 'mp3', + speed: 1.1, + extraOptions: { + instruction: '温柔、克制', + }, + }, + { + keyPlaintext: Buffer.from('step-plan-key', 'utf8'), + baseURL: 'https://api.stepfun.com', + unspeechBaseURL: 'http://unspeech.local:5933', + adapterParams: { + endpointProfile: 'step-plan', + model: 'stepaudio-2.5-tts', + }, + fetchImpl, + }, + ) + + const [calledURL, init] = (fetchImpl as unknown as { mock: { calls: [string, RequestInit][] } }).mock.calls[0] + expect(calledURL).toBe('http://unspeech.local:5933/v1/audio/speech') + expect(init.method).toBe('POST') + expect(init.headers).toMatchObject({ + 'Authorization': 'Bearer step-plan-key', + 'Content-Type': 'application/json', + }) + expect(JSON.parse(init.body as string)).toEqual({ + model: 'stepfun/stepaudio-2.5-tts', + input: '你好', + voice: 'cixingnansheng', + response_format: 'mp3', + speed: 1.1, + extra_body: { + endpoint_profile: 'step-plan', + instruction: '温柔、克制', + }, + }) + expect(result.contentType).toBe('audio/mpeg') + expect(result.body).toBeInstanceOf(ArrayBuffer) + }) + it('passes voice_label through to unspeech for provider-level validation', async () => { const adapter = getAdapter('stepfun') const fetchImpl = vi.fn(async () => new Response(new Uint8Array([1]), { @@ -395,7 +447,7 @@ describe('stepfunAdapter', () => { }, { keyPlaintext: Buffer.from('step-key', 'utf8'), - baseURL: 'https://api.stepfun.com/v1/audio/speech', + baseURL: 'https://api.stepfun.com', unspeechBaseURL: 'http://unspeech.local', adapterParams: { model: 'stepaudio-2.5-tts' }, fetchImpl, @@ -415,13 +467,35 @@ describe('stepfunAdapter', () => { { text: 'hi', voice: 'cixingnansheng' }, { keyPlaintext: Buffer.from('bad-key', 'utf8'), - baseURL: 'https://api.stepfun.com/v1/audio/speech', + baseURL: 'https://api.stepfun.com', unspeechBaseURL: 'http://unspeech.local', adapterParams: { model: 'stepaudio-2.5-tts' }, fetchImpl, }, )).rejects.toMatchObject({ status: 401 }) }) + + it('preserves an unspeech request abort for router timeout classification', async () => { + const adapter = getAdapter('stepfun') + const abortController = new AbortController() + const abortError = new Error('attempt-timeout') + abortController.abort(abortError) + const fetchImpl = vi.fn(async (_input: string | URL | Request, init?: RequestInit) => { + throw init?.signal?.reason ?? new Error('aborted') + }) as unknown as typeof fetch + + await expect(adapter.send( + { text: 'hi', voice: 'cixingnansheng' }, + { + keyPlaintext: Buffer.from('step-key', 'utf8'), + baseURL: 'https://api.stepfun.com', + unspeechBaseURL: 'http://unspeech.local', + adapterParams: { model: 'stepaudio-2.5-tts' }, + fetchImpl, + abortSignal: abortController.signal, + }, + )).rejects.toBe(abortError) + }) }) describe('volcengineAdapter.send', () => { diff --git a/apps/server/src/services/adapters/tts/stepfun.ts b/apps/server/src/services/adapters/tts/stepfun.ts index 064151332..44d676720 100644 --- a/apps/server/src/services/adapters/tts/stepfun.ts +++ b/apps/server/src/services/adapters/tts/stepfun.ts @@ -15,8 +15,8 @@ const STEPFUN_DEFAULT_VOICE = 'cixingnansheng' * StepFun TTS adapter. * * Use when: - * - Routing hosted speech synthesis to StepFun through unspeech's - * OpenAI-compatible `stepfun/*` backend. + * - Routing speech synthesis to StepFun through unspeech's OpenAI-compatible + * `stepfun/*` backend. * * Expects: * - `ctx.unspeechBaseURL` points at an unspeech deployment that includes the @@ -24,6 +24,8 @@ const STEPFUN_DEFAULT_VOICE = 'cixingnansheng' * - `ctx.keyPlaintext` is the StepFun API key. * - `ctx.adapterParams.model` optionally selects `stepaudio-2.5-tts`, * `step-tts-2`, or `step-tts-mini`. + * - `ctx.adapterParams.endpointProfile` optionally selects a provider-owned + * endpoint profile such as `step-plan`; AIRI never owns the endpoint URL. * * Returns: * - {@link TtsResult} with the upstream audio body and content type. @@ -41,6 +43,7 @@ export const stepfunAdapter: TtsAdapter = { const responseFormat = input.responseFormat ?? (typeof ctx.adapterParams.responseFormat === 'string' && ctx.adapterParams.responseFormat ? ctx.adapterParams.responseFormat : STEPFUN_DEFAULT_FORMAT) + const extraBody = buildExtraBody(input, ctx) return sendSpeechViaUnSpeech({ ctx, @@ -49,7 +52,7 @@ export const stepfunAdapter: TtsAdapter = { voice, speed: input.speed, responseFormat, - extraBody: buildExtraBody(input, ctx), + extraBody, fallbackContentType: audioMimeFromFormat(responseFormat), providerLabel: 'stepfun', }) @@ -68,6 +71,9 @@ function buildExtraBody(input: TtsInput, ctx: TtsAdapterContext): Record = {} + if (typeof ctx.adapterParams.endpointProfile === 'string' && ctx.adapterParams.endpointProfile) + body.endpoint_profile = ctx.adapterParams.endpointProfile + if (typeof extraOptions.volume === 'number' && Number.isFinite(extraOptions.volume)) body.volume = extraOptions.volume else if (typeof ctx.adapterParams.volume === 'number' && Number.isFinite(ctx.adapterParams.volume)) diff --git a/apps/server/src/services/adapters/tts/types.ts b/apps/server/src/services/adapters/tts/types.ts index 176d567a2..1c52ef44c 100644 --- a/apps/server/src/services/adapters/tts/types.ts +++ b/apps/server/src/services/adapters/tts/types.ts @@ -42,16 +42,11 @@ export interface TtsAdapterContext { /** * Per-upstream baseURL from `LLM_ROUTER_CONFIG.tts.upstreams[i].baseURL`. * - * Historically the upstream provider URL (e.g. - * `https://eastasia.tts.speech.microsoft.com/cognitiveservices/v1`). After - * the Phase-B unspeech migration, adapters no longer call upstreams - * directly — every `send()` forwards through unspeech REST — so this field - * is informational only and adapters MAY ignore it. Kept on the context so - * existing operator configs continue to validate (the schema requires a - * non-empty string). + * Adapters forward through unspeech and may use this as provider metadata. + * Provider endpoint selection belongs to unspeech, not this URL. */ baseURL: string - /** unspeech REST base URL (no trailing slash) — adapters POST to `/v1/audio/speech`. */ + /** unspeech REST base URL (no trailing slash). */ unspeechBaseURL: string /** Free-form adapter-specific params from `tts.upstreams[i].adapterParams` (e.g. Volcengine `appid` / `cluster`). */ adapterParams: Record @@ -90,12 +85,10 @@ export type TtsAdapterId = 'azure' | 'dashscope-cosyvoice' | 'stepfun' | 'volcen * `keyPlaintext` and `region` are mandatory for live providers (Azure) that * proxy through unspeech and call the upstream provider with a subscription * key; the router decrypts the envelope key and forwards `adapterParams.region` - * verbatim. Providers with static, credential-less catalogs (DashScope - * cosyvoice, Volcengine) ignore both fields. + * verbatim. Unspeech-backed static catalogs ignore both fields. * - * `unspeechBaseURL` is `UNSPEECH_UPSTREAM.restBaseURL` resolved by the - * router. Passing it through the context keeps adapters free of configKV - * coupling — they receive a fully-resolved URL string. + * `unspeechBaseURL` is `UNSPEECH_UPSTREAM.restBaseURL` resolved by the router. + * Passing it through the context keeps adapters free of configKV coupling. */ export interface TtsVoiceCatalogContext { /** Decrypted upstream credential (live providers only). */ @@ -139,10 +132,9 @@ export interface TtsAdapter { /** * Returns the voice catalog for the provider. * - * Live providers (Azure) call upstream via unspeech using the supplied - * region + plaintext key. Static providers (dashscope-cosyvoice, volcengine) - * return their compiled-in JSON and ignore the context fields. Adapters - * MUST throw on upstream failure — no empty-array fallback. + * Live providers (Azure) call upstream through unspeech using the supplied + * region + plaintext key. Static provider catalogs are also owned and served + * by unspeech. Adapters MUST throw on upstream failure — no empty fallback. */ getVoiceCatalog: (ctx: TtsVoiceCatalogContext) => Promise } diff --git a/apps/server/src/services/adapters/tts/unspeech.ts b/apps/server/src/services/adapters/tts/unspeech.ts index 7ad388582..9d8117262 100644 --- a/apps/server/src/services/adapters/tts/unspeech.ts +++ b/apps/server/src/services/adapters/tts/unspeech.ts @@ -66,6 +66,11 @@ export async function sendSpeechViaUnSpeech(options: SendSpeechOptions): Promise } } catch (error) { + // Keep abort identity intact so the router can apply `onTimeout` + // independently from HTTP 500 fallback policy. + if (ctx.abortSignal?.aborted) + 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 diff --git a/apps/server/src/services/domain/admin/router-config/index.ts b/apps/server/src/services/domain/admin/router-config/index.ts index 4b34cb14d..0ddca8faf 100644 --- a/apps/server/src/services/domain/admin/router-config/index.ts +++ b/apps/server/src/services/domain/admin/router-config/index.ts @@ -391,7 +391,7 @@ export function buildStepfunSlice(input: StepfunSliceInput, envelope: EnvelopeCr model: { provider: 'stepfun', upstreams: [{ - baseURL: 'https://api.stepfun.com/v1/audio/speech', + baseURL: 'https://api.stepfun.com', keys: [{ id: keyEntryId, ciphertext }], adapterParams: { model: input.upstreamModel ?? 'stepaudio-2.5-tts', @@ -604,7 +604,7 @@ function buildStepfunSlicePreservingKey(input: StepfunSliceInput, envelope: Enve model: { provider: 'stepfun', upstreams: [{ - baseURL: 'https://api.stepfun.com/v1/audio/speech', + baseURL: 'https://api.stepfun.com', keys: [key], adapterParams: { model: input.upstreamModel ?? 'stepaudio-2.5-tts', @@ -715,6 +715,11 @@ export function buildSlice( * - The next config tree, ready to feed `configKV.set('LLM_ROUTER_CONFIG', ...)`. * `defaults` is preserved verbatim when merging — the admin endpoint does * not currently re-tune timeouts via this path. + * + * Throws: + * - When merge mode targets a grouped LLM/TTS model. The legacy slice contract + * cannot identify one upstream or represent routing groups, so replacing that + * model would silently erase its routing policy. */ export function buildNextRouterConfig( mode: 'merge' | 'reset', @@ -729,12 +734,25 @@ export function buildNextRouterConfig( = mode === 'merge' && existing?.asr?.models ? { ...existing.asr.models } : {} for (const slice of slices) { - if (slice.surface === 'llm') + if (slice.surface === 'llm') { + if (mode === 'merge' && llmModels[slice.modelName]?.routing != null) { + throw new Error( + `Legacy admin config cannot update grouped llm model ${slice.modelName}; use a group-aware router config update`, + ) + } llmModels[slice.modelName] = slice.model - else if (slice.surface === 'tts') + } + else if (slice.surface === 'tts') { + if (mode === 'merge' && ttsModels[slice.modelName]?.routing != null) { + throw new Error( + `Legacy admin config cannot update grouped tts model ${slice.modelName}; use a group-aware router config update`, + ) + } ttsModels[slice.modelName] = slice.model - else + } + else { asrModels[slice.modelName] = slice.model + } } // Defaults live alongside the models but aren't editable through this diff --git a/apps/server/src/services/domain/admin/router-config/tests/admin-router-config.test.ts b/apps/server/src/services/domain/admin/router-config/tests/admin-router-config.test.ts index 82e7cbde5..a873d0f99 100644 --- a/apps/server/src/services/domain/admin/router-config/tests/admin-router-config.test.ts +++ b/apps/server/src/services/domain/admin/router-config/tests/admin-router-config.test.ts @@ -271,7 +271,7 @@ describe('buildStepfunSlice', () => { expect(built.kind).toBe('stepfun') expect(built.model.provider).toBe('stepfun') - expect(built.model.upstreams[0].baseURL).toBe('https://api.stepfun.com/v1/audio/speech') + expect(built.model.upstreams[0].baseURL).toBe('https://api.stepfun.com') expect(built.model.upstreams[0].adapterParams).toEqual({ model: 'stepaudio-2.5-tts', defaultVoice: 'cixingnansheng', @@ -582,6 +582,70 @@ describe('createAdminRouterConfigService', () => { expect(Object.keys(written.tts.models)).toEqual(['microsoft/v1']) }) + it('rejects legacy merge updates that would replace a grouped TTS model', async () => { + const modelName = 'stepfun/stepaudio-2.5-tts' + const existingConfig = { + llm: { models: {} }, + tts: { + models: { + [modelName]: { + provider: 'stepfun' as const, + upstreams: [ + { + id: 'plan', + baseURL: 'https://api.stepfun.com', + keys: [{ id: 'plan-key', ciphertext: 'plan-ciphertext' }], + adapterParams: { endpointProfile: 'step-plan', model: 'stepaudio-2.5-tts' }, + }, + { + id: 'paygo', + baseURL: 'https://api.stepfun.com', + keys: [{ id: 'paygo-key', ciphertext: 'paygo-ciphertext' }], + adapterParams: { endpointProfile: 'default', model: 'stepaudio-2.5-tts' }, + }, + ], + routing: { + groups: [ + { + id: 'plan', + upstreamIds: ['plan'], + strategy: 'ordered' as const, + retryOn: { httpCodes: [402], onTimeout: false }, + continueOn: { httpCodes: [402], onTimeout: false }, + }, + { + id: 'paygo', + upstreamIds: ['paygo'], + strategy: 'ordered' as const, + retryOn: { httpCodes: [429, 500], onTimeout: true }, + }, + ], + }, + fallbackTriggers: DEFAULT_FALLBACK_TRIGGERS, + }, + }, + }, + defaults: { perAttemptTimeoutMs: 30000, fullChainTimeoutMs: 60000, fallbackHttpCodes: [500] }, + } + kv.store.set('LLM_ROUTER_CONFIG', existingConfig) + + const service = createAdminRouterConfigService({ configKV: kv.service, envelope, redis }) + + await expect(service.apply({ + mode: 'merge', + dryRun: false, + slices: [{ + kind: 'stepfun', + modelName, + upstreamModel: 'stepaudio-2.5-tts', + plaintextKey: 'rotated-key', + }], + })).rejects.toThrow(/cannot update grouped tts model/i) + + expect(kv.store.get('LLM_ROUTER_CONFIG')).toBe(existingConfig) + expect(captured).toEqual([]) + }) + it('current returns editable slices from configKV without exposing raw ciphertext', async () => { kv.store.set('LLM_ROUTER_CONFIG', { llm: { diff --git a/apps/server/src/services/domain/llm-router/router.ts b/apps/server/src/services/domain/llm-router/router.ts index 99150dd35..9e530f871 100644 --- a/apps/server/src/services/domain/llm-router/router.ts +++ b/apps/server/src/services/domain/llm-router/router.ts @@ -8,7 +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 { LlmRouteContext, LlmRouteRequest, LlmUpstream, TtsUpstream } from './types' +import type { LlmRouteContext, LlmRouteRequest, LlmRoutingGroup, LlmUpstream, RouteFailureTriggers, TtsRoutingGroup, TtsUpstream } from './types' import { Buffer as NodeBuffer } from 'node:buffer' @@ -95,13 +95,34 @@ function deriveProviderTag(baseURL: string): string { /** * Identity of the pool (concurrency pool) one TTS upstream belongs to. One * upstream == one app_id, so the Volcengine `adapterParams.appid` is the pool - * key when present; the baseURL is a stable fallback for providers without an - * app_id concept. Two upstreams sharing an app_id would (correctly) share one - * concurrency budget, though thetypical config gives each app_id its own upstream. + * key when present. Other providers use a model-scoped upstream id, then a + * model-scoped baseURL for configs without ids. This prevents model-local ids + * such as `plan` from sharing Redis counters across unrelated models. Two + * upstreams sharing an app_id correctly share one global concurrency budget. */ -function ttsPoolId(upstream: TtsUpstream): string { +function ttsPoolId(upstream: TtsUpstream, modelName: string): string { const appid = upstream.adapterParams?.appid - return typeof appid === 'string' && appid.length > 0 ? appid : upstream.baseURL + if (typeof appid === 'string' && appid.length > 0) + return appid + return `model:${JSON.stringify([ + modelName, + upstream.id == null ? 'baseURL' : 'id', + upstream.id ?? upstream.baseURL, + ])}` +} + +function failuresMatch( + statuses: ReadonlyArray, + triggers: RouteFailureTriggers | undefined, +): boolean { + if (statuses.length === 0 || triggers == null) + return false + + return statuses.every((status) => { + if (status === 'timeout') + return triggers.onTimeout + return triggers.httpCodes.includes(status) + }) } export interface CreateLlmRouterServiceOptions { @@ -311,10 +332,10 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { status_code: status, }) if (!fallbackHttpCodes.includes(status)) { - // Status not in the fallback whitelist — surface as the last - // status and stop walking this upstream. We still let the outer - // loop try the next upstream (KTD-13: cross-upstream fallback - // happens in the same request, regardless of per-status policy). + // Status not in the key-level fallback whitelist — surface as the + // last status and stop walking this upstream. A configured provider + // group decides separately whether another account may be attempted; + // models without groups preserve the historical KTD-13 behavior. attemptIndex += 1 break } @@ -367,14 +388,14 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { throw new Error(`Expected llm model slice for ${req.modelName}, got ${slice.kind}`) } + const llmModel = slice.model const defaults = slice.defaults ?? { perAttemptTimeoutMs: 30000, fullChainTimeoutMs: 60000, fallbackHttpCodes: [401, 402, 403, 429, 500, 502, 503, 504] } - const fallbackHttpCodes = slice.model.fallbackTriggers?.httpCodes ?? defaults.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 }> = [] let triedUpstreams = 0 - for (let i = 0; i < slice.model.upstreams.length; i += 1) { - const upstream = slice.model.upstreams[i] + async function attemptUpstream(upstream: LlmUpstream, index: number) { const provider = deriveProviderTag(upstream.baseURL) triedUpstreams += 1 // Surface the current upstream so the caller can label success metrics @@ -387,7 +408,7 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { const result = await dispatchOneUpstream( upstream, - i, + index, req, perAttemptTimeoutMs, fallbackHttpCodes, @@ -397,14 +418,70 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { if (result.kind === 'ok') { if (ctx) ctx.upstreamModel = result.upstreamModel - return result.response + return { kind: 'ok' as const, response: result.response } } - // This upstream exhausted; record and continue. options.gatewayMetrics?.keyExhaustedCount.add(1, { provider }) + return { + kind: 'exhausted' as const, + statuses: result.failures.map(failure => failure.status), + } } - // FULL exhaustion: every upstream's every key failed. + async function routeGroup(group: LlmRoutingGroup): Promise< + | { kind: 'ok', response: Response } + | { kind: 'exhausted', statuses: Array, transitionBlocked: boolean } + > { + const statuses: Array = [] + for (let groupCandidateIndex = 0; groupCandidateIndex < group.upstreamIds.length; groupCandidateIndex += 1) { + const upstreamId = group.upstreamIds[groupCandidateIndex] + const index = llmModel.upstreams.findIndex(upstream => upstream.id === upstreamId) + if (index === -1) { + throw new Error( + `LLM routing group ${group.id} references unknown upstream ${upstreamId} for model ${req.modelName}`, + ) + } + + const result = await attemptUpstream(llmModel.upstreams[index], index) + if (result.kind === 'ok') + return result + 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: false } + } + + if (llmModel.routing != null) { + for (let groupIndex = 0; groupIndex < llmModel.routing.groups.length; groupIndex += 1) { + const group = llmModel.routing.groups[groupIndex] + const result = await routeGroup(group) + if (result.kind === 'ok') + return result.response + + const hasNextGroup = groupIndex < llmModel.routing.groups.length - 1 + if ( + result.transitionBlocked + || !hasNextGroup + || !failuresMatch(result.statuses, group.continueOn) + ) { + break + } + } + } + else { + for (let index = 0; index < llmModel.upstreams.length; index += 1) { + const result = await attemptUpstream(llmModel.upstreams[index], index) + if (result.kind === 'ok') + return result.response + } + } + + // Terminal exhaustion: every transition allowed by the active provider + // route has failed. A policy boundary may intentionally leave later + // upstreams untouched. const lastFailure = allFailures.at(-1) if (lastFailure == null) { // Should not happen: schema guarantees ≥1 upstream and ≥1 key. Treat @@ -412,9 +489,9 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { throw new Error(`Router exhausted with no recorded failures for model ${req.modelName}`) } - // Same-status exhaustion: every recorded failure shares the same - // status (or 'timeout'). Strong signal of an account-level / shared- - // backend cap that ordinary fallback cannot recover from. + // Same-status exhaustion: every recorded failure shares one status (or + // timeout). This is a strong signal of a shared upstream constraint that + // ordinary candidate fallback cannot recover from. const distinctStatuses = new Set(allFailures.map(f => f.status)) if (distinctStatuses.size === 1) { const status = allFailures[0].status @@ -553,8 +630,9 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { status_code: rawStatus, }) if (!fallbackHttpCodes.includes(rawStatus)) { - // Same policy as chat: non-fallback status stops this upstream - // but the outer loop still tries the next upstream. + // 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 } @@ -572,14 +650,14 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { } /** - * Capacity-aware layer over {@link dispatchOneTtsUpstream}: spreads one TTS - * request across the model's pool (one app_id per upstream) by least-loaded - * ordering, gating each dispatch on an atomic concurrency-slot acquire. + * Capacity-aware layer over {@link dispatchOneTtsUpstream}: gates each capped + * dispatch on an atomic concurrency-slot acquire. It can preserve configured + * order or rank equivalent accounts by current in-flight usage. * * Returns: - * - the 2xx `Response` on success, - * - `null` when every dispatched upstream exhausted (caller maps the recorded - * failures to an upstream error via the shared exhaustion path), + * - an `ok` result with the 2xx `Response`, + * - an `exhausted` result with every attempted status and whether a retry + * policy blocked the remaining peers, * - throws 503 `TTS_POOL_SATURATED` when every pool was at capacity or in a * 429 cool-down so nothing was dispatched - fail-fast with context, never a * silent stall (origin R3). @@ -589,9 +667,14 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { modelName: string, attemptUpstream: (upstream: TtsUpstream, index: number) => Promise< | { kind: 'ok', response: Response } - | { kind: 'exhausted', sawTooManyRequests: boolean } + | { kind: 'exhausted', sawTooManyRequests: boolean, statuses: Array } >, - ): Promise { + retryOn?: RouteFailureTriggers, + strategy: 'least-inflight' | 'ordered' = 'least-inflight', + ): Promise< + | { kind: 'ok', response: Response } + | { kind: 'exhausted', statuses: Array, transitionBlocked: boolean } + > { async function markSaturated(upstream: TtsUpstream, poolId: string): Promise { await ledger.markSaturated(poolId, ttsPoolSaturationTtlSeconds) options.gatewayMetrics?.poolSaturationMarked.add(1, { @@ -600,12 +683,11 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { }) } - // Best-effort pre-read: order pools least-loaded-first (spreads load) and - // drop pools already full or in a saturation cool-down. tryAcquire below is - // the authoritative gate against the cross-replica race — ordering only - // decides *preference*, not correctness. - const ranked = (await Promise.all(upstreams.map(async (upstream, index) => { - const poolId = ttsPoolId(upstream) + // Best-effort pre-read drops pools already full or in a saturation + // cool-down. tryAcquire below remains the authoritative gate against the + // cross-replica race. + const candidates = await Promise.all(upstreams.map(async (upstream, index) => { + const poolId = ttsPoolId(upstream, modelName) const maxConcurrency = typeof upstream.maxConcurrency === 'number' ? upstream.maxConcurrency : null const saturated = await ledger.isSaturated(poolId) if (saturated) { @@ -614,30 +696,39 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { index, poolId, maxConcurrency, - remaining: maxConcurrency == null ? Number.POSITIVE_INFINITY : 0, + inflight: Number.POSITIVE_INFINITY, eligible: false, } } if (maxConcurrency == null) - return { upstream, index, poolId, maxConcurrency, remaining: Number.POSITIVE_INFINITY, eligible: true } + return { upstream, index, poolId, maxConcurrency, inflight: 0, eligible: true } const inflight = await ledger.currentInflight(poolId) - const remaining = maxConcurrency - inflight - return { upstream, index, poolId, maxConcurrency, remaining, eligible: remaining > 0 } - }))) - .filter(c => c.eligible) - .sort((a, b) => b.remaining - a.remaining) + return { upstream, index, poolId, maxConcurrency, inflight, eligible: inflight < maxConcurrency } + })) + const eligible = candidates.filter(c => c.eligible) + const ranked = strategy === 'least-inflight' + ? eligible.sort((a, b) => a.inflight - b.inflight) + : eligible let dispatchedAny = false - for (const { upstream, index, poolId, maxConcurrency } of ranked) { + let attemptedPools = 0 + const statuses: Array = [] + for (let rankedIndex = 0; rankedIndex < ranked.length; rankedIndex += 1) { + const { upstream, index, poolId, maxConcurrency } = ranked[rankedIndex] + const hasNextCandidate = rankedIndex < ranked.length - 1 if (maxConcurrency == null) { // Unlimited pool — dispatch without occupying a slot. dispatchedAny = true + attemptedPools += 1 const result = await attemptUpstream(upstream, index) if (result.kind === 'ok') - return result.response + return result + statuses.push(...result.statuses) if (result.sawTooManyRequests) await markSaturated(upstream, poolId) + if (hasNextCandidate && retryOn != null && !failuresMatch(result.statuses, retryOn)) + return { kind: 'exhausted', statuses, transitionBlocked: true } continue } @@ -652,12 +743,16 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { } dispatchedAny = true + attemptedPools += 1 try { const result = await attemptUpstream(upstream, index) if (result.kind === 'ok') - return result.response + return result + statuses.push(...result.statuses) if (result.sawTooManyRequests) await markSaturated(upstream, poolId) + if (hasNextCandidate && retryOn != null && !failuresMatch(result.statuses, retryOn)) + return { kind: 'exhausted', statuses, transitionBlocked: true } } finally { await ledger.release(poolId) @@ -672,7 +767,15 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { ) } - return null + // Advancing past a group requires evidence from every candidate. A pool + // skipped because it was full, circuit-broken, or lost the acquire race has + // not produced a failure status, so it must block the transition even when + // every dispatched pool returned an allowed status. + return { + kind: 'exhausted', + statuses, + transitionBlocked: attemptedPools !== upstreams.length, + } } async function routeTts(req: { modelName: string, input: TtsInput, abortSignal?: AbortSignal }, ctx?: LlmRouteContext): Promise { @@ -694,9 +797,6 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { const defaults = slice.defaults ?? { perAttemptTimeoutMs: 30000, fullChainTimeoutMs: 60000, fallbackHttpCodes: [401, 402, 403, 429, 500, 502, 503, 504] } const fallbackHttpCodes = ttsModel.fallbackTriggers?.httpCodes ?? defaults.fallbackHttpCodes ?? [401, 402, 403, 429, 500, 502, 503, 504] - // Adapters POST to unspeech `/v1/audio/speech`; resolve the base URL once - // per request rather than per upstream attempt so a single configKV miss - // surfaces as a clean 503 before any key rotation happens. const unspeechBaseURL = (await options.configKV.getOrThrow('UNSPEECH_UPSTREAM')).restBaseURL const allFailures: Array<{ provider: string, keyId: string, status: number | 'timeout', errorMessage?: string }> = [] @@ -712,7 +812,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 } + | { kind: 'exhausted', sawTooManyRequests: boolean, statuses: Array } > { const providerTag = deriveProviderTag(upstream.baseURL) triedUpstreams += 1 @@ -741,25 +841,93 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { } options.gatewayMetrics?.keyExhaustedCount.add(1, { provider: providerTag }) - return { kind: 'exhausted', sawTooManyRequests: result.failures.some(f => f.status === 429) } + return { + kind: 'exhausted', + sawTooManyRequests: result.failures.some(f => f.status === 429), + statuses: result.failures.map(failure => failure.status), + } } - // A model "uses the pool" when any upstream declares a concurrency cap. Models - // without one keep the original fixed-order fallback and make zero Redis - // calls — no behavior change for existing single-app configs. - const poolingEnabled = ttsModel.upstreams.some(u => typeof u.maxConcurrency === 'number') + async function routeGroup(group: TtsRoutingGroup): Promise< + | { kind: 'ok', response: Response } + | { kind: 'exhausted', statuses: Array, transitionBlocked: boolean } + > { + const indexedUpstreams = group.upstreamIds.map((upstreamId) => { + const index = ttsModel.upstreams.findIndex(upstream => upstream.id === upstreamId) + if (index === -1) { + throw new Error( + `TTS routing group ${group.id} references unknown upstream ${upstreamId} for model ${req.modelName}`, + ) + } + return { upstream: ttsModel.upstreams[index], index } + }) - if (!poolingEnabled) { - for (let i = 0; i < ttsModel.upstreams.length; i += 1) { - const result = await attemptUpstream(ttsModel.upstreams[i], i) + const capacityManaged = indexedUpstreams.some(({ upstream }) => upstream.maxConcurrency != null) + if (group.strategy === 'least-inflight' || capacityManaged) { + const indexByUpstream = new Map(indexedUpstreams.map(({ upstream, index }) => [upstream, index])) + return routeTtsAcrossPools( + indexedUpstreams.map(({ upstream }) => upstream), + req.modelName, + (upstream) => { + const index = indexByUpstream.get(upstream) + if (index == null) + throw new Error(`TTS routing lost upstream index for model ${req.modelName}`) + return attemptUpstream(upstream, index) + }, + group.retryOn, + group.strategy, + ) + } + + const statuses: Array = [] + for (let groupCandidateIndex = 0; groupCandidateIndex < indexedUpstreams.length; groupCandidateIndex += 1) { + const { upstream, index } = indexedUpstreams[groupCandidateIndex] + const result = await attemptUpstream(upstream, index) + if (result.kind === 'ok') + return result + 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: false } + } + + if (ttsModel.routing != null) { + for (let groupIndex = 0; groupIndex < ttsModel.routing.groups.length; groupIndex += 1) { + const group = ttsModel.routing.groups[groupIndex] + const result = await routeGroup(group) if (result.kind === 'ok') return result.response + + const hasNextGroup = groupIndex < ttsModel.routing.groups.length - 1 + if ( + result.transitionBlocked + || !hasNextGroup + || !failuresMatch(result.statuses, group.continueOn) + ) { + break + } } } else { - const served = await routeTtsAcrossPools(ttsModel.upstreams, req.modelName, attemptUpstream) - if (served != null) - return served + // Models without an explicit provider route preserve the established + // behavior: fixed order unless any upstream declares a concurrency cap. + const poolingEnabled = ttsModel.upstreams.some(u => typeof u.maxConcurrency === 'number') + + if (!poolingEnabled) { + for (let i = 0; i < ttsModel.upstreams.length; i += 1) { + const result = await attemptUpstream(ttsModel.upstreams[i], i) + if (result.kind === 'ok') + return result.response + } + } + else { + const result = await routeTtsAcrossPools(ttsModel.upstreams, req.modelName, attemptUpstream) + if (result.kind === 'ok') + return result.response + } } const lastFailure = allFailures.at(-1) diff --git a/apps/server/src/services/domain/llm-router/tests/router.test.ts b/apps/server/src/services/domain/llm-router/tests/router.test.ts index c9627ef4c..a6546d262 100644 --- a/apps/server/src/services/domain/llm-router/tests/router.test.ts +++ b/apps/server/src/services/domain/llm-router/tests/router.test.ts @@ -77,7 +77,7 @@ function makeLedger(overrides: Partial = {}): ConcurrencyLedg function makeConfigKV(config: RouterConfig | null): ConfigKVService { return { getOptional: vi.fn(async (key: string) => (key === 'LLM_ROUTER_CONFIG' ? config : null)), - // routeTts reads UNSPEECH_UPSTREAM once per request via getOrThrow. + // routeTts resolves UNSPEECH_UPSTREAM lazily when the chosen adapter needs it. // LLM-side tests never invoke routeTts so the value is irrelevant; TTS // tests need a populated restBaseURL. getOrThrow: vi.fn(async (key: string) => { @@ -691,6 +691,121 @@ describe('createLlmRouterService', () => { expect((configKV.getOptional as ReturnType).mock.calls.length).toBe(2) }) + describe('route LLM provider groups', () => { + function makeGroupedLlmRouter(fetchImpl: typeof fetch) { + const { config, crypto } = makeConfig({ + upstreams: [ + { baseURL: 'https://api.stepfun.com/step_plan/v1', keyIds: ['plan-a'] }, + { baseURL: 'https://api.stepfun.com/step_plan/v1', keyIds: ['plan-b'] }, + { baseURL: 'https://api.stepfun.com/v1', keyIds: ['paygo'] }, + ], + }) + const model = config.llm.models['openai/gpt-5-mini'] + Object.assign(model.upstreams[0], { id: 'plan-a' }) + Object.assign(model.upstreams[1], { id: 'plan-b' }) + Object.assign(model.upstreams[2], { id: 'paygo' }) + Object.assign(model, { + routing: { + groups: [ + { + id: 'plan', + upstreamIds: ['plan-a', 'plan-b'], + retryOn: { + httpCodes: [402, 429, 500, 502, 503, 504], + onTimeout: true, + }, + continueOn: { + httpCodes: [402], + onTimeout: false, + }, + }, + { + id: 'paygo', + upstreamIds: ['paygo'], + retryOn: { + httpCodes: [429, 500, 502, 503, 504], + onTimeout: true, + }, + }, + ], + }, + }) + + return createLlmRouterService({ + configKV: makeConfigKV(config), + envelopeCrypto: crypto, + gatewayMetrics: makeMetrics(), + fetchImpl, + redis: makeRedisStub(), + concurrencyLedger: makeLedger(), + }) + } + + it('uses the ordinary LLM API only after every Plan account returns 402', async () => { + const calledURLs: string[] = [] + const fetchImpl = vi.fn(async (input: string | URL | Request) => { + const url = String(input) + calledURLs.push(url) + if (url.includes('/step_plan/')) + return failResponse(402, { error: { code: 'quota_exceeded' } }) + return happyResponse({ id: 'completion' }) + }) as unknown as typeof fetch + + const router = makeGroupedLlmRouter(fetchImpl) + const response = await router.route({ + modelName: 'openai/gpt-5-mini', + body: { messages: [] }, + }) + + expect(response.status).toBe(200) + expect(calledURLs).toEqual([ + 'https://api.stepfun.com/step_plan/v1/chat/completions', + 'https://api.stepfun.com/step_plan/v1/chat/completions', + 'https://api.stepfun.com/v1/chat/completions', + ]) + }) + + it('does not spend ordinary LLM API balance when the Plan group is rate-limited', async () => { + const calledURLs: string[] = [] + const fetchImpl = vi.fn(async (input: string | URL | Request) => { + calledURLs.push(String(input)) + return failResponse(429) + }) as unknown as typeof fetch + + const router = makeGroupedLlmRouter(fetchImpl) + + await expect(router.route({ + modelName: 'openai/gpt-5-mini', + body: { messages: [] }, + })).rejects.toBeInstanceOf(ApiError) + + expect(calledURLs).toEqual([ + 'https://api.stepfun.com/step_plan/v1/chat/completions', + 'https://api.stepfun.com/step_plan/v1/chat/completions', + ]) + expect(calledURLs).not.toContain('https://api.stepfun.com/v1/chat/completions') + }) + + it('stops the LLM provider route immediately on Plan authentication failure', async () => { + const calledURLs: string[] = [] + const fetchImpl = vi.fn(async (input: string | URL | Request) => { + calledURLs.push(String(input)) + return failResponse(401) + }) as unknown as typeof fetch + + const router = makeGroupedLlmRouter(fetchImpl) + + await expect(router.route({ + modelName: 'openai/gpt-5-mini', + body: { messages: [] }, + })).rejects.toBeInstanceOf(ApiError) + + expect(calledURLs).toEqual([ + 'https://api.stepfun.com/step_plan/v1/chat/completions', + ]) + }) + }) + // --- routeTts adapter error contract ------------------------------------- // // ROOT CAUSE: @@ -899,9 +1014,305 @@ describe('createLlmRouterService', () => { }) }) + describe('routeTts provider groups', () => { + function endpointProfileFrom(init?: RequestInit): string { + const body = JSON.parse(String(init?.body)) as { + extra_body?: { endpoint_profile?: string } + } + return body.extra_body?.endpoint_profile ?? 'default' + } + + function makeGroupedStepfunConfig(): { config: RouterConfig, crypto: ReturnType } { + const crypto = createEnvelopeCrypto({ masterKey: freshMasterKey() }) + const modelName = 'stepfun/stepaudio-2.5-tts' + const upstreams = [ + { + id: 'plan-a', + baseURL: 'https://api.stepfun.com', + keyId: 'plan-key-a', + endpointProfile: 'step-plan', + }, + { + id: 'plan-b', + baseURL: 'https://api.stepfun.com', + keyId: 'plan-key-b', + endpointProfile: 'step-plan', + }, + { + id: 'paygo', + baseURL: 'https://api.stepfun.com', + keyId: 'paygo-key', + endpointProfile: 'default', + }, + ].map(upstream => ({ + id: upstream.id, + baseURL: upstream.baseURL, + keys: [{ + id: upstream.keyId, + ciphertext: crypto.encryptKey(`sk-${upstream.keyId}`, { + modelName, + keyEntryId: upstream.keyId, + }), + }], + adapterParams: { + endpointProfile: upstream.endpointProfile, + model: 'stepaudio-2.5-tts', + }, + })) + + const config = { + llm: { models: {} }, + tts: { + models: { + [modelName]: { + provider: 'stepfun', + upstreams, + routing: { + groups: [ + { + id: 'plan', + upstreamIds: ['plan-a', 'plan-b'], + strategy: 'ordered', + retryOn: { + httpCodes: [402, 429, 500, 502, 503, 504], + onTimeout: true, + }, + continueOn: { + httpCodes: [402], + onTimeout: false, + }, + }, + { + id: 'paygo', + upstreamIds: ['paygo'], + strategy: 'ordered', + retryOn: { + httpCodes: [429, 500, 502, 503, 504], + onTimeout: true, + }, + }, + ], + }, + fallbackTriggers: { + httpCodes: [401, 402, 429, 500, 502, 503, 504], + onTimeout: true, + }, + }, + }, + }, + defaults: { + perAttemptTimeoutMs: 5000, + fullChainTimeoutMs: 10000, + fallbackHttpCodes: [401, 402, 429, 500, 502, 503, 504], + }, + } as unknown as RouterConfig + + return { config, crypto } + } + + function makeGroupedStepfunRouter( + fetchImpl: typeof fetch, + ): ReturnType { + const { config, crypto } = makeGroupedStepfunConfig() + return createLlmRouterService({ + configKV: makeConfigKV(config), + envelopeCrypto: crypto, + gatewayMetrics: makeMetrics(), + fetchImpl, + redis: makeRedisStub(), + concurrencyLedger: makeLedger(), + }) + } + + it('uses pay-as-you-go only after every Plan account reports quota exhaustion', async () => { + const calledProfiles: string[] = [] + const fetchImpl = vi.fn(async (input: string | URL | Request, init?: RequestInit) => { + expect(String(input)).toBe('http://unspeech.local:5933/v1/audio/speech') + const profile = endpointProfileFrom(init) + calledProfiles.push(profile) + if (profile === 'step-plan') + return failResponse(402, { error: { code: 'quota_exceeded' } }) + return new Response(new Uint8Array([0x01]), { + status: 200, + headers: { 'content-type': 'audio/mpeg' }, + }) + }) as unknown as typeof fetch + + const router = makeGroupedStepfunRouter(fetchImpl) + const response = await router.routeTts({ + modelName: 'stepfun/stepaudio-2.5-tts', + input: { text: '你好' }, + }) + + expect(response.status).toBe(200) + expect(calledProfiles).toEqual(['step-plan', 'step-plan', 'default']) + }) + + it('stays inside the Plan group when a Plan account succeeds', async () => { + const calledProfiles: string[] = [] + const fetchImpl = vi.fn(async (input: string | URL | Request, init?: RequestInit) => { + expect(String(input)).toBe('http://unspeech.local:5933/v1/audio/speech') + calledProfiles.push(endpointProfileFrom(init)) + return new Response(new Uint8Array([0x01]), { + status: 200, + headers: { 'content-type': 'audio/mpeg' }, + }) + }) as unknown as typeof fetch + + const router = makeGroupedStepfunRouter(fetchImpl) + const response = await router.routeTts({ + modelName: 'stepfun/stepaudio-2.5-tts', + input: { text: '你好' }, + }) + + expect(response.status).toBe(200) + expect(calledProfiles).toEqual(['step-plan']) + }) + + it('requires UNSPEECH_UPSTREAM for an endpoint-profile request', async () => { + const { config, crypto } = makeGroupedStepfunConfig() + const configKV = makeConfigKV(config) + const getOrThrow = vi.fn(async () => { + throw new Error('UNSPEECH_UPSTREAM not configured') + }) + Object.assign(configKV, { getOrThrow }) + const fetchImpl = vi.fn(async () => new Response(new Uint8Array([0x01]), { + status: 200, + headers: { 'content-type': 'audio/mpeg' }, + })) as unknown as typeof fetch + const router = createLlmRouterService({ + configKV, + envelopeCrypto: crypto, + gatewayMetrics: makeMetrics(), + fetchImpl, + redis: makeRedisStub(), + concurrencyLedger: makeLedger(), + }) + + await expect(router.routeTts({ + modelName: 'stepfun/stepaudio-2.5-tts', + input: { text: '你好' }, + })).rejects.toThrow('UNSPEECH_UPSTREAM not configured') + + expect(fetchImpl).not.toHaveBeenCalled() + expect(getOrThrow).toHaveBeenCalledWith('UNSPEECH_UPSTREAM') + }) + + it('requires UNSPEECH_UPSTREAM for the StepFun voice catalog', async () => { + const { config, crypto } = makeGroupedStepfunConfig() + const configKV = makeConfigKV(config) + const getOrThrow = vi.fn(async () => { + throw new Error('UNSPEECH_UPSTREAM not configured') + }) + Object.assign(configKV, { getOrThrow }) + const fetchImpl = vi.fn() as unknown as typeof fetch + const router = createLlmRouterService({ + configKV, + envelopeCrypto: crypto, + gatewayMetrics: makeMetrics(), + fetchImpl, + redis: makeRedisStub(), + concurrencyLedger: makeLedger(), + }) + + await expect( + router.listTtsVoices('stepfun/stepaudio-2.5-tts'), + ).rejects.toThrow('UNSPEECH_UPSTREAM not configured') + + expect(fetchImpl).not.toHaveBeenCalled() + expect(getOrThrow).toHaveBeenCalledWith('UNSPEECH_UPSTREAM') + }) + + it('keeps a Step Plan attempt timeout distinct from HTTP 500 at the paid boundary', async () => { + const { config, crypto } = makeGroupedStepfunConfig() + config.defaults!.perAttemptTimeoutMs = 10 + const model = config.tts.models['stepfun/stepaudio-2.5-tts'] + model.routing!.groups[0].continueOn = { + httpCodes: [500], + onTimeout: false, + } + const calledProfiles: string[] = [] + const fetchImpl = vi.fn((input: string | URL | Request, init?: RequestInit) => { + expect(String(input)).toBe('http://unspeech.local:5933/v1/audio/speech') + const profile = endpointProfileFrom(init) + calledProfiles.push(profile) + if (profile !== 'step-plan') { + return Promise.resolve(new Response(new Uint8Array([0x01]), { + status: 200, + headers: { 'content-type': 'audio/mpeg' }, + })) + } + + return new Promise((_, reject) => { + const rejectAbort = () => reject(init?.signal?.reason ?? new Error('aborted')) + if (init?.signal?.aborted) + rejectAbort() + else + init?.signal?.addEventListener('abort', rejectAbort, { once: true }) + }) + }) as unknown as typeof fetch + const router = createLlmRouterService({ + configKV: makeConfigKV(config), + envelopeCrypto: crypto, + gatewayMetrics: makeMetrics(), + fetchImpl, + redis: makeRedisStub(), + concurrencyLedger: makeLedger(), + }) + + await expect(router.routeTts({ + modelName: 'stepfun/stepaudio-2.5-tts', + input: { text: '你好' }, + })).rejects.toMatchObject({ + statusCode: 504, + details: expect.objectContaining({ lastStatusCode: 'timeout' }), + }) + + expect(calledProfiles).toEqual(['step-plan', 'step-plan']) + expect(calledProfiles).not.toContain('default') + }) + + it('does not cross the paid boundary when the Plan group is rate-limited', async () => { + const calledProfiles: string[] = [] + const fetchImpl = vi.fn(async (input: string | URL | Request, init?: RequestInit) => { + expect(String(input)).toBe('http://unspeech.local:5933/v1/audio/speech') + calledProfiles.push(endpointProfileFrom(init)) + return failResponse(429) + }) as unknown as typeof fetch + + const router = makeGroupedStepfunRouter(fetchImpl) + + await expect(router.routeTts({ + modelName: 'stepfun/stepaudio-2.5-tts', + input: { text: '你好' }, + })).rejects.toBeInstanceOf(ApiError) + + expect(calledProfiles).toEqual(['step-plan', 'step-plan']) + expect(calledProfiles).not.toContain('default') + }) + + it('stops the provider route immediately on a Plan authentication failure', async () => { + const calledProfiles: string[] = [] + const fetchImpl = vi.fn(async (input: string | URL | Request, init?: RequestInit) => { + expect(String(input)).toBe('http://unspeech.local:5933/v1/audio/speech') + calledProfiles.push(endpointProfileFrom(init)) + return failResponse(401) + }) as unknown as typeof fetch + + const router = makeGroupedStepfunRouter(fetchImpl) + + await expect(router.routeTts({ + modelName: 'stepfun/stepaudio-2.5-tts', + input: { text: '你好' }, + })).rejects.toBeInstanceOf(ApiError) + + expect(calledProfiles).toEqual(['step-plan']) + }) + }) + describe('routeTtspool capacity-aware routing', () => { // One app_id == one upstream (Volcengine `adapterParams.appid`), each capped - // at `maxConcurrency`. The router spreads load least-loaded-first across pools + // at `maxConcurrency`. The router spreads load least-inflight-first across pools // and circuit-breaks a pool on 429 (app_id concurrency exceeded upstream-side). function makePoolConfig( upstreams: Array<{ baseURL: string, appid: string, maxConcurrency?: number }>, @@ -934,7 +1345,7 @@ describe('createLlmRouterService', () => { return { config, crypto } } - // Stateful in-memory ledger so least-loaded ordering and capacity gating are + // Stateful in-memory ledger so least-inflight ordering and capacity gating are // observable. `seed` pre-loads inflight counts to drive deterministic ranking. function makeStatefulLedger(seed: Record = {}, saturatedSeed: string[] = []) { const inflight = new Map(Object.entries(seed)) @@ -974,7 +1385,7 @@ describe('createLlmRouterService', () => { }) } - it('routes to the least-loadedpool (covers AE1 — load spread, not first-fill)', async () => { + it('routes to the least-inflight pool (covers AE1 — load spread, not first-fill)', async () => { // @example two app_ids cap 10, seeded 8 vs 2 in-flight -> the new request // goes to the freer pool (app-2), not the config-first pool (app-1). const { config, crypto } = makePoolConfig([ @@ -992,6 +1403,208 @@ describe('createLlmRouterService', () => { expect(tryAcquire.mock.calls[0][0]).toBe('app-2') }) + it('ranks least-inflight accounts by current usage when concurrency caps differ', async () => { + const { config, crypto } = makePoolConfig([ + { baseURL: 'https://up-a.example', appid: 'app-1', maxConcurrency: 100 }, + { baseURL: 'https://up-b.example', appid: 'app-2', maxConcurrency: 10 }, + ]) + const { ledger, tryAcquire } = makeStatefulLedger({ 'app-1': 50, 'app-2': 0 }) + const fetchImpl = vi.fn(async () => happyResponse({ ok: 1 })) as unknown as typeof fetch + + const router = makePoolRouter(config, crypto, ledger, fetchImpl) + const response = await router.routeTts({ modelName: 'tts-pool', input: { text: 'hi' } }) + + expect(response.status).toBe(200) + expect(tryAcquire).toHaveBeenCalledTimes(1) + expect(tryAcquire).toHaveBeenCalledWith('app-2', 10) + }) + + it('namespaces a non-appid pool by model and upstream id', async () => { + const crypto = createEnvelopeCrypto({ masterKey: freshMasterKey() }) + const modelName = 'stepfun/stepaudio-2.5-tts' + const keyEntryId = 'plan-key' + const config = { + llm: { models: {} }, + tts: { + models: { + [modelName]: { + provider: 'stepfun', + upstreams: [{ + id: 'plan', + baseURL: 'https://api.stepfun.com', + keys: [{ + id: keyEntryId, + ciphertext: crypto.encryptKey('sk-plan', { modelName, keyEntryId }), + }], + adapterParams: { + endpointProfile: 'step-plan', + model: 'stepaudio-2.5-tts', + }, + maxConcurrency: 1, + }], + routing: { + groups: [{ + id: 'plan', + upstreamIds: ['plan'], + strategy: 'least-inflight', + retryOn: { httpCodes: [402, 429, 500, 502, 503, 504], onTimeout: true }, + }], + }, + fallbackTriggers: { httpCodes: [402, 429, 500, 502, 503, 504], onTimeout: true }, + }, + }, + }, + defaults: { + perAttemptTimeoutMs: 5000, + fullChainTimeoutMs: 10000, + fallbackHttpCodes: [402, 429, 500, 502, 503, 504], + }, + } as RouterConfig + const { ledger, tryAcquire } = makeStatefulLedger() + const fetchImpl = vi.fn(async () => new Response(new Uint8Array([0x01]), { + status: 200, + headers: { 'content-type': 'audio/mpeg' }, + })) as unknown as typeof fetch + + const router = makePoolRouter(config, crypto, ledger, fetchImpl) + const response = await router.routeTts({ modelName, input: { text: 'hi' } }) + + expect(response.status).toBe(200) + expect(tryAcquire).toHaveBeenCalledWith( + 'model:["stepfun/stepaudio-2.5-tts","id","plan"]', + 1, + ) + }) + + it('enforces maxConcurrency for an ordered provider group', async () => { + const { config, crypto } = makePoolConfig([ + { baseURL: 'https://up-a.example', appid: 'app-1', maxConcurrency: 10 }, + ]) + const model = config.tts.models['tts-pool'] + Object.assign(model.upstreams[0], { id: 'primary' }) + Object.assign(model, { + routing: { + groups: [{ + id: 'primary', + upstreamIds: ['primary'], + strategy: 'ordered', + retryOn: { httpCodes: [429, 500, 502, 503, 504], onTimeout: true }, + }], + }, + }) + const { ledger, tryAcquire, release } = makeStatefulLedger() + const fetchImpl = vi.fn(async () => happyResponse({ ok: 1 })) as unknown as typeof fetch + + const router = makePoolRouter(config, crypto, ledger, fetchImpl) + const response = await router.routeTts({ modelName: 'tts-pool', input: { text: 'hi' } }) + + expect(response.status).toBe(200) + expect(tryAcquire).toHaveBeenCalledWith('app-1', 10) + expect(release).toHaveBeenCalledWith('app-1') + }) + + it('keeps least-inflight selection inside the active group before considering pay-as-you-go', async () => { + const { config, crypto } = makePoolConfig([ + { baseURL: 'https://plan-a.example', appid: 'plan-a', maxConcurrency: 10 }, + { baseURL: 'https://plan-b.example', appid: 'plan-b', maxConcurrency: 10 }, + { baseURL: 'https://paygo.example', appid: 'paygo' }, + ]) + const model = config.tts.models['tts-pool'] + Object.assign(model.upstreams[0], { id: 'plan-a' }) + Object.assign(model.upstreams[1], { id: 'plan-b' }) + Object.assign(model.upstreams[2], { id: 'paygo' }) + Object.assign(model, { + routing: { + groups: [ + { + id: 'plan', + upstreamIds: ['plan-a', 'plan-b'], + strategy: 'least-inflight', + retryOn: { httpCodes: [429, 500, 502, 503, 504], onTimeout: true }, + continueOn: { httpCodes: [402], onTimeout: false }, + }, + { + id: 'paygo', + upstreamIds: ['paygo'], + strategy: 'ordered', + retryOn: { httpCodes: [429, 500, 502, 503, 504], onTimeout: true }, + }, + ], + }, + }) + const { ledger, tryAcquire } = makeStatefulLedger({ 'plan-a': 8, 'plan-b': 2 }) + const selectedAppIds: string[] = [] + const fetchImpl = vi.fn(async (_input: string | URL | Request, init?: RequestInit) => { + const body = JSON.parse(String(init?.body)) as { extra_body?: { app?: { appid?: string } } } + selectedAppIds.push(body.extra_body?.app?.appid ?? 'unknown') + return new Response(new Uint8Array([0x01]), { + status: 200, + headers: { 'content-type': 'audio/mpeg' }, + }) + }) as unknown as typeof fetch + + const router = makePoolRouter(config, crypto, ledger, fetchImpl) + const response = await router.routeTts({ modelName: 'tts-pool', input: { text: 'hi' } }) + + expect(response.status).toBe(200) + expect(selectedAppIds).toEqual(['plan-b']) + expect(tryAcquire).toHaveBeenCalledTimes(1) + expect(tryAcquire).toHaveBeenCalledWith('plan-b', 10) + }) + + it('does not cross groups when a Plan account was skipped at its concurrency limit', async () => { + const { config, crypto } = makePoolConfig([ + { baseURL: 'https://plan-a.example', appid: 'plan-a', maxConcurrency: 10 }, + { baseURL: 'https://plan-b.example', appid: 'plan-b', maxConcurrency: 10 }, + { baseURL: 'https://paygo.example', appid: 'paygo' }, + ]) + const model = config.tts.models['tts-pool'] + Object.assign(model.upstreams[0], { id: 'plan-a' }) + Object.assign(model.upstreams[1], { id: 'plan-b' }) + Object.assign(model.upstreams[2], { id: 'paygo' }) + Object.assign(model, { + routing: { + groups: [ + { + id: 'plan', + upstreamIds: ['plan-a', 'plan-b'], + strategy: 'least-inflight', + retryOn: { httpCodes: [402, 429, 500, 502, 503, 504], onTimeout: true }, + continueOn: { httpCodes: [402], onTimeout: false }, + }, + { + id: 'paygo', + upstreamIds: ['paygo'], + strategy: 'ordered', + retryOn: { httpCodes: [429, 500, 502, 503, 504], onTimeout: true }, + }, + ], + }, + }) + const { ledger } = makeStatefulLedger({ 'plan-a': 10, 'plan-b': 0 }) + const selectedAppIds: string[] = [] + const fetchImpl = vi.fn(async (_input: string | URL | Request, init?: RequestInit) => { + const body = JSON.parse(String(init?.body)) as { extra_body?: { app?: { appid?: string } } } + const appid = body.extra_body?.app?.appid ?? 'unknown' + selectedAppIds.push(appid) + if (appid === 'plan-b') + return failResponse(402, { error: { code: 'quota_exceeded' } }) + return new Response(new Uint8Array([0x01]), { + status: 200, + headers: { 'content-type': 'audio/mpeg' }, + }) + }) 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) + + expect(selectedAppIds).toEqual(['plan-b']) + }) + it('skips a fullpool and dispatches to one with capacity', async () => { // @example app-1 at cap (10/10) -> filtered out; app-2 (0/10) serves. const { config, crypto } = makePoolConfig([ diff --git a/apps/server/src/services/domain/llm-router/types.ts b/apps/server/src/services/domain/llm-router/types.ts index 6c8e72fab..ebce6fd4d 100644 --- a/apps/server/src/services/domain/llm-router/types.ts +++ b/apps/server/src/services/domain/llm-router/types.ts @@ -14,8 +14,13 @@ import type { llmModelSchema, llmRouterConfigSchema, llmRouterDefaultsSchema, + llmRoutingGroupSchema, + llmRoutingSchema, llmUpstreamSchema, + routeFailureTriggersSchema, ttsModelSchema, + ttsRoutingGroupSchema, + ttsRoutingSchema, ttsUpstreamSchema, } from '../../adapters/config-kv' @@ -31,25 +36,45 @@ export type RouterConfig = InferOutput export type RouterDefaults = InferOutput /** - * LLM upstream — one provider endpoint with its ordered key list. + * LLM upstream — one candidate endpoint with its ordered key list. */ export type LlmUpstream = InferOutput /** - * LLM model entry — ordered list of upstreams to try in fallback order. + * LLM model entry — upstream candidates plus an optional grouped route. */ export type LlmModel = InferOutput /** - * TTS upstream — one provider endpoint with adapter params + key list. + * LLM route composed from ordered candidate groups. + */ +export type LlmRouting = InferOutput + +/** + * One ordered group of interchangeable LLM candidates. + */ +export type LlmRoutingGroup = InferOutput + +/** + * TTS upstream — one candidate endpoint with adapter params + key list. */ export type TtsUpstream = InferOutput /** - * TTS model entry — provider tag + ordered upstreams. + * TTS model entry — provider tag, upstream candidates, and optional grouped route. */ export type TtsModel = InferOutput +/** + * TTS route composed from ordered candidate groups. + */ +export type TtsRouting = InferOutput + +/** + * One group of interchangeable TTS candidates. + */ +export type TtsRoutingGroup = InferOutput + /** * ASR model entry — provider tag + ordered upstreams for realtime transcription. */ @@ -66,6 +91,11 @@ export type AsrUpstream = InferOutput */ export type FallbackTriggers = InferOutput +/** + * Failure allow-list that authorizes a routing transition. + */ +export type RouteFailureTriggers = InferOutput + /** * One entry in `upstream.keys`: stable id + at-rest envelope ciphertext. * The plaintext key is only produced lazily by the key-rotator at call time. diff --git a/apps/server/src/utils/redis-keys.ts b/apps/server/src/utils/redis-keys.ts index 435db6d86..5146ebb34 100644 --- a/apps/server/src/utils/redis-keys.ts +++ b/apps/server/src/utils/redis-keys.ts @@ -46,9 +46,10 @@ export function lockRedisKey(domain: string, ...identifiers: RedisKeyPart[]): st /** * In-flight request counter for one TTSpool (per app_id concurrency pool). - * `poolId` is the upstream's `adapterParams.appid` (or baseURL fallback). The - * counter is INCR'd on slot acquire and DECR'd on release; a short TTL bounds - * leakage if a replica crashes between acquire and release. + * `poolId` is the upstream's global `adapterParams.appid` or a model-scoped + * upstream identity for providers without app ids. The counter is INCR'd on + * slot acquire and DECR'd on release; a short TTL bounds leakage if a replica + * crashes between acquire and release. */ export function ttsPoolInflightRedisKey(poolId: string): string { return redisKeyFrom('tts', 'pool', 'inflight', poolId)