diff --git a/apps/server/package.json b/apps/server/package.json index f35ee598e..ca1c1bab9 100644 --- a/apps/server/package.json +++ b/apps/server/package.json @@ -64,6 +64,7 @@ "posthog-node": "catalog:", "resend": "^6.12.2", "stripe": "^22.0.2", + "unspeech": "catalog:xsai", "valibot": "catalog:", "ws": "catalog:", "zod": "catalog:" @@ -72,7 +73,6 @@ "@better-auth/cli": "^1.4.21", "@types/pg": "^8.20.0", "@types/ws": "^8.18.1", - "drizzle-kit": "catalog:", - "unspeech": "catalog:xsai" + "drizzle-kit": "catalog:" } } diff --git a/apps/server/src/routes/admin/config/router/index.ts b/apps/server/src/routes/admin/config/router/index.ts index fa11e0b6c..a06cc7dda 100644 --- a/apps/server/src/routes/admin/config/router/index.ts +++ b/apps/server/src/routes/admin/config/router/index.ts @@ -54,6 +54,7 @@ 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)), + defaultVoice: optional(pipe(string(), nonEmpty('defaultVoice must not be empty'), maxLength(200))), plaintextKey: pipe(string(), nonEmpty('plaintextKey is required'), maxLength(MAX_KEY_LENGTH)), keyEntryId: optional(pipe(string(), nonEmpty(), maxLength(200), NO_PIPE)), }) diff --git a/apps/server/src/routes/openai/v1/speech.ts b/apps/server/src/routes/openai/v1/speech.ts index 11c4486e3..bd72855d2 100644 --- a/apps/server/src/routes/openai/v1/speech.ts +++ b/apps/server/src/routes/openai/v1/speech.ts @@ -3,150 +3,28 @@ import type { Context, Handler } from 'hono' import type { HonoEnv } from '../../../types/hono' import type { V1RouteDeps } from './types' -import { useLogger } from '@guiiai/logg' - -import { nanoid } from '../../../utils/id' -import { createOpenAiRouteBilling } from './billing' -import { buildSafeResponseHeaders } from './response' -import { createRouteTelemetry, newRouteContext } from './telemetry' +import { createOpenAiSpeechService } from '../../../services/domain/openai-speech' export function createSpeechHandler(deps: V1RouteDeps): Handler { - const logger = useLogger('v1-completions').useGlobalConfig() - const telemetry = createRouteTelemetry({ + const speechService = createOpenAiSpeechService({ + configKV: deps.configKV, + fluxService: deps.fluxService, genAi: deps.genAi, + llmRouter: deps.llmRouter, + llmTracing: deps.llmTracing, requestLogService: deps.requestLogService, + ttsMeter: deps.ttsMeter, }) - const billing = createOpenAiRouteBilling(deps) return async function handleTTS(c: Context) { const user = c.get('user')! - const requestId = nanoid() + const body = await c.req.json() as Record - const body = await c.req.json() - 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 = await deps.configKV.getOrThrow('DEFAULT_TTS_MODEL') - } - - logger.withFields({ - requestId, - userId: user.id, - model: requestModel, - inputChars: inputText.length, - voice: typeof body.voice === 'string' ? body.voice : undefined, - }).log('tts speech request') - - const billingAuthorization = await billing.authorizeTts(user.id, inputText) - - // 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 generationTrace = deps.llmTracing.startTtsGeneration({ - input: ttsInput, - model: requestModel, - requestId, + return speechService.handleSpeechRequest({ userId: user.id, + body, sessionId: c.req.header('x-airi-session-id'), - }) - - const span = telemetry.startTtsSpan({ model: requestModel }) - - const startedAt = Date.now() - - const routeCtx = newRouteContext() - let response: Response - try { - response = await telemetry.runWithSpan(span, () => - deps.llmRouter.routeTts({ modelName: requestModel, input: ttsInput, abortSignal: c.req.raw.signal }, routeCtx)) - } - catch (err) { - telemetry.failSpan(span, 'TTS router exhausted or unknown model') - generationTrace.fail('TTS router exhausted or unknown model') - telemetry.recordMetrics({ model: requestModel, status: 502, type: 'tts', provider: routeCtx.provider, durationMs: Date.now() - startedAt, fluxConsumed: 0 }) - throw err - } - - const durationMs = Date.now() - startedAt - telemetry.setHttpStatus(span, response.status) - - if (!response.ok) { - telemetry.failSpan(span, `Gateway ${response.status}`) - generationTrace.fail(`Gateway ${response.status}`) - telemetry.recordMetrics({ model: requestModel, status: response.status, type: 'tts', provider: routeCtx.provider, durationMs, fluxConsumed: 0 }) - logger.withFields({ requestId, userId: user.id, model: requestModel, status: response.status, durationMs }) - .warn('tts speech delivered with upstream error status') - return new Response(response.body, { - status: response.status, - headers: buildSafeResponseHeaders(response), - }) - } - - // 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. - // - // 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 billing.settleTts({ - userId: user.id, - inputText, - currentBalance: billingAuthorization.balance, - requestId, - model: requestModel, - }) - fluxConsumed = result.fluxDebited - telemetry.recordTtsBillingOnSpan(span, fluxConsumed) - generationTrace.succeed({ - inputChars: inputText.length, - fluxConsumed, - output: { contentType: response.headers.get('content-type') }, - }) - } - catch (err) { - generationTrace.fail('TTS billing failed') - throw err - } - finally { - telemetry.endSpan(span) - } - telemetry.recordMetrics({ model: requestModel, status: response.status, type: 'tts', provider: routeCtx.provider, durationMs, fluxConsumed }) - - telemetry.recordRequestLog({ - userId: user.id, - model: requestModel, - status: response.status, - durationMs, - fluxConsumed, - }) - - logger.withFields({ - requestId, - userId: user.id, - model: requestModel, - status: response.status, - durationMs, - inputChars: inputText.length, - fluxConsumed, - }).log('tts speech delivered') - - return new Response(response.body, { - status: response.status, - headers: buildSafeResponseHeaders(response), + abortSignal: c.req.raw.signal, }) } } diff --git a/apps/server/src/services/adapters/tts/azure.ts b/apps/server/src/services/adapters/tts/azure.ts index 2fe16b86c..add94bee3 100644 --- a/apps/server/src/services/adapters/tts/azure.ts +++ b/apps/server/src/services/adapters/tts/azure.ts @@ -2,133 +2,10 @@ import type { Voice } from 'unspeech' import type { TtsAdapter, TtsAdapterContext, TtsInput, TtsResult, TtsVoiceCatalogContext } from './types' -import { errorMessageFrom } from '@moeru/std' +import { buildMicrosoftSsml, inferMicrosoftContentType, isMicrosoftVoiceId, resolveMicrosoftOutputFormat } from 'unspeech' -import { createBadGatewayError, createBadRequestError, createInternalError, createServiceUnavailableError } from '../../../utils/error' - -// NOTICE: -// Voice IDs Azure accepts are stable strings like `en-US-AvaMultilingualNeural`. -// We allow the canonical Microsoft pattern only — letters/digits/hyphens. -// Without this guard a malicious `voice` field can break out of `name='...'` -// in the SSML envelope and inject arbitrary `` / `` elements, -// running under our Azure credential. -// Source: codex review 2026-05-15 HIGH #3. -const AZURE_VOICE_ID = /^[a-z0-9-]+$/i - -/** - * Default Azure voice when the caller doesn't pick one. Microsoft markets this - * as a general-purpose multilingual neural voice, which matches our hosted - * default behavior for unspecified voice. - * - * NOTICE: - * Hardcoded near use because there is no operator-tunable default voice yet; - * promote to configKV when ops need per-tenant defaults. - */ -const DEFAULT_AZURE_VOICE = 'en-US-AvaMultilingualNeural' - -/** - * Default Azure output format header value. - * - * 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' - -/** - * Resolves a caller's `responseFormat` to the Azure `X-Microsoft-OutputFormat` - * header value Azure expects. - * - * Before: - * - `"mp3"` - * - `"wav"` - * - `"audio-24khz-48kbitrate-mono-mp3"` (already an Azure format key) - * - * After: - * - `"audio-24khz-48kbitrate-mono-mp3"` - * - `"riff-24khz-16bit-mono-pcm"` - * - `"audio-24khz-48kbitrate-mono-mp3"` - */ -function resolveAzureFormat(responseFormat: string | undefined): string { - if (!responseFormat) - return DEFAULT_AZURE_FORMAT - // Caller already supplied an Azure-native format key; pass through. - if (responseFormat.includes('-')) - return responseFormat - if (responseFormat === 'mp3') - return 'audio-24khz-48kbitrate-mono-mp3' - if (responseFormat === 'wav') - return 'riff-24khz-16bit-mono-pcm' - if (responseFormat === 'opus') - return 'ogg-24khz-16bit-mono-opus' - // Unknown short codes: pass through verbatim — Azure will 400 if invalid and - // the router maps that error. - return responseFormat -} - -/** - * Normalizes a numeric speed multiplier into Azure's SSML `prosody rate` - * percent string. `1.0` returns empty (caller skips the `` wrapper). - * - * Before: - * - `1.0` - * - `1.2` - * - `0.8` - * - * After: - * - `""` - * - `"+20%"` - * - `"-20%"` - */ -function speedToProsodyRate(speed: number | undefined): string { - if (speed == null || speed === 1) - return '' - // Math: SSML accepts non-zero percentages relative to native rate; - // `(speed - 1) * 100` gives the delta. Sign prefix is required. - const delta = Math.round((speed - 1) * 100) - if (delta === 0) - return '' - return delta > 0 ? `+${delta}%` : `${delta}%` -} - -/** - * Minimal XML escape for text injected into SSML. Azure rejects malformed XML - * (raw `<`, `&`, etc.) — we escape only the five XML-mandated entities and - * leave the rest of the text intact. - */ -function escapeForSsml(text: string): string { - return text - .replaceAll('&', '&') - .replaceAll('<', '<') - .replaceAll('>', '>') - .replaceAll('"', '"') - .replaceAll('\'', ''') -} - -/** - * Builds an Azure-compatible SSML envelope for a `{text, voice, speed}` triple. - * - * Use when: - * - Wrapping a hosted user prompt before POSTing to Azure REST TTS. - * - * Expects: - * - `text` is plain text (already-built SSML must be sent via - * `extraOptions.disableSsml = true` and bypass this function). - * - * Returns: - * - A self-contained `` document string Azure accepts as - * `Content-Type: application/ssml+xml`. - */ -function buildAzureSsml(text: string, voice: string, speed: number | undefined): string { - const safe = escapeForSsml(text) - const rate = speedToProsodyRate(speed) - const inner = rate - ? `${safe}` - : safe - // xml:lang on is required by Azure; voice's own language wins for - // pronunciation, but the root attribute must still be present. - return `${inner}` -} +import { createBadRequestError, createInternalError, createServiceUnavailableError } from '../../../utils/error' +import { listVoicesViaUnSpeech, sendSpeechViaUnSpeech } from './unspeech' /** * Azure Cognitive Services REST adapter. @@ -151,69 +28,35 @@ export const azureAdapter: TtsAdapter = { id: 'azure', async send(input: TtsInput, ctx: TtsAdapterContext): Promise { - const voice = input.voice ?? DEFAULT_AZURE_VOICE - if (!AZURE_VOICE_ID.test(voice)) + const defaultVoice = typeof ctx.adapterParams.defaultVoice === 'string' + ? ctx.adapterParams.defaultVoice + : undefined + const voice = input.voice ?? defaultVoice + if (!voice) + throw createBadRequestError('azure voice is required when adapterParams.defaultVoice is not configured', 'BAD_REQUEST') + if (!isMicrosoftVoiceId(voice)) throw createBadRequestError(`azure voice id contains unsupported characters: ${voice}`, 'BAD_REQUEST', { voice }) - const outputFormat = resolveAzureFormat(input.responseFormat) + const outputFormat = resolveMicrosoftOutputFormat(input.responseFormat) const disableSsml = input.extraOptions?.disableSsml === true - // We build the SSML envelope on our side rather than letting unspeech do - // it, because (a) unspeech's `processSSML` does not honor `` - // so speed multipliers would be silently dropped, and (b) the SSML escape - // (`escapeForSsml`) is a security boundary — keeping it in this process - // means a misbehaving unspeech can never accidentally accept raw text and - // surface an injection. unspeech detects a pre-built `` document - // in the `input` field and forwards it verbatim. const ssml = disableSsml ? input.text - : buildAzureSsml(input.text, voice, input.speed) + : buildMicrosoftSsml(input.text, voice, input.speed) const region = ctx.adapterParams?.region if (typeof region !== 'string' || !region) throw createInternalError('azure tts upstream is missing adapterParams.region') - const body = JSON.stringify({ + return sendSpeechViaUnSpeech({ + ctx, model: 'microsoft/v1', input: ssml, voice, - response_format: outputFormat, - extra_body: { region }, + responseFormat: outputFormat, + extraBody: { region, disable_ssml: true }, + fallbackContentType: inferMicrosoftContentType(outputFormat), + providerLabel: 'azure', }) - - let response: Response - try { - response = await ctx.fetchImpl(`${ctx.unspeechBaseURL.replace(/\/+$/, '')}/v1/audio/speech`, { - method: 'POST', - headers: { - // unspeech's microsoft backend reads the bearer token verbatim as - // the Azure subscription key (unspeech/pkg/backend/microsoft/speech.go:264). - 'Authorization': `Bearer ${ctx.keyPlaintext.toString('utf8')}`, - 'Content-Type': 'application/json', - }, - body, - signal: ctx.abortSignal, - }) - } - catch (error) { - // Network-level failure (DNS, connection reset, abort). Re-throw so the - // router can decide to fall back or surface as 502/504. - throw createInternalError(`azure tts fetch failed: ${errorMessageFrom(error) ?? 'unknown'}`) - } - - if (!response.ok) { - // Bubble the upstream status. Router (U3) maps to fallback or 5xx. - // unspeech preserves the upstream Azure status code on its - // `apierrors.NewUpstreamError(status)` envelope. - const text = await response.text().catch(() => '') - const err = new Error(`azure tts upstream ${response.status}: ${text.slice(0, 256)}`) as Error & { status?: number } - err.status = response.status - throw err - } - - const arrayBuffer = await response.arrayBuffer() - const contentType = response.headers.get('content-type') ?? inferContentTypeFromAzureFormat(outputFormat) - - return { contentType, body: arrayBuffer } }, async getVoiceCatalog(ctx: TtsVoiceCatalogContext): Promise { @@ -228,51 +71,10 @@ export const azureAdapter: TtsAdapter = { if (!ctx.keyPlaintext) throw createServiceUnavailableError('azure tts key not configured', 'AZURE_TTS_NOT_CONFIGURED') - const url = `${ctx.unspeechBaseURL.replace(/\/+$/, '')}/api/voices?provider=microsoft®ion=${encodeURIComponent(ctx.region)}` - - let response: Response - try { - response = await ctx.fetchImpl(url, { - method: 'GET', - headers: { - // unspeech's microsoft backend reads the bearer token verbatim as - // the Azure subscription key (see unspeech/pkg/backend/microsoft/voices.go). - Authorization: `Bearer ${ctx.keyPlaintext.toString('utf8')}`, - Accept: 'application/json', - }, - signal: ctx.abortSignal, - }) - } - catch (error) { - throw createBadGatewayError(`azure voices fetch failed: ${errorMessageFrom(error) ?? 'unknown'}`) - } - - if (!response.ok) { - const text = await response.text().catch(() => '') - throw createBadGatewayError( - `azure voices upstream ${response.status}: ${text.slice(0, 256)}`, - { lastStatusCode: response.status }, - ) - } - - const data = await response.json() as { voices: Voice[] } - if (!Array.isArray(data.voices)) - throw createBadGatewayError('azure voices upstream missing voices[]') - - return data.voices + return listVoicesViaUnSpeech({ + ctx, + query: `provider=microsoft®ion=${encodeURIComponent(ctx.region)}`, + providerLabel: 'azure', + }) }, } - -/** - * Maps Azure's output format key to a MIME type for the gateway response when - * the upstream omits `content-type` (rare but defensive). - */ -function inferContentTypeFromAzureFormat(format: string): string { - if (format.includes('mp3')) - return 'audio/mpeg' - if (format.includes('opus')) - return 'audio/ogg' - if (format.includes('pcm') || format.startsWith('riff')) - return 'audio/wav' - return 'application/octet-stream' -} diff --git a/apps/server/src/services/adapters/tts/dashscope-cosyvoice.ts b/apps/server/src/services/adapters/tts/dashscope-cosyvoice.ts index 1300499d6..837455e8d 100644 --- a/apps/server/src/services/adapters/tts/dashscope-cosyvoice.ts +++ b/apps/server/src/services/adapters/tts/dashscope-cosyvoice.ts @@ -2,9 +2,8 @@ import type { Voice } from 'unspeech' import type { TtsAdapter, TtsAdapterContext, TtsInput, TtsResult, TtsVoiceCatalogContext } from './types' -import { errorMessageFrom } from '@moeru/std' - -import { createBadGatewayError, createBadRequestError, createInternalError } from '../../../utils/error' +import { createBadRequestError } from '../../../utils/error' +import { listVoicesViaUnSpeech, sendSpeechViaUnSpeech } from './unspeech' /** * Default cosyvoice audio format. Mirrors the OpenAI `mp3` default expected by @@ -58,42 +57,15 @@ export const dashscopeCosyvoiceAdapter: TtsAdapter = { const voice = input.voice const format = input.responseFormat ?? DEFAULT_COSYVOICE_FORMAT - const body = JSON.stringify({ + return sendSpeechViaUnSpeech({ + ctx, model: `alibaba/${model}`, input: input.text, voice, - response_format: format, + responseFormat: format, + fallbackContentType: formatToMime(format), + providerLabel: 'dashscope-cosyvoice', }) - - let response: Response - try { - response = await ctx.fetchImpl(`${ctx.unspeechBaseURL.replace(/\/+$/, '')}/v1/audio/speech`, { - method: 'POST', - headers: { - 'Authorization': `Bearer ${ctx.keyPlaintext.toString('utf8')}`, - 'Content-Type': 'application/json', - }, - body, - signal: ctx.abortSignal, - }) - } - catch (error) { - throw createInternalError(`dashscope-cosyvoice tts fetch failed: ${errorMessageFrom(error) ?? 'unknown'}`) - } - - if (!response.ok) { - const text = await response.text().catch(() => '') - const err = new Error(`dashscope-cosyvoice tts upstream ${response.status}: ${text.slice(0, 256)}`) as Error & { status?: number } - err.status = response.status - throw err - } - - // unspeech aggregates the WS binary frames and returns the audio buffer - // directly, so we no longer parse a JSON envelope or follow a signed URL. - const audioBytes = await response.arrayBuffer() - const contentType = response.headers.get('content-type') ?? formatToMime(format) - - return { contentType, body: audioBytes } }, async getVoiceCatalog(ctx: TtsVoiceCatalogContext): Promise { @@ -104,33 +76,11 @@ export const dashscopeCosyvoiceAdapter: TtsAdapter = { const params = new URLSearchParams({ provider: 'alibaba' }) if (typeof ctx.adapterParams.model === 'string') params.set('model', ctx.adapterParams.model) - const url = `${ctx.unspeechBaseURL.replace(/\/+$/, '')}/api/voices?${params.toString()}` - - let response: Response - try { - response = await ctx.fetchImpl(url, { - method: 'GET', - headers: { Accept: 'application/json' }, - signal: ctx.abortSignal, - }) - } - catch (error) { - throw createBadGatewayError(`cosyvoice voices fetch failed: ${errorMessageFrom(error) ?? 'unknown'}`) - } - - if (!response.ok) { - const text = await response.text().catch(() => '') - throw createBadGatewayError( - `cosyvoice voices upstream ${response.status}: ${text.slice(0, 256)}`, - { lastStatusCode: response.status }, - ) - } - - const data = await response.json() as { voices: Voice[] } - if (!Array.isArray(data.voices)) - throw createBadGatewayError('cosyvoice voices upstream missing voices[]') - - return data.voices + return listVoicesViaUnSpeech({ + ctx, + query: params.toString(), + providerLabel: 'cosyvoice', + }) }, } diff --git a/apps/server/src/services/adapters/tts/index.test.ts b/apps/server/src/services/adapters/tts/index.test.ts index 1503c215f..b71f960bc 100644 --- a/apps/server/src/services/adapters/tts/index.test.ts +++ b/apps/server/src/services/adapters/tts/index.test.ts @@ -217,6 +217,7 @@ describe('azureAdapter.send', () => { expect(body.model).toBe('microsoft/v1') expect(body.voice).toBe('en-US-AvaMultilingualNeural') expect((body.extra_body as { region?: string }).region).toBe('eastasia') + expect((body.extra_body as { disable_ssml?: boolean }).disable_ssml).toBe(true) // SSML is built on our side so speed survives — verify the prosody tag is in // the input field unspeech receives. expect(body.input).toContain('') @@ -225,6 +226,47 @@ describe('azureAdapter.send', () => { expect(headers.Authorization).toBe('Bearer azure-sub-key') }) + it('uses adapterParams.defaultVoice when the request omits voice', async () => { + const adapter = getAdapter('azure') + const fetchImpl = vi.fn(async () => new Response(new Uint8Array([1, 2, 3]), { + status: 200, + headers: { 'content-type': 'audio/mpeg' }, + })) as unknown as typeof fetch + + await adapter.send( + { text: 'hi there' }, + { + keyPlaintext: Buffer.from('azure-sub-key', 'utf8'), + baseURL: 'https://eastasia.tts.speech.microsoft.com/cognitiveservices/v1', + unspeechBaseURL: 'http://unspeech.local:5933', + adapterParams: { region: 'eastasia', defaultVoice: 'en-US-AvaMultilingualNeural' }, + fetchImpl, + }, + ) + + const [, init] = (fetchImpl as unknown as { mock: { calls: [string, RequestInit][] } }).mock.calls[0] + const body = JSON.parse(init.body as string) as Record + expect(body.voice).toBe('en-US-AvaMultilingualNeural') + }) + + it('rejects missing voice when adapterParams.defaultVoice is not configured', async () => { + const adapter = getAdapter('azure') + const fetchImpl = vi.fn() as unknown as typeof fetch + + await expect(adapter.send( + { text: 'hi' }, + { + keyPlaintext: Buffer.from('k', 'utf8'), + baseURL: 'https://eastasia.tts.speech.microsoft.com/cognitiveservices/v1', + unspeechBaseURL: 'http://unspeech.local:5933', + adapterParams: { region: 'eastasia' }, + fetchImpl, + }, + )).rejects.toMatchObject({ statusCode: 400 }) + + expect(fetchImpl).not.toHaveBeenCalled() + }) + it('throws Error with .status when unspeech non-2xx', async () => { const adapter = getAdapter('azure') const fetchImpl = vi.fn(async () => new Response('upstream rejected', { status: 401 })) as unknown as typeof fetch diff --git a/apps/server/src/services/adapters/tts/unspeech.ts b/apps/server/src/services/adapters/tts/unspeech.ts new file mode 100644 index 000000000..8b2c3cd46 --- /dev/null +++ b/apps/server/src/services/adapters/tts/unspeech.ts @@ -0,0 +1,118 @@ +import type { Voice } from 'unspeech' + +import type { TtsAdapterContext, TtsResult, TtsVoiceCatalogContext } from './types' + +import { errorMessageFrom } from '@moeru/std' +import { generateSpeechResponse, listVoices, UnSpeechAPIError } from 'unspeech' + +import { createBadGatewayError, createInternalError } from '../../../utils/error' + +interface SendSpeechOptions { + ctx: TtsAdapterContext + model: string + input: string + voice: string + responseFormat: string + extraBody?: Record + fallbackContentType: string + providerLabel: string +} + +/** + * Sends one OpenAI-shaped speech request through the unspeech SDK. + * + * Use when: + * - A TTS adapter has resolved AIRI's provider policy and needs to delegate the + * actual HTTP request to unspeech. + * + * Expects: + * - `model`, `voice`, `responseFormat`, and `extraBody` already match the + * provider-specific unspeech contract. + * + * Returns: + * - The binary audio payload plus a content type for the OpenAI route. + */ +export async function sendSpeechViaUnSpeech(options: SendSpeechOptions): Promise { + const { + ctx, + extraBody, + fallbackContentType, + input, + model, + providerLabel, + responseFormat, + voice, + } = options + + try { + const result = await generateSpeechResponse({ + apiKey: ctx.keyPlaintext.toString('utf8'), + baseURL: `${ctx.unspeechBaseURL.replace(/\/+$/, '')}/v1/`, + fetch: ctx.fetchImpl, + input, + model, + responseFormat, + voice, + abortSignal: ctx.abortSignal, + extraBody, + }) + + return { + contentType: result.contentType ?? fallbackContentType, + body: result.body, + } + } + catch (error) { + if (error instanceof UnSpeechAPIError) { + const err = new Error(`${providerLabel} tts upstream ${error.status}: ${error.responseBody.slice(0, 256)}`) as Error & { status?: number } + err.status = error.status + throw err + } + + throw createInternalError(`${providerLabel} tts fetch failed: ${errorMessageFrom(error) ?? 'unknown'}`) + } +} + +interface ListVoicesOptions { + ctx: TtsVoiceCatalogContext + query: string + providerLabel: string +} + +/** + * Lists unspeech voices and maps SDK failures into AIRI gateway errors. + * + * Use when: + * - A TTS adapter needs unspeech's normalized `Voice[]` catalog. + * + * Expects: + * - `query` is an unspeech `/api/voices` query string such as + * `provider=microsoft®ion=eastasia`. + * + * Returns: + * - The parsed voice catalog. + */ +export async function listVoicesViaUnSpeech(options: ListVoicesOptions): Promise { + const { ctx, providerLabel, query } = options + + try { + return await listVoices({ + apiKey: ctx.keyPlaintext?.toString('utf8'), + baseURL: ctx.unspeechBaseURL.replace(/\/+$/, ''), + fetch: ctx.fetchImpl, + query, + abortSignal: ctx.abortSignal, + headers: { Accept: 'application/json' }, + }) + } + catch (error) { + if (error instanceof UnSpeechAPIError) { + throw createBadGatewayError( + `${providerLabel} voices upstream ${error.status}: ${error.responseBody.slice(0, 256)}`, + { lastStatusCode: error.status }, + ) + } + + throw createBadGatewayError(`${providerLabel} voices fetch failed: ${errorMessageFrom(error) ?? 'unknown'}`) + } +} diff --git a/apps/server/src/services/adapters/tts/volcengine.ts b/apps/server/src/services/adapters/tts/volcengine.ts index e2083568f..24677f975 100644 --- a/apps/server/src/services/adapters/tts/volcengine.ts +++ b/apps/server/src/services/adapters/tts/volcengine.ts @@ -2,10 +2,9 @@ import type { Voice } from 'unspeech' import type { TtsAdapter, TtsAdapterContext, TtsInput, TtsResult, TtsVoiceCatalogContext } from './types' -import { errorMessageFrom } from '@moeru/std' - -import { createBadGatewayError, createInternalError } from '../../../utils/error' +import { createInternalError } from '../../../utils/error' import { nanoid } from '../../../utils/id' +import { listVoicesViaUnSpeech, sendSpeechViaUnSpeech } from './unspeech' /** * Default Volcengine TTS voice id. `BV001_streaming` is Volcengine's standard @@ -69,48 +68,21 @@ export const volcengineAdapter: TtsAdapter = { // - takes `app.appid`, `app.cluster`, `user.uid`, `request.reqid`, // `audio.encoding`, `audio.speed_ratio` from `extra_body` jsonpath. // - decodes the upstream base64 audio frame itself and returns binary. - const body = JSON.stringify({ + return sendSpeechViaUnSpeech({ + ctx, model: apiResourceId ? `volcengine/${apiResourceId}` : 'volcengine', input: input.text, voice, - response_format: encoding, - extra_body: { + responseFormat: encoding, + extraBody: { app: { appid, cluster }, user: { uid: 'airi-server' }, audio: { speed_ratio: speed }, request: { reqid: nanoid(), operation: 'query' }, }, + fallbackContentType: encodingToMime(encoding), + providerLabel: 'volcengine', }) - - let response: Response - try { - response = await ctx.fetchImpl(`${ctx.unspeechBaseURL.replace(/\/+$/, '')}/v1/audio/speech`, { - method: 'POST', - headers: { - 'Authorization': `Bearer ${ctx.keyPlaintext.toString('utf8')}`, - 'Content-Type': 'application/json', - }, - body, - signal: ctx.abortSignal, - }) - } - catch (error) { - throw createInternalError(`volcengine tts fetch failed: ${errorMessageFrom(error) ?? 'unknown'}`) - } - - if (!response.ok) { - const text = await response.text().catch(() => '') - const err = new Error(`volcengine tts upstream ${response.status}: ${text.slice(0, 256)}`) as Error & { status?: number } - err.status = response.status - throw err - } - - // unspeech decodes the base64 audio frame and returns binary audio - // directly — no more JSON envelope on this side. - const arrayBuffer = await response.arrayBuffer() - const contentType = response.headers.get('content-type') ?? encodingToMime(encoding) - - return { contentType, body: arrayBuffer } }, async getVoiceCatalog(ctx: TtsVoiceCatalogContext): Promise { @@ -119,39 +91,18 @@ export const volcengineAdapter: TtsAdapter = { // streaming-compatible voices. Passing `model=` narrows // further by `compatible_models` — adapterParams.model is the operator- // configured resource id (e.g. `seed-tts-2.0`). - const url = new URL(`${ctx.unspeechBaseURL.replace(/\/+$/, '')}/api/voices`) - url.searchParams.set('provider', 'volcengine') + const params = new URLSearchParams({ provider: 'volcengine' }) const apiResourceId = typeof ctx.adapterParams?.model === 'string' ? ctx.adapterParams.model : undefined if (apiResourceId) - url.searchParams.set('model', apiResourceId) + params.set('model', apiResourceId) - let response: Response - try { - response = await ctx.fetchImpl(url.toString(), { - method: 'GET', - headers: { Accept: 'application/json' }, - signal: ctx.abortSignal, - }) - } - catch (error) { - throw createBadGatewayError(`volcengine voices fetch failed: ${errorMessageFrom(error) ?? 'unknown'}`) - } - - if (!response.ok) { - const text = await response.text().catch(() => '') - throw createBadGatewayError( - `volcengine voices upstream ${response.status}: ${text.slice(0, 256)}`, - { lastStatusCode: response.status }, - ) - } - - const data = await response.json() as { voices: Voice[] } - if (!Array.isArray(data.voices)) - throw createBadGatewayError('volcengine voices upstream missing voices[]') - - return data.voices + return listVoicesViaUnSpeech({ + ctx, + query: params.toString(), + providerLabel: 'volcengine', + }) }, } diff --git a/apps/server/src/services/domain/admin/router-config/index.ts b/apps/server/src/services/domain/admin/router-config/index.ts index 2c31b93af..5021411ee 100644 --- a/apps/server/src/services/domain/admin/router-config/index.ts +++ b/apps/server/src/services/domain/admin/router-config/index.ts @@ -66,6 +66,8 @@ export interface AzureSliceInput { modelName: string /** Azure Speech region, used in baseURL and `adapterParams.region`. */ region: string + /** Default Microsoft voice used when `/audio/speech` omits `voice`. */ + defaultVoice?: string plaintextKey: string /** @default 'azure-tts-prod-1' */ keyEntryId?: string @@ -187,7 +189,10 @@ export function buildAzureSlice(input: AzureSliceInput, envelope: EnvelopeCrypto upstreams: [{ baseURL: `https://${input.region}.tts.speech.microsoft.com/cognitiveservices/v1`, keys: [{ id: keyEntryId, ciphertext }], - adapterParams: { region: input.region }, + adapterParams: { + region: input.region, + ...(input.defaultVoice ? { defaultVoice: input.defaultVoice } : {}), + }, }], } as TtsModel, } diff --git a/apps/server/src/services/domain/admin/router-config/tests/admin-router-config.test.ts b/apps/server/src/services/domain/admin/router-config/tests/admin-router-config.test.ts index 0a6ef6a99..595bd96bd 100644 --- a/apps/server/src/services/domain/admin/router-config/tests/admin-router-config.test.ts +++ b/apps/server/src/services/domain/admin/router-config/tests/admin-router-config.test.ts @@ -150,13 +150,17 @@ describe('buildAzureSlice', () => { kind: 'azure', modelName: 'microsoft/v1', region: 'eastasia', + defaultVoice: 'en-US-AvaMultilingualNeural', 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' }) + expect(built.model.upstreams[0].adapterParams).toEqual({ + region: 'eastasia', + defaultVoice: 'en-US-AvaMultilingualNeural', + }) const decrypted = envelope.decryptKey(built.model.upstreams[0].keys[0].ciphertext, { modelName: 'microsoft/v1', diff --git a/apps/server/src/services/domain/openai-speech/index.ts b/apps/server/src/services/domain/openai-speech/index.ts new file mode 100644 index 000000000..22259db0b --- /dev/null +++ b/apps/server/src/services/domain/openai-speech/index.ts @@ -0,0 +1,227 @@ +import type { GenAiMetrics } from '../../../otel' +import type { ConfigKVService } from '../../adapters/config-kv' +import type { FluxMeter } from '../billing/flux-meter' +import type { FluxService } from '../flux' +import type { LlmRouterService } from '../llm-router' +import type { startTtsGeneration, TtsGenerationTrace } from '../llm-tracing' +import type { RequestLogService } from '../request-log' + +import { useLogger } from '@guiiai/logg' +import { context, SpanStatusCode, trace } from '@opentelemetry/api' + +import { createPaymentRequiredError } from '../../../utils/error' +import { nanoid } from '../../../utils/id' +import { + AIRI_ATTR_BILLING_FLUX_CONSUMED, + AIRI_ATTR_GEN_AI_OPERATION_KIND, + GEN_AI_ATTR_REQUEST_MODEL, +} from '../../../utils/observability' + +const tracer = trace.getTracer('v1-completions') + +const SAFE_RESPONSE_HEADERS = new Set([ + 'content-type', + 'content-length', + 'transfer-encoding', + 'cache-control', +]) + +export interface OpenAiSpeechServiceDeps { + fluxService: FluxService + configKV: ConfigKVService + requestLogService: RequestLogService + ttsMeter: FluxMeter + llmRouter: LlmRouterService + genAi?: GenAiMetrics | null + llmTracing: { + startTtsGeneration: (input: Parameters[0]) => TtsGenerationTrace + } +} + +export interface OpenAiSpeechRequest { + userId: string + body: Record + sessionId?: string + abortSignal?: AbortSignal +} + +/** + * Runs the OpenAI-shaped text-to-speech gateway flow. + * + * Use when: + * - The HTTP route has parsed an authenticated `/audio/speech` request and + * needs domain orchestration for billing, routing, tracing, and logging. + * + * Expects: + * - `body` is the parsed JSON request body. + * - Auth and route guards have already run. + * + * Returns: + * - A gateway `Response` with safe upstream headers and audio body. + */ +export function createOpenAiSpeechService(deps: OpenAiSpeechServiceDeps) { + const logger = useLogger('v1-completions').useGlobalConfig() + + async function handleSpeechRequest(input: OpenAiSpeechRequest): Promise { + const requestId = nanoid() + let requestModel = typeof input.body.model === 'string' ? input.body.model : 'auto' + const inputText = typeof input.body.input === 'string' ? input.body.input : '' + + if (requestModel === 'auto') + requestModel = await deps.configKV.getOrThrow('DEFAULT_TTS_MODEL') + + logger.withFields({ + requestId, + userId: input.userId, + model: requestModel, + inputChars: inputText.length, + voice: typeof input.body.voice === 'string' ? input.body.voice : undefined, + }).log('tts speech request') + + const flux = await deps.fluxService.getFlux(input.userId) + if (flux.flux <= 0) + throw createPaymentRequiredError('Insufficient flux') + await deps.ttsMeter.assertCanAfford(input.userId, inputText.length, flux.flux) + + const ttsInput = { + text: inputText, + voice: typeof input.body.voice === 'string' ? input.body.voice : undefined, + speed: typeof input.body.speed === 'number' ? input.body.speed : undefined, + responseFormat: typeof input.body.response_format === 'string' ? input.body.response_format : undefined, + } + + const generationTrace = deps.llmTracing.startTtsGeneration({ + input: ttsInput, + model: requestModel, + requestId, + userId: input.userId, + sessionId: input.sessionId, + }) + + const span = tracer.startSpan('llm.gateway.tts', { + attributes: { + [GEN_AI_ATTR_REQUEST_MODEL]: requestModel, + [AIRI_ATTR_GEN_AI_OPERATION_KIND]: 'text_to_speech', + }, + }) + + const startedAt = Date.now() + const routeCtx = { provider: 'unknown', triedUpstreams: 0, triedKeys: 0, lastStatus: null } + let response: Response + try { + response = await context.with(trace.setSpan(context.active(), span), () => + deps.llmRouter.routeTts({ + modelName: requestModel, + input: ttsInput, + abortSignal: input.abortSignal, + }, routeCtx)) + } + catch (err) { + span.setStatus({ code: SpanStatusCode.ERROR, message: 'TTS router exhausted or unknown model' }) + span.end() + generationTrace.fail('TTS router exhausted or unknown model') + recordMetrics({ + durationMs: Date.now() - startedAt, + fluxConsumed: 0, + model: requestModel, + provider: routeCtx.provider, + status: 502, + }) + throw err + } + + const durationMs = Date.now() - startedAt + span.setAttribute('http.response.status_code', response.status) + + if (!response.ok) { + span.setStatus({ code: SpanStatusCode.ERROR, message: `Gateway ${response.status}` }) + span.end() + generationTrace.fail(`Gateway ${response.status}`) + recordMetrics({ model: requestModel, status: response.status, provider: routeCtx.provider, durationMs, fluxConsumed: 0 }) + logger.withFields({ requestId, userId: input.userId, model: requestModel, status: response.status, durationMs }) + .warn('tts speech delivered with upstream error status') + return new Response(response.body, { + status: response.status, + headers: buildSafeResponseHeaders(response), + }) + } + + let fluxConsumed = 0 + try { + const result = await deps.ttsMeter.accumulate({ + userId: input.userId, + units: inputText.length, + currentBalance: flux.flux, + requestId, + metadata: { model: requestModel }, + }) + fluxConsumed = result.fluxDebited + span.setAttribute(AIRI_ATTR_BILLING_FLUX_CONSUMED, fluxConsumed) + generationTrace.succeed({ + inputChars: inputText.length, + fluxConsumed, + output: { contentType: response.headers.get('content-type') }, + }) + } + catch (err) { + generationTrace.fail('TTS billing failed') + throw err + } + finally { + span.end() + } + + recordMetrics({ model: requestModel, status: response.status, provider: routeCtx.provider, durationMs, fluxConsumed }) + deps.requestLogService.logRequest({ + userId: input.userId, + model: requestModel, + status: response.status, + durationMs, + fluxConsumed, + }).catch(err => logger.withError(err).warn('Failed to write llm_request_log row')) + + logger.withFields({ + requestId, + userId: input.userId, + model: requestModel, + status: response.status, + durationMs, + inputChars: inputText.length, + fluxConsumed, + }).log('tts speech delivered') + + return new Response(response.body, { + status: response.status, + headers: buildSafeResponseHeaders(response), + }) + } + + function recordMetrics(input: { + model: string + status: number + provider: string + durationMs: number + fluxConsumed: number + }): void { + const attrs = { + [GEN_AI_ATTR_REQUEST_MODEL]: input.model, + [AIRI_ATTR_GEN_AI_OPERATION_KIND]: 'tts', + 'http.response.status_code': input.status, + 'provider': input.provider, + } + deps.genAi?.operationCount.add(1, attrs) + deps.genAi?.operationDuration.record(input.durationMs / 1000, attrs) + deps.genAi?.fluxConsumed.add(input.fluxConsumed, attrs) + } + + return { handleSpeechRequest } +} + +function buildSafeResponseHeaders(response: Response): Headers { + const headers = new Headers() + response.headers.forEach((value, key) => { + if (SAFE_RESPONSE_HEADERS.has(key.toLowerCase())) + headers.set(key, value) + }) + return headers +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 6f20d8e23..5717260dd 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -399,8 +399,8 @@ catalogs: version: 4.1.4 xsai: unspeech: - specifier: ^0.1.13 - version: 0.1.13 + specifier: ^0.1.14 + version: 0.1.14 overrides: array-flatten: npm:@nolyfill/array-flatten@^1.0.44 @@ -776,6 +776,9 @@ importers: stripe: specifier: ^22.0.2 version: 22.0.2(@types/node@25.6.0) + unspeech: + specifier: catalog:xsai + version: 0.1.14 valibot: specifier: 'catalog:' version: 1.2.0(typescript@5.9.3) @@ -798,9 +801,6 @@ importers: drizzle-kit: specifier: 'catalog:' version: 0.31.10 - unspeech: - specifier: catalog:xsai - version: 0.1.13 apps/stage-pocket: dependencies: @@ -1022,7 +1022,7 @@ importers: version: 11.0.5 unspeech: specifier: catalog:xsai - version: 0.1.13 + version: 0.1.14 uuid: specifier: ^13.0.0 version: 13.0.0 @@ -1458,7 +1458,7 @@ importers: version: 11.0.5 unspeech: specifier: catalog:xsai - version: 0.1.13 + version: 0.1.14 uqr: specifier: 'catalog:' version: 0.1.3 @@ -1915,7 +1915,7 @@ importers: version: 11.0.5 unspeech: specifier: catalog:xsai - version: 0.1.13 + version: 0.1.14 uuid: specifier: ^13.0.0 version: 13.0.0 @@ -3001,7 +3001,7 @@ importers: version: 2.9.6(vue@3.5.32(typescript@5.9.3)) unspeech: specifier: catalog:xsai - version: 0.1.13 + version: 0.1.14 vue: specifier: 'catalog:' version: 3.5.32(typescript@5.9.3) @@ -3128,7 +3128,7 @@ importers: version: 4.0.4(vue@3.5.32(typescript@5.9.3)) unspeech: specifier: catalog:xsai - version: 0.1.13 + version: 0.1.14 vue: specifier: 'catalog:' version: 3.5.32(typescript@5.9.3) @@ -3460,7 +3460,7 @@ importers: version: 5.1.0 unspeech: specifier: catalog:xsai - version: 0.1.13 + version: 0.1.14 unstorage: specifier: 'catalog:' version: 1.17.5(aws4fetch@1.0.20)(idb-keyval@6.2.2)(ioredis@5.10.1) @@ -17562,8 +17562,8 @@ packages: synckit: optional: true - unspeech@0.1.13: - resolution: {integrity: sha512-2xBvi5mbQBUHV8x1rHz8zCDNHQkfkTL6F7YuDUyyNWiM4hYJV4YIxbM+NjG5U3SVUKvkoHeCnzSTkX0nnPcTEQ==} + unspeech@0.1.14: + resolution: {integrity: sha512-+iXL6ZC4ZEIwcUM3JEjNRNVX6z/tIaJyPgMwSE2Y8ewyt7aej+PaqP1PZVaMMLOWKT5OEv4lLsiGDZgVLCxNzA==} unstorage@1.17.5: resolution: {integrity: sha512-0i3iqvRfx29hkNntHyQvJTpf5W9dQ9ZadSoRU8+xVlhVtT7jAX57fazYO9EHvcRCfBCyi5YRya7XCDOsbTgkPg==} @@ -24569,9 +24569,9 @@ snapshots: obug: 2.1.1 std-env: 4.1.0 tinyrainbow: 3.1.0 - vitest: 4.1.4(@opentelemetry/api@1.9.1)(@types/node@25.6.0)(@vitest/browser-playwright@4.1.4)(@vitest/coverage-v8@4.1.4)(jsdom@29.1.1(@noble/hashes@2.0.1)(canvas@3.2.3))(vite@8.0.8(@types/node@25.6.0)(esbuild@0.27.2)(jiti@2.6.1)(less@4.6.4)(terser@5.46.1)(tsx@4.21.0)(yaml@2.8.3)) + vitest: 4.1.4(@opentelemetry/api@1.9.1)(@types/node@24.12.2)(@vitest/browser-playwright@4.1.4)(@vitest/coverage-v8@4.1.4)(jsdom@29.1.1(@noble/hashes@2.0.1)(canvas@3.2.3))(vite@8.0.8(@types/node@24.12.2)(esbuild@0.27.2)(jiti@2.6.1)(less@4.6.4)(terser@5.46.1)(tsx@4.21.0)(yaml@2.8.3)) optionalDependencies: - '@vitest/browser': 4.1.4(bufferutil@4.1.0)(utf-8-validate@5.0.10)(vite@8.0.8(@types/node@25.6.0)(esbuild@0.27.2)(jiti@2.6.1)(less@4.6.4)(terser@5.46.1)(tsx@4.21.0)(yaml@2.8.3))(vitest@4.1.4) + '@vitest/browser': 4.1.4(bufferutil@4.1.0)(utf-8-validate@5.0.10)(vite@8.0.8(@types/node@24.12.2)(esbuild@0.27.2)(jiti@2.6.1)(less@4.6.4)(terser@5.46.1)(tsx@4.21.0)(yaml@2.8.3))(vitest@4.1.4) '@vitest/eslint-plugin@1.6.15(@typescript-eslint/eslint-plugin@8.58.1(@typescript-eslint/parser@8.58.1(eslint@10.2.1(jiti@2.6.1))(typescript@5.9.3))(eslint@10.2.1(jiti@2.6.1))(typescript@5.9.3))(eslint@10.2.1(jiti@2.6.1))(typescript@5.9.3)(vitest@4.1.4)': dependencies: @@ -33231,7 +33231,7 @@ snapshots: optionalDependencies: synckit: 0.11.12 - unspeech@0.1.13: + unspeech@0.1.14: dependencies: '@xsai-ext/providers': 0.4.4 '@xsai/shared': 0.4.4 diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index 04522bed2..43a4c5c97 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -163,7 +163,7 @@ catalogs: '@vitest/coverage-v8': ^4.1.4 vitest: ^4.1.4 xsai: - unspeech: ^0.1.13 + unspeech: ^0.1.14 ignoredBuiltDependencies: - '@ax-llm/ax'