feat(server): finalize in-process LLM/TTS router cutover
End-state of the multi-step KTD-5 / KTD-6 / U8 work. The knoway sidecar is no longer reachable from server code; the router is required at boot and now owns chat completions, TTS synthesis, and voice catalog listing. Highlights: - LLM_ROUTER_MASTER_KEY becomes required; app.ts drops the graceful- skip branch and the chat fallback fetch path is gone. - /audio/speech and /audio/voices route through new routeTts / listTtsVoices entries that reuse the chat key-rotator + per-attempt timeout + abort propagation. - DEFAULT_CHAT_MODEL / DEFAULT_TTS_MODEL move from env to configKV so default-model swaps are hot-reloadable via Pub/Sub. - GATEWAY_BASE_URL removed from env schema, .env, .env.local, smoke, verification harness. Redis upstream-voices cache deleted — catalogs come from in-process adapter JSON. - routeTts splits adapter error contract by ApiError statusCode: 4xx propagates without fallback; 5xx folds into the network-failure fallback path. handleTTS wraps billing + span attribute in try/finally to plug a span leak when ttsMeter.accumulate() throws. - seed-router-config.ts rewritten with --merge (default) / --reset / --dry-run modes and env-var key handoff (OPENROUTER_KEY / AZURE_KEY / DASHSCOPE_KEY) so prod seed flows never put plaintext on the CLI. Adds DashScope CosyVoice seeding. Docs (CLAUDE.md, architecture-overview.md, transport-and-routes.md) reflect the new boundary. verifications/llm-router.md replaces the overstated "U1-U9 shipped" line with an evidence-vs-pending table. Tests: full 40-file / 343-case server suite green. New regressions pin ApiError 4xx → no-fallback, ApiError 5xx → fallback, TTS billing failure → span closed and error propagated.
This commit is contained in:
RainbowBird
+15
-3
@@ -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=""
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 有多种职责，但都不是余额真相源
|
||||
|
||||
|
||||
@@ -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()`
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 "<plaintext>" \
|
||||
* [--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()
|
||||
}
|
||||
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
|
||||
|
||||
+24
-32
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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' }],
|
||||
})
|
||||
|
||||
@@ -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')
|
||||
})
|
||||
})
|
||||
|
||||
|
||||
@@ -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')
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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' })
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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'
|
||||
|
||||
|
||||
Reference in New Issue
Block a user