diff --git a/apps/server/.env b/apps/server/.env index 2eff879ef..7bb02234f 100644 --- a/apps/server/.env +++ b/apps/server/.env @@ -21,6 +21,18 @@ API_SERVER_URL="" # OTEL_EXPORTER_OTLP_ENDPOINT="http://localhost:4318" -GATEWAY_BASE_URL="http://localhost:18080" -DEFAULT_CHAT_MODEL="openai/gpt-5-mini" -DEFAULT_TTS_MODEL="microsoft/v1" +# Master key for the in-process LLM/TTS router envelope crypto. +# Base64 of 32 random bytes. Required at boot: missing or wrong-length value +# fails env parsing and the server refuses to start (no graceful skip). +# Do NOT commit a real key here — put the value in `.env.local` (gitignored). +# +# Generate: +# openssl rand -base64 32 +# +# Rotation: when replacing the key, copy the old value to +# LLM_ROUTER_MASTER_KEY_PREVIOUS first, set LLM_ROUTER_MASTER_KEY to the new +# one, redeploy, then run a re-wrap of every stored ciphertext before +# dropping PREVIOUS. See `apps/server/src/utils/envelope-crypto.ts`. +LLM_ROUTER_MASTER_KEY="" +# LLM_ROUTER_MASTER_KEY_PREVIOUS="" + diff --git a/apps/server/CLAUDE.md b/apps/server/CLAUDE.md index add43d7d7..3192c9919 100644 --- a/apps/server/CLAUDE.md +++ b/apps/server/CLAUDE.md @@ -49,7 +49,7 @@ Local observability: `docker compose -f apps/server/docker-compose.otel.yml up - - **Flux read/write separation**: `FluxService` reads (Redis cache-aside), `BillingService` writes (single Postgres tx that mutates `user_flux` and writes the matching `flux_transaction` ledger row). Never put write-balance logic in `flux.ts`. - **No async billing pipeline**: debits and credits update balance + ledger in one transaction. The `(user_id, request_id)` partial unique index gives DB-level idempotency for retries; LLM `request log` rows are written best-effort right after the response is delivered. -- **LLM gateway proxy**: `/api/v1/openai` forwards to `GATEWAY_BASE_URL`. Server handles auth/billing/logging — not model execution. +- **In-process LLM/TTS router**: `/api/v1/openai` is dispatched by `services/llm-router` reading `LLM_ROUTER_CONFIG` (per-model upstream chain + envelope-encrypted keys). `chat/completions` walks LLM upstreams with key fallback; `audio/speech` delegates to a TTS adapter (`azure` / `dashscope-cosyvoice` / `volcengine`); `audio/voices` returns the adapter's compiled-in catalog. Server handles auth/billing/logging, not model execution. - **Redis is cache + pub/sub, not truth**: balance cache, app_settings read cache, WebSocket cross-instance pub/sub. Truth is always Postgres. - **Auth**: Better Auth + OIDC. `sessionMiddleware` fills context but doesn't block; `authGuard` returns 401. - **Multi-instance safe**: all writes go through Postgres transactions; cross-instance messaging uses Redis Pub/Sub. No async work, no in-process singletons — admin flux grants happen synchronously inside the POST that triggered them. diff --git a/apps/server/docs/ai-context/architecture-overview.md b/apps/server/docs/ai-context/architecture-overview.md index 4567a557d..758469975 100644 --- a/apps/server/docs/ai-context/architecture-overview.md +++ b/apps/server/docs/ai-context/architecture-overview.md @@ -138,12 +138,12 @@ CLI 入口在 `src/bin/run.ts`,只有一种角色: 这是服务端最重要的边界之一,尽量不要把写余额逻辑重新塞回 `flux.ts`。 -### LLM 网关代理而不是本地 provider 编排 +### LLM/TTS 路由在进程内,而不是本地 provider 编排 -`/api/v1/openai` 并不直接调具体模型 provider,而是转发到 `config: GATEWAY_BASE_URL`。因此: +`/api/v1/openai` 由 `services/llm-router` 读取 `LLM_ROUTER_CONFIG` 后按 upstream 链路 + key rotator 直接调 provider(OpenRouter、Azure Speech、阿里云 DashScope、火山引擎 等),不再依赖外部 knoway sidecar。因此: -- 服务端关心的是鉴权、限流、计费、日志、观测 -- 具体模型执行和 usage 返回格式由 gateway 决定 +- 服务端关心的是鉴权、限流、计费、日志、观测、上游路由与 key 健康 +- 具体模型协议翻译由 `services/llm-router` 与 `services/tts-adapters` 的 adapter 完成 ### Redis 有多种职责,但都不是余额真相源 diff --git a/apps/server/docs/ai-context/transport-and-routes.md b/apps/server/docs/ai-context/transport-and-routes.md index 79f835c4a..d073b1d6b 100644 --- a/apps/server/docs/ai-context/transport-and-routes.md +++ b/apps/server/docs/ai-context/transport-and-routes.md @@ -175,7 +175,7 @@ 1. 校验已登录 2. 检查相关配置是否存在 3. 检查用户 Flux 是否大于 0 -4. 代理请求到 `GATEWAY_BASE_URL` +4. 交给 `llmRouter.route` / `llmRouter.routeTts`:读 `LLM_ROUTER_CONFIG`,按 upstream 链路 + key rotator + envelope crypto 调上游 5. 解析 usage,计算扣费 6. 记录 metrics 7. 调 `billingService.debitFlux()` diff --git a/apps/server/docs/ai-context/verifications/llm-router.md b/apps/server/docs/ai-context/verifications/llm-router.md index d54962052..8ee1d2326 100644 --- a/apps/server/docs/ai-context/verifications/llm-router.md +++ b/apps/server/docs/ai-context/verifications/llm-router.md @@ -1,8 +1,27 @@ # LLM/TTS router replacing knoway — verification -Verification artifacts for the in-process router shipped across U1-U9 of +Verification artifacts for the in-process router. Scope tracked against `docs/plans/2026-05-15-001-feat-llm-tts-router-replacing-knoway-plan.md`. +## Coverage status + +| User path | Code wired | Has fresh evidence | +|---|---|---| +| chat completions happy (router → OpenRouter) | ✅ | ✅ commit `3a88f4225`, 2026-05-15 | +| chat completions fallback (key/upstream exhaustion) | ✅ | ❌ unit-test only, needs real-wire run | +| TTS speech (Azure) via `routeTts` | ✅ | ⏳ pending (was knoway-fetch until 2026-05-15) | +| TTS speech (dashscope-cosyvoice) via `routeTts` | ✅ | ⏳ pending | +| TTS speech (Volcengine) via `routeTts` | ✅ | ⏳ pending | +| `/audio/voices` from adapter catalog (no upstream) | ✅ | ⏳ pending (sanity curl) | +| `/livez` | ✅ | ✅ commit `cfad87757`, 2026-05-15 | +| `/readyz` | ✅ | ✅ commit `cfad87757`, 2026-05-15 | + +The TTS paths and `/audio/voices` were missing from the prior revision of +this doc (which claimed `shipped across U1-U9` while the route handlers +were still hitting `GATEWAY_BASE_URL`). The router-side wiring landed +2026-05-15; the table above tracks the real evidence backlog so the doc +stops asserting completion ahead of measurement. + ## E2E: chat completion through router service - **Scenario**: operator seeds `LLM_ROUTER_CONFIG` with one OpenRouter LLM @@ -90,10 +109,10 @@ Verification artifacts for the in-process router shipped across U1-U9 of - **U9 admin HTTP endpoint**: bootstrap currently goes through the `scripts/seed-router-config.ts` CLI. The plan's full HTTP admin endpoint with ETag + audit log + HMAC publish is deferred; tracked in plan U9. -- **GATEWAY_BASE_URL**: still required in env schema. The chat completions - route reads it for the legacy knoway fall-through path when `llmRouter` - is `null`. Remove once all deployments have rotated in - `LLM_ROUTER_MASTER_KEY` and `LLM_ROUTER_CONFIG`. +- ~~**GATEWAY_BASE_URL**: still required in env schema~~. Resolved + 2026-05-15: env entry removed, all routes go through `llmRouter.route` / + `routeTts` / `listTtsVoices`. The `LLM_ROUTER_MASTER_KEY` env var is + now required (no graceful skip). - **Grafana dashboard JSON updates**: the new `airi.gen_ai.gateway.*` counters are emitted from `apps/server/src/otel/index.ts` but the Grafana dashboard JSON in `otel/grafana/dashboards/` does not yet have diff --git a/apps/server/scripts/seed-router-config.ts b/apps/server/scripts/seed-router-config.ts index d6a2875cf..77df8479f 100644 --- a/apps/server/scripts/seed-router-config.ts +++ b/apps/server/scripts/seed-router-config.ts @@ -1,28 +1,38 @@ #!/usr/bin/env tsx +/* eslint-disable no-console */ /** - * Seed LLM_ROUTER_CONFIG into configKV (Postgres truth + Redis cache). + * Seed / patch LLM_ROUTER_CONFIG in configKV (Postgres truth + Redis cache). * * Use when: * - Bootstrapping a new deployment before U9's full admin endpoint ships. - * - Local E2E testing — populate one OpenRouter LLM model and one Azure TTS - * model so the router has something to dispatch. + * - Adding one TTS provider on top of an existing config without touching + * the others (the default `--merge` mode). + * - Previewing a write before committing it (`--dry-run`). * * Expects: - * - `.env.local` (or env) provides REDIS_URL + LLM_ROUTER_MASTER_KEY. - * - The plaintext provider keys come from positional args / env, not flags - * (so they never land in shell history with a leading `--key=`). + * - `.env.local` (or the deployment env) provides `REDIS_URL` and + * `LLM_ROUTER_MASTER_KEY`. + * - Provider plaintext keys come from env vars — never CLI flags — so they + * stay out of shell history and `ps`: + * OPENROUTER_KEY="sk-or-..." + * AZURE_KEY="..." + * DASHSCOPE_KEY="..." * - * Usage: - * pnpm -F @proj-airi/server exec tsx scripts/seed-router-config.ts \ - * --openrouter-key "" \ - * [--openrouter-model "openai/gpt-4o-mini"] \ - * [--azure-key "<plaintext>" --azure-region "eastasia"] \ - * [--default-chat-model "chat-default"] + * Modes: + * - default: merge. Reads existing `LLM_ROUTER_CONFIG`, upserts entries for + * providers whose key env var is set, leaves the rest untouched. + * - `--reset`: full overwrite. The new config contains only the providers + * you supplied keys for; everything else is dropped. + * - `--dry-run`: compute the final config and print it (ciphertext redacted) + * without writing or publishing. * - * On write the script publishes `configkv:invalidate` so any running - * instance picks up the new config within Pub/Sub propagation time - * (R16 / KTD-4, ≤5s under healthy Redis). + * On a real write (non dry-run) the script publishes `configkv:invalidate` + * so any running instance picks the new config up within Pub/Sub propagation + * time (R16 / KTD-4, ≤5s under healthy Redis). */ +import type { ConfigKVService } from '../src/services/config-kv' +import type { EnvelopeCrypto } from '../src/utils/envelope-crypto' + import { env, exit } from 'node:process' import Redis from 'ioredis' @@ -31,23 +41,192 @@ import { parseEnv } from '../src/libs/env' import { createConfigKVService } from '../src/services/config-kv' import { createEnvelopeCrypto } from '../src/utils/envelope-crypto' -function parseArgs(argv: string[]): Record<string, string> { - const out: Record<string, string> = {} +interface Args { + mode: 'merge' | 'reset' + dryRun: boolean + openrouterModel: string + defaultChatModel: string + azureRegion: string + azureTtsModel: string + dashscopeTtsModel: string + defaultTtsModel: string | undefined +} + +function parseArgs(argv: string[]): Args { + const flags = new Set<string>() + const values: Record<string, string> = {} for (let i = 0; i < argv.length; i++) { const arg = argv[i] - if (arg.startsWith('--')) { - const key = arg.slice(2) - const next = argv[i + 1] - if (next == null || next.startsWith('--')) { - out[key] = 'true' - } - else { - out[key] = next - i++ - } + if (!arg.startsWith('--')) + continue + const key = arg.slice(2) + const next = argv[i + 1] + if (next == null || next.startsWith('--')) { + flags.add(key) + } + else { + values[key] = next + i++ } } - return out + return { + mode: flags.has('reset') ? 'reset' : 'merge', + dryRun: flags.has('dry-run'), + openrouterModel: values['openrouter-model'] ?? 'openai/gpt-4o-mini', + defaultChatModel: values['default-chat-model'] ?? 'chat-default', + azureRegion: values['azure-region'] ?? 'eastasia', + azureTtsModel: values['azure-tts-model'] ?? 'microsoft/v1', + dashscopeTtsModel: values['dashscope-tts-model'] ?? 'alibaba/cosyvoice-v1', + defaultTtsModel: values['default-tts-model'], + } +} + +interface ProviderSlice { + llmModelName?: string + llmModel?: Record<string, unknown> + ttsModelName?: string + ttsModel?: Record<string, unknown> +} + +function buildOpenRouter(args: Args, plaintext: string, envelope: EnvelopeCrypto): ProviderSlice { + const keyEntryId = 'openrouter-prod-1' + const ciphertext = envelope.encryptKey(plaintext, { + modelName: args.defaultChatModel, + keyEntryId, + }) + return { + llmModelName: args.defaultChatModel, + llmModel: { + upstreams: [{ + baseURL: 'https://openrouter.ai/api/v1', + overrideModel: args.openrouterModel, + keys: [{ id: keyEntryId, ciphertext }], + headerTemplate: 'Bearer {KEY}', + }], + }, + } +} + +function buildAzure(args: Args, plaintext: string, envelope: EnvelopeCrypto): ProviderSlice { + const keyEntryId = 'azure-tts-prod-1' + const ciphertext = envelope.encryptKey(plaintext, { + modelName: args.azureTtsModel, + keyEntryId, + }) + return { + ttsModelName: args.azureTtsModel, + ttsModel: { + provider: 'azure', + upstreams: [{ + baseURL: `https://${args.azureRegion}.tts.speech.microsoft.com/cognitiveservices/v1`, + keys: [{ id: keyEntryId, ciphertext }], + adapterParams: { region: args.azureRegion }, + }], + }, + } +} + +function buildDashscope(args: Args, plaintext: string, envelope: EnvelopeCrypto): ProviderSlice { + const keyEntryId = 'dashscope-tts-prod-1' + const ciphertext = envelope.encryptKey(plaintext, { + modelName: args.dashscopeTtsModel, + keyEntryId, + }) + return { + ttsModelName: args.dashscopeTtsModel, + ttsModel: { + provider: 'dashscope-cosyvoice', + upstreams: [{ + baseURL: 'https://dashscope-intl.aliyuncs.com/api/v1', + keys: [{ id: keyEntryId, ciphertext }], + adapterParams: {}, + }], + }, + } +} + +interface BuiltConfig { + config: { llm: { models: Record<string, unknown> }, tts: { models: Record<string, unknown> } } + appliedSlices: ProviderSlice[] + defaultChatModel: string | undefined + defaultTtsModel: string | undefined +} + +/** + * Decides the next `LLM_ROUTER_CONFIG` shape from current args + (optionally) + * the existing config. + * + * Use when: + * - About to write a new config tree, before serializing it. + * + * Returns: + * - `config` — the next tree to write. + * - `appliedSlices` — which providers were upserted this run (for log output). + * - `defaultChatModel` / `defaultTtsModel` — the alias values to write into + * their dedicated configKV entries (undefined means "don't change"). + */ +function buildNextConfig(args: Args, existing: any, slices: ProviderSlice[]): BuiltConfig { + // `merge`: start from existing (or empty if none yet); `reset`: start fresh. + // `existing` is the parsed `LLM_ROUTER_CONFIG` value (or null when absent). + const llmModels: Record<string, unknown> + = args.mode === 'merge' && existing?.llm?.models ? { ...existing.llm.models } : {} + const ttsModels: Record<string, unknown> + = args.mode === 'merge' && existing?.tts?.models ? { ...existing.tts.models } : {} + + for (const slice of slices) { + if (slice.llmModelName && slice.llmModel) + llmModels[slice.llmModelName] = slice.llmModel + if (slice.ttsModelName && slice.ttsModel) + ttsModels[slice.ttsModelName] = slice.ttsModel + } + + // Default chat alias: explicit flag wins; otherwise only set on first-time + // bootstrap (no existing alias) and only if we just added the chat slice. + const llmSlice = slices.find(s => s.llmModelName) + let defaultChatModel: string | undefined + if (args.defaultChatModel && llmSlice && llmSlice.llmModelName === args.defaultChatModel) + defaultChatModel = args.defaultChatModel + + // Default TTS alias: explicit `--default-tts-model` wins; otherwise pick the + // first TTS slice we added this run, but never silently override an alias + // the operator already chose in `merge` mode. + let defaultTtsModel: string | undefined = args.defaultTtsModel + if (!defaultTtsModel && args.mode === 'reset') { + const ttsSlice = slices.find(s => s.ttsModelName) + defaultTtsModel = ttsSlice?.ttsModelName + } + + return { + config: { llm: { models: llmModels }, tts: { models: ttsModels } }, + appliedSlices: slices, + defaultChatModel, + defaultTtsModel, + } +} + +/** + * Redacts every `ciphertext` field down to its length for safe printing. + * + * Before: + * - `{ "keys": [{ "id": "k1", "ciphertext": "aGVsbG8=...long..." }] }` + * + * After: + * - `{ "keys": [{ "id": "k1", "ciphertext": "<ciphertext: 1024 chars>" }] }` + */ +function redactCiphertext(value: unknown): unknown { + if (Array.isArray(value)) + return value.map(redactCiphertext) + if (value && typeof value === 'object') { + const out: Record<string, unknown> = {} + for (const [k, v] of Object.entries(value as Record<string, unknown>)) { + if (k === 'ciphertext' && typeof v === 'string') + out[k] = `<ciphertext: ${v.length} chars>` + else + out[k] = redactCiphertext(v) + } + return out + } + return value } async function main() { @@ -59,94 +238,91 @@ async function main() { const args = parseArgs(process.argv.slice(2)) - const openrouterKey = args['openrouter-key'] - if (!openrouterKey) { - console.error('error: --openrouter-key <plaintext> is required') - console.error('example: pnpm -F @proj-airi/server exec tsx scripts/seed-router-config.ts \\') - console.error(' --openrouter-key sk-or-v1-XXXX \\') - console.error(' --openrouter-model openai/gpt-4o-mini') + const openrouterKey = process.env.OPENROUTER_KEY + const azureKey = process.env.AZURE_KEY + const dashscopeKey = process.env.DASHSCOPE_KEY + + if (!openrouterKey && !azureKey && !dashscopeKey) { + console.error('error: at least one provider key env var must be set:') + console.error(' OPENROUTER_KEY=... pnpm ... seed-router-config.ts # chat') + console.error(' AZURE_KEY=... pnpm ... seed-router-config.ts # tts (azure)') + console.error(' DASHSCOPE_KEY=... pnpm ... seed-router-config.ts # tts (cosyvoice)') + console.error('keys are read from env so they never appear in shell history or `ps`.') exit(1) } - const openrouterModel = args['openrouter-model'] ?? 'openai/gpt-4o-mini' - const defaultChatModel = args['default-chat-model'] ?? 'chat-default' const envelope = createEnvelopeCrypto({ masterKey: parsedEnv.LLM_ROUTER_MASTER_KEY, previousMasterKey: parsedEnv.LLM_ROUTER_MASTER_KEY_PREVIOUS, }) - // LLM upstreams ------------------------------------------------------------ + const slices: ProviderSlice[] = [] + if (openrouterKey) + slices.push(buildOpenRouter(args, openrouterKey, envelope)) + if (azureKey) + slices.push(buildAzure(args, azureKey, envelope)) + if (dashscopeKey) + slices.push(buildDashscope(args, dashscopeKey, envelope)) - const openrouterKeyEntryId = 'openrouter-prod-1' - const openrouterCiphertext = envelope.encryptKey(openrouterKey, { - modelName: defaultChatModel, - keyEntryId: openrouterKeyEntryId, - }) + // Connect to Redis before reading existing config (merge mode) and before + // writing. In dry-run we still connect because `merge` needs to read. + const redis = new Redis(parsedEnv.REDIS_URL) + const configKV: ConfigKVService = createConfigKVService(redis) - const config = { - llm: { - models: { - [defaultChatModel]: { - upstreams: [ - { - baseURL: 'https://openrouter.ai/api/v1', - overrideModel: openrouterModel, - keys: [{ id: openrouterKeyEntryId, ciphertext: openrouterCiphertext }], - headerTemplate: 'Bearer {KEY}', - }, - ], - }, - }, - }, - tts: { - models: {} as Record<string, unknown>, - }, - } as const + const existing = args.mode === 'merge' + ? await configKV.getOptional('LLM_ROUTER_CONFIG') + : null - // Optional Azure TTS entry ------------------------------------------------ + const built = buildNextConfig(args, existing, slices) - const azureKey = args['azure-key'] - if (azureKey) { - const azureModel = args['azure-tts-model'] ?? 'tts-default' - const azureRegion = args['azure-region'] ?? 'eastasia' - const azureKeyEntryId = 'azure-tts-prod-1' - const azureCiphertext = envelope.encryptKey(azureKey, { - modelName: azureModel, - keyEntryId: azureKeyEntryId, - }) + // Summary header — same in both real and dry-run paths so output is easy to + // diff between modes. + console.log(`mode: ${args.mode}${args.dryRun ? ' (dry-run)' : ''}`) + console.log(`providers: ${slices.map((s) => { + if (s.llmModelName) + return `openrouter→${s.llmModelName}` + if (s.ttsModel && (s.ttsModel as any).provider === 'azure') + return `azure→${s.ttsModelName}` + if (s.ttsModel && (s.ttsModel as any).provider === 'dashscope-cosyvoice') + return `dashscope→${s.ttsModelName}` + return s.ttsModelName ?? '?' + }).join(', ') || '(none)'}`) + console.log(`llm.models: [${Object.keys(built.config.llm.models).join(', ')}]`) + console.log(`tts.models: [${Object.keys(built.config.tts.models).join(', ')}]`) + if (built.defaultChatModel) + console.log(`DEFAULT_CHAT_MODEL → ${built.defaultChatModel}`) + if (built.defaultTtsModel) + console.log(`DEFAULT_TTS_MODEL → ${built.defaultTtsModel}`) - ;(config.tts.models as Record<string, unknown>)[azureModel] = { - provider: 'azure', - upstreams: [ - { - baseURL: `https://${azureRegion}.tts.speech.microsoft.com/cognitiveservices/v1`, - keys: [{ id: azureKeyEntryId, ciphertext: azureCiphertext }], - adapterParams: { region: azureRegion }, - }, - ], - } + if (args.dryRun) { + console.log('') + console.log('--- LLM_ROUTER_CONFIG (ciphertext redacted) ---') + console.log(JSON.stringify(redactCiphertext(built.config), null, 2)) + console.log('--- end ---') + console.log('dry-run: no writes, no publish.') + await redis.quit() + return } - // Write + publish invalidation ------------------------------------------- + // Real write path. configKV.set runs the valibot validator on the way in, + // so a malformed slice fails here before we publish invalidation. + await configKV.set('LLM_ROUTER_CONFIG', built.config as never) + if (built.defaultChatModel) + await configKV.set('DEFAULT_CHAT_MODEL', built.defaultChatModel) + if (built.defaultTtsModel) + await configKV.set('DEFAULT_TTS_MODEL', built.defaultTtsModel) - const redis = new Redis(parsedEnv.REDIS_URL) - const configKV = createConfigKVService(redis) - - await configKV.set('LLM_ROUTER_CONFIG', config as never) - - const payload = JSON.stringify({ - key: 'LLM_ROUTER_CONFIG', - version: Date.now(), - publishedAt: Date.now(), - }) - await redis.publish('configkv:invalidate', payload) - - console.log('LLM_ROUTER_CONFIG seeded:') - console.log(` default chat model: ${defaultChatModel} → ${openrouterModel} (1 key)`) - if (azureKey) - console.log(` azure tts: ${args['azure-tts-model'] ?? 'tts-default'} (1 key, ${args['azure-region'] ?? 'eastasia'})`) - console.log(`Published configkv:invalidate (key=LLM_ROUTER_CONFIG)`) + const keysToInvalidate: string[] = ['LLM_ROUTER_CONFIG'] + if (built.defaultChatModel) + keysToInvalidate.push('DEFAULT_CHAT_MODEL') + if (built.defaultTtsModel) + keysToInvalidate.push('DEFAULT_TTS_MODEL') + for (const key of keysToInvalidate) { + const payload = JSON.stringify({ key, version: Date.now(), publishedAt: Date.now() }) + await redis.publish('configkv:invalidate', payload) + } + console.log(`Published configkv:invalidate for ${keysToInvalidate.length} keys.`) await redis.quit() } diff --git a/apps/server/src/app.test.ts b/apps/server/src/app.test.ts index 38ce661fb..308a56854 100644 --- a/apps/server/src/app.test.ts +++ b/apps/server/src/app.test.ts @@ -68,7 +68,10 @@ function createTestDeps() { } as any, otel: null, userDeletionService: {} as any, - llmRouter: null, + llmRouter: { + route: vi.fn(async () => new Response('{}', { status: 200 })), + invalidateConfig: vi.fn(), + } as any, posthog: null, } diff --git a/apps/server/src/app.ts b/apps/server/src/app.ts index 82d6cd214..70db21869 100644 --- a/apps/server/src/app.ts +++ b/apps/server/src/app.ts @@ -93,7 +93,7 @@ interface AppDeps { env: Env otel: OtelInstance | null userDeletionService: UserDeletionService - llmRouter: LlmRouterService | null + llmRouter: LlmRouterService posthog: PostHog | null } @@ -169,29 +169,27 @@ export async function buildApp(deps: AppDeps) { // (R16 / KTD-4). Falls back to the in-memory cache TTL on missed messages. // Uses a dedicated ioredis subscriber connection (subscribe mode requires // a separate connection per ioredis docs). - if (deps.llmRouter) { - const configSub = deps.redis.duplicate() - configSub.on('message', (channel, message) => { - if (channel !== 'configkv:invalidate') + const configSub = deps.redis.duplicate() + configSub.on('message', (channel, message) => { + if (channel !== 'configkv:invalidate') + return + try { + const payload = JSON.parse(message) as { key?: unknown } + if (payload?.key !== 'LLM_ROUTER_CONFIG') return - try { - const payload = JSON.parse(message) as { key?: unknown } - if (payload?.key !== 'LLM_ROUTER_CONFIG') - return - deps.llmRouter?.invalidateConfig() - deps.otel?.gateway?.configReload.add(1, { - source: 'pubsub', - service_instance_id: deps.env.OTEL_SERVICE_NAME, - }) - } - catch (err) { - logger.withError(err).warn('Failed to parse configkv:invalidate payload') - } - }) - configSub.subscribe('configkv:invalidate').catch((err: unknown) => { - logger.withError(err).warn('Failed to subscribe to configkv:invalidate channel') - }) - } + deps.llmRouter.invalidateConfig() + deps.otel?.gateway?.configReload.add(1, { + source: 'pubsub', + service_instance_id: deps.env.OTEL_SERVICE_NAME, + }) + } + catch (err) { + logger.withError(err).warn('Failed to parse configkv:invalidate payload') + } + }) + configSub.subscribe('configkv:invalidate').catch((err: unknown) => { + logger.withError(err).warn('Failed to subscribe to configkv:invalidate channel') + }) const builtApp = app .use('*', sessionMiddleware(deps.auth, deps.env)) @@ -296,7 +294,7 @@ export async function buildApp(deps: AppDeps) { /** * V1 routes for official provider. */ - .route('/api/v1/openai', createV1CompletionsRoutes(deps.fluxService, deps.billingService, deps.configKV, deps.requestLogService, deps.ttsMeter, deps.redis, deps.env, deps.llmRouter, deps.otel?.genAi, deps.otel?.revenue, deps.otel?.rateLimit)) + .route('/api/v1/openai', createV1CompletionsRoutes(deps.fluxService, deps.billingService, deps.configKV, deps.requestLogService, deps.ttsMeter, deps.llmRouter, deps.otel?.genAi, deps.otel?.revenue, deps.otel?.rateLimit)) /** * Flux routes. @@ -572,17 +570,11 @@ export async function createApp() { }) // LLM router (KTD-5 in-process replacement for the knoway sidecar). - // Graceful skip when LLM_ROUTER_MASTER_KEY is unset: the chat-completions - // route falls back to the legacy GATEWAY_BASE_URL fetch path. This branch - // exists only for the U4→U8 transition window — U8 makes the master key - // (and the router) non-optional and deletes the legacy fallback. + // LLM_ROUTER_MASTER_KEY is required at env-parse time, so this provider + // always builds a real router — the legacy `null` fallback path is gone. const llmRouter = injeca.provide('services:llmRouter', { dependsOn: { configKV, env: parsedEnv, otel }, build: ({ dependsOn }) => { - if (dependsOn.env.LLM_ROUTER_MASTER_KEY == null) { - logger.warn('LLM_ROUTER_MASTER_KEY is not set; LLM router is disabled and chat completions will use the legacy GATEWAY_BASE_URL path (U4 transition)') - return null - } const envelopeCrypto = createEnvelopeCrypto({ masterKey: dependsOn.env.LLM_ROUTER_MASTER_KEY, previousMasterKey: dependsOn.env.LLM_ROUTER_MASTER_KEY_PREVIOUS, diff --git a/apps/server/src/libs/env.test.ts b/apps/server/src/libs/env.test.ts index f1a0313f4..ea3177f45 100644 --- a/apps/server/src/libs/env.test.ts +++ b/apps/server/src/libs/env.test.ts @@ -1,3 +1,5 @@ +import { Buffer } from 'node:buffer' + import { describe, expect, it } from 'vitest' import { parseAdditionalTrustedOriginsEnv, parseEnv } from './env' @@ -11,9 +13,8 @@ function baseEnv(): Record<string, string> { AUTH_GOOGLE_CLIENT_SECRET: 'google-secret', AUTH_GITHUB_CLIENT_ID: 'github-client', AUTH_GITHUB_CLIENT_SECRET: 'github-secret', - GATEWAY_BASE_URL: 'http://localhost:18080', - DEFAULT_CHAT_MODEL: 'openai/gpt-5-mini', - DEFAULT_TTS_MODEL: 'microsoft/v1', + // Required: a deterministic 32-byte base64 value so env parse succeeds. + LLM_ROUTER_MASTER_KEY: Buffer.alloc(32, 0xAA).toString('base64'), } } @@ -63,12 +64,7 @@ describe('parseEnv', () => { }) expect(Buffer.isBuffer(env.LLM_ROUTER_MASTER_KEY)).toBe(true) - expect(env.LLM_ROUTER_MASTER_KEY?.length).toBe(32) - }) - - it('lLM_ROUTER_MASTER_KEY is undefined when not set', () => { - const env = parseEnv(baseEnv()) - expect(env.LLM_ROUTER_MASTER_KEY).toBeUndefined() + expect(env.LLM_ROUTER_MASTER_KEY.length).toBe(32) }) // NOTICE: @@ -90,8 +86,8 @@ describe('parseEnv', () => { LLM_ROUTER_MASTER_KEY_PREVIOUS: previous, }) - expect(env.LLM_ROUTER_MASTER_KEY?.length).toBe(32) + expect(env.LLM_ROUTER_MASTER_KEY.length).toBe(32) expect(env.LLM_ROUTER_MASTER_KEY_PREVIOUS?.length).toBe(32) - expect(env.LLM_ROUTER_MASTER_KEY?.equals(env.LLM_ROUTER_MASTER_KEY_PREVIOUS!)).toBe(false) + expect(env.LLM_ROUTER_MASTER_KEY.equals(env.LLM_ROUTER_MASTER_KEY_PREVIOUS!)).toBe(false) }) }) diff --git a/apps/server/src/libs/env.ts b/apps/server/src/libs/env.ts index 0a0ef80f0..dc19de6c4 100644 --- a/apps/server/src/libs/env.ts +++ b/apps/server/src/libs/env.ts @@ -1,5 +1,6 @@ import type { InferOutput } from 'valibot' +import { Buffer } from 'node:buffer' import { env, exit } from 'node:process' import { useLogger } from '@guiiai/logg' @@ -119,23 +120,24 @@ const EnvSchema = object({ POSTHOG_API_KEY: optional(string(), ''), POSTHOG_HOST: optional(string(), 'https://us.i.posthog.com'), - // LLM gateway (infrastructure config — baked per deployment). - // GATEWAY_BASE_URL is retained for the knoway path during cutover; U8 removes - // it once the in-process router fully replaces the sidecar. - GATEWAY_BASE_URL: pipe(string(), nonEmpty('GATEWAY_BASE_URL is required')), - DEFAULT_CHAT_MODEL: pipe(string(), nonEmpty('DEFAULT_CHAT_MODEL is required')), - DEFAULT_TTS_MODEL: pipe(string(), nonEmpty('DEFAULT_TTS_MODEL is required')), + // LLM/TTS gateway is fully internalised by the in-process router; provider + // baseURLs live per-upstream inside LLM_ROUTER_CONFIG, and the default chat / + // tts model aliases moved to configKV (DEFAULT_CHAT_MODEL / DEFAULT_TTS_MODEL) + // so they're hot-swappable via Pub/Sub invalidation. No env entries needed + // here. // Envelope-encryption master key for in-process LLM/TTS router (KTD-5). // Stored as base64-encoded 32 random bytes. Validator decodes + asserts the // 32-byte length at parse time so a misconfigured key fails the deploy // rather than passing readiness and breaking on first router request. - LLM_ROUTER_MASTER_KEY: optional(pipe( + // Required: the router has no fallback path, so an unset master key means + // chat completions cannot serve at all. + LLM_ROUTER_MASTER_KEY: pipe( string(), - nonEmpty('LLM_ROUTER_MASTER_KEY must not be empty'), + nonEmpty('LLM_ROUTER_MASTER_KEY is required'), transform(b64 => Buffer.from(b64, 'base64')), check(buf => buf.length === 32, 'LLM_ROUTER_MASTER_KEY must decode to exactly 32 bytes (base64-encoded 32-byte random)'), - )), + ), // Optional second master key used only during rotation: encrypts under // LLM_ROUTER_MASTER_KEY (new), retries decrypt against LLM_ROUTER_MASTER_KEY_PREVIOUS // (old). Drop after re-encrypting every stored ciphertext. diff --git a/apps/server/src/routes/openai/v1/index.ts b/apps/server/src/routes/openai/v1/index.ts index 8155188ed..07bc39911 100644 --- a/apps/server/src/routes/openai/v1/index.ts +++ b/apps/server/src/routes/openai/v1/index.ts @@ -1,8 +1,6 @@ import type { Context } from 'hono' -import type Redis from 'ioredis' import type { PostHog } from 'posthog-node' -import type { Env } from '../../../libs/env' import type { GenAiMetrics, RateLimitMetrics, RevenueMetrics } from '../../../otel' import type { UsageInfo } from '../../../services/billing/billing' import type { BillingService } from '../../../services/billing/billing-service' @@ -33,21 +31,10 @@ import { GEN_AI_ATTR_REQUEST_MODEL, GEN_AI_ATTR_USAGE_INPUT_TOKENS, GEN_AI_ATTR_USAGE_OUTPUT_TOKENS, - getServerConnectionAttributes, } from '../../../utils/observability' -import { getCompressed, setCompressed } from '../../../utils/redis-compressed' -import { ttsVoicesUpstreamCacheRedisKey } from '../../../utils/redis-keys' const tracer = trace.getTracer('v1-completions') -// Upstream /audio/voices only changes when the gateway onboards a new backend -// (days-to-weeks cadence). 24h TTL cuts per-request gateway calls to ~1/day -// per model; operators can bump DEFAULT_TTS_VOICES immediately via configKV -// (that map is fetched fresh per request, not cached here) and can wipe the -// stale voice list with `DEL tts:voices:upstream:<model>` when a backend swap -// needs to show up before the TTL expires. -const TTS_VOICES_CACHE_TTL_SECONDS = 24 * 60 * 60 - const SAFE_RESPONSE_HEADERS = new Set([ 'content-type', 'content-length', @@ -64,10 +51,6 @@ function buildSafeResponseHeaders(response: Response): Headers { return headers } -function normalizeBaseUrl(gatewayBaseUrl: string): string { - return gatewayBaseUrl.endsWith('/') ? gatewayBaseUrl : `${gatewayBaseUrl}/` -} - function getLlmMetricAttributes(opts: { model: string, type: string, status: number }): Record<string, string | number> { if (opts.type === 'chat') { return { @@ -90,9 +73,7 @@ export function createV1CompletionsRoutes( configKV: ConfigKVService, requestLogService: RequestLogService, ttsMeter: FluxMeter, - redis: Redis, - env: Env, - llmRouter: LlmRouterService | null, + llmRouter: LlmRouterService, genAi?: GenAiMetrics | null, revenue?: RevenueMetrics | null, rateLimitMetrics?: RateLimitMetrics | null, @@ -149,26 +130,20 @@ export function createV1CompletionsRoutes( } const body = await c.req.json() - // NOTICE: - // GATEWAY_BASE_URL is still required by env validation (U8 removes it after - // router is fully wired). When `llmRouter` is null (graceful skip — no - // LLM_ROUTER_MASTER_KEY), this falls back to the legacy knoway fetch path - // unchanged. Server-connection attributes still tag the legacy path; the - // router enriches the active span with its own gateway.* attrs on success. - const baseUrl = normalizeBaseUrl(env.GATEWAY_BASE_URL) - const serverAttributes = getServerConnectionAttributes(baseUrl) let requestModel = body.model || 'auto' if (requestModel === 'auto') { - requestModel = env.DEFAULT_CHAT_MODEL + requestModel = await configKV.getOrThrow('DEFAULT_CHAT_MODEL') } + // Server-connection attrs come from the router (which knows the actual + // upstream baseURL it dispatched to) — it enriches the active span with + // its own `airi.gen_ai.gateway.*` attrs on success. const span = tracer.startSpan('llm.gateway.chat', { attributes: { [GEN_AI_ATTR_OPERATION_NAME]: 'chat', [GEN_AI_ATTR_REQUEST_MODEL]: requestModel, [AIRI_ATTR_GEN_AI_STREAM]: !!body.stream, - ...serverAttributes, }, }) @@ -187,14 +162,7 @@ export function createV1CompletionsRoutes( let response: Response try { response = await context.with(trace.setSpan(context.active(), span), () => - llmRouter - ? llmRouter.route({ modelName: requestModel, body, headers: {}, abortSignal: clientAbort }) - : fetch(`${baseUrl}chat/completions`, { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ ...body, model: requestModel }), - signal: clientAbort, - })) + llmRouter.route({ modelName: requestModel, body, headers: {}, abortSignal: clientAbort })) } catch (err) { span.setStatus({ code: SpanStatusCode.ERROR, message: 'Router exhausted or unknown model' }) @@ -486,15 +454,13 @@ export function createV1CompletionsRoutes( } const body = await c.req.json() - const baseUrl = normalizeBaseUrl(env.GATEWAY_BASE_URL) - const serverAttributes = getServerConnectionAttributes(baseUrl) let requestModel = body.model || 'auto' // NOTICE: Guard against non-string body.input — upstream would reject it // anyway, but billing math (.length → INCRBY) turns NaN into a Redis error. const inputText: string = typeof body.input === 'string' ? body.input : '' if (requestModel === 'auto') { - requestModel = env.DEFAULT_TTS_MODEL + requestModel = await configKV.getOrThrow('DEFAULT_TTS_MODEL') } // Pre-flight: refuse before hitting upstream if this segment would push the @@ -502,22 +468,36 @@ export function createV1CompletionsRoutes( // still pass when the user has at least 1 Flux. await ttsMeter.assertCanAfford(user.id, inputText.length, flux.flux) + // Map OpenAI-shaped /audio/speech body → adapter-neutral TtsInput. Speed + // / response_format / extra fields stay in adapterParams for adapters that + // care (Azure SSML rate, Volcengine audio_params, etc.). + const ttsInput = { + text: inputText, + voice: typeof body.voice === 'string' ? body.voice : undefined, + speed: typeof body.speed === 'number' ? body.speed : undefined, + responseFormat: typeof body.response_format === 'string' ? body.response_format : undefined, + } + const span = tracer.startSpan('llm.gateway.tts', { attributes: { [GEN_AI_ATTR_REQUEST_MODEL]: requestModel, [AIRI_ATTR_GEN_AI_OPERATION_KIND]: 'text_to_speech', - ...serverAttributes, }, }) const startedAt = Date.now() - const response = await context.with(trace.setSpan(context.active(), span), () => - fetch(`${baseUrl}audio/speech`, { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ ...body, model: requestModel }), - })) + let response: Response + try { + response = await context.with(trace.setSpan(context.active(), span), () => + llmRouter.routeTts({ modelName: requestModel, input: ttsInput, abortSignal: c.req.raw.signal })) + } + catch (err) { + span.setStatus({ code: SpanStatusCode.ERROR, message: 'TTS router exhausted or unknown model' }) + span.end() + recordMetrics({ model: requestModel, status: 502, type: 'tts', durationMs: Date.now() - startedAt, fluxConsumed: 0 }) + throw err + } const durationMs = Date.now() - startedAt span.setAttribute('http.response.status_code', response.status) @@ -535,16 +515,27 @@ export function createV1CompletionsRoutes( // Debt-ledger billing: accumulate chars in Redis; only debit when we // cross a whole-Flux boundary. Sub-threshold requests cost 0 Flux at this // call site — the cost is realised on a later request that crosses. - const { fluxDebited: fluxConsumed } = await ttsMeter.accumulate({ - userId: user.id, - units: inputText.length, - currentBalance: flux.flux, - requestId: nanoid(), - metadata: { model: requestModel }, - }) - - span.setAttribute(AIRI_ATTR_BILLING_FLUX_CONSUMED, fluxConsumed) - span.end() + // + // Wrapped in try/finally so a Redis blip inside `accumulate()` (or any + // throw before `span.end()`) doesn't leak the active span. Falling-through + // to `throw` reaches the global ApiError handler — billing failure on a + // 200 upstream is rare but observable, and a dropped span would have + // hidden it. + let fluxConsumed = 0 + try { + const result = await ttsMeter.accumulate({ + userId: user.id, + units: inputText.length, + currentBalance: flux.flux, + requestId: nanoid(), + metadata: { model: requestModel }, + }) + fluxConsumed = result.fluxDebited + span.setAttribute(AIRI_ATTR_BILLING_FLUX_CONSUMED, fluxConsumed) + } + finally { + span.end() + } recordMetrics({ model: requestModel, status: response.status, type: 'tts', durationMs, fluxConsumed }) recordRequestLog({ @@ -563,46 +554,27 @@ export function createV1CompletionsRoutes( async function handleListVoices(c: Context<HonoEnv>) { // Voice catalogs are per-model (different TTS models expose different - // voices), so cache key and upstream query are both keyed by model. - // `auto` and missing values fall back to the server's default. + // voices). Catalog content comes from the adapter's compiled-in JSON + // (apps/server/src/services/tts-adapters/voices/*.json), so there's + // nothing to fetch and the Redis upstream cache is no longer needed — + // adapter-side JSON is already in-process. Recommended map stays in + // configKV so operators can edit it without a deploy. const requested = c.req.query('model') - const model = (!requested || requested === 'auto') ? env.DEFAULT_TTS_MODEL : requested - const cacheKey = ttsVoicesUpstreamCacheRedisKey(model) + const model = (!requested || requested === 'auto') + ? await configKV.getOrThrow('DEFAULT_TTS_MODEL') + : requested - let body: Record<string, unknown> - const cached = await getCompressed(redis, cacheKey) - if (cached != null) { - body = JSON.parse(cached) as Record<string, unknown> - } - else { - const url = new URL(`${normalizeBaseUrl(env.GATEWAY_BASE_URL)}audio/voices`) - url.searchParams.set('model', model) - const response = await fetch(url, { method: 'GET', headers: { Accept: 'application/json' } }) - if (!response.ok) { - return new Response(response.body, { - status: response.status, - headers: buildSafeResponseHeaders(response), - }) - } - // Cache the raw upstream bytes so the write path skips a parse→stringify - // round-trip. Body is parsed below for merging with `recommended`. - const text = await response.text() - body = JSON.parse(text) as Record<string, unknown> - await setCompressed(redis, cacheKey, text, TTS_VOICES_CACHE_TTL_SECONDS) - } - - // Recommended map is read fresh from configKV (Redis-backed) so operator - // edits take effect immediately even while the upstream list is cached. + const voices = await llmRouter.listTtsVoices(model) const recommended = (await configKV.getOptional('DEFAULT_TTS_VOICES')) ?? {} - return Response.json({ ...body, recommended }) + return Response.json({ voices, recommended }) } async function handleListTTSModels(_c: Context<HonoEnv>) { // Mirror the chat provider: expose a single 'auto' routing alias instead // of the concrete DEFAULT_TTS_MODEL id. Keeps clients insulated from // backend model swaps and stays symmetric with /chat listModels. - // /audio/speech and /audio/voices already translate 'auto' into - // env.DEFAULT_TTS_MODEL before hitting upstream. + // /audio/speech and /audio/voices already translate 'auto' into the + // configKV DEFAULT_TTS_MODEL alias before hitting upstream. return Response.json({ models: [{ id: 'auto', name: 'Auto' }], }) diff --git a/apps/server/src/routes/openai/v1/route.test.ts b/apps/server/src/routes/openai/v1/route.test.ts index c073a08ab..642a09e67 100644 --- a/apps/server/src/routes/openai/v1/route.test.ts +++ b/apps/server/src/routes/openai/v1/route.test.ts @@ -1,4 +1,3 @@ -import type { Env } from '../../../libs/env' import type { BillingService } from '../../../services/billing/billing-service' import type { ConfigKVService } from '../../../services/config-kv' import type { FluxService } from '../../../services/flux' @@ -6,8 +5,6 @@ import type { LlmRouterService } from '../../../services/llm-router' import type { RequestLogService } from '../../../services/request-log' import type { HonoEnv } from '../../../types/hono' -import { Buffer } from 'node:buffer' - import { Hono } from 'hono' import { afterAll, describe, expect, it, vi } from 'vitest' @@ -41,20 +38,13 @@ function createMockBillingService(flux = 100): BillingService { } as any } -function createMockEnv(overrides: Partial<Env> = {}): Env { - return { - GATEWAY_BASE_URL: 'http://mock-gateway/', - DEFAULT_CHAT_MODEL: 'openai/gpt-5-mini', - DEFAULT_TTS_MODEL: 'tts-1', - ...overrides, - } as Env -} - function createMockConfigKV(overrides: Record<string, any> = {}): ConfigKVService { const defaults: Record<string, any> = { FLUX_PER_REQUEST: 1, FLUX_PER_1K_CHARS_TTS: 2, TTS_DEBT_TTL_SECONDS: 86400, + DEFAULT_CHAT_MODEL: 'openai/gpt-5-mini', + DEFAULT_TTS_MODEL: 'tts-1', ...overrides, } return { @@ -75,22 +65,6 @@ function createMockRequestLogService(): RequestLogService { } } -function createMockRedis() { - const store = new Map<string, Buffer>() - return { - get: vi.fn(async (key: string) => { - const v = store.get(key) - return v ? v.toString('utf8') : null - }), - getBuffer: vi.fn(async (key: string) => store.get(key) ?? null), - set: vi.fn(async (key: string, value: string | Buffer) => { - store.set(key, Buffer.isBuffer(value) ? value : Buffer.from(value, 'utf8')) - return 'OK' - }), - _store: store, - } -} - // NOTE: a router-mock helper used to live here but was removed because the // existing route tests all exercise the legacy fetch path (llmRouter = null). // Router internals are exhaustively covered in @@ -112,15 +86,41 @@ function createMockTtsMeter(unitsPerFlux = 1000) { } as any } +function createMockLlmRouter(impl?: Partial<LlmRouterService>): LlmRouterService { + return { + // Default: forward to globalThis.fetch so existing chat tests that mock + // fetch keep working. Per-test overrides can replace `route` directly. + route: vi.fn(async ({ modelName, body, abortSignal }) => { + return globalThis.fetch('http://mock-gateway/chat/completions', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ ...body, model: modelName }), + signal: abortSignal, + }) + }), + // TTS default also forwards to fetch, against a stable path tests can + // assert on. The mocked response body becomes the audio payload. + routeTts: vi.fn(async ({ modelName, input, abortSignal }) => { + return globalThis.fetch('http://mock-gateway/audio/speech', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ model: modelName, input: input.text, voice: input.voice }), + signal: abortSignal, + }) + }), + listTtsVoices: vi.fn(async () => []), + invalidateConfig: vi.fn(), + ...impl, + } as LlmRouterService +} + function createTestApp( fluxService: FluxService, configKV: ConfigKVService, billingService?: BillingService, requestLogService?: RequestLogService, ttsMeter?: ReturnType<typeof createMockTtsMeter>, - env?: Env, - redis?: ReturnType<typeof createMockRedis>, - llmRouter?: LlmRouterService | null, + llmRouter?: LlmRouterService, ) { const routes = createV1CompletionsRoutes( fluxService, @@ -128,9 +128,7 @@ function createTestApp( configKV, requestLogService ?? createMockRequestLogService(), ttsMeter ?? createMockTtsMeter(), - (redis ?? createMockRedis()) as any, - env ?? createMockEnv(), - llmRouter ?? null, + llmRouter ?? createMockLlmRouter(), null, ) const app = new Hono<HonoEnv>() @@ -338,11 +336,7 @@ describe('v1CompletionsRoutes', () => { const app = createTestApp( createMockFluxService(), - createMockConfigKV(), - undefined, - undefined, - undefined, - createMockEnv({ DEFAULT_CHAT_MODEL: 'anthropic/claude-sonnet' }), + createMockConfigKV({ DEFAULT_CHAT_MODEL: 'anthropic/claude-sonnet' }), ) await app.fetch( @@ -507,11 +501,7 @@ describe('v1CompletionsRoutes', () => { const app = createTestApp( createMockFluxService(), - createMockConfigKV(), - undefined, - undefined, - undefined, - createMockEnv({ DEFAULT_TTS_MODEL: 'tts-1-hd' }), + createMockConfigKV({ DEFAULT_TTS_MODEL: 'tts-1-hd' }), ) await app.fetch( @@ -553,17 +543,22 @@ describe('v1CompletionsRoutes', () => { expect(billingService.consumeFluxForLLM).not.toHaveBeenCalled() }) - it('should not charge when upstream returns error', async () => { - globalThis.fetch = vi.fn(async () => new Response('{"error":"service down"}', { - status: 500, - headers: { 'Content-Type': 'application/json' }, - })) - + it('should not charge when routeTts upstream returns error', async () => { + const llmRouter = createMockLlmRouter({ + routeTts: vi.fn(async () => new Response('{"error":"service down"}', { + status: 500, + headers: { 'Content-Type': 'application/json' }, + })) as any, + }) const billingService = createMockBillingService(100) - const app = createTestApp(createMockFluxService(), createMockConfigKV(), billingService) + const app = createTestApp(createMockFluxService(), createMockConfigKV(), billingService, undefined, undefined, llmRouter) const res = await app.fetch( - new Request('http://localhost/api/v1/openai/audio/voices', { method: 'GET' }), + new Request('http://localhost/api/v1/openai/audio/speech', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ model: 'auto', input: 'hello', voice: 'alloy' }), + }), { user: testUser } as any, ) @@ -643,16 +638,72 @@ describe('v1CompletionsRoutes', () => { expect(res.status).toBe(401) }) - it('should forward gateway error status', async () => { - globalThis.fetch = vi.fn(async () => new Response('{"error":"bad"}', { - status: 502, - headers: { 'Content-Type': 'application/json' }, - })) + // ROOT CAUSE: + // + // Before patch, `handleTTS` ran `ttsMeter.accumulate()` outside any + // try/finally and set the billing attribute + called `span.end()` + // *afterwards*. If `accumulate()` rejected (e.g. Redis blip on + // INCRBY), the call site threw straight to `app.onError` and the + // active span was never closed — OTel batched-span buffer leaked one + // span per failed TTS billing event, and `recordRequestLog` was + // skipped silently. + // + // After patch (apps/server/src/routes/openai/v1/index.ts:471-493): + // `accumulate()` + `span.setAttribute()` are wrapped in try/finally, + // span.end() runs unconditionally, and the error propagates to the + // global handler. recordRequestLog is still skipped (we can't log a + // billing-failed request without a fluxConsumed value), but the + // failure is now observable instead of hidden by a leaked span. + it('tTS billing failure closes the span and surfaces error to onError (regression)', async () => { + const requestLogService = createMockRequestLogService() + const ttsMeter = createMockTtsMeter() + // Override accumulate to simulate a Redis INCRBY failure mid-billing. + ttsMeter.accumulate = vi.fn(async () => { + throw new Error('redis INCRBY timeout') + }) - const app = createTestApp(createMockFluxService(), createMockConfigKV()) + const app = createTestApp( + createMockFluxService(), + createMockConfigKV(), + undefined, + requestLogService, + ttsMeter, + ) const res = await app.fetch( - new Request('http://localhost/api/v1/openai/audio/voices', { method: 'GET' }), + new Request('http://localhost/api/v1/openai/audio/speech', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ model: 'auto', input: 'hi', voice: 'en-US-AvaMultilingualNeural' }), + }), + { user: testUser } as any, + ) + + // Generic Error (not ApiError) → onError renders 500. + expect(res.status).toBe(500) + // recordRequestLog never reached, by design (no fluxConsumed to log). + expect(requestLogService.logRequest).not.toHaveBeenCalled() + // accumulate was actually attempted (proves we walked into the billing + // block, not the upstream-error branch). + expect(ttsMeter.accumulate).toHaveBeenCalledTimes(1) + }) + + it('should forward routeTts error status (502)', async () => { + const llmRouter = createMockLlmRouter({ + routeTts: vi.fn(async () => new Response('{"error":"bad"}', { + status: 502, + headers: { 'Content-Type': 'application/json' }, + })) as any, + }) + + const app = createTestApp(createMockFluxService(), createMockConfigKV(), undefined, undefined, undefined, llmRouter) + + const res = await app.fetch( + new Request('http://localhost/api/v1/openai/audio/speech', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ model: 'auto', input: 'hi', voice: 'alloy' }), + }), { user: testUser } as any, ) expect(res.status).toBe(502) @@ -663,11 +714,7 @@ describe('v1CompletionsRoutes', () => { it('exposes only the auto routing alias regardless of DEFAULT_TTS_MODEL', async () => { const app = createTestApp( createMockFluxService(), - createMockConfigKV(), - undefined, - undefined, - undefined, - createMockEnv({ DEFAULT_TTS_MODEL: 'microsoft/v1' }), + createMockConfigKV({ DEFAULT_TTS_MODEL: 'microsoft/v1' }), ) const res = await app.fetch( @@ -689,17 +736,19 @@ describe('v1CompletionsRoutes', () => { }) describe('gET /api/v1/openai/audio/voices', () => { - it('should proxy voice list from gateway', async () => { - const voicesResponse = { voices: [ - { id: 'en-US-JennyNeural', name: 'Jenny', provider: 'MICROSOFT_SPEECH_SERVICE_V1', locale: 'en-US', gender: 'Female' }, - { id: 'alloy', name: 'Alloy', provider: 'OPEN_AI', locale: '', gender: '' }, - ] } - globalThis.fetch = vi.fn(async () => new Response(JSON.stringify(voicesResponse), { - status: 200, - headers: { 'Content-Type': 'application/json' }, - })) + it('returns the adapter voice catalog plus recommended map for the resolved model', async () => { + const voices = [ + { id: 'en-US-JennyNeural', name: 'Jenny', provider: 'azure', locale: 'en-US', gender: 'Female' }, + { id: 'en-US-AvaMultilingualNeural', name: 'Ava', provider: 'azure', locale: 'en-US', gender: 'Female' }, + ] + const llmRouter = createMockLlmRouter({ + listTtsVoices: vi.fn(async () => voices) as any, + }) + const configKV = createMockConfigKV({ + DEFAULT_TTS_VOICES: { 'en-US': 'en-US-AvaMultilingualNeural' }, + }) - const app = createTestApp(createMockFluxService(), createMockConfigKV()) + const app = createTestApp(createMockFluxService(), configKV, undefined, undefined, undefined, llmRouter) const res = await app.fetch( new Request('http://localhost/api/v1/openai/audio/voices', { method: 'GET' }), @@ -707,87 +756,33 @@ describe('v1CompletionsRoutes', () => { ) expect(res.status).toBe(200) - const data = await res.json() as typeof voicesResponse - expect(data.voices).toHaveLength(2) - expect(data.voices[0].id).toBe('en-US-JennyNeural') - - const [calledUrl, calledInit] = (globalThis.fetch as any).mock.calls[0] - expect(String(calledUrl)).toBe('http://mock-gateway/audio/voices?model=tts-1') - expect(calledInit).toMatchObject({ method: 'GET' }) + const data = await res.json() as { voices: typeof voices, recommended: Record<string, string> } + expect(data.voices).toEqual(voices) + expect(data.recommended).toEqual({ 'en-US': 'en-US-AvaMultilingualNeural' }) + expect(llmRouter.listTtsVoices).toHaveBeenCalledWith('tts-1') }) - it('serves the second request from Redis without re-hitting the gateway', async () => { - const voicesResponse = { voices: [{ id: 'alloy', name: 'Alloy' }] } - globalThis.fetch = vi.fn(async () => new Response(JSON.stringify(voicesResponse), { - status: 200, - headers: { 'Content-Type': 'application/json' }, - })) - const redis = createMockRedis() - const app = createTestApp(createMockFluxService(), createMockConfigKV(), undefined, undefined, undefined, undefined, redis) + it('uses the explicit ?model= query when provided instead of DEFAULT_TTS_MODEL', async () => { + const llmRouter = createMockLlmRouter({ + listTtsVoices: vi.fn(async (model: string) => [{ id: `${model}-v`, name: model } as any]) as any, + }) - await app.fetch(new Request('http://localhost/api/v1/openai/audio/voices'), { user: testUser } as any) - await app.fetch(new Request('http://localhost/api/v1/openai/audio/voices'), { user: testUser } as any) + const app = createTestApp(createMockFluxService(), createMockConfigKV(), undefined, undefined, undefined, llmRouter) - expect(globalThis.fetch).toHaveBeenCalledTimes(1) - expect(redis.set).toHaveBeenCalledTimes(1) + await app.fetch(new Request('http://localhost/api/v1/openai/audio/voices?model=alibaba/cosyvoice-v1'), { user: testUser } as any) + expect(llmRouter.listTtsVoices).toHaveBeenCalledWith('alibaba/cosyvoice-v1') }) - it('stores the cached body gzipped (first two bytes are the gzip magic)', async () => { - // Big payload so gzip actually compresses below plain json size. Using - // a single-voice body would hit the gzip header overhead and be larger. - const voices = Array.from({ length: 200 }, (_, i) => ({ - id: `voice-${i}`, - name: `Voice ${i}`, - description: 'Microsoft Server Speech Text to Speech Voice (zh-CN, XiaozhenNeural)', - })) - const voicesResponse = { voices } - const rawJson = JSON.stringify(voicesResponse) - globalThis.fetch = vi.fn(async () => new Response(rawJson, { - status: 200, - headers: { 'Content-Type': 'application/json' }, - })) - const redis = createMockRedis() - const app = createTestApp(createMockFluxService(), createMockConfigKV(), undefined, undefined, undefined, undefined, redis) + it('resolves `auto` model to configKV DEFAULT_TTS_MODEL', async () => { + const llmRouter = createMockLlmRouter({ + listTtsVoices: vi.fn(async () => []) as any, + }) + const configKV = createMockConfigKV({ DEFAULT_TTS_MODEL: 'microsoft/v1' }) - await app.fetch(new Request('http://localhost/api/v1/openai/audio/voices'), { user: testUser } as any) + const app = createTestApp(createMockFluxService(), configKV, undefined, undefined, undefined, llmRouter) - const stored = redis._store.get('tts:voices:upstream:tts-1') - expect(stored).toBeDefined() - expect(stored![0]).toBe(0x1F) - expect(stored![1]).toBe(0x8B) - expect(stored!.length).toBeLessThan(rawJson.length) - }) - - it('transparently reads legacy plain-JSON cache entries left over from before compression', async () => { - globalThis.fetch = vi.fn() - const redis = createMockRedis() - redis._store.set( - 'tts:voices:upstream:tts-1', - Buffer.from(JSON.stringify({ voices: [{ id: 'legacy', name: 'Legacy' }] }), 'utf8'), - ) - const app = createTestApp(createMockFluxService(), createMockConfigKV(), undefined, undefined, undefined, undefined, redis) - - const res = await app.fetch(new Request('http://localhost/api/v1/openai/audio/voices'), { user: testUser } as any) - const data = await res.json() as { voices: { id: string }[] } - - expect(data.voices).toEqual([{ id: 'legacy', name: 'Legacy' }]) - expect(globalThis.fetch).not.toHaveBeenCalled() - }) - - it('caches per-model — different models each hit the gateway once', async () => { - const fetchMock = vi.fn(async (url: any) => new Response( - JSON.stringify({ voices: [{ id: `${new URL(url).searchParams.get('model')}-v` }] }), - { status: 200, headers: { 'Content-Type': 'application/json' } }, - )) - globalThis.fetch = fetchMock as any - const redis = createMockRedis() - const app = createTestApp(createMockFluxService(), createMockConfigKV(), undefined, undefined, undefined, undefined, redis) - - await app.fetch(new Request('http://localhost/api/v1/openai/audio/voices?model=tts-1'), { user: testUser } as any) - await app.fetch(new Request('http://localhost/api/v1/openai/audio/voices?model=tts-hd'), { user: testUser } as any) - await app.fetch(new Request('http://localhost/api/v1/openai/audio/voices?model=tts-1'), { user: testUser } as any) - - expect(fetchMock).toHaveBeenCalledTimes(2) + await app.fetch(new Request('http://localhost/api/v1/openai/audio/voices?model=auto'), { user: testUser } as any) + expect(llmRouter.listTtsVoices).toHaveBeenCalledWith('microsoft/v1') }) }) diff --git a/apps/server/src/scripts/otel/smoke.ts b/apps/server/src/scripts/otel/smoke.ts index b09ea43b7..0fbc4f9b0 100644 --- a/apps/server/src/scripts/otel/smoke.ts +++ b/apps/server/src/scripts/otel/smoke.ts @@ -18,6 +18,7 @@ * Usage: * pnpm -F @proj-airi/server exec node --import tsx ./src/scripts/otel/smoke.ts */ +import { Buffer } from 'node:buffer' import { env, exit } from 'node:process' import { metrics } from '@opentelemetry/api' @@ -38,9 +39,9 @@ env.AUTH_GOOGLE_CLIENT_ID ??= 'test' env.AUTH_GOOGLE_CLIENT_SECRET ??= 'test' env.AUTH_GITHUB_CLIENT_ID ??= 'test' env.AUTH_GITHUB_CLIENT_SECRET ??= 'test' -env.GATEWAY_BASE_URL ??= 'http://test' -env.DEFAULT_CHAT_MODEL ??= 'test' -env.DEFAULT_TTS_MODEL ??= 'test' +// 32 deterministic bytes is enough to satisfy env validation; the smoke +// script never actually hits the router. +env.LLM_ROUTER_MASTER_KEY ??= Buffer.alloc(32, 0xAA).toString('base64') env.OTEL_EXPORTER_OTLP_ENDPOINT ??= 'http://localhost:4318' const { initOtel } = await import('../../otel/index') diff --git a/apps/server/src/services/config-kv.ts b/apps/server/src/services/config-kv.ts index c67918423..69e6a6003 100644 --- a/apps/server/src/services/config-kv.ts +++ b/apps/server/src/services/config-kv.ts @@ -108,6 +108,16 @@ const ConfigEntrySchemas = { // BCP-47 locale → recommended voice id for the default TTS model. // Consumed by the client to preselect a voice matching UI locale. DEFAULT_TTS_VOICES: optional(record(string(), string()), {}), + // Server-side alias resolution for `model: 'auto'` in /chat/completions and + // /audio/speech. The modelName written here must exist as a key in + // LLM_ROUTER_CONFIG.{llm,tts}.models — the router itself doesn't understand + // `auto`, this layer translates before dispatch. No default: missing entry + // surfaces CONFIG_NOT_SET (resolveWithDefault swallows ValiError) so a + // misconfigured deploy fails the request instead of silently routing to an + // empty modelName. Naked schema (not wrapped in optional) keeps the inferred + // type tight (`string` rather than `string | undefined`) for call sites. + DEFAULT_CHAT_MODEL: pipe(string(), nonEmpty('DEFAULT_CHAT_MODEL must not be empty')), + DEFAULT_TTS_MODEL: pipe(string(), nonEmpty('DEFAULT_TTS_MODEL must not be empty')), // No default — the router throws CONFIG_NOT_SET when this entry is absent // so the admin endpoint (U9) is forced to populate it before traffic flows. LLM_ROUTER_CONFIG: optional(llmRouterConfigSchema), diff --git a/apps/server/src/services/llm-router/router.test.ts b/apps/server/src/services/llm-router/router.test.ts index 26f5c7c15..2f75d593c 100644 --- a/apps/server/src/services/llm-router/router.test.ts +++ b/apps/server/src/services/llm-router/router.test.ts @@ -497,4 +497,162 @@ describe('createLlmRouterService', () => { expect(fetchImpl.mock.calls.length).toBe(2) expect((configKV.getOptional as ReturnType<typeof vi.fn>).mock.calls.length).toBe(2) }) + + // --- routeTts adapter error contract ------------------------------------- + // + // ROOT CAUSE: + // + // Before patch, `dispatchOneTtsUpstream` read `err.status` to decide + // fallback, but `ApiError.statusCode` (not `.status`) is the canonical + // field. Every adapter-internal `ApiError` (invalid voice, missing + // adapter params, network wrap) was silently coerced to `'timeout'` and + // 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. + describe('routeTts adapter error handling', () => { + function makeTtsConfig(opts: { + provider?: 'azure' + upstreams?: Array<{ baseURL: string, keyIds: string[], adapterParams?: Record<string, unknown> }> + }): { config: RouterConfig, crypto: ReturnType<typeof createEnvelopeCrypto> } { + const crypto = createEnvelopeCrypto({ masterKey: freshMasterKey() }) + const modelName = 'tts-test' + const upstreams = opts.upstreams ?? [{ baseURL: 'https://up-a.example', keyIds: ['kA1'] }] + const upstreamConfigs = upstreams.map(u => ({ + baseURL: u.baseURL, + keys: u.keyIds.map((id) => { + const plaintext = `sk-${id}` + const ct = crypto.encryptKey(plaintext, { modelName, keyEntryId: id }) + return { id, ciphertext: ct } + }), + adapterParams: u.adapterParams ?? {}, + })) + const config: RouterConfig = { + llm: { models: {} }, + tts: { + models: { + [modelName]: { + provider: opts.provider ?? 'azure', + upstreams: upstreamConfigs, + fallbackTriggers: { httpCodes: [401, 429, 500, 502, 503, 504], onTimeout: true }, + }, + }, + }, + defaults: { + perAttemptTimeoutMs: 5000, + fullChainTimeoutMs: 10000, + fallbackHttpCodes: [401, 429, 500, 502, 503, 504], + }, + } as RouterConfig + return { config, crypto } + } + + it('apiError 4xx (invalid voice) propagates without touching the second key', async () => { + // azure adapter validates `voice` against AZURE_VOICE_ID before any + // network call; an invalid voice throws createBadRequestError(400). + // Two keys are configured: the second must NEVER be tried. + const { config, crypto } = makeTtsConfig({ upstreams: [{ baseURL: 'https://az.example', keyIds: ['kA1', 'kA2'] }] }) + const fetchImpl = vi.fn(async () => happyResponse({ ok: 1 })) + const metrics = makeMetrics() + + const router = createLlmRouterService({ + configKV: makeConfigKV(config), + envelopeCrypto: crypto, + gatewayMetrics: metrics, + fetchImpl, + }) + + let caught: unknown + try { + await router.routeTts({ + modelName: 'tts-test', + input: { text: 'hi', voice: 'bogus voice with spaces' }, + }) + } + catch (err) { + caught = err + } + + expect(caught).toBeInstanceOf(ApiError) + expect((caught as ApiError).statusCode).toBe(400) + // The adapter rejects before fetch; with the bug this would have walked + // both keys (and pushed fallback counters). After the fix: zero fetch, + // zero fallback bookkeeping. + expect(fetchImpl).not.toHaveBeenCalled() + expect((metrics.fallbackCount.add as ReturnType<typeof vi.fn>)).not.toHaveBeenCalled() + }) + + it('apiError 5xx (adapter-wrapped network failure) walks to the next key', async () => { + // azure adapter wraps a fetch reject as createInternalError(500). + // The router should treat that as a fallback-eligible network failure + // and try the second key — not propagate the 500 as a final error. + const { config, crypto } = makeTtsConfig({ upstreams: [{ baseURL: 'https://az.example', keyIds: ['kA1', 'kA2'] }] }) + + let callIdx = 0 + const fetchImpl = vi.fn(async () => { + callIdx += 1 + if (callIdx === 1) + throw new TypeError('network unreachable') + return new Response(new Uint8Array([0x01]), { status: 200, headers: { 'content-type': 'audio/mpeg' } }) + }) + const metrics = makeMetrics() + + const router = createLlmRouterService({ + configKV: makeConfigKV(config), + envelopeCrypto: crypto, + gatewayMetrics: metrics, + fetchImpl, + }) + + const res = await router.routeTts({ + modelName: 'tts-test', + input: { text: 'hi', voice: 'en-US-AvaMultilingualNeural' }, + }) + + expect(res.status).toBe(200) + expect(fetchImpl).toHaveBeenCalledTimes(2) + // Adapter-wrapped 500 is in the fallback list, so one fallback hop is + // recorded between key 1 and key 2. + expect((metrics.fallbackCount.add as ReturnType<typeof vi.fn>).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. + const { config, crypto } = makeTtsConfig({ upstreams: [{ baseURL: 'https://az.example', keyIds: ['kA1', 'kA2'] }] }) + + let callIdx = 0 + const fetchImpl = vi.fn(async () => { + callIdx += 1 + if (callIdx === 1) + return failResponse(401) + return new Response(new Uint8Array([0x01]), { status: 200, headers: { 'content-type': 'audio/mpeg' } }) + }) + const metrics = makeMetrics() + + const router = createLlmRouterService({ + configKV: makeConfigKV(config), + envelopeCrypto: crypto, + gatewayMetrics: metrics, + fetchImpl, + }) + + const res = await router.routeTts({ + modelName: 'tts-test', + input: { text: 'hi', voice: 'en-US-AvaMultilingualNeural' }, + }) + + expect(res.status).toBe(200) + expect(fetchImpl).toHaveBeenCalledTimes(2) + const fallbackCalls = (metrics.fallbackCount.add as ReturnType<typeof vi.fn>).mock.calls + expect(fallbackCalls.length).toBe(1) + // Recorded reason matches the upstream status, not 'timeout' — that's + // the regression: pre-fix this would have been 'timeout' because the + // adapter's `Error & { status }` was read as undefined. + expect(fallbackCalls[0][1]).toMatchObject({ reason: '401' }) + }) + }) }) diff --git a/apps/server/src/services/llm-router/router.ts b/apps/server/src/services/llm-router/router.ts index c36d99d0c..3cb7d4b79 100644 --- a/apps/server/src/services/llm-router/router.ts +++ b/apps/server/src/services/llm-router/router.ts @@ -3,17 +3,20 @@ import type { Buffer } from 'node:buffer' import type { GatewayMetrics } from '../../otel' import type { EnvelopeCrypto } from '../../utils/envelope-crypto' import type { ConfigKVService } from '../config-kv' -import type { LlmRouteRequest, LlmUpstream } from './types' +import type { TtsAdapterId, TtsInput } from '../tts-adapters/types' +import type { LlmRouteRequest, LlmUpstream, TtsUpstream } from './types' import { useLogger } from '@guiiai/logg' import { trace } from '@opentelemetry/api' +import { ApiError } from '../../utils/error' import { AIRI_ATTR_GEN_AI_GATEWAY_FALLBACK_DEPTH, AIRI_ATTR_GEN_AI_GATEWAY_KEY_ID, AIRI_ATTR_GEN_AI_GATEWAY_UPSTREAM_INDEX, AIRI_ATTR_GEN_AI_GATEWAY_UPSTREAM_URL, } from '../../utils/observability' +import { getAdapter } from '../tts-adapters' import { createConfigLoader } from './config-loader' import { mapUpstreamError } from './error-mapping' import { createKeyRotator } from './key-rotator' @@ -304,8 +307,217 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) { }) } + /** + * 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. + */ + async function dispatchOneTtsUpstream( + upstream: TtsUpstream, + upstreamIndex: number, + providerId: TtsAdapterId, + input: TtsInput, + modelName: string, + abortSignal: AbortSignal | undefined, + perAttemptTimeoutMs: number, + fallbackHttpCodes: number[], + onAttemptFailure: (failure: { keyId: string, status: number | 'timeout' }) => void, + ): Promise< + | { kind: 'ok', contentType: string, body: ArrayBuffer | ReadableStream<Uint8Array>, attemptIndex: number } + | { kind: 'exhausted', failures: Array<{ keyId: string, status: number | 'timeout' }> } + > { + 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' }> = [] + let attemptIndex = 0 + + for (const key of rotator) { + try { + // Per-attempt timeout composed with caller abort — same shape as chat. + // See chat dispatch for the AbortSignal.any rationale. + const attemptCtrl = new AbortController() + const timeoutHandle = setTimeout(() => attemptCtrl.abort(new Error('attempt-timeout')), perAttemptTimeoutMs) + const callerOnAbort = () => attemptCtrl.abort(abortSignal?.reason) + if (abortSignal != null) { + if (abortSignal.aborted) + attemptCtrl.abort(abortSignal.reason) + else + abortSignal.addEventListener('abort', callerOnAbort, { once: true }) + } + + let result + try { + result = await adapter.send(input, { + keyPlaintext: key.plaintext, + baseURL: upstream.baseURL.replace(/\/+$/, ''), + adapterParams: upstream.adapterParams ?? {}, + fetchImpl, + abortSignal: attemptCtrl.signal, + }) + } + finally { + clearTimeout(timeoutHandle) + if (abortSignal != null) + abortSignal.removeEventListener('abort', callerOnAbort) + } + + trace.getActiveSpan()?.setAttributes({ + [AIRI_ATTR_GEN_AI_GATEWAY_UPSTREAM_URL]: upstream.baseURL, + [AIRI_ATTR_GEN_AI_GATEWAY_UPSTREAM_INDEX]: upstreamIndex, + [AIRI_ATTR_GEN_AI_GATEWAY_KEY_ID]: key.id, + [AIRI_ATTR_GEN_AI_GATEWAY_FALLBACK_DEPTH]: attemptIndex, + }) + return { kind: 'ok', contentType: result.contentType, body: result.body, attemptIndex } + } + catch (err) { + if (abortSignal?.aborted) { + logger.withError(err).withFields({ keyId: key.id }).debug('Caller aborted upstream tts fetch; propagating without fallback') + throw err + } + + // Adapter contract (see `apps/server/src/services/tts-adapters/types.ts` + // and the three impls): + // + // - `ApiError` 4xx — adapter rejected the *request* before talking to + // the upstream (e.g. azure invalid voice id, volcengine missing + // `adapterParams.appid`). Every key on every upstream would reject + // the same way, so propagate without fallback. + // - `ApiError` 5xx — adapter caught a network failure and wrapped it + // in `createInternalError(...)`. Different keys / upstreams may + // succeed, so fold into the same fallback path as a plain network + // error. + // - `Error & { status: number }` — upstream answered non-2xx; we own + // the fallback decision and consult `fallbackHttpCodes`. + // - plain `Error` — network failure or per-attempt timeout (our + // AbortController fired); treat as `'timeout'` for KTD-1 mapping. + 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' + failures.push({ keyId: key.id, status: failureStatus }) + onAttemptFailure({ keyId: key.id, status: failureStatus }) + 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 policy as chat: non-fallback status stops this upstream + // but the outer loop still tries the next upstream. + attemptIndex += 1 + break + } + } + logger.withError(err).withFields({ keyId: key.id, upstream: upstream.baseURL }).warn('Upstream TTS attempt failed') + } + finally { + key.plaintext.fill(0) + } + + attemptIndex += 1 + } + + return { kind: 'exhausted', failures } + } + + async function routeTts(req: { modelName: string, input: TtsInput, abortSignal?: AbortSignal }): Promise<Response> { + if (req.abortSignal?.aborted) + throw req.abortSignal.reason ?? new Error('aborted') + + const slice = await configLoader.getModelConfig('tts', req.modelName) + if (slice.kind !== 'tts') { + // Defensive — config-loader returns 'tts' when kind='tts', but a future + // schema change could broaden this. Surface as 500. + throw new Error(`Expected tts model slice for ${req.modelName}, got ${slice.kind}`) + } + + 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 allFailures: Array<{ provider: string, keyId: string, status: number | 'timeout' }> = [] + let triedUpstreams = 0 + + for (let i = 0; i < slice.model.upstreams.length; i += 1) { + const upstream = slice.model.upstreams[i] + const providerTag = deriveProviderTag(upstream.baseURL) + triedUpstreams += 1 + + // tts upstream schema has no per-upstream timeoutMs (see ttsUpstreamSchema); + // we use the defaults bucket alone here. + const perAttemptTimeoutMs = defaults.perAttemptTimeoutMs ?? 30000 + + const result = await dispatchOneTtsUpstream( + upstream, + i, + slice.model.provider, + req.input, + req.modelName, + req.abortSignal, + perAttemptTimeoutMs, + fallbackHttpCodes, + (failure) => { allFailures.push({ provider: providerTag, ...failure }) }, + ) + + if (result.kind === 'ok') { + return new Response(result.body, { + status: 200, + headers: { 'content-type': result.contentType }, + }) + } + + options.gatewayMetrics?.keyExhaustedCount.add(1, { provider: providerTag }) + } + + const lastFailure = allFailures.at(-1) + if (lastFailure == null) { + throw new Error(`Router exhausted with no recorded failures for tts model ${req.modelName}`) + } + + const distinctStatuses = new Set(allFailures.map(f => f.status)) + if (distinctStatuses.size === 1) { + const status = allFailures[0].status + const providersHit = new Set(allFailures.map(f => f.provider)) + for (const provider of providersHit) { + options.gatewayMetrics?.sameStatusExhaustion.add(1, { + provider, + status_code: typeof status === 'number' ? status : 'timeout', + }) + } + } + + throw mapUpstreamError(lastFailure.status, { + triedKeys: allFailures.length, + triedUpstreams, + lastStatusCode: lastFailure.status, + }) + } + + /** + * Returns the static voice catalog for one TTS provider model. Read from + * the adapter's compiled-in JSON — no network call, no envelope decrypt, + * no per-upstream variation (voice lists are provider-wide). + */ + async function listTtsVoices(modelName: string) { + const slice = await configLoader.getModelConfig('tts', modelName) + if (slice.kind !== 'tts') + throw new Error(`Expected tts model slice for ${modelName}, got ${slice.kind}`) + return getAdapter(slice.model.provider).getVoiceCatalog() + } + return { route, + routeTts, + listTtsVoices, /** * Expose the loader's invalidate hook so U7's Pub/Sub subscriber and * the admin endpoint (U9) can flush the cache without a separate diff --git a/apps/server/src/services/tts-adapters/azure.ts b/apps/server/src/services/tts-adapters/azure.ts index 9666d5af2..402a8a9a3 100644 --- a/apps/server/src/services/tts-adapters/azure.ts +++ b/apps/server/src/services/tts-adapters/azure.ts @@ -31,8 +31,9 @@ const DEFAULT_AZURE_VOICE = 'en-US-AvaMultilingualNeural' /** * Default Azure output format header value. * - * Mirrors the current knoway hosted default. Maps to OpenAI's `mp3` response - * format at the adapter boundary (callers can still override). + * Maps to OpenAI's `mp3` response format at the adapter boundary so callers + * who don't pin `response_format` get a sensible mp3 stream. Callers can + * still override via `input.responseFormat`. */ const DEFAULT_AZURE_FORMAT = 'audio-24khz-48kbitrate-mono-mp3'