diff --git a/apps/server/src/app.test.ts b/apps/server/src/app.test.ts index c2f98b379..027a764e6 100644 --- a/apps/server/src/app.test.ts +++ b/apps/server/src/app.test.ts @@ -47,6 +47,7 @@ function createTestDeps() { stripeService: {} as any, billingService: {} as any, adminFluxGrantsService: {} as any, + adminRouterConfigService: {} as any, ttsMeter: {} as any, requestLogService: {} as any, configKV: { diff --git a/apps/server/src/app.ts b/apps/server/src/app.ts index 9d3a1ab22..1cab72fd3 100644 --- a/apps/server/src/app.ts +++ b/apps/server/src/app.ts @@ -6,6 +6,7 @@ import type { Database } from './libs/db' import type { Env } from './libs/env' import type { OtelInstance } from './otel' import type { AdminFluxGrantsService } from './services/admin-flux-grants' +import type { AdminRouterConfigService } from './services/admin-router-config' import type { BillingService } from './services/billing/billing-service' import type { FluxMeter } from './services/billing/flux-meter' import type { CharacterService } from './services/characters' @@ -46,6 +47,7 @@ import { sessionMiddleware } from './middlewares/auth' import { emitOtelLog, initOtel } from './otel' import { registerActiveSessionsGauge } from './otel/gauges/active-sessions' import { registerDistinctActiveUsersGauge } from './otel/gauges/distinct-active-users' +import { createAdminRouterConfigRoutes } from './routes/admin/config/router' import { createAdminFluxGrantsRoutes } from './routes/admin/flux-grants' import { createAudioSpeechWsHandlers } from './routes/audio-speech-ws' import { createAuthRoutes } from './routes/auth' @@ -57,6 +59,7 @@ import { createV1Routes } from './routes/openai/v1' import { createProviderRoutes } from './routes/providers' import { createStripeRoutes } from './routes/stripe' import { createAdminFluxGrantsService } from './services/admin-flux-grants' +import { createAdminRouterConfigService } from './services/admin-router-config' import { createBillingService } from './services/billing/billing-service' import { createFluxMeter } from './services/billing/flux-meter' import { createCharacterService } from './services/characters' @@ -71,7 +74,6 @@ import { createProviderService } from './services/providers' import { createRequestLogService } from './services/request-log' import { createStripeService } from './services/stripe' import { createUserDeletionService } from './services/user-deletion' -import { ApiError, createInternalError } from './utils/error' import { createEnvelopeCrypto } from './utils/envelope-crypto' import { ApiError, createInternalError, createUnauthorizedError } from './utils/error' import { nanoid } from './utils/id' @@ -88,6 +90,7 @@ interface AppDeps { stripeService: StripeService billingService: BillingService adminFluxGrantsService: AdminFluxGrantsService + adminRouterConfigService: AdminRouterConfigService ttsMeter: FluxMeter requestLogService: RequestLogService configKV: ConfigKVService @@ -340,6 +343,14 @@ export async function buildApp(deps: AppDeps) { */ .route('/api/admin/flux-grants', createAdminFluxGrantsRoutes(deps.adminFluxGrantsService, deps.env)) + /** + * Admin LLM router config seeding/patching. Replaces the + * `scripts/seed-router-config.ts` and `scripts/seed-streaming-tts.ts` + * one-off scripts for in-cluster use; the scripts stay as break-glass + * tools. See `routes/admin/config/router/index.ts` for the body shape. + */ + .route('/api/admin/config/router', createAdminRouterConfigRoutes(deps.adminRouterConfigService, deps.env)) + /** * Catch-all 404 in JSON. Replaces hono's default `text/html` "404 Not * Found" so unmatched routes (typos, stale email links, scanners) get a @@ -608,6 +619,18 @@ export async function createApp() { }), }) + // Admin router-config seeding service. Reuses the shared envelope crypto + // so written ciphertexts decrypt cleanly under the same master key the + // gateway already uses. Mounted at POST /api/admin/config/router. + const adminRouterConfigService = injeca.provide('services:adminRouterConfig', { + dependsOn: { configKV, envelopeCrypto, redis }, + build: ({ dependsOn }) => createAdminRouterConfigService({ + configKV: dependsOn.configKV, + envelope: dependsOn.envelopeCrypto, + redis: dependsOn.redis, + }), + }) + // LLM router (KTD-5 in-process replacement for the knoway sidecar). // 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. @@ -633,6 +656,7 @@ export async function createApp() { stripeService, billingService, adminFluxGrantsService, + adminRouterConfigService, ttsMeter, configKV, envelopeCrypto, @@ -668,6 +692,7 @@ export async function createApp() { stripeService: resolved.stripeService, billingService: resolved.billingService, adminFluxGrantsService: resolved.adminFluxGrantsService, + adminRouterConfigService: resolved.adminRouterConfigService, ttsMeter: resolved.ttsMeter, requestLogService: resolved.requestLogService, configKV: resolved.configKV, diff --git a/apps/server/src/routes/admin/config/router/index.ts b/apps/server/src/routes/admin/config/router/index.ts new file mode 100644 index 000000000..fed8b3b83 --- /dev/null +++ b/apps/server/src/routes/admin/config/router/index.ts @@ -0,0 +1,198 @@ +import type { Env } from '../../../../libs/env' +import type { AdminRouterConfigService, SliceInput } from '../../../../services/admin-router-config' +import type { HonoEnv } from '../../../../types/hono' + +import { Hono } from 'hono' +import { + array, + boolean, + literal, + maxLength, + minLength, + nonEmpty, + object, + optional, + picklist, + pipe, + regex, + safeParse, + string, + url, + variant, +} from 'valibot' + +import { adminGuard } from '../../../../middlewares/admin-guard' +import { authGuard } from '../../../../middlewares/auth' +import { createBadRequestError } from '../../../../utils/error' + +/** + * Hard cap on slices per request. The envelope crypto is cheap (~1ms each), + * so the cap exists to bound request body size and audit log noise, not CPU. + * Realistic admin calls touch 1–3 providers at a time. + */ +const MAX_SLICES_PER_REQUEST = 20 + +/** + * Hard cap on plaintext key length. Real provider keys are 30–200 chars; + * 1KB leaves headroom for unusual formats while keeping the body lean. + */ +const MAX_KEY_LENGTH = 1024 + +/** AAD separator constraint mirrored from `keyEntrySchema` in config-kv. */ +const NO_PIPE = regex(/^[^|]+$/, 'must not contain "|" (reserved AAD separator)') + +const OpenRouterSliceSchema = object({ + kind: literal('openrouter'), + modelName: pipe(string(), nonEmpty('modelName is required'), maxLength(200), NO_PIPE), + overrideModel: pipe(string(), nonEmpty('overrideModel is required'), maxLength(200)), + plaintextKey: pipe(string(), nonEmpty('plaintextKey is required'), maxLength(MAX_KEY_LENGTH)), + baseURL: optional(pipe(string(), url('baseURL must be a valid URL'))), + keyEntryId: optional(pipe(string(), nonEmpty(), maxLength(200), NO_PIPE)), + headerTemplate: optional(pipe(string(), nonEmpty(), maxLength(200))), +}) + +const AzureSliceSchema = object({ + kind: literal('azure'), + modelName: pipe(string(), nonEmpty('modelName is required'), maxLength(200), NO_PIPE), + region: pipe(string(), nonEmpty('region is required'), maxLength(64)), + plaintextKey: pipe(string(), nonEmpty('plaintextKey is required'), maxLength(MAX_KEY_LENGTH)), + keyEntryId: optional(pipe(string(), nonEmpty(), maxLength(200), NO_PIPE)), +}) + +const DashscopeSliceSchema = object({ + kind: literal('dashscope-cosyvoice'), + modelName: pipe(string(), nonEmpty('modelName is required'), maxLength(200), NO_PIPE), + region: picklist(['intl', 'cn'], 'region must be "intl" or "cn"'), + upstreamModel: pipe(string(), nonEmpty('upstreamModel is required'), maxLength(200)), + plaintextKey: pipe(string(), nonEmpty('plaintextKey is required'), maxLength(MAX_KEY_LENGTH)), + keyEntryId: optional(pipe(string(), nonEmpty(), maxLength(200), NO_PIPE)), +}) + +/** + * `upstreamURL` must be ws:// or wss://. Mirrors the soft-validate in the + * seed-streaming-tts script — http(s):// here is almost always a copy-paste + * of the unspeech REST endpoint, which would fail at `new WebSocket()` + * inside the audio-speech-ws proxy with no actionable error for the admin. + */ +const StreamingTtsSliceSchema = object({ + kind: literal('streaming-tts'), + upstreamURL: pipe( + string(), + nonEmpty('upstreamURL is required'), + regex(/^wss?:\/\/\S+$/, 'upstreamURL must start with ws:// or wss://'), + maxLength(500), + ), + plaintextKey: pipe(string(), nonEmpty('plaintextKey is required'), maxLength(MAX_KEY_LENGTH)), + keyEntryId: optional(pipe(string(), nonEmpty(), maxLength(200), NO_PIPE)), +}) + +const SliceSchema = variant('kind', [ + OpenRouterSliceSchema, + AzureSliceSchema, + DashscopeSliceSchema, + StreamingTtsSliceSchema, +]) + +const BodySchema = object({ + mode: optional(picklist(['merge', 'reset']), 'merge'), + dryRun: optional(boolean(), false), + slices: pipe( + array(SliceSchema), + minLength(1, 'slices must not be empty'), + maxLength(MAX_SLICES_PER_REQUEST, `slices must be at most ${MAX_SLICES_PER_REQUEST} entries`), + ), + defaults: optional(object({ + chatModel: optional(pipe(string(), nonEmpty('defaults.chatModel must not be empty'), maxLength(200))), + ttsModel: optional(pipe(string(), nonEmpty('defaults.ttsModel must not be empty'), maxLength(200))), + })), +}) + +/** + * Admin route for seeding / patching the LLM router config tree. + * + * Mounted at `POST /api/admin/config/router`. Replaces the + * `scripts/seed-router-config.ts` and `scripts/seed-streaming-tts.ts` + * one-off scripts for routine in-cluster operation; the scripts are kept + * as break-glass tools for cold-boot / disaster recovery. + * + * Body shape (discriminated on `slices[].kind`): + * + * { + * "mode": "merge" | "reset", // defaults to "merge" + * "dryRun": false, // when true, returns redacted preview + * // and skips writes + invalidation + * "slices": [ + * { "kind": "openrouter", "modelName": "chat-default", + * "overrideModel": "openai/gpt-4o-mini", "plaintextKey": "..." }, + * { "kind": "azure", "modelName": "microsoft/v1", + * "region": "eastasia", "plaintextKey": "..." }, + * { "kind": "dashscope-cosyvoice", "modelName": "alibaba/cosyvoice-v2", + * "region": "intl", "upstreamModel": "cosyvoice-v2", + * "plaintextKey": "..." }, + * { "kind": "streaming-tts", + * "upstreamURL": "ws://airi-unspeech.railway.internal:5933/v1/audio/speech/stream", + * "plaintextKey": "..." } + * ], + * "defaults": { + * "chatModel": "chat-default", // writes DEFAULT_CHAT_MODEL + * "ttsModel": "alibaba/cosyvoice-v2" // writes DEFAULT_TTS_MODEL + * } + * } + * + * Response: + * + * { + * "applied": [{ kind, target, modelName?, keyEntryId, surface? }, ...], + * "invalidatedKeys": ["LLM_ROUTER_CONFIG", "DEFAULT_CHAT_MODEL", ...], + * "preview": { // ciphertext redacted to "" + * "LLM_ROUTER_CONFIG": { ... }, + * "STREAMING_TTS_UPSTREAM": { ... }, + * "DEFAULT_CHAT_MODEL": "chat-default", + * "DEFAULT_TTS_MODEL": "alibaba/cosyvoice-v2" + * } + * } + * + * Security notes: + * - `plaintextKey` is consumed in-process and never returned. The preview + * only ever contains length-redacted ciphertext. + * - The route relies on `bodyLimit(1MB)` from the global middleware chain; + * no per-route bumps. + */ +export function createAdminRouterConfigRoutes( + service: AdminRouterConfigService, + env: Env, +) { + return new Hono() + .use('*', authGuard) + .use('*', adminGuard(env)) + .post('/', async (c) => { + const user = c.get('user')! + + const raw = await c.req.json().catch(() => null) + if (raw == null) + throw createBadRequestError('Request body must be JSON', 'INVALID_BODY') + + const parsed = safeParse(BodySchema, raw) + if (!parsed.success) { + throw createBadRequestError( + 'Invalid request body', + 'INVALID_BODY', + parsed.issues.map(i => ({ + path: i.path?.map(p => p.key).join('.'), + message: i.message, + })), + ) + } + + const body = parsed.output + const result = await service.apply({ + mode: body.mode, + dryRun: body.dryRun, + slices: body.slices as SliceInput[], + defaults: body.defaults, + actorUserId: user.id, + }) + + return c.json(result) + }) +} diff --git a/apps/server/src/services/admin-router-config/index.ts b/apps/server/src/services/admin-router-config/index.ts new file mode 100644 index 000000000..a54d8a291 --- /dev/null +++ b/apps/server/src/services/admin-router-config/index.ts @@ -0,0 +1,502 @@ +import type Redis from 'ioredis' +import type { InferOutput } from 'valibot' + +import type { EnvelopeCrypto } from '../../utils/envelope-crypto' +import type { ConfigKVService, llmModelSchema, llmRouterConfigSchema, ttsModelSchema, ttsUpstreamSchema } from '../config-kv' + +import { useLogger } from '@guiiai/logg' + +import { createBadRequestError } from '../../utils/error' + +/** + * AAD label used when encrypting/decrypting the streaming TTS upstream key. + * Must match `STREAM_MODEL_LABEL_FALLBACK` in + * apps/server/src/routes/audio-speech-ws/index.ts — the ws proxy decrypts + * with this label, so writing under a different one surfaces as + * `DECRYPT_FAILED` at session start. + */ +const STREAMING_TTS_AAD_MODEL_NAME = 'streaming-tts' + +/** Default key entry id per provider. Operator can override per request. */ +const DEFAULT_KEY_ENTRY_IDS = { + 'openrouter': 'openrouter-prod-1', + 'azure': 'azure-tts-prod-1', + 'dashscope-cosyvoice': 'dashscope-tts-prod-1', + 'streaming-tts': 'volcengine-prod-1', +} as const + +type LlmRouterConfig = InferOutput +type LlmModel = InferOutput +type TtsModel = InferOutput +type TtsUpstream = InferOutput + +/** + * Per-provider input. The admin route validates the shape with Valibot + * (discriminated on `kind`) before handing the slice to the service. + * + * `plaintextKey` enters the process here, gets envelope-encrypted in + * {@link buildSlice}, and is dropped from memory before the response is + * built — it never reaches logs, ciphertext previews, or audit records. + */ +export type SliceInput + = | OpenRouterSliceInput + | AzureSliceInput + | DashscopeSliceInput + | StreamingTtsSliceInput + +export interface OpenRouterSliceInput { + kind: 'openrouter' + /** Key under `LLM_ROUTER_CONFIG.llm.models`. */ + modelName: string + /** Upstream model id sent to OpenRouter (e.g. `openai/gpt-4o-mini`). */ + overrideModel: string + /** Plaintext provider key. Encrypted in-place; never echoed back. */ + plaintextKey: string + /** @default 'https://openrouter.ai/api/v1' */ + baseURL?: string + /** @default 'openrouter-prod-1' */ + keyEntryId?: string + /** @default 'Bearer {KEY}' */ + headerTemplate?: string +} + +export interface AzureSliceInput { + kind: 'azure' + /** Key under `LLM_ROUTER_CONFIG.tts.models` (e.g. `microsoft/v1`). */ + modelName: string + /** Azure Speech region, used in baseURL and `adapterParams.region`. */ + region: string + plaintextKey: string + /** @default 'azure-tts-prod-1' */ + keyEntryId?: string +} + +export interface DashscopeSliceInput { + kind: 'dashscope-cosyvoice' + /** Key under `LLM_ROUTER_CONFIG.tts.models` (e.g. `alibaba/cosyvoice-v2`). */ + modelName: string + /** `intl` → dashscope-intl.aliyuncs.com (Singapore); `cn` → dashscope.aliyuncs.com (Beijing). */ + region: 'intl' | 'cn' + /** Concrete cosyvoice variant the adapter calls upstream. Independent from `modelName`. */ + upstreamModel: string + plaintextKey: string + /** @default 'dashscope-tts-prod-1' */ + keyEntryId?: string +} + +export interface StreamingTtsSliceInput { + kind: 'streaming-tts' + /** unspeech ws endpoint: `ws://airi-unspeech.railway.internal:5933/v1/audio/speech/stream` etc. */ + upstreamURL: string + /** Upstream provider key (Volcengine `X-Api-Key`), not an unspeech token. */ + plaintextKey: string + /** @default 'volcengine-prod-1' */ + keyEntryId?: string +} + +interface LlmModelSlice { + target: 'llm-router' + surface: 'llm' + kind: 'openrouter' + modelName: string + model: LlmModel + keyEntryId: string +} + +interface TtsModelSlice { + target: 'llm-router' + surface: 'tts' + kind: 'azure' | 'dashscope-cosyvoice' + modelName: string + model: TtsModel + keyEntryId: string +} + +interface StreamingTtsSlice { + target: 'streaming-tts' + kind: 'streaming-tts' + value: TtsUpstream + keyEntryId: string +} + +type BuiltSlice = LlmModelSlice | TtsModelSlice | StreamingTtsSlice + +/** + * Encrypts an OpenRouter slice into the LLM_ROUTER_CONFIG.llm shape. + * + * Use when: + * - Admin posts an `openrouter` slice; called by {@link buildSlice}. + * + * Returns: + * - A `BuiltSlice` whose `model.upstreams[0].keys[0].ciphertext` is the + * envelope-encrypted plaintext key with AAD `{modelName, keyEntryId}`. + */ +export function buildOpenRouterSlice(input: OpenRouterSliceInput, envelope: EnvelopeCrypto): LlmModelSlice { + const keyEntryId = input.keyEntryId ?? DEFAULT_KEY_ENTRY_IDS.openrouter + const ciphertext = envelope.encryptKey(input.plaintextKey, { + modelName: input.modelName, + keyEntryId, + }) + return { + target: 'llm-router', + surface: 'llm', + kind: 'openrouter', + modelName: input.modelName, + keyEntryId, + model: { + upstreams: [{ + baseURL: input.baseURL ?? 'https://openrouter.ai/api/v1', + overrideModel: input.overrideModel, + keys: [{ id: keyEntryId, ciphertext }], + headerTemplate: input.headerTemplate ?? 'Bearer {KEY}', + }], + } as LlmModel, + } +} + +/** + * Encrypts an Azure TTS slice into the LLM_ROUTER_CONFIG.tts shape. + * + * Use when: + * - Admin posts an `azure` slice; called by {@link buildSlice}. + */ +export function buildAzureSlice(input: AzureSliceInput, envelope: EnvelopeCrypto): TtsModelSlice { + const keyEntryId = input.keyEntryId ?? DEFAULT_KEY_ENTRY_IDS.azure + const ciphertext = envelope.encryptKey(input.plaintextKey, { + modelName: input.modelName, + keyEntryId, + }) + return { + target: 'llm-router', + surface: 'tts', + kind: 'azure', + modelName: input.modelName, + keyEntryId, + model: { + provider: 'azure', + upstreams: [{ + baseURL: `https://${input.region}.tts.speech.microsoft.com/cognitiveservices/v1`, + keys: [{ id: keyEntryId, ciphertext }], + adapterParams: { region: input.region }, + }], + } as TtsModel, + } +} + +/** + * Encrypts a DashScope cosyvoice slice into the LLM_ROUTER_CONFIG.tts shape. + * + * Use when: + * - Admin posts a `dashscope-cosyvoice` slice; called by {@link buildSlice}. + * + * Expects: + * - The dashscope-cosyvoice adapter does NOT append + * `/services/audio/tts/SpeechSynthesizer`; the full non-streaming endpoint + * path must be baked into `baseURL` here. A bare `/api/v1` baseURL was the + * root cause of the 404 storm during the v1→v2 migration. + */ +export function buildDashscopeSlice(input: DashscopeSliceInput, envelope: EnvelopeCrypto): TtsModelSlice { + const keyEntryId = input.keyEntryId ?? DEFAULT_KEY_ENTRY_IDS['dashscope-cosyvoice'] + const ciphertext = envelope.encryptKey(input.plaintextKey, { + modelName: input.modelName, + keyEntryId, + }) + const host = input.region === 'cn' + ? 'dashscope.aliyuncs.com' + : 'dashscope-intl.aliyuncs.com' + return { + target: 'llm-router', + surface: 'tts', + kind: 'dashscope-cosyvoice', + modelName: input.modelName, + keyEntryId, + model: { + provider: 'dashscope-cosyvoice', + upstreams: [{ + baseURL: `https://${host}/api/v1/services/audio/tts/SpeechSynthesizer`, + keys: [{ id: keyEntryId, ciphertext }], + adapterParams: { model: input.upstreamModel }, + }], + } as TtsModel, + } +} + +/** + * Encrypts a streaming TTS slice into the STREAMING_TTS_UPSTREAM shape. + * + * Use when: + * - Admin posts a `streaming-tts` slice; called by {@link buildSlice}. + * + * Expects: + * - `upstreamURL` starts with `ws://` or `wss://`. http:// is almost always a + * copy-paste of the unspeech REST endpoint and fails at `new WebSocket()` + * inside the audio-speech-ws proxy. + */ +export function buildStreamingTtsSlice(input: StreamingTtsSliceInput, envelope: EnvelopeCrypto): StreamingTtsSlice { + const keyEntryId = input.keyEntryId ?? DEFAULT_KEY_ENTRY_IDS['streaming-tts'] + const ciphertext = envelope.encryptKey(input.plaintextKey, { + modelName: STREAMING_TTS_AAD_MODEL_NAME, + keyEntryId, + }) + return { + target: 'streaming-tts', + kind: 'streaming-tts', + keyEntryId, + value: { + baseURL: input.upstreamURL, + keys: [{ id: keyEntryId, ciphertext }], + adapterParams: {}, + }, + } +} + +/** + * Encrypts a slice input. Routes to the per-kind builder. + * + * Use when: + * - The service main path needs to turn an admin-supplied slice into a + * ready-to-write configKV fragment. Tests dispatch the same way. + */ +export function buildSlice(input: SliceInput, envelope: EnvelopeCrypto): BuiltSlice { + switch (input.kind) { + case 'openrouter': + return buildOpenRouterSlice(input, envelope) + case 'azure': + return buildAzureSlice(input, envelope) + case 'dashscope-cosyvoice': + return buildDashscopeSlice(input, envelope) + case 'streaming-tts': + return buildStreamingTtsSlice(input, envelope) + } +} + +/** + * Computes the next `LLM_ROUTER_CONFIG` tree. + * + * Use when: + * - One or more LLM/TTS slices need to be merged into (or reset on top of) + * the existing configKV entry. + * + * Expects: + * - `existing` is the current parsed `LLM_ROUTER_CONFIG` (or `null` if the + * entry is absent). `mode: 'merge'` preserves models not touched this run; + * `mode: 'reset'` drops every prior entry and keeps only what is in + * `slices`. + * + * Returns: + * - The next config tree, ready to feed `configKV.set('LLM_ROUTER_CONFIG', ...)`. + * `defaults` is preserved verbatim when merging — the admin endpoint does + * not currently re-tune timeouts via this path. + */ +export function buildNextRouterConfig( + mode: 'merge' | 'reset', + existing: LlmRouterConfig | null | undefined, + slices: (LlmModelSlice | TtsModelSlice)[], +): LlmRouterConfig { + const llmModels: Record + = mode === 'merge' && existing?.llm?.models ? { ...existing.llm.models } : {} + const ttsModels: Record + = mode === 'merge' && existing?.tts?.models ? { ...existing.tts.models } : {} + + for (const slice of slices) { + if (slice.surface === 'llm') + llmModels[slice.modelName] = slice.model + else + ttsModels[slice.modelName] = slice.model + } + + // Defaults live alongside the models but aren't editable through this + // endpoint yet; keep the existing tree in merge mode so we don't blow them + // away. In reset mode, fall back to the schema default object. + const defaults = mode === 'merge' && existing?.defaults + ? existing.defaults + : { perAttemptTimeoutMs: 30000, fullChainTimeoutMs: 60000, fallbackHttpCodes: [401, 402, 403, 429, 500, 502, 503, 504] } + + return { + llm: { models: llmModels }, + tts: { models: ttsModels }, + defaults, + } +} + +/** + * Redacts every `ciphertext` field down to its byte length for safe response + * preview. + * + * Before: + * - `{ "keys": [{ "id": "k1", "ciphertext": "aGVsbG8=...long..." }] }` + * + * After: + * - `{ "keys": [{ "id": "k1", "ciphertext": "" }] }` + */ +export function redactCiphertext(value: unknown): unknown { + if (Array.isArray(value)) + return value.map(redactCiphertext) + if (value && typeof value === 'object') { + const out: Record = {} + for (const [k, v] of Object.entries(value as Record)) { + if (k === 'ciphertext' && typeof v === 'string') + out[k] = `` + else + out[k] = redactCiphertext(v) + } + return out + } + return value +} + +export interface ApplyInput { + mode: 'merge' | 'reset' + dryRun: boolean + slices: SliceInput[] + defaults?: { + chatModel?: string + ttsModel?: string + } + /** Admin user id for audit logging only. Not part of the persisted config. */ + actorUserId?: string +} + +export interface AppliedSummary { + kind: SliceInput['kind'] + target: 'llm-router' | 'streaming-tts' + surface?: 'llm' | 'tts' + modelName?: string + keyEntryId: string +} + +export interface ApplyResult { + applied: AppliedSummary[] + invalidatedKeys: string[] + preview: { + LLM_ROUTER_CONFIG?: unknown + STREAMING_TTS_UPSTREAM?: unknown + DEFAULT_CHAT_MODEL?: string + DEFAULT_TTS_MODEL?: string + } +} + +interface AdminRouterConfigDeps { + configKV: ConfigKVService + envelope: EnvelopeCrypto + redis: Redis +} + +/** + * Admin service for seeding / patching the LLM router configKV tree. + * + * Use when: + * - Mounting `POST /api/admin/config/router`. The HTTP layer parses the body + * and forwards it here; this layer owns encryption, merge semantics, + * validation, and cross-instance invalidation. + * + * Expects: + * - `envelope` is wired with the same master key the LLM router decrypts + * under, otherwise written ciphertexts will surface as `DECRYPT_FAILED` at + * the first /chat/completions or /audio/speech request. + * + * Returns: + * - `apply()` resolves to the redacted preview, the list of touched configKV + * keys, and a per-slice summary suitable for an audit row. Plaintext keys + * are dropped from memory before this resolves. + */ +export function createAdminRouterConfigService(deps: AdminRouterConfigDeps) { + const logger = useLogger('admin-router-config').useGlobalConfig() + + /** + * Applies an admin request, returning the redacted preview either way. + * + * Expects: + * - At most one `streaming-tts` slice per request. The streaming surface + * is a single unspeech instance per deployment, so multiple entries are + * almost always an admin mistake. + */ + async function apply(input: ApplyInput): Promise { + const streamingCount = input.slices.filter(s => s.kind === 'streaming-tts').length + if (streamingCount > 1) + throw createBadRequestError('At most one streaming-tts slice per request', 'INVALID_BODY') + + // Step 1: encrypt every slice. Throws (via envelope) only on malformed + // master key, which means the deployment is broken; surface as 500. + const built = input.slices.map(s => buildSlice(s, deps.envelope)) + + const llmTtsSlices = built.filter((s): s is LlmModelSlice | TtsModelSlice => s.target === 'llm-router') + const streamingSlice = built.find((s): s is StreamingTtsSlice => s.target === 'streaming-tts') + + // Step 2: build the next LLM_ROUTER_CONFIG tree if any LLM/TTS slice + // was supplied. `merge` reads existing first; `reset` skips the read. + let nextRouterConfig: LlmRouterConfig | undefined + if (llmTtsSlices.length > 0) { + const existing = input.mode === 'merge' + ? await deps.configKV.getOptional('LLM_ROUTER_CONFIG') + : null + nextRouterConfig = buildNextRouterConfig(input.mode, existing, llmTtsSlices) + } + + const preview: ApplyResult['preview'] = {} + if (nextRouterConfig) + preview.LLM_ROUTER_CONFIG = redactCiphertext(nextRouterConfig) + if (streamingSlice) + preview.STREAMING_TTS_UPSTREAM = redactCiphertext(streamingSlice.value) + if (input.defaults?.chatModel) + preview.DEFAULT_CHAT_MODEL = input.defaults.chatModel + if (input.defaults?.ttsModel) + preview.DEFAULT_TTS_MODEL = input.defaults.ttsModel + + const applied: AppliedSummary[] = built.map(s => s.target === 'streaming-tts' + ? { kind: s.kind, target: s.target, keyEntryId: s.keyEntryId } + : { kind: s.kind, target: s.target, surface: s.surface, modelName: s.modelName, keyEntryId: s.keyEntryId }) + + if (input.dryRun) { + logger.withFields({ + actorUserId: input.actorUserId, + mode: input.mode, + applied, + dryRun: true, + }).log('admin-router-config dry-run') + return { applied, invalidatedKeys: [], preview } + } + + // Step 3: real writes. configKV.set runs the per-key valibot validator, + // so a malformed shape fails here BEFORE we publish invalidation. + const invalidatedKeys: string[] = [] + if (nextRouterConfig) { + await deps.configKV.set('LLM_ROUTER_CONFIG', nextRouterConfig as never) + invalidatedKeys.push('LLM_ROUTER_CONFIG') + } + if (streamingSlice) { + await deps.configKV.set('STREAMING_TTS_UPSTREAM', streamingSlice.value as never) + invalidatedKeys.push('STREAMING_TTS_UPSTREAM') + } + if (input.defaults?.chatModel) { + await deps.configKV.set('DEFAULT_CHAT_MODEL', input.defaults.chatModel) + invalidatedKeys.push('DEFAULT_CHAT_MODEL') + } + if (input.defaults?.ttsModel) { + await deps.configKV.set('DEFAULT_TTS_MODEL', input.defaults.ttsModel) + invalidatedKeys.push('DEFAULT_TTS_MODEL') + } + + // Step 4: cross-instance invalidation. config-sync-subscriber currently + // only acts on `LLM_ROUTER_CONFIG` (audio-speech-ws reads + // `STREAMING_TTS_UPSTREAM` fresh on every connection), but we publish + // all touched keys for forward compatibility. + for (const key of invalidatedKeys) { + const payload = JSON.stringify({ key, version: Date.now(), publishedAt: Date.now() }) + await deps.redis.publish('configkv:invalidate', payload) + } + + logger.withFields({ + actorUserId: input.actorUserId, + mode: input.mode, + applied, + invalidatedKeys, + }).log('admin-router-config applied') + + return { applied, invalidatedKeys, preview } + } + + return { apply } +} + +export type AdminRouterConfigService = ReturnType diff --git a/apps/server/src/services/admin-router-config/tests/admin-router-config.test.ts b/apps/server/src/services/admin-router-config/tests/admin-router-config.test.ts new file mode 100644 index 000000000..1baa849ae --- /dev/null +++ b/apps/server/src/services/admin-router-config/tests/admin-router-config.test.ts @@ -0,0 +1,429 @@ +import type Redis from 'ioredis' + +import type { ConfigKVService } from '../../config-kv' + +import { randomBytes } from 'node:crypto' + +import { beforeEach, describe, expect, it, vi } from 'vitest' + +import { + buildAzureSlice, + buildDashscopeSlice, + buildNextRouterConfig, + buildOpenRouterSlice, + buildStreamingTtsSlice, + createAdminRouterConfigService, + redactCiphertext, +} from '..' +import { createEnvelopeCrypto } from '../../../utils/envelope-crypto' + +function freshEnvelope() { + return createEnvelopeCrypto({ masterKey: randomBytes(32) }) +} + +interface FakeConfigKV { + store: Map + service: ConfigKVService + publishedChannels: { channel: string, payload: string }[] +} + +/** + * In-memory fake for ConfigKVService + a thin Redis publish stub. + * + * Use when: + * - Service tests need to assert which configKV entries got written and + * which channel publishes fired, without touching real Redis. + */ +function fakeConfigKV(): FakeConfigKV { + const store = new Map() + const service: Partial = { + async getOptional(key: string) { + return (store.get(key) ?? null) as never + }, + async getOrThrow(key: string) { + const v = store.get(key) + if (v === undefined) + throw new Error(`fake getOrThrow missing ${key}`) + return v as never + }, + async get(key: string) { + return this.getOrThrow!(key as never) + }, + async set(key: string, value: unknown) { + store.set(key, value) + }, + } + return { store, service: service as ConfigKVService, publishedChannels: [] } +} + +function fakeRedis(captured: { channel: string, payload: string }[]): Redis { + return { + publish: vi.fn(async (channel: string, payload: string) => { + captured.push({ channel, payload }) + return 1 + }), + } as unknown as Redis +} + +describe('redactCiphertext', () => { + it('replaces ciphertext strings with a length tag', () => { + const input = { + keys: [{ id: 'k1', ciphertext: 'a'.repeat(100) }], + adapterParams: { nested: { ciphertext: 'b'.repeat(50) } }, + } + expect(redactCiphertext(input)).toEqual({ + keys: [{ id: 'k1', ciphertext: '' }], + adapterParams: { nested: { ciphertext: '' } }, + }) + }) + + it('leaves non-ciphertext fields unchanged', () => { + expect(redactCiphertext({ baseURL: 'https://x', count: 3, flag: true })).toEqual({ + baseURL: 'https://x', + count: 3, + flag: true, + }) + }) + + it('walks arrays', () => { + expect(redactCiphertext([{ ciphertext: 'xx' }, { ciphertext: 'yyy' }])).toEqual([ + { ciphertext: '' }, + { ciphertext: '' }, + ]) + }) +}) + +describe('buildOpenRouterSlice', () => { + it('encrypts the plaintext key under {modelName, keyEntryId} AAD', () => { + const envelope = freshEnvelope() + const built = buildOpenRouterSlice({ + kind: 'openrouter', + modelName: 'chat-default', + overrideModel: 'openai/gpt-4o-mini', + plaintextKey: 'sk-or-secret', + }, envelope) + + expect(built.target).toBe('llm-router') + expect(built.surface).toBe('llm') + expect(built.modelName).toBe('chat-default') + expect(built.keyEntryId).toBe('openrouter-prod-1') + + const upstream = built.model.upstreams[0] + expect(upstream.baseURL).toBe('https://openrouter.ai/api/v1') + expect(upstream.overrideModel).toBe('openai/gpt-4o-mini') + expect(upstream.headerTemplate).toBe('Bearer {KEY}') + + // Round-trip the ciphertext under the same AAD — guards against the + // AAD getting silently changed (which would surface as DECRYPT_FAILED + // at gateway runtime, but never in tests like this if we only checked + // the ciphertext length). + const decrypted = envelope.decryptKey(upstream.keys[0].ciphertext, { + modelName: 'chat-default', + keyEntryId: 'openrouter-prod-1', + }) + expect(decrypted.toString('utf8')).toBe('sk-or-secret') + }) + + it('respects custom baseURL, keyEntryId, and headerTemplate', () => { + const envelope = freshEnvelope() + const built = buildOpenRouterSlice({ + kind: 'openrouter', + modelName: 'chat-default', + overrideModel: 'openai/gpt-4o-mini', + plaintextKey: 'sk', + baseURL: 'https://proxy.example/api/v1', + keyEntryId: 'openrouter-prod-2', + headerTemplate: 'X-Custom-Token {KEY}', + }, envelope) + + expect(built.keyEntryId).toBe('openrouter-prod-2') + expect(built.model.upstreams[0].baseURL).toBe('https://proxy.example/api/v1') + expect(built.model.upstreams[0].headerTemplate).toBe('X-Custom-Token {KEY}') + expect(built.model.upstreams[0].keys[0].id).toBe('openrouter-prod-2') + }) +}) + +describe('buildAzureSlice', () => { + it('builds the cognitiveservices baseURL from region and surfaces region in adapterParams', () => { + const envelope = freshEnvelope() + const built = buildAzureSlice({ + kind: 'azure', + modelName: 'microsoft/v1', + region: 'eastasia', + plaintextKey: 'azure-key', + }, envelope) + + expect(built.kind).toBe('azure') + expect(built.model.provider).toBe('azure') + expect(built.model.upstreams[0].baseURL).toBe('https://eastasia.tts.speech.microsoft.com/cognitiveservices/v1') + expect(built.model.upstreams[0].adapterParams).toEqual({ region: 'eastasia' }) + + const decrypted = envelope.decryptKey(built.model.upstreams[0].keys[0].ciphertext, { + modelName: 'microsoft/v1', + keyEntryId: 'azure-tts-prod-1', + }) + expect(decrypted.toString('utf8')).toBe('azure-key') + }) +}) + +describe('buildDashscopeSlice', () => { + it.each([ + ['intl', 'dashscope-intl.aliyuncs.com'], + ['cn', 'dashscope.aliyuncs.com'], + ] as const)('uses the %s region host', (region, host) => { + const envelope = freshEnvelope() + const built = buildDashscopeSlice({ + kind: 'dashscope-cosyvoice', + modelName: 'alibaba/cosyvoice-v2', + region, + upstreamModel: 'cosyvoice-v2', + plaintextKey: 'sk-dash', + }, envelope) + + expect(built.model.upstreams[0].baseURL).toBe(`https://${host}/api/v1/services/audio/tts/SpeechSynthesizer`) + expect(built.model.upstreams[0].adapterParams).toEqual({ model: 'cosyvoice-v2' }) + }) +}) + +describe('buildStreamingTtsSlice', () => { + it('encrypts under the streaming-tts AAD model label (must match audio-speech-ws decrypt)', () => { + const envelope = freshEnvelope() + const built = buildStreamingTtsSlice({ + kind: 'streaming-tts', + upstreamURL: 'ws://airi-unspeech.railway.internal:5933/v1/audio/speech/stream', + plaintextKey: 'volc-key', + }, envelope) + + expect(built.target).toBe('streaming-tts') + expect(built.value.baseURL).toBe('ws://airi-unspeech.railway.internal:5933/v1/audio/speech/stream') + + // The AAD modelName MUST be the literal 'streaming-tts' — anything else + // surfaces as DECRYPT_FAILED at session start in audio-speech-ws. + const decrypted = envelope.decryptKey(built.value.keys[0].ciphertext, { + modelName: 'streaming-tts', + keyEntryId: 'volcengine-prod-1', + }) + expect(decrypted.toString('utf8')).toBe('volc-key') + }) +}) + +describe('buildNextRouterConfig', () => { + it('merge mode preserves models not touched this run', () => { + const envelope = freshEnvelope() + const existing = { + llm: { models: { 'untouched-chat': { upstreams: [{ baseURL: 'https://old', keys: [{ id: 'k', ciphertext: 'c' }], headerTemplate: 'Bearer {KEY}' }] } } }, + tts: { models: { 'untouched-tts': { provider: 'azure' as const, upstreams: [{ baseURL: 'https://old-tts', keys: [{ id: 'k', ciphertext: 'c' }], adapterParams: {} }] } } }, + defaults: { perAttemptTimeoutMs: 12345, fullChainTimeoutMs: 60000, fallbackHttpCodes: [500] }, + } + const newSlice = buildOpenRouterSlice({ + kind: 'openrouter', + modelName: 'chat-default', + overrideModel: 'openai/gpt-4o-mini', + plaintextKey: 'sk', + }, envelope) + + const next = buildNextRouterConfig('merge', existing, [newSlice]) + + expect(Object.keys(next.llm.models).sort()).toEqual(['chat-default', 'untouched-chat']) + expect(Object.keys(next.tts.models)).toEqual(['untouched-tts']) + // Defaults preserved verbatim in merge mode — the admin endpoint does + // not currently re-tune timeouts, and zeroing them would silently break + // gateway timeouts. + expect(next.defaults?.perAttemptTimeoutMs).toBe(12345) + }) + + it('reset mode drops every prior entry and uses default timeouts', () => { + const envelope = freshEnvelope() + const existing = { + llm: { models: { old: { upstreams: [{ baseURL: 'https://old', keys: [{ id: 'k', ciphertext: 'c' }], headerTemplate: 'Bearer {KEY}' }] } } }, + tts: { models: {} }, + defaults: { perAttemptTimeoutMs: 12345, fullChainTimeoutMs: 60000, fallbackHttpCodes: [500] }, + } + const newSlice = buildOpenRouterSlice({ + kind: 'openrouter', + modelName: 'chat-default', + overrideModel: 'openai/gpt-4o-mini', + plaintextKey: 'sk', + }, envelope) + + const next = buildNextRouterConfig('reset', existing, [newSlice]) + + expect(Object.keys(next.llm.models)).toEqual(['chat-default']) + expect(next.defaults?.perAttemptTimeoutMs).toBe(30000) + }) + + it('starts from empty when existing is null (first-time bootstrap)', () => { + const envelope = freshEnvelope() + const newSlice = buildAzureSlice({ + kind: 'azure', + modelName: 'microsoft/v1', + region: 'eastasia', + plaintextKey: 'azure-key', + }, envelope) + + const next = buildNextRouterConfig('merge', null, [newSlice]) + expect(Object.keys(next.llm.models)).toEqual([]) + expect(Object.keys(next.tts.models)).toEqual(['microsoft/v1']) + }) +}) + +describe('createAdminRouterConfigService', () => { + let kv: FakeConfigKV + let captured: { channel: string, payload: string }[] + let redis: Redis + let envelope: ReturnType + + beforeEach(() => { + kv = fakeConfigKV() + captured = [] + redis = fakeRedis(captured) + envelope = freshEnvelope() + }) + + it('dry-run returns redacted preview without touching the store or pub/sub', async () => { + const service = createAdminRouterConfigService({ configKV: kv.service, envelope, redis }) + const result = await service.apply({ + mode: 'merge', + dryRun: true, + slices: [{ + kind: 'openrouter', + modelName: 'chat-default', + overrideModel: 'openai/gpt-4o-mini', + plaintextKey: 'sk-or-secret', + }], + defaults: { chatModel: 'chat-default' }, + }) + + expect(kv.store.size).toBe(0) + expect(captured).toEqual([]) + expect(result.invalidatedKeys).toEqual([]) + + // Preview must redact ciphertext — leaking plaintext OR raw ciphertext + // back to admin response would be a regression. + const preview = result.preview.LLM_ROUTER_CONFIG as { llm: { models: Record } } + const ct = preview.llm.models['chat-default'].upstreams[0].keys[0].ciphertext + expect(ct).toMatch(/^$/) + expect(ct).not.toContain('sk-or-secret') + + expect(result.preview.DEFAULT_CHAT_MODEL).toBe('chat-default') + }) + + it('writes LLM_ROUTER_CONFIG, DEFAULT_CHAT_MODEL, and publishes invalidation', async () => { + const service = createAdminRouterConfigService({ configKV: kv.service, envelope, redis }) + const result = await service.apply({ + mode: 'reset', + dryRun: false, + slices: [{ + kind: 'openrouter', + modelName: 'chat-default', + overrideModel: 'openai/gpt-4o-mini', + plaintextKey: 'sk', + }], + defaults: { chatModel: 'chat-default' }, + }) + + expect(kv.store.has('LLM_ROUTER_CONFIG')).toBe(true) + expect(kv.store.get('DEFAULT_CHAT_MODEL')).toBe('chat-default') + expect(result.invalidatedKeys.sort()).toEqual(['DEFAULT_CHAT_MODEL', 'LLM_ROUTER_CONFIG']) + expect(captured.map(p => JSON.parse(p.payload).key).sort()).toEqual(['DEFAULT_CHAT_MODEL', 'LLM_ROUTER_CONFIG']) + }) + + it('writes STREAMING_TTS_UPSTREAM and publishes invalidation when a streaming-tts slice is included', async () => { + const service = createAdminRouterConfigService({ configKV: kv.service, envelope, redis }) + const result = await service.apply({ + mode: 'merge', + dryRun: false, + slices: [{ + kind: 'streaming-tts', + upstreamURL: 'wss://unspeech.example/v1/audio/speech/stream', + plaintextKey: 'volc', + }], + }) + + expect(kv.store.has('STREAMING_TTS_UPSTREAM')).toBe(true) + expect(kv.store.has('LLM_ROUTER_CONFIG')).toBe(false) + expect(result.invalidatedKeys).toEqual(['STREAMING_TTS_UPSTREAM']) + expect(captured.map(p => JSON.parse(p.payload).key)).toEqual(['STREAMING_TTS_UPSTREAM']) + }) + + it('rejects multiple streaming-tts slices', async () => { + const service = createAdminRouterConfigService({ configKV: kv.service, envelope, redis }) + await expect(service.apply({ + mode: 'merge', + dryRun: true, + slices: [ + { kind: 'streaming-tts', upstreamURL: 'ws://a/x', plaintextKey: 'a' }, + { kind: 'streaming-tts', upstreamURL: 'ws://b/x', plaintextKey: 'b' }, + ], + })).rejects.toThrow(/At most one streaming-tts/i) + }) + + it('merge mode reads existing LLM_ROUTER_CONFIG and preserves untouched models', async () => { + // Seed an existing entry directly into the fake store, matching the + // shape configKV.getOptional would have returned after a prior admin call. + kv.store.set('LLM_ROUTER_CONFIG', { + llm: { models: { 'preexisting-chat': { upstreams: [{ baseURL: 'https://x', keys: [{ id: 'k', ciphertext: 'c' }], headerTemplate: 'Bearer {KEY}' }] } } }, + tts: { models: {} }, + defaults: { perAttemptTimeoutMs: 30000, fullChainTimeoutMs: 60000, fallbackHttpCodes: [500] }, + }) + + const service = createAdminRouterConfigService({ configKV: kv.service, envelope, redis }) + await service.apply({ + mode: 'merge', + dryRun: false, + slices: [{ + kind: 'azure', + modelName: 'microsoft/v1', + region: 'eastasia', + plaintextKey: 'azure-key', + }], + }) + + const written = kv.store.get('LLM_ROUTER_CONFIG') as { llm: { models: Record }, tts: { models: Record } } + expect(Object.keys(written.llm.models)).toEqual(['preexisting-chat']) + expect(Object.keys(written.tts.models)).toEqual(['microsoft/v1']) + }) + + it('reset mode skips the existing read and drops prior entries', async () => { + kv.store.set('LLM_ROUTER_CONFIG', { + llm: { models: { 'should-be-dropped': { upstreams: [{ baseURL: 'https://x', keys: [{ id: 'k', ciphertext: 'c' }], headerTemplate: 'Bearer {KEY}' }] } } }, + tts: { models: {} }, + defaults: { perAttemptTimeoutMs: 30000, fullChainTimeoutMs: 60000, fallbackHttpCodes: [500] }, + }) + + const getOptionalSpy = vi.spyOn(kv.service, 'getOptional') + + const service = createAdminRouterConfigService({ configKV: kv.service, envelope, redis }) + await service.apply({ + mode: 'reset', + dryRun: false, + slices: [{ + kind: 'openrouter', + modelName: 'chat-default', + overrideModel: 'openai/gpt-4o-mini', + plaintextKey: 'sk', + }], + }) + + expect(getOptionalSpy).not.toHaveBeenCalled() + const written = kv.store.get('LLM_ROUTER_CONFIG') as { llm: { models: Record } } + expect(Object.keys(written.llm.models)).toEqual(['chat-default']) + }) + + it('returns per-slice applied summaries that the audit log can use', async () => { + const service = createAdminRouterConfigService({ configKV: kv.service, envelope, redis }) + const result = await service.apply({ + mode: 'merge', + dryRun: true, + slices: [ + { kind: 'openrouter', modelName: 'chat-default', overrideModel: 'openai/gpt-4o-mini', plaintextKey: 'sk' }, + { kind: 'dashscope-cosyvoice', modelName: 'alibaba/cosyvoice-v2', region: 'intl', upstreamModel: 'cosyvoice-v2', plaintextKey: 'sk' }, + ], + }) + + expect(result.applied).toEqual([ + { kind: 'openrouter', target: 'llm-router', surface: 'llm', modelName: 'chat-default', keyEntryId: 'openrouter-prod-1' }, + { kind: 'dashscope-cosyvoice', target: 'llm-router', surface: 'tts', modelName: 'alibaba/cosyvoice-v2', keyEntryId: 'dashscope-tts-prod-1' }, + ]) + }) +})