refactor(api-server): harden streaming TTS providers

server/apps/api/src/routes/admin/config/router/index.ts

@@ -24,6 +24,7 @@ import {
 
 import { adminGuard } from '../../../../middlewares/admin-guard'
 import { authGuard } from '../../../../middlewares/auth'
+import { STEPFUN_STREAMING_TTS_MODEL_IDS } from '../../../../services/adapters/config-kv'
 import { createBadRequestError } from '../../../../utils/error'
 
 /**
@@ -106,18 +107,16 @@ const StepfunSliceSchema = object({
   existingKeyEntryId: optional(pipe(string(), nonEmpty(), maxLength(200), NO_PIPE)),
 })
 
-const STEPFUN_STREAMING_MODEL_ID = /^stepfun\/(?:stepaudio-2\.5-tts|step-tts-2|step-tts-mini)$/
-
 const StepfunStreamingSliceSchema = pipe(object({
   kind: literal('stepfun-streaming'),
-  enabled: boolean(),
+  rollout: picklist(['disabled', 'available', 'default']),
   upstreamURL: pipe(string(), regex(/^wss?:\/\/\S+$/, 'upstreamURL must start with ws:// or wss://'), maxLength(500)),
   models: pipe(array(object({
-    id: pipe(string(), regex(STEPFUN_STREAMING_MODEL_ID, 'models[].id must be a supported StepFun streaming model'), maxLength(200)),
+    id: picklist(STEPFUN_STREAMING_TTS_MODEL_IDS, 'models[].id must be a supported StepFun streaming model'),
     name: optional(pipe(string(), nonEmpty(), maxLength(200))),
     description: optional(pipe(string(), nonEmpty(), maxLength(500))),
   })), minLength(1, 'models must not be empty')),
-  defaultModel: pipe(string(), regex(STEPFUN_STREAMING_MODEL_ID, 'defaultModel must be a supported StepFun streaming model'), maxLength(200)),
+  defaultModel: picklist(STEPFUN_STREAMING_TTS_MODEL_IDS, 'defaultModel must be a supported StepFun streaming model'),
   voices: pipe(array(object({
     id: pipe(string(), nonEmpty('voices[].id is required'), maxLength(200)),
     name: optional(pipe(string(), nonEmpty(), maxLength(200))),
@@ -129,7 +128,7 @@ const StepfunStreamingSliceSchema = pipe(object({
   plaintextKey: optional(pipe(string(), nonEmpty('plaintextKey must not be empty when provided'), maxLength(MAX_KEY_LENGTH))),
   keyEntryId: optional(pipe(string(), nonEmpty(), maxLength(200), NO_PIPE)),
   existingKeyEntryId: optional(pipe(string(), nonEmpty(), maxLength(200), NO_PIPE)),
-}), check(config => new Set(config.models.map(model => model.id)).size === config.models.length, 'models[].id must be unique'), check(config => config.models.some(model => model.id === config.defaultModel), 'defaultModel must be present in models'))
+}), check(config => new Set(config.models.map(model => model.id)).size === config.models.length, 'models[].id must be unique'), check(config => config.models.some(model => model.id === config.defaultModel), 'defaultModel must be present in models'), check(config => new Set(config.voices.map(voice => voice.id)).size === config.voices.length, 'voices[].id must be unique'))
 
 const AliyunNlsAsrSliceSchema = object({
   kind: literal('aliyun-nls-asr'),
@@ -220,7 +219,8 @@ const BodySchema = object({
 /**
  * Admin route for seeding / patching the LLM router config tree. Mounted
  * at `POST /api/admin/config/router`; the only supported way to write
- * `LLM_ROUTER_CONFIG`, `UNSPEECH_UPSTREAM`, and the
+ * `LLM_ROUTER_CONFIG`, `UNSPEECH_UPSTREAM`,
+ * `STEPFUN_STREAMING_TTS_UPSTREAM`, and the
  * `DEFAULT_{CHAT,TTS}_MODEL` aliases.
  *
  * Body shape (discriminated on `slices[].kind`):
@@ -246,6 +246,11 @@ const BodySchema = object({
  *       { "kind": "stepfun", "modelName": "stepfun/stepaudio-2.5-tts",
  *         "upstreamModel": "stepaudio-2.5-tts",
  *         "defaultVoice": "cixingnansheng", "plaintextKey": "..." },
+ *       { "kind": "stepfun-streaming", "rollout": "available",
+ *         "upstreamURL": "wss://api.stepfun.com/v1/realtime/audio",
+ *         "models": [{ "id": "stepfun/step-tts-2" }],
+ *         "defaultModel": "stepfun/step-tts-2",
+ *         "voices": [{ "id": "lively-girl" }], "plaintextKey": "..." },
  *       { "kind": "aliyun-nls-asr", "modelName": "auto",
  *         "accessKeyId": "...", "appKey": "...", "plaintextKey": "..." },
  *       { "kind": "unspeech",

server/apps/api/src/routes/audio-speech-ws/provider.ts

@@ -0,0 +1,126 @@
+import type { ConfigKVService, StepfunStreamingTtsUpstream, UnspeechUpstream } from '../../services/adapters/config-kv'
+import type { StreamingTtsStartCommand } from './providers/types'
+
+import { ofetch } from 'ofetch'
+
+import { STEPFUN_STREAMING_TTS_KEY_CONTEXT } from '../../services/adapters/config-kv'
+import { isUnspeechStreamingModelEnabled, streamingTtsModelResourceId } from '../../services/domain/streaming-tts-policy'
+
+export interface ResolvedStreamingTtsProvider {
+  /** Provider selected by exact public model id. */
+  kind: 'unspeech' | 'stepfun'
+  /** Provider websocket URL, including provider-specific model parameters. */
+  upstreamURL: string
+  /** Ordered encrypted credentials available to this provider. */
+  keys: Array<{ id: string, ciphertext: string }>
+  /** Envelope encryption context that must match the configuration writer. */
+  keyContext: string
+  /** Operator-level StepFun instruction used when the client does not supply one. */
+  instruction?: string
+}
+
+/** A client-visible policy failure found while resolving a streaming provider. */
+export class StreamingTtsResolutionError extends Error {
+  constructor(
+    readonly code: string,
+    readonly closeCode: number,
+    options?: ErrorOptions,
+  ) {
+    super(code, options)
+    this.name = 'StreamingTtsResolutionError'
+  }
+}
+
+/**
+ * Resolves one client start command to exactly one configured provider.
+ *
+ * StepFun in `available` mode only handles its explicit model ids; unSpeech
+ * remains available for all of its configured models. `default` affects the
+ * catalog's selected model, not this deterministic model-to-provider mapping.
+ */
+export async function resolveStreamingTtsProvider(
+  start: StreamingTtsStartCommand,
+  configKV: ConfigKVService,
+): Promise<ResolvedStreamingTtsProvider> {
+  let stepfun: StepfunStreamingTtsUpstream | null
+  let unspeech: UnspeechUpstream | null
+  try {
+    const [loadedStepfun, loadedUnspeech] = await Promise.all([
+      configKV.getOptional('STEPFUN_STREAMING_TTS_UPSTREAM'),
+      configKV.getOptional('UNSPEECH_UPSTREAM'),
+    ])
+    stepfun = loadedStepfun ?? null
+    unspeech = loadedUnspeech ?? null
+  }
+  catch (error) {
+    throw new StreamingTtsResolutionError('config_unavailable', 1011, { cause: error })
+  }
+
+  if (stepfun && stepfun.rollout !== 'disabled' && stepfun.models.some(model => model.id === start.model)) {
+    if (!stepfun.voices.some(voice => voice.id === start.voice))
+      throw new StreamingTtsResolutionError('streaming_tts_voice_not_enabled', 1008)
+    if (!isStepfunResponseFormatSupported(start.responseFormat))
+      throw new StreamingTtsResolutionError('streaming_tts_response_format_not_supported', 1008)
+    return {
+      kind: 'stepfun',
+      upstreamURL: stepfunURL(stepfun.baseURL, start.model),
+      keys: stepfun.keys,
+      keyContext: STEPFUN_STREAMING_TTS_KEY_CONTEXT,
+      instruction: stepfun.instruction,
+    }
+  }
+
+  const streaming = unspeech?.streaming
+  if (!unspeech?.restBaseURL || !streaming?.baseURL || streaming.keys.length === 0) {
+    const hasAvailableStepfun = stepfun != null && stepfun.rollout !== 'disabled'
+    throw new StreamingTtsResolutionError(
+      hasAvailableStepfun ? 'streaming_tts_model_not_enabled' : 'streaming_tts_not_configured',
+      1008,
+    )
+  }
+  if (!isUnspeechStreamingModelEnabled(streaming.models ?? [], start.model))
+    throw new StreamingTtsResolutionError('streaming_tts_model_not_enabled', 1008)
+
+  const voicesURL = unspeechVoicesURL(unspeech.restBaseURL, streamingTtsModelResourceId(start.model))
+  let voices: unknown[]
+  try {
+    const data = await ofetch(voicesURL, { timeout: 5000 }) as { voices?: unknown[] }
+    voices = Array.isArray(data.voices) ? data.voices : []
+  }
+  catch (error) {
+    throw new StreamingTtsResolutionError('streaming_tts_voice_catalog_unavailable', 1011, { cause: error })
+  }
+  if (!voices.some(voice => voiceId(voice) === start.voice))
+    throw new StreamingTtsResolutionError('streaming_tts_voice_not_enabled', 1008)
+
+  return {
+    kind: 'unspeech',
+    upstreamURL: streaming.baseURL,
+    keys: streaming.keys,
+    keyContext: 'streaming-tts',
+  }
+}
+
+function isStepfunResponseFormatSupported(format: string | undefined): boolean {
+  return format == null || format === 'mp3' || format === 'opus' || format === 'flac'
+}
+
+function stepfunURL(baseURL: string, model: string): string {
+  const url = new URL(baseURL)
+  url.searchParams.set('model', streamingTtsModelResourceId(model))
+  return url.toString()
+}
+
+function unspeechVoicesURL(restBaseURL: string, resourceId: string): string {
+  const url = new URL(restBaseURL)
+  url.pathname = '/api/voices'
+  url.search = new URLSearchParams({ provider: 'volcengine', model: resourceId }).toString()
+  return url.toString()
+}
+
+function voiceId(voice: unknown): string | null {
+  if (typeof voice !== 'object' || voice == null)
+    return null
+  const id = (voice as { id?: unknown }).id
+  return typeof id === 'string' && id.length > 0 ? id : null
+}

server/apps/api/src/routes/audio-speech-ws/providers/stepfun.ts

@@ -0,0 +1,341 @@
+import type { RawData } from 'ws'
+
+import type { StreamingTtsCommand, StreamingTtsProviderEvent, StreamingTtsTransport, StreamingTtsTransportOptions } from './types'
+
+import { Buffer } from 'node:buffer'
+
+import WebSocket from 'ws'
+
+import { errorMessageFrom } from '@moeru/std'
+import { looseObject, optional, record, safeParse, string, unknown as unknownValue } from 'valibot'
+
+const StepfunServerEventSchema = looseObject({
+  type: string(),
+  data: optional(record(string(), unknownValue())),
+})
+
+const DEFAULT_HANDSHAKE_TIMEOUT_MS = 10000
+const DEFAULT_COMPLETION_TIMEOUT_MS = 30000
+
+/**
+ * `connecting -> creating -> ready -> finishing -> terminal` mirrors the
+ * StepFun handshake. Error, close, and abort may enter `terminal` from any
+ * phase; buffered AIRI commands are released only after `response.created`.
+ */
+type StepfunPhase = 'connecting' | 'creating' | 'ready' | 'finishing' | 'terminal'
+
+/**
+ * Creates a native StepFun streaming TTS transport.
+ *
+ * The adapter owns StepFun's ordered handshake, session correlation, JSON
+ * envelope, Base64 decoding, format selection, and command buffering. Callers
+ * only observe AIRI's provider-neutral streaming events.
+ */
+export function createStepfunTransport(options: StreamingTtsTransportOptions & { instruction?: string }): StreamingTtsTransport {
+  let phase: StepfunPhase = 'connecting'
+  let sessionId: string | null = null
+  const pendingCommands: StreamingTtsCommand[] = []
+  const handshakeTimeoutMs = options.timeouts?.handshakeMs ?? DEFAULT_HANDSHAKE_TIMEOUT_MS
+  const completionTimeoutMs = options.timeouts?.completionMs ?? DEFAULT_COMPLETION_TIMEOUT_MS
+  let handshakeTimer: ReturnType<typeof setTimeout> | undefined
+  let completionTimer: ReturnType<typeof setTimeout> | undefined
+
+  let ws: WebSocket
+  try {
+    ws = new WebSocket(options.upstreamURL, {
+      headers: { Authorization: `Bearer ${options.keyPlaintext.toString('utf8')}` },
+    })
+  }
+  finally {
+    options.keyPlaintext.fill(0)
+  }
+
+  function emit(event: StreamingTtsProviderEvent) {
+    options.onEvent(event)
+  }
+
+  function clearDeadlines() {
+    if (handshakeTimer)
+      clearTimeout(handshakeTimer)
+    if (completionTimer)
+      clearTimeout(completionTimer)
+    handshakeTimer = undefined
+    completionTimer = undefined
+  }
+
+  function refreshCompletionDeadline() {
+    if (completionTimer)
+      clearTimeout(completionTimer)
+    completionTimer = setTimeout(() => {
+      fail('stepfun_completion_timeout', 'StepFun did not complete the streaming TTS session in time')
+    }, completionTimeoutMs)
+  }
+
+  function fail(code: string, message: string) {
+    if (phase === 'terminal')
+      return
+    phase = 'terminal'
+    clearDeadlines()
+    pendingCommands.length = 0
+    emit({ type: 'failed', code, message })
+    try {
+      ws.terminate()
+    }
+    catch {}
+  }
+
+  function sendJson(value: Record<string, unknown>): boolean {
+    try {
+      ws.send(JSON.stringify(value))
+      return true
+    }
+    catch (error) {
+      fail('stepfun_send_failed', errorMessageFrom(error) ?? 'StepFun websocket send failed')
+      return false
+    }
+  }
+
+  function sendCommand(command: StreamingTtsCommand) {
+    if (!sessionId || (phase !== 'ready' && phase !== 'finishing')) {
+      pendingCommands.push(command)
+      return
+    }
+
+    if (command.type === 'text') {
+      for (const text of splitText(command.text)) {
+        if (!sendJson({ type: 'tts.text.delta', data: { session_id: sessionId, text } }))
+          return
+        emit({ type: 'input-accepted', chars: text.length })
+      }
+      return
+    }
+
+    if (sendJson({ type: 'tts.text.done', data: { session_id: sessionId } })) {
+      phase = 'finishing'
+      refreshCompletionDeadline()
+    }
+  }
+
+  function flushPendingCommands() {
+    const commands = pendingCommands.splice(0)
+    for (const command of commands) {
+      if (phase === 'terminal')
+        return
+      sendCommand(command)
+    }
+  }
+
+  function requireCurrentSession(data: Record<string, unknown> | undefined): boolean {
+    const eventSessionId = stringField(data, 'session_id')
+    if (sessionId && eventSessionId === sessionId)
+      return true
+    fail('stepfun_session_mismatch', 'StepFun returned an event for an unexpected session')
+    return false
+  }
+
+  function handleMessage(data: RawData, isBinary: boolean) {
+    if (phase === 'terminal')
+      return
+    if (isBinary) {
+      fail('stepfun_invalid_binary_event', 'StepFun unexpectedly returned a binary websocket frame')
+      return
+    }
+
+    const parsed = safeParse(StepfunServerEventSchema, JSON.parse(bufferToString(data)))
+    if (!parsed.success) {
+      fail('stepfun_invalid_event', 'StepFun returned a malformed websocket event')
+      return
+    }
+
+    const event = parsed.output
+    switch (event.type) {
+      case 'tts.connection.done': {
+        if (phase !== 'connecting') {
+          fail('stepfun_invalid_transition', `Unexpected ${event.type} while ${phase}`)
+          return
+        }
+        sessionId = stringField(event.data, 'session_id') ?? null
+        if (!sessionId) {
+          fail('stepfun_invalid_event', 'StepFun connection event omitted session_id')
+          return
+        }
+        phase = 'creating'
+        sendJson(createFrame(sessionId, options))
+        return
+      }
+      case 'tts.response.created': {
+        if (phase !== 'creating' || !requireCurrentSession(event.data))
+          return
+        phase = 'ready'
+        if (handshakeTimer)
+          clearTimeout(handshakeTimer)
+        handshakeTimer = undefined
+        emit({ type: 'started' })
+        flushPendingCommands()
+        return
+      }
+      case 'tts.response.sentence.start':
+      case 'tts.response.sentence.end':
+      case 'tts.response.subtitle': {
+        if ((phase !== 'ready' && phase !== 'finishing') || !requireCurrentSession(event.data))
+          return
+        if (phase === 'finishing')
+          refreshCompletionDeadline()
+        const controlEvent = event.type.replace('tts.response.', '') as 'sentence.start' | 'sentence.end' | 'subtitle'
+        emit({ type: 'control', event: controlEvent, payload: event.data ?? {} })
+        return
+      }
+      case 'tts.response.audio.delta': {
+        if ((phase !== 'ready' && phase !== 'finishing') || !requireCurrentSession(event.data))
+          return
+        if (phase === 'finishing')
+          refreshCompletionDeadline()
+        const audio = stringField(event.data, 'audio')
+        if (!audio) {
+          fail('stepfun_invalid_audio_event', 'StepFun audio delta omitted audio')
+          return
+        }
+        const bytes = Buffer.from(audio, 'base64')
+        if (bytes.byteLength === 0) {
+          fail('stepfun_invalid_audio_event', 'StepFun audio delta contained invalid Base64')
+          return
+        }
+        emit({ type: 'audio', data: bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.byteLength) as ArrayBuffer })
+        return
+      }
+      case 'tts.response.audio.done': {
+        if ((phase !== 'ready' && phase !== 'finishing') || !requireCurrentSession(event.data))
+          return
+        phase = 'terminal'
+        clearDeadlines()
+        emit({ type: 'completed' })
+        return
+      }
+      case 'tts.response.error': {
+        const eventSessionId = stringField(event.data, 'session_id')
+        if (eventSessionId && sessionId && eventSessionId !== sessionId) {
+          fail('stepfun_session_mismatch', 'StepFun returned an error for an unexpected session')
+          return
+        }
+        fail(
+          stringField(event.data, 'code') ?? 'stepfun_upstream_error',
+          stringField(event.data, 'message') ?? 'StepFun streaming TTS failed',
+        )
+        return
+      }
+      default:
+        fail('stepfun_unknown_event', `Unsupported StepFun event: ${event.type}`)
+    }
+  }
+
+  ws.on('message', (data, isBinary) => {
+    try {
+      handleMessage(data, isBinary)
+    }
+    catch (error) {
+      fail('stepfun_invalid_event', errorMessageFrom(error) ?? 'StepFun returned an invalid event')
+    }
+  })
+  ws.on('error', error => fail('stepfun_upstream_error', error.message))
+  ws.on('close', (code, reason) => {
+    if (phase === 'terminal')
+      return
+    phase = 'terminal'
+    clearDeadlines()
+    pendingCommands.length = 0
+    emit({ type: 'closed', code, reason: reason.toString() || 'stepfun_upstream_closed' })
+  })
+
+  handshakeTimer = setTimeout(() => {
+    fail('stepfun_handshake_timeout', 'StepFun did not create the streaming TTS session in time')
+  }, handshakeTimeoutMs)
+
+  return {
+    kind: 'stepfun',
+    upstreamURL: options.upstreamURL,
+    keyEntryId: options.keyEntryId,
+    send(command) {
+      if (phase === 'terminal')
+        return
+      sendCommand(command)
+    },
+    abort() {
+      phase = 'terminal'
+      clearDeadlines()
+      pendingCommands.length = 0
+      try {
+        ws.terminate()
+      }
+      catch {}
+    },
+  }
+}
+
+function createFrame(sessionId: string, options: StreamingTtsTransportOptions & { instruction?: string }): Record<string, unknown> {
+  const extraBody = options.start.extraBody ?? {}
+  const sampleRate = numberField(extraBody, 'sample_rate')
+    ?? numberField(recordField(extraBody, 'audio'), 'sample_rate')
+  const speedRatio = numberField(extraBody, 'speed_ratio')
+  const volumeRatio = numberField(extraBody, 'volume_ratio')
+  const instruction = stringField(extraBody, 'instruction') ?? options.instruction
+
+  return {
+    type: 'tts.create',
+    data: {
+      session_id: sessionId,
+      voice_id: options.start.voice,
+      response_format: streamingFormat(options.start.responseFormat),
+      text_normalization: 'standard',
+      mode: 'default',
+      ...(sampleRate !== undefined ? { sample_rate: sampleRate } : {}),
+      ...(speedRatio !== undefined ? { speed_ratio: speedRatio } : {}),
+      ...(volumeRatio !== undefined ? { volume_ratio: volumeRatio } : {}),
+      ...(instruction ? { instruction } : {}),
+    },
+  }
+}
+
+function streamingFormat(value: string | undefined): 'mp3_stream' | 'opus_stream' | 'flac_stream' {
+  switch (value) {
+    case 'opus':
+      return 'opus_stream'
+    case 'flac':
+      return 'flac_stream'
+    default:
+      return 'mp3_stream'
+  }
+}
+
+/** Splits at Unicode code-point boundaries because StepFun caps one delta at 1000 characters. */
+function splitText(text: string): string[] {
+  const characters = Array.from(text)
+  const chunks: string[] = []
+  for (let start = 0; start < characters.length; start += 1000)
+    chunks.push(characters.slice(start, start + 1000).join(''))
+  return chunks
+}
+
+function bufferToString(data: RawData): string {
+  if (Array.isArray(data))
+    return Buffer.concat(data).toString('utf8')
+  if (data instanceof ArrayBuffer)
+    return Buffer.from(data).toString('utf8')
+  return data.toString('utf8')
+}
+
+function stringField(record: Record<string, unknown> | undefined, key: string): string | undefined {
+  const value = record?.[key]
+  return typeof value === 'string' && value.length > 0 ? value : undefined
+}
+
+function numberField(record: Record<string, unknown> | undefined, key: string): number | undefined {
+  const value = record?.[key]
+  return typeof value === 'number' && Number.isFinite(value) ? value : undefined
+}
+
+function recordField(record: Record<string, unknown> | undefined, key: string): Record<string, unknown> | undefined {
+  const value = record?.[key]
+  return typeof value === 'object' && value != null && !Array.isArray(value)
+    ? value as Record<string, unknown>
+    : undefined
+}

server/apps/api/src/routes/audio-speech-ws/providers/types.ts

@@ -0,0 +1,67 @@
+import type { Buffer } from 'node:buffer'
+
+export interface StreamingTtsStartCommand {
+  /** Canonical public model id used for provider resolution and billing. */
+  model: string
+  /** Provider voice id already selected by the client. */
+  voice: string
+  /** Client audio container request; providers map it to their wire format. */
+  responseFormat?: string
+  /** Provider-neutral optional controls forwarded through the selected adapter. */
+  extraBody?: Record<string, unknown>
+}
+
+export type StreamingTtsCommand
+  = | { type: 'text', text: string }
+    | { type: 'finish' }
+
+export type StreamingTtsProviderEvent
+  = | { type: 'started' }
+    | { type: 'input-accepted', chars: number }
+    | { type: 'audio', data: ArrayBuffer }
+    | { type: 'control', event: 'sentence.start' | 'sentence.end' | 'subtitle', payload: Record<string, unknown> }
+    | { type: 'completed', usageChars?: number }
+    | { type: 'failed', code: string, message: string }
+    | { type: 'closed', code: number, reason: string }
+
+/**
+ * Hides an upstream provider's websocket protocol behind AIRI's streaming TTS
+ * lifecycle. Commands may be submitted while connecting and are delivered in
+ * order once the provider session is ready.
+ */
+export interface StreamingTtsTransport {
+  /** Resolved provider identity used by telemetry. */
+  readonly kind: 'unspeech' | 'stepfun'
+  /** Exact upstream endpoint used for this connection. */
+  readonly upstreamURL: string
+  /** Encrypted key entry identifier used for observability, never the secret. */
+  readonly keyEntryId: string
+  /** Queues or sends one ordered client command until the provider is ready. */
+  send: (command: StreamingTtsCommand) => void
+  /** Immediately stops generation and releases the upstream websocket. */
+  abort: () => void
+}
+
+/** Inputs shared by provider-specific streaming websocket adapters. */
+export interface StreamingTtsTransportOptions {
+  /** Validated client start command. */
+  start: StreamingTtsStartCommand
+  /** Provider endpoint selected by configuration policy. */
+  upstreamURL: string
+  /** Credential identifier recorded in traces. */
+  keyEntryId: string
+  /** Decrypted credential; the adapter must zero it after constructing the websocket. */
+  keyPlaintext: Buffer
+  /** Synchronous provider-neutral event sink owned by the client session. */
+  onEvent: (event: StreamingTtsProviderEvent) => void
+  /** Optional runtime deadlines; adapters apply production defaults when omitted. */
+  timeouts?: Partial<StreamingTtsTransportTimeouts>
+}
+
+/** Deadlines that prevent stalled upstream sessions from retaining resources. */
+export interface StreamingTtsTransportTimeouts {
+  /** Maximum milliseconds from dialing until the provider acknowledges the session. */
+  handshakeMs: number
+  /** Maximum milliseconds without provider progress after finish is sent. */
+  completionMs: number
+}

server/apps/api/src/routes/audio-speech-ws/providers/unspeech.ts

@@ -0,0 +1,213 @@
+import type { RawData } from 'ws'
+
+import type { StreamingTtsCommand, StreamingTtsProviderEvent, StreamingTtsTransport, StreamingTtsTransportOptions } from './types'
+
+import WebSocket from 'ws'
+
+import { errorMessageFrom } from '@moeru/std'
+
+import { bufferToString, readUsageChars, toBufferLike } from '../protocol'
+
+const DEFAULT_HANDSHAKE_TIMEOUT_MS = 10000
+const DEFAULT_COMPLETION_TIMEOUT_MS = 30000
+
+/**
+ * unSpeech becomes `ready` when the websocket opens, then remains ready until
+ * finish, completion, failure, close, or abort makes the transport terminal.
+ */
+type UnspeechPhase = 'connecting' | 'ready' | 'finishing' | 'terminal'
+
+/** Wraps unSpeech's AIRI-compatible websocket protocol as a normalized transport. */
+export function createUnspeechTransport(options: StreamingTtsTransportOptions): StreamingTtsTransport {
+  let phase: UnspeechPhase = 'connecting'
+  const pendingCommands: StreamingTtsCommand[] = []
+  const handshakeTimeoutMs = options.timeouts?.handshakeMs ?? DEFAULT_HANDSHAKE_TIMEOUT_MS
+  const completionTimeoutMs = options.timeouts?.completionMs ?? DEFAULT_COMPLETION_TIMEOUT_MS
+  let handshakeTimer: ReturnType<typeof setTimeout> | undefined
+  let completionTimer: ReturnType<typeof setTimeout> | undefined
+
+  let ws: WebSocket
+  try {
+    ws = new WebSocket(options.upstreamURL, {
+      headers: { Authorization: `Bearer ${options.keyPlaintext.toString('utf8')}` },
+    })
+  }
+  finally {
+    options.keyPlaintext.fill(0)
+  }
+
+  function emit(event: StreamingTtsProviderEvent) {
+    options.onEvent(event)
+  }
+
+  function clearDeadlines() {
+    if (handshakeTimer)
+      clearTimeout(handshakeTimer)
+    if (completionTimer)
+      clearTimeout(completionTimer)
+    handshakeTimer = undefined
+    completionTimer = undefined
+  }
+
+  function refreshCompletionDeadline() {
+    if (completionTimer)
+      clearTimeout(completionTimer)
+    completionTimer = setTimeout(() => {
+      fail('unspeech_completion_timeout', 'unSpeech did not complete the streaming TTS session in time')
+    }, completionTimeoutMs)
+  }
+
+  function fail(code: string, message: string) {
+    if (phase === 'terminal')
+      return
+    phase = 'terminal'
+    clearDeadlines()
+    pendingCommands.length = 0
+    emit({ type: 'failed', code, message })
+    try {
+      ws.terminate()
+    }
+    catch {}
+  }
+
+  function sendJson(value: Record<string, unknown>): boolean {
+    try {
+      ws.send(JSON.stringify(value))
+      return true
+    }
+    catch (error) {
+      fail('unspeech_send_failed', errorMessageFrom(error) ?? 'unSpeech websocket send failed')
+      return false
+    }
+  }
+
+  function sendCommand(command: StreamingTtsCommand) {
+    if (phase === 'connecting') {
+      pendingCommands.push(command)
+      return
+    }
+    if (phase === 'terminal')
+      return
+
+    if (command.type === 'text') {
+      if (sendJson({ event: 'text', text: command.text }))
+        emit({ type: 'input-accepted', chars: command.text.length })
+      return
+    }
+
+    if (sendJson({ event: 'finish' })) {
+      phase = 'finishing'
+      refreshCompletionDeadline()
+    }
+  }
+
+  function handleControl(data: RawData) {
+    let event: { event?: unknown, payload?: unknown }
+    try {
+      event = JSON.parse(bufferToString(data)) as { event?: unknown, payload?: unknown }
+    }
+    catch {
+      fail('unspeech_invalid_event', 'unSpeech returned malformed JSON')
+      return
+    }
+
+    const payload = isRecord(event.payload) ? event.payload : undefined
+    if (phase === 'finishing' && event.event !== 'session.finished' && event.event !== 'error')
+      refreshCompletionDeadline()
+    switch (event.event) {
+      case 'session.started':
+        if (handshakeTimer)
+          clearTimeout(handshakeTimer)
+        handshakeTimer = undefined
+        emit({ type: 'started' })
+        return
+      case 'sentence.start':
+      case 'sentence.end':
+      case 'subtitle':
+        emit({ type: 'control', event: event.event, payload: payload ?? {} })
+        return
+      case 'session.finished':
+        phase = 'terminal'
+        clearDeadlines()
+        emit({ type: 'completed', usageChars: readUsageChars(payload) ?? undefined })
+        return
+      case 'error':
+        fail(
+          typeof payload?.code === 'string' ? payload.code : 'unspeech_upstream_error',
+          typeof payload?.message === 'string' ? payload.message : 'unSpeech streaming TTS failed',
+        )
+    }
+  }
+
+  ws.on('open', () => {
+    if (phase !== 'connecting')
+      return
+    phase = 'ready'
+    if (!sendJson({
+      event: 'start',
+      model: options.start.model,
+      voice: options.start.voice,
+      ...(options.start.responseFormat ? { response_format: options.start.responseFormat } : {}),
+      ...(options.start.extraBody ? { extra_body: options.start.extraBody } : {}),
+    })) {
+      return
+    }
+
+    const commands = pendingCommands.splice(0)
+    for (const command of commands)
+      sendCommand(command)
+  })
+  ws.on('message', (data, isBinary) => {
+    if (phase === 'terminal')
+      return
+    if (isBinary) {
+      if (phase === 'finishing')
+        refreshCompletionDeadline()
+      emit({ type: 'audio', data: toBufferLike(data) })
+      return
+    }
+    handleControl(data)
+  })
+  ws.on('error', error => fail('unspeech_upstream_error', error.message))
+  ws.on('close', (code, reason) => {
+    if (phase === 'terminal')
+      return
+    phase = 'terminal'
+    clearDeadlines()
+    pendingCommands.length = 0
+    emit({ type: 'closed', code, reason: reason.toString() || 'unspeech_upstream_closed' })
+  })
+
+  handshakeTimer = setTimeout(() => {
+    fail('unspeech_handshake_timeout', 'unSpeech did not start the streaming TTS session in time')
+  }, handshakeTimeoutMs)
+
+  return {
+    kind: 'unspeech',
+    upstreamURL: options.upstreamURL,
+    keyEntryId: options.keyEntryId,
+    send(command) {
+      sendCommand(command)
+    },
+    abort() {
+      const canCancel = phase === 'ready' || phase === 'finishing'
+      phase = 'terminal'
+      clearDeadlines()
+      pendingCommands.length = 0
+      if (canCancel) {
+        try {
+          ws.send(JSON.stringify({ event: 'cancel' }))
+        }
+        catch {}
+      }
+      try {
+        ws.terminate()
+      }
+      catch {}
+    },
+  }
+}
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === 'object' && value != null && !Array.isArray(value)
+}

server/apps/api/src/routes/audio-speech-ws/route.test.ts

@@ -2,6 +2,8 @@ import type { AddressInfo } from 'node:net'
 
 import type { WSContext, WSEvents } from 'hono/ws'
 
+import type { AudioSpeechWsHandlersOptions } from './types'
+
 import { Buffer } from 'node:buffer'
 import { createServer } from 'node:http'
 
@@ -22,6 +24,7 @@ interface MockUpstream {
   receivedFrames: Array<{ kind: 'text' | 'binary', data: string | Buffer }>
   /** Auth header observed during handshake. */
   observedAuth: string | undefined
+  disconnectedClients: number
   close: () => Promise<void>
 }
 
@@ -29,9 +32,16 @@ async function startMockUpstream(
   scriptedResponses: MockUpstream['scriptedResponses'],
   voices: Array<{ id: string, name?: string }> = [{ id: 'mock', name: 'Mock Voice' }],
   protocol: 'unspeech' | 'stepfun' = 'unspeech',
+  options: {
+    stepfunCreatedDelayMs?: number
+    suppressStepfunConnectionDone?: boolean
+    suppressStepfunCreated?: boolean
+    scriptedResponseDelayMs?: number
+  } = {},
 ): Promise<MockUpstream> {
   const receivedFrames: MockUpstream['receivedFrames'] = []
   let observedAuth: string | undefined
+  let disconnectedClients = 0
 
   const httpServer = createServer((req, res) => {
     if (req.url?.startsWith('/api/voices')) {
@@ -46,13 +56,16 @@ async function startMockUpstream(
 
   wss.on('connection', (ws, req) => {
     observedAuth = req.headers.authorization
-    if (protocol === 'stepfun') {
+    if (protocol === 'stepfun' && !options.suppressStepfunConnectionDone) {
       ws.send(JSON.stringify({
         type: 'tts.connection.done',
         data: { session_id: 'stepfun-session' },
       }))
     }
     let replayed = false
+    ws.on('close', () => {
+      disconnectedClients += 1
+    })
     ws.on('message', async (data, isBinary) => {
       const buf = Buffer.isBuffer(data) ? data : Buffer.from(data as ArrayBuffer)
       const decoded = isBinary ? buf : buf.toString('utf8')
@@ -74,7 +87,12 @@ async function startMockUpstream(
         if (!isBinary) {
           const event = JSON.parse(decoded as string) as { type?: string }
           if (event.type === 'tts.create') {
-            ws.send(JSON.stringify({ type: 'tts.response.created', data: { session_id: 'stepfun-session' } }))
+            if (options.suppressStepfunCreated)
+              return
+            if (options.stepfunCreatedDelayMs)
+              await new Promise(resolve => setTimeout(resolve, options.stepfunCreatedDelayMs))
+            if (ws.readyState === 1)
+              ws.send(JSON.stringify({ type: 'tts.response.created', data: { session_id: 'stepfun-session' } }))
             return
           }
           if (event.type === 'tts.text.done')
@@ -104,7 +122,7 @@ async function startMockUpstream(
 
       replayed = true
       for (const resp of scriptedResponses) {
-        await new Promise(resolve => setTimeout(resolve, 5))
+        await new Promise(resolve => setTimeout(resolve, options.scriptedResponseDelayMs ?? 5))
 
         if (resp.kind === 'json')
           ws.send(JSON.stringify(resp.payload), { binary: false })
@@ -128,6 +146,9 @@ async function startMockUpstream(
     get observedAuth() {
       return observedAuth
     },
+    get disconnectedClients() {
+      return disconnectedClients
+    },
     async close() {
       wss.close()
       await new Promise<void>(resolve => httpServer.close(() => resolve()))
@@ -164,7 +185,7 @@ function makeMockClientWs(): MockClientWs {
     },
     readyState: 1,
     binaryType: 'arraybuffer',
-    raw: {} as any,
+    raw: {},
     protocol: '',
     url: null,
   } as unknown as WSContext
@@ -185,19 +206,20 @@ function makeFakeDeps(overrides: {
   decryptedKey?: string
   streamingModels?: Array<{ id: string, name?: string, description?: string }>
   stepfunStreaming?: {
-    enabled: boolean
+    rollout: 'disabled' | 'available' | 'default'
     baseURL: string
     models: Array<{ id: string, name?: string, description?: string }>
     defaultModel: string
     voices: Array<{ id: string, name?: string }>
   }
+  streamingTtsTimeouts?: { handshakeMs?: number, completionMs?: number }
 }) {
   const ttsMeter = {
     assertCanAfford: vi.fn(async (_userId: string, _newUnits: number, currentBalance: number) => {
       if (currentBalance <= 0)
         throw Object.assign(new Error('Insufficient flux'), { statusCode: 402 })
     }),
-    accumulate: vi.fn(async () => ({
+    accumulate: vi.fn(async (_input: Parameters<AudioSpeechWsHandlersOptions['ttsMeter']['accumulate']>[0]) => ({
       fluxDebited: 1,
       debtAfter: 0,
       balanceAfter: overrides.fluxBalance - 1,
@@ -208,7 +230,7 @@ function makeFakeDeps(overrides: {
     getFlux: vi.fn(async () => ({ flux: overrides.fluxBalance })),
   }
   const requestLogService = {
-    logRequest: vi.fn(async () => undefined),
+    logRequest: vi.fn(async (_input: Parameters<AudioSpeechWsHandlersOptions['requestLogService']['logRequest']>[0]) => undefined),
   }
   const productEventService = {
     track: vi.fn(async () => undefined),
@@ -248,7 +270,21 @@ function makeFakeDeps(overrides: {
     decryptKey: vi.fn(() => Buffer.from(overrides.decryptedKey ?? 'mock-upstream-token', 'utf8')),
   }
 
-  return { configKV, envelopeCrypto, fluxService, ttsMeter, requestLogService, productEventService }
+  return {
+    configKV,
+    envelopeCrypto,
+    fluxService,
+    ttsMeter,
+    requestLogService,
+    productEventService,
+    streamingTtsTimeouts: overrides.streamingTtsTimeouts,
+  }
+}
+
+function createTestHandlers(deps: ReturnType<typeof makeFakeDeps>) {
+  // The fixture intentionally implements only the service methods exercised by
+  // this route. Keep the partial-service adaptation at this single test boundary.
+  return createAudioSpeechWsHandlers(deps as unknown as AudioSpeechWsHandlersOptions)
 }
 
 /** Drives the WSEvents lifecycle as if a real client had connected. */
@@ -256,13 +292,13 @@ async function driveClientSession(events: WSEvents, client: MockClientWs, client
   // onOpen handles the initial dial. The route fires `void dialUpstream()`
   // which is async, so we await a microtask tick to let the upstream
   // dialing kick off.
-  events.onOpen?.(new Event('open') as any, client.ctx)
+  events.onOpen?.(new Event('open'), client.ctx)
   await new Promise(r => setTimeout(r, 50))
 
   for (const frame of clientFrames) {
     const isBinary = Buffer.isBuffer(frame)
     const data = isBinary ? frame : String(frame)
-    events.onMessage?.({ data } as any, client.ctx)
+    events.onMessage?.(new MessageEvent('message', { data }), client.ctx)
     await new Promise(r => setTimeout(r, 20))
   }
 }
@@ -285,7 +321,7 @@ describe('audio-speech-ws route', () => {
     ])
 
     const deps = makeFakeDeps({ upstreamURL: upstream.url, restBaseURL: upstream.restBaseURL, fluxBalance: 100 })
-    const handlers = createAudioSpeechWsHandlers(deps as any)
+    const handlers = createTestHandlers(deps)
     const events = handlers('user-123', { voiceType: 'official_selected' })
     const client = makeMockClientWs()
 
@@ -320,7 +356,7 @@ describe('audio-speech-ws route', () => {
     // sniff-from-text-frame fallback (which would be the input string
     // length of "hello streaming tts" = 19).
     expect(deps.ttsMeter.accumulate).toHaveBeenCalledTimes(1)
-    expect((deps.ttsMeter.accumulate.mock.calls[0] as any[])[0]).toMatchObject({
+    expect(deps.ttsMeter.accumulate.mock.calls[0]?.[0]).toMatchObject({
       userId: 'user-123',
       units: 42,
       metadata: { model: 'volcengine/seed-tts-2.0' },
@@ -329,7 +365,7 @@ describe('audio-speech-ws route', () => {
     // Request log gets the model label from the start frame, not the
     // hardcoded fallback.
     expect(deps.requestLogService.logRequest).toHaveBeenCalledTimes(1)
-    expect((deps.requestLogService.logRequest.mock.calls[0] as any[])[0]).toMatchObject({
+    expect(deps.requestLogService.logRequest.mock.calls[0]?.[0]).toMatchObject({
       userId: 'user-123',
       model: 'volcengine/seed-tts-2.0',
       status: 200,
@@ -363,14 +399,14 @@ describe('audio-speech-ws route', () => {
       restBaseURL: upstream.restBaseURL,
       fluxBalance: 100,
       stepfunStreaming: {
-        enabled: true,
+        rollout: 'default',
         baseURL: upstream.url,
         models: [{ id: 'stepfun/step-tts-2', name: 'Step TTS 2' }],
         defaultModel: 'stepfun/step-tts-2',
         voices: [{ id: 'lively-girl', name: 'Lively Girl' }],
       },
     })
-    const handlers = createAudioSpeechWsHandlers(deps as any)
+    const handlers = createTestHandlers(deps)
     const events = handlers('user-stepfun')
     const client = makeMockClientWs()
 
@@ -397,40 +433,426 @@ describe('audio-speech-ws route', () => {
     expect(deps.ttsMeter.accumulate).toHaveBeenCalledWith(expect.objectContaining({ userId: 'user-stepfun', units: 2001 }))
   })
 
-  it('settles StepFun text usage when the client cancels before audio completion', async () => {
+  it('closes StepFun immediately and records cancellation separately from billing', async () => {
     upstream = await startMockUpstream([], [{ id: 'lively-girl', name: 'Lively Girl' }], 'stepfun')
     const deps = makeFakeDeps({
       upstreamURL: upstream.url,
       restBaseURL: upstream.restBaseURL,
       fluxBalance: 100,
       stepfunStreaming: {
-        enabled: true,
+        rollout: 'default',
         baseURL: upstream.url,
         models: [{ id: 'stepfun/step-tts-2', name: 'Step TTS 2' }],
         defaultModel: 'stepfun/step-tts-2',
         voices: [{ id: 'lively-girl', name: 'Lively Girl' }],
       },
     })
-    const events = createAudioSpeechWsHandlers(deps as any)('user-stepfun-cancel')
+    const events = createTestHandlers(deps)('user-stepfun-cancel')
     const client = makeMockClientWs()
+    let releaseBilling: (() => void) | undefined
+    deps.ttsMeter.accumulate.mockImplementation(async () => new Promise((resolve) => {
+      releaseBilling = () => resolve({
+        fluxDebited: 1,
+        debtAfter: 0,
+        balanceAfter: 99,
+        unbilledFlux: 0,
+      })
+    }))
 
     await driveClientSession(events, client, [
       JSON.stringify({ event: 'start', model: 'stepfun/step-tts-2', voice: 'lively-girl' }),
       JSON.stringify({ event: 'text', text: 'paid text' }),
     ])
-    events.onClose?.(new Event('close') as any, client.ctx)
-    await new Promise(r => setTimeout(r, 100))
+    events.onClose?.(new CloseEvent('close'), client.ctx)
+    await new Promise(r => setTimeout(r, 30))
 
+    expect(upstream.disconnectedClients).toBe(1)
     expect(deps.ttsMeter.accumulate).toHaveBeenCalledWith(expect.objectContaining({
       userId: 'user-stepfun-cancel',
       units: 9,
     }))
+    expect(deps.requestLogService.logRequest).not.toHaveBeenCalled()
+
+    releaseBilling?.()
+    await new Promise(r => setTimeout(r, 30))
+    expect(deps.requestLogService.logRequest).toHaveBeenCalledWith(expect.objectContaining({ status: 499 }))
+    expect(deps.productEventService.track).toHaveBeenCalledWith(expect.objectContaining({
+      action: 'speech_cancelled',
+      status: 'cancelled',
+      reason: 'client_disconnected',
+    }))
+    expect(deps.productEventService.track).not.toHaveBeenCalledWith(expect.objectContaining({ action: 'speech_succeeded' }))
+  })
+
+  it('cancels immediately while start configuration is still pending', async () => {
+    upstream = await startMockUpstream([], [{ id: 'lively-girl' }], 'stepfun')
+    const deps = makeFakeDeps({
+      upstreamURL: upstream.url,
+      fluxBalance: 100,
+      stepfunStreaming: {
+        rollout: 'available',
+        baseURL: upstream.url,
+        models: [{ id: 'stepfun/step-tts-2' }],
+        defaultModel: 'stepfun/step-tts-2',
+        voices: [{ id: 'lively-girl' }],
+      },
+    })
+    const readConfig = deps.configKV.getOptional.getMockImplementation()
+    let releaseConfig: (() => void) | undefined
+    const configGate = new Promise<void>((resolve) => {
+      releaseConfig = resolve
+    })
+    deps.configKV.getOptional.mockImplementation(async (key) => {
+      await configGate
+      return readConfig?.(key) ?? null
+    })
+    const events = createTestHandlers(deps)('user-stepfun-explicit-cancel')
+    const client = makeMockClientWs()
+
+    events.onOpen?.(new Event('open'), client.ctx)
+    events.onMessage?.(new MessageEvent('message', {
+      data: JSON.stringify({ event: 'start', model: 'stepfun/step-tts-2', voice: 'lively-girl' }),
+    }), client.ctx)
+    await new Promise(resolve => setTimeout(resolve, 10))
+    events.onMessage?.(new MessageEvent('message', {
+      data: JSON.stringify({ event: 'cancel' }),
+    }), client.ctx)
+    await new Promise(resolve => setTimeout(resolve, 30))
+
+    expect(client.closed).toBe(true)
+    expect(upstream.observedAuth).toBeUndefined()
+    expect(deps.requestLogService.logRequest).toHaveBeenCalledWith(expect.objectContaining({ status: 499 }))
+    expect(deps.productEventService.track).toHaveBeenCalledWith(expect.objectContaining({
+      action: 'speech_cancelled',
+      status: 'cancelled',
+      reason: 'client_cancelled',
+    }))
+
+    releaseConfig?.()
+    await new Promise(resolve => setTimeout(resolve, 30))
+    expect(upstream.observedAuth).toBeUndefined()
+  })
+
+  it('fails and releases a StepFun connection that never completes its handshake', async () => {
+    upstream = await startMockUpstream([], [{ id: 'lively-girl' }], 'stepfun', { suppressStepfunConnectionDone: true })
+    const deps = makeFakeDeps({
+      upstreamURL: upstream.url,
+      fluxBalance: 100,
+      streamingTtsTimeouts: { handshakeMs: 25, completionMs: 100 },
+      stepfunStreaming: {
+        rollout: 'available',
+        baseURL: upstream.url,
+        models: [{ id: 'stepfun/step-tts-2' }],
+        defaultModel: 'stepfun/step-tts-2',
+        voices: [{ id: 'lively-girl' }],
+      },
+    })
+    const events = createTestHandlers(deps)('user-stepfun-timeout')
+    const client = makeMockClientWs()
+
+    await driveClientSession(events, client, [
+      JSON.stringify({ event: 'start', model: 'stepfun/step-tts-2', voice: 'lively-girl' }),
+    ])
+    await new Promise(resolve => setTimeout(resolve, 50))
+
+    expect(client.sent.filter(frame => frame.kind === 'text').map(frame => JSON.parse(frame.data as string))).toContainEqual(
+      expect.objectContaining({ event: 'error', code: 'stepfun_handshake_timeout' }),
+    )
+    expect(upstream.disconnectedClients).toBe(1)
+    expect(deps.requestLogService.logRequest).toHaveBeenCalledWith(expect.objectContaining({ status: 502 }))
+  })
+
+  it('fails and releases an unSpeech connection that never completes after finish', async () => {
+    upstream = await startMockUpstream([
+      { kind: 'json', payload: { event: 'session.started' } },
+    ])
+    const deps = makeFakeDeps({
+      upstreamURL: upstream.url,
+      restBaseURL: upstream.restBaseURL,
+      fluxBalance: 100,
+      streamingTtsTimeouts: { handshakeMs: 100, completionMs: 25 },
+    })
+    const events = createTestHandlers(deps)('user-unspeech-timeout')
+    const client = makeMockClientWs()
+
+    await driveClientSession(events, client, [
+      JSON.stringify({ event: 'start', model: 'volcengine/seed-tts-2.0', voice: 'mock' }),
+      JSON.stringify({ event: 'text', text: 'timeout input' }),
+      JSON.stringify({ event: 'finish' }),
+    ])
+    await new Promise(resolve => setTimeout(resolve, 80))
+
+    expect(client.sent.filter(frame => frame.kind === 'text').map(frame => JSON.parse(frame.data as string))).toContainEqual(
+      expect.objectContaining({ event: 'error', code: 'unspeech_completion_timeout' }),
+    )
+    expect(upstream.disconnectedClients).toBe(1)
+    expect(deps.ttsMeter.accumulate).toHaveBeenCalledWith(expect.objectContaining({ units: 13 }))
+  })
+
+  it('keeps a long StepFun completion alive while audio progress continues', async () => {
+    const audio = Buffer.from('progress').toString('base64')
+    upstream = await startMockUpstream([
+      { kind: 'json', payload: { type: 'tts.response.audio.delta', data: { session_id: 'stepfun-session', audio } } },
+      { kind: 'json', payload: { type: 'tts.response.audio.delta', data: { session_id: 'stepfun-session', audio } } },
+      { kind: 'json', payload: { type: 'tts.response.audio.delta', data: { session_id: 'stepfun-session', audio } } },
+      { kind: 'json', payload: { type: 'tts.response.audio.done', data: { session_id: 'stepfun-session' } } },
+    ], [{ id: 'lively-girl' }], 'stepfun', { scriptedResponseDelayMs: 15 })
+    const deps = makeFakeDeps({
+      upstreamURL: upstream.url,
+      fluxBalance: 100,
+      streamingTtsTimeouts: { handshakeMs: 100, completionMs: 25 },
+      stepfunStreaming: {
+        rollout: 'available',
+        baseURL: upstream.url,
+        models: [{ id: 'stepfun/step-tts-2' }],
+        defaultModel: 'stepfun/step-tts-2',
+        voices: [{ id: 'lively-girl' }],
+      },
+    })
+    const events = createTestHandlers(deps)('user-stepfun-progress')
+    const client = makeMockClientWs()
+
+    await driveClientSession(events, client, [
+      JSON.stringify({ event: 'start', model: 'stepfun/step-tts-2', voice: 'lively-girl' }),
+      JSON.stringify({ event: 'text', text: 'long healthy output' }),
+      JSON.stringify({ event: 'finish' }),
+    ])
+    await new Promise(resolve => setTimeout(resolve, 100))
+
+    expect(client.sent.filter(frame => frame.kind === 'binary')).toHaveLength(3)
+    expect(deps.requestLogService.logRequest).toHaveBeenCalledWith(expect.objectContaining({ status: 200 }))
+    expect(deps.productEventService.track).not.toHaveBeenCalledWith(expect.objectContaining({ reason: 'stepfun_completion_timeout' }))
+  })
+
+  it('never bills less than text accepted by unSpeech when upstream usage under-reports', async () => {
+    upstream = await startMockUpstream([
+      { kind: 'json', payload: { event: 'session.started' } },
+      { kind: 'json', payload: { event: 'session.finished', payload: { usage: { text_words: 0 } } } },
+    ])
+    const deps = makeFakeDeps({ upstreamURL: upstream.url, restBaseURL: upstream.restBaseURL, fluxBalance: 100 })
+    const events = createTestHandlers(deps)('user-unspeech-under-report')
+    const client = makeMockClientWs()
+
+    await driveClientSession(events, client, [
+      JSON.stringify({ event: 'start', model: 'volcengine/seed-tts-2.0', voice: 'mock' }),
+      JSON.stringify({ event: 'text', text: 'accepted text' }),
+      JSON.stringify({ event: 'finish' }),
+    ])
+    await new Promise(resolve => setTimeout(resolve, 80))
+
+    expect(deps.ttsMeter.accumulate).toHaveBeenCalledWith(expect.objectContaining({ units: 13 }))
+  })
+
+  it('allows unSpeech to validate the model when the curated model list is empty', async () => {
+    upstream = await startMockUpstream([
+      { kind: 'json', payload: { event: 'session.started' } },
+      { kind: 'json', payload: { event: 'session.finished', payload: {} } },
+    ])
+    const deps = makeFakeDeps({
+      upstreamURL: upstream.url,
+      restBaseURL: upstream.restBaseURL,
+      fluxBalance: 100,
+      streamingModels: [],
+    })
+    const events = createTestHandlers(deps)('user-unspeech-upstream-policy')
+    const client = makeMockClientWs()
+
+    await driveClientSession(events, client, [
+      JSON.stringify({ event: 'start', model: 'volcengine/upstream-model', voice: 'mock' }),
+      JSON.stringify({ event: 'text', text: 'hello' }),
+      JSON.stringify({ event: 'finish' }),
+    ])
+    await new Promise(resolve => setTimeout(resolve, 80))
+
+    expect(upstream.observedAuth).toBe('Bearer mock-upstream-token')
+    expect(deps.requestLogService.logRequest).toHaveBeenCalledWith(expect.objectContaining({ status: 200 }))
+  })
+
+  it('does not bill text that never reached a created StepFun session', async () => {
+    upstream = await startMockUpstream([], [{ id: 'lively-girl' }], 'stepfun', { stepfunCreatedDelayMs: 200 })
+    const deps = makeFakeDeps({
+      upstreamURL: upstream.url,
+      fluxBalance: 100,
+      stepfunStreaming: {
+        rollout: 'available',
+        baseURL: upstream.url,
+        models: [{ id: 'stepfun/step-tts-2' }],
+        defaultModel: 'stepfun/step-tts-2',
+        voices: [{ id: 'lively-girl' }],
+      },
+    })
+    const events = createTestHandlers(deps)('user-stepfun-unaccepted')
+    const client = makeMockClientWs()
+
+    await driveClientSession(events, client, [
+      JSON.stringify({ event: 'start', model: 'stepfun/step-tts-2', voice: 'lively-girl' }),
+      JSON.stringify({ event: 'text', text: 'not accepted' }),
+    ])
+    events.onClose?.(new CloseEvent('close'), client.ctx)
+    await new Promise(r => setTimeout(r, 30))
+
+    expect(deps.ttsMeter.accumulate).not.toHaveBeenCalled()
+    expect(upstream.receivedFrames.map(frame => JSON.parse(frame.data as string).type)).toEqual(['tts.create'])
+  })
+
+  it('records a StepFun response error as failure while charging only accepted text', async () => {
+    upstream = await startMockUpstream([
+      { kind: 'json', payload: { type: 'tts.response.error', data: { session_id: 'stepfun-session', code: 'provider_busy', message: 'busy' } } },
+    ], [{ id: 'lively-girl' }], 'stepfun')
+    const deps = makeFakeDeps({
+      upstreamURL: upstream.url,
+      fluxBalance: 100,
+      stepfunStreaming: {
+        rollout: 'available',
+        baseURL: upstream.url,
+        models: [{ id: 'stepfun/step-tts-2' }],
+        defaultModel: 'stepfun/step-tts-2',
+        voices: [{ id: 'lively-girl' }],
+      },
+    })
+    const events = createTestHandlers(deps)('user-stepfun-error')
+    const client = makeMockClientWs()
+
+    await driveClientSession(events, client, [
+      JSON.stringify({ event: 'start', model: 'stepfun/step-tts-2', voice: 'lively-girl' }),
+      JSON.stringify({ event: 'text', text: 'bill accepted text' }),
+      JSON.stringify({ event: 'finish' }),
+    ])
+    await new Promise(r => setTimeout(r, 80))
+
+    expect(deps.ttsMeter.accumulate).toHaveBeenCalledWith(expect.objectContaining({ units: 18 }))
+    expect(deps.requestLogService.logRequest).toHaveBeenCalledWith(expect.objectContaining({ status: 502 }))
+    expect(deps.productEventService.track).toHaveBeenCalledWith(expect.objectContaining({ action: 'speech_failed', reason: 'provider_busy' }))
+    expect(deps.productEventService.track).not.toHaveBeenCalledWith(expect.objectContaining({ action: 'speech_succeeded' }))
+  })
+
+  it('rejects StepFun events that do not belong to the active session', async () => {
+    upstream = await startMockUpstream([
+      {
+        kind: 'json',
+        payload: {
+          type: 'tts.response.audio.delta',
+          data: { session_id: 'another-session', audio: Buffer.from('wrong-session').toString('base64') },
+        },
+      },
+    ], [{ id: 'lively-girl' }], 'stepfun')
+    const deps = makeFakeDeps({
+      upstreamURL: upstream.url,
+      fluxBalance: 100,
+      stepfunStreaming: {
+        rollout: 'available',
+        baseURL: upstream.url,
+        models: [{ id: 'stepfun/step-tts-2' }],
+        defaultModel: 'stepfun/step-tts-2',
+        voices: [{ id: 'lively-girl' }],
+      },
+    })
+    const events = createTestHandlers(deps)('user-stepfun-correlation')
+    const client = makeMockClientWs()
+
+    await driveClientSession(events, client, [
+      JSON.stringify({ event: 'start', model: 'stepfun/step-tts-2', voice: 'lively-girl' }),
+      JSON.stringify({ event: 'text', text: 'hello' }),
+      JSON.stringify({ event: 'finish' }),
+    ])
+    await new Promise(r => setTimeout(r, 80))
+
+    const controlFrames = client.sent.filter(frame => frame.kind === 'text').map(frame => JSON.parse(frame.data as string))
+    expect(controlFrames).toContainEqual(expect.objectContaining({ event: 'error', code: 'stepfun_session_mismatch' }))
+    expect(client.sent.filter(frame => frame.kind === 'binary')).toHaveLength(0)
+    expect(deps.requestLogService.logRequest).toHaveBeenCalledWith(expect.objectContaining({ status: 502 }))
+  })
+
+  it('rejects an oversized streaming session before forwarding text upstream', async () => {
+    upstream = await startMockUpstream([], [{ id: 'lively-girl' }], 'stepfun')
+    const deps = makeFakeDeps({
+      upstreamURL: upstream.url,
+      fluxBalance: 100,
+      stepfunStreaming: {
+        rollout: 'available',
+        baseURL: upstream.url,
+        models: [{ id: 'stepfun/step-tts-2' }],
+        defaultModel: 'stepfun/step-tts-2',
+        voices: [{ id: 'lively-girl' }],
+      },
+    })
+    const events = createTestHandlers(deps)('user-stepfun-limit')
+    const client = makeMockClientWs()
+
+    await driveClientSession(events, client, [
+      JSON.stringify({ event: 'start', model: 'stepfun/step-tts-2', voice: 'lively-girl' }),
+      JSON.stringify({ event: 'text', text: 'x'.repeat(20001) }),
+    ])
+    await new Promise(r => setTimeout(r, 30))
+
+    expect(client.closeCode).toBe(1009)
+    expect(upstream.receivedFrames.map(frame => JSON.parse(frame.data as string).type)).toEqual(['tts.create'])
+    expect(deps.ttsMeter.accumulate).not.toHaveBeenCalled()
+  })
+
+  it('rejects a StepFun response format that cannot preserve the client contract', async () => {
+    upstream = await startMockUpstream([], [{ id: 'lively-girl' }], 'stepfun')
+    const deps = makeFakeDeps({
+      upstreamURL: upstream.url,
+      fluxBalance: 100,
+      stepfunStreaming: {
+        rollout: 'available',
+        baseURL: upstream.url,
+        models: [{ id: 'stepfun/step-tts-2' }],
+        defaultModel: 'stepfun/step-tts-2',
+        voices: [{ id: 'lively-girl' }],
+      },
+    })
+    const events = createTestHandlers(deps)('user-stepfun-format')
+    const client = makeMockClientWs()
+
+    await driveClientSession(events, client, [
+      JSON.stringify({ event: 'start', model: 'stepfun/step-tts-2', voice: 'lively-girl', response_format: 'aac' }),
+    ])
+
+    expect(upstream.observedAuth).toBeUndefined()
+    expect(client.sent.filter(frame => frame.kind === 'text').map(frame => JSON.parse(frame.data as string))).toContainEqual(
+      expect.objectContaining({ event: 'error', code: 'streaming_tts_response_format_not_supported' }),
+    )
+  })
+
+  it('bills accepted text when a later affordability window is blocked', async () => {
+    upstream = await startMockUpstream([], [{ id: 'lively-girl' }], 'stepfun')
+    const deps = makeFakeDeps({
+      upstreamURL: upstream.url,
+      fluxBalance: 100,
+      stepfunStreaming: {
+        rollout: 'available',
+        baseURL: upstream.url,
+        models: [{ id: 'stepfun/step-tts-2' }],
+        defaultModel: 'stepfun/step-tts-2',
+        voices: [{ id: 'lively-girl' }],
+      },
+    })
+    deps.ttsMeter.assertCanAfford.mockImplementation(async (_userId, units) => {
+      if (units > 2000)
+        throw Object.assign(new Error('Insufficient flux'), { statusCode: 402 })
+    })
+    const events = createTestHandlers(deps)('user-stepfun-window')
+    const client = makeMockClientWs()
+
+    await driveClientSession(events, client, [
+      JSON.stringify({ event: 'start', model: 'stepfun/step-tts-2', voice: 'lively-girl' }),
+      JSON.stringify({ event: 'text', text: 'x'.repeat(2000) }),
+      JSON.stringify({ event: 'text', text: 'x' }),
+    ])
+    await new Promise(r => setTimeout(r, 50))
+
+    expect(client.closeCode).toBe(1008)
+    expect(deps.ttsMeter.accumulate).toHaveBeenCalledWith(expect.objectContaining({ units: 2000 }))
+    expect(deps.requestLogService.logRequest).toHaveBeenCalledWith(expect.objectContaining({ status: 402 }))
+    expect(deps.productEventService.track).toHaveBeenCalledWith(expect.objectContaining({ action: 'speech_blocked' }))
   })
 
   it('refuses the session with insufficient_flux when the user is broke', async () => {
     upstream = await startMockUpstream([])
     const deps = makeFakeDeps({ upstreamURL: upstream.url, restBaseURL: upstream.restBaseURL, fluxBalance: 0 })
-    const handlers = createAudioSpeechWsHandlers(deps as any)
+    const handlers = createTestHandlers(deps)
     const events = handlers('user-broke', { trigger: 'auto', source: 'chat_auto_tts' })
     const client = makeMockClientWs()
 
@@ -468,9 +890,9 @@ describe('audio-speech-ws route', () => {
 
   it('refuses with streaming_tts_not_configured when UNSPEECH_UPSTREAM.streaming is empty', async () => {
     const deps = makeFakeDeps({ upstreamURL: 'ws://unused', fluxBalance: 100 })
-    deps.configKV.getOptional = vi.fn(async () => null) as any
+    deps.configKV.getOptional.mockImplementation(async () => null)
 
-    const handlers = createAudioSpeechWsHandlers(deps as any)
+    const handlers = createTestHandlers(deps)
     const events = handlers('user-noconf')
     const client = makeMockClientWs()
 
@@ -495,7 +917,7 @@ describe('audio-speech-ws route', () => {
       fluxBalance: 100,
       streamingModels: [{ id: 'volcengine/seed-tts-2.0', name: 'Seed-TTS 2.0' }],
     })
-    const handlers = createAudioSpeechWsHandlers(deps as any)
+    const handlers = createTestHandlers(deps)
     const events = handlers('user-disabled-model')
     const client = makeMockClientWs()
 
@@ -521,7 +943,7 @@ describe('audio-speech-ws route', () => {
   it('refuses an unknown streaming voice before dialing upstream', async () => {
     upstream = await startMockUpstream([], [{ id: 'enabled-voice', name: 'Enabled Voice' }])
     const deps = makeFakeDeps({ upstreamURL: upstream.url, restBaseURL: upstream.restBaseURL, fluxBalance: 100 })
-    const handlers = createAudioSpeechWsHandlers(deps as any)
+    const handlers = createTestHandlers(deps)
     const events = handlers('user-disabled-voice')
     const client = makeMockClientWs()
 
@@ -554,7 +976,7 @@ describe('audio-speech-ws route', () => {
     ])
 
     const deps = makeFakeDeps({ upstreamURL: upstream.url, restBaseURL: upstream.restBaseURL, fluxBalance: 100 })
-    const handlers = createAudioSpeechWsHandlers(deps as any)
+    const handlers = createTestHandlers(deps)
     const events = handlers('user-no-usage')
     const client = makeMockClientWs()
 
@@ -567,7 +989,7 @@ describe('audio-speech-ws route', () => {
     await new Promise(r => setTimeout(r, 200))
 
     expect(deps.ttsMeter.accumulate).toHaveBeenCalledTimes(1)
-    expect((deps.ttsMeter.accumulate.mock.calls[0] as any[])[0]).toMatchObject({
+    expect(deps.ttsMeter.accumulate.mock.calls[0]?.[0]).toMatchObject({
       userId: 'user-no-usage',
       units: 10, // "hello" + "world" = 10 chars
     })

server/apps/api/src/routes/audio-speech-ws/types.ts

@@ -4,6 +4,7 @@ import type { FluxService } from '../../services/domain/flux'
 import type { ProductEventService } from '../../services/domain/product-events'
 import type { RequestLogService } from '../../services/domain/request-log'
 import type { EnvelopeCrypto } from '../../utils/envelope-crypto'
+import type { StreamingTtsTransportOptions } from './providers/types'
 
 /**
  * Dependencies required by the streaming speech websocket proxy.
@@ -21,4 +22,6 @@ export interface AudioSpeechWsHandlersOptions {
   requestLogService: RequestLogService
   /** Writes first-party product analytics for distinct-user aggregation. */
   productEventService: ProductEventService
+  /** Overrides provider deadlines, primarily for constrained deployments and deterministic tests. */
+  streamingTtsTimeouts?: StreamingTtsTransportOptions['timeouts']
 }

server/apps/api/src/routes/openai/v1/operations/speech-catalog/index.ts

@@ -5,6 +5,7 @@ import { useLogger } from '@guiiai/logg'
 import { ofetch } from 'ofetch'
 
 import { catalogVoiceResponse } from '../../../../../services/domain/provider-catalog/provider-voices'
+import { isUnspeechStreamingModelEnabled, streamingTtsModelResourceId } from '../../../../../services/domain/streaming-tts-policy'
 import { createBadGatewayError, createBadRequestError, createServiceUnavailableError } from '../../../../../utils/error'
 
 const VOICE_PACK_MODEL_ID = 'voice-pack'
@@ -81,35 +82,42 @@ export function createSpeechCatalogOperation(deps: V1RouteDeps): SpeechCatalogOp
    * No empty-array fallback: the UI surfaces a real failure state.
    */
   async function listStreamingVoices(input: ListStreamingVoicesInput) {
-    const stepfun = await deps.configKV.getOptional('STEPFUN_STREAMING_TTS_UPSTREAM')
-    if (stepfun?.enabled) {
-      const model = input.model
-      const matchedModel = model
-        ? stepfun.models.find(item => item.id === model || item.id === `stepfun/${model}`)
-        : undefined
-      if (model && !matchedModel)
-        throw createBadRequestError('streaming voices: model is not enabled', 'STREAMING_TTS_MODEL_NOT_ENABLED')
-
-      const recommended = (await deps.configKV.getOptional('DEFAULT_TTS_VOICES'))?.[matchedModel?.id ?? stepfun.defaultModel] ?? {}
+    const [stepfun, unspeech] = await Promise.all([
+      deps.configKV.getOptional('STEPFUN_STREAMING_TTS_UPSTREAM'),
+      deps.configKV.getOptional('UNSPEECH_UPSTREAM'),
+    ])
+    const model = input.model
+    const stepfunAvailable = stepfun != null && stepfun.rollout !== 'disabled'
+    const selectedStepfunModel = stepfunAvailable
+      ? model
+        ? stepfun.models.find(item => item.id === model)
+        : stepfun.rollout === 'default'
+          ? stepfun.models.find(item => item.id === stepfun.defaultModel)
+          : undefined
+      : undefined
+    if (stepfun && selectedStepfunModel) {
+      const recommended = (await deps.configKV.getOptional('DEFAULT_TTS_VOICES'))?.[selectedStepfunModel.id] ?? {}
       return Response.json({ voices: stepfun.voices, recommended })
     }
 
-    const unspeech = await deps.configKV.getOptional('UNSPEECH_UPSTREAM')
+    const unspeechModels = unspeech?.streaming?.models ?? []
+    const unspeechModelEnabled = model == null
+      || (unspeech?.streaming != null && isUnspeechStreamingModelEnabled(unspeechModels, model))
+    if (!unspeechModelEnabled)
+      throw createBadRequestError('streaming voices: model is not enabled', 'STREAMING_TTS_MODEL_NOT_ENABLED')
     if (!unspeech?.streaming?.baseURL)
       throw createServiceUnavailableError('streaming tts upstream not configured', 'STREAMING_TTS_NOT_CONFIGURED')
 
     // Pass through the api_resource_id (e.g. `seed-tts-2.0`). unspeech
     // filters the embedded Volcengine catalogue server-side; absent model
     // means "return everything streaming-safe".
-    const model = input.model
-
     let voicesURL: string
     try {
       const u = new URL(unspeech.restBaseURL)
       u.pathname = '/api/voices'
       const params = new URLSearchParams({ provider: 'volcengine' })
       if (model)
-        params.set('model', model)
+        params.set('model', streamingTtsModelResourceId(model))
       u.search = `?${params.toString()}`
       voicesURL = u.toString()
     }
@@ -173,34 +181,28 @@ export function createSpeechCatalogOperation(deps: V1RouteDeps): SpeechCatalogOp
   }
 
   async function listStreamingSpeechModels() {
-    const stepfun = await deps.configKV.getOptional('STEPFUN_STREAMING_TTS_UPSTREAM')
-    if (stepfun?.enabled) {
-      return Response.json({
-        available: true,
-        models: stepfun.models.map(m => ({
-          id: m.id,
-          name: m.name ?? m.id,
-          description: m.description,
-        })),
-        default: stepfun.defaultModel,
-      })
-    }
-
-    const unspeech = await deps.configKV.getOptional('UNSPEECH_UPSTREAM')
-    const models = unspeech?.streaming?.models ?? []
+    const [stepfun, unspeech] = await Promise.all([
+      deps.configKV.getOptional('STEPFUN_STREAMING_TTS_UPSTREAM'),
+      deps.configKV.getOptional('UNSPEECH_UPSTREAM'),
+    ])
+    const unspeechModels = unspeech?.streaming?.models ?? []
+    const stepfunModels = stepfun?.rollout === 'disabled' ? [] : (stepfun?.models ?? [])
+    const models = [...unspeechModels, ...stepfunModels]
     // `available` is the operator-controlled visibility switch the client gates
     // the streaming provider on. It tracks whether `UNSPEECH_UPSTREAM.streaming`
     // is configured at all — not whether `models[]` happens to be empty — so an
     // operator who has wired the upstream but not yet curated models still
     // surfaces the provider rather than silently hiding it.
     return Response.json({
-      available: !!unspeech?.streaming?.baseURL,
+      available: !!unspeech?.streaming?.baseURL || stepfunModels.length > 0,
       models: models.map(m => ({
         id: m.id,
         name: m.name ?? m.id,
         description: m.description,
       })),
-      default: unspeech?.streaming?.defaultModel ?? null,
+      default: stepfun?.rollout === 'default'
+        ? stepfun.defaultModel
+        : (unspeech?.streaming?.defaultModel ?? null),
     })
   }
 

server/apps/api/src/routes/openai/v1/route.test.ts

@@ -1964,6 +1964,62 @@ describe('v1CompletionsRoutes', () => {
       expect(data.default).toBe('volcengine/seed-tts-2.0')
     })
 
+    it('exposes StepFun for explicit selection without replacing the current default', async () => {
+      const app = createTestApp(createMockFluxService(), createMockConfigKV({
+        UNSPEECH_UPSTREAM: {
+          restBaseURL: 'http://unspeech.local:5933',
+          streaming: {
+            baseURL: 'wss://unspeech.local',
+            keys: [{ id: 'volc-key', ciphertext: 'enc' }],
+            models: [{ id: 'volcengine/seed-tts-2.0' }],
+            defaultModel: 'volcengine/seed-tts-2.0',
+          },
+        },
+        STEPFUN_STREAMING_TTS_UPSTREAM: {
+          rollout: 'available',
+          baseURL: 'wss://api.stepfun.com/v1/realtime/audio',
+          keys: [{ id: 'step-key', ciphertext: 'enc' }],
+          models: [{ id: 'stepfun/step-tts-2' }],
+          defaultModel: 'stepfun/step-tts-2',
+          voices: [{ id: 'lively-girl', labels: {}, languages: [] }],
+        },
+      }))
+
+      const res = await app.fetch(new Request('http://localhost/api/v1/audio/models/streaming'), { user: testUser } as any)
+      const data = await res.json() as { models: Array<{ id: string }>, default: string }
+
+      expect(data.models.map(model => model.id)).toEqual(['volcengine/seed-tts-2.0', 'stepfun/step-tts-2'])
+      expect(data.default).toBe('volcengine/seed-tts-2.0')
+    })
+
+    it('switches only the default when StepFun rollout is default', async () => {
+      const app = createTestApp(createMockFluxService(), createMockConfigKV({
+        UNSPEECH_UPSTREAM: {
+          restBaseURL: 'http://unspeech.local:5933',
+          streaming: {
+            baseURL: 'wss://unspeech.local',
+            keys: [{ id: 'volc-key', ciphertext: 'enc' }],
+            models: [{ id: 'volcengine/seed-tts-2.0' }],
+            defaultModel: 'volcengine/seed-tts-2.0',
+          },
+        },
+        STEPFUN_STREAMING_TTS_UPSTREAM: {
+          rollout: 'default',
+          baseURL: 'wss://api.stepfun.com/v1/realtime/audio',
+          keys: [{ id: 'step-key', ciphertext: 'enc' }],
+          models: [{ id: 'stepfun/step-tts-2' }],
+          defaultModel: 'stepfun/step-tts-2',
+          voices: [{ id: 'lively-girl', labels: {}, languages: [] }],
+        },
+      }))
+
+      const res = await app.fetch(new Request('http://localhost/api/v1/audio/models/streaming'), { user: testUser } as any)
+      const data = await res.json() as { models: Array<{ id: string }>, default: string }
+
+      expect(data.models.map(model => model.id)).toEqual(['volcengine/seed-tts-2.0', 'stepfun/step-tts-2'])
+      expect(data.default).toBe('stepfun/step-tts-2')
+    })
+
     it('returns default: null when operator has not set a streaming default', async () => {
       const app = createTestApp(
         createMockFluxService(),
@@ -1988,6 +2044,28 @@ describe('v1CompletionsRoutes', () => {
       expect(data.default).toBeNull()
     })
 
+    it('keeps default null when StepFun is only available for explicit selection', async () => {
+      const app = createTestApp(createMockFluxService(), createMockConfigKV({
+        STEPFUN_STREAMING_TTS_UPSTREAM: {
+          rollout: 'available',
+          baseURL: 'wss://api.stepfun.com/v1/realtime/audio',
+          keys: [{ id: 'step-key', ciphertext: 'enc' }],
+          models: [{ id: 'stepfun/step-tts-2' }],
+          defaultModel: 'stepfun/step-tts-2',
+          voices: [{ id: 'lively-girl', labels: {}, languages: [] }],
+        },
+      }))
+
+      const res = await app.fetch(
+        new Request('http://localhost/api/v1/audio/models/streaming', { method: 'GET' }),
+        { user: testUser } as any,
+      )
+      const data = await res.json() as { available: boolean, default: string | null }
+
+      expect(data.available).toBe(true)
+      expect(data.default).toBeNull()
+    })
+
     it('returns an empty list when UNSPEECH_UPSTREAM is unset', async () => {
       const app = createTestApp(createMockFluxService(), createMockConfigKV())
 
@@ -2408,6 +2486,33 @@ describe('v1CompletionsRoutes', () => {
       globalThis.fetch = vi.fn(async () => new Response(body, { status })) as any
     }
 
+    it('returns StepFun voices only for the canonical configured model id', async () => {
+      const configKV = createMockConfigKV({
+        STEPFUN_STREAMING_TTS_UPSTREAM: {
+          rollout: 'available',
+          baseURL: 'wss://api.stepfun.com/v1/realtime/audio',
+          keys: [{ id: 'step-key', ciphertext: 'enc' }],
+          models: [{ id: 'stepfun/step-tts-2' }],
+          defaultModel: 'stepfun/step-tts-2',
+          voices: [{ id: 'lively-girl', name: 'Lively Girl', labels: {}, languages: [] }],
+        },
+      })
+      const app = createTestApp(createMockFluxService(), configKV)
+
+      const exact = await app.fetch(
+        new Request('http://localhost/api/v1/audio/voices/streaming?model=stepfun/step-tts-2'),
+        { user: testUser } as any,
+      )
+      const alias = await app.fetch(
+        new Request('http://localhost/api/v1/audio/voices/streaming?model=step-tts-2'),
+        { user: testUser } as any,
+      )
+
+      expect(exact.status).toBe(200)
+      expect(await exact.json()).toMatchObject({ voices: [{ id: 'lively-girl' }] })
+      expect(alias.status).toBe(400)
+    })
+
     it('returns the streaming-model bucket of DEFAULT_TTS_VOICES when ?model= matches', async () => {
       mockUnspeechVoices([{ id: 'zh_female_vv_uranus_bigtts', name: 'Vivi 2.0' }])
       const configKV = createMockConfigKV({
@@ -2430,6 +2535,36 @@ describe('v1CompletionsRoutes', () => {
       expect(data.recommended).toEqual({ 'zh-cn': 'zh_female_vv_uranus_bigtts' })
     })
 
+    it('normalizes a canonical unSpeech model id before querying its voice catalog', async () => {
+      let requestedURL = ''
+      globalThis.fetch = vi.fn(async (input) => {
+        requestedURL = String(input)
+        return new Response(JSON.stringify({ voices: [] }), {
+          status: 200,
+          headers: { 'Content-Type': 'application/json' },
+        })
+      }) as any
+      const configKV = createMockConfigKV({
+        UNSPEECH_UPSTREAM: {
+          restBaseURL: 'http://unspeech.local:5933',
+          streaming: {
+            baseURL: 'ws://unspeech.local:5933/v1/audio/speech/stream',
+            keys: [{ id: 'k1', ciphertext: 'enc' }],
+            models: [{ id: 'volcengine/seed-tts-2.0' }],
+          },
+        },
+      })
+      const app = createTestApp(createMockFluxService(), configKV)
+
+      const res = await app.fetch(
+        new Request('http://localhost/api/v1/audio/voices/streaming?model=volcengine/seed-tts-2.0'),
+        { user: testUser } as any,
+      )
+
+      expect(res.status).toBe(200)
+      expect(new URL(requestedURL).searchParams.get('model')).toBe('seed-tts-2.0')
+    })
+
     it('returns empty recommended when ?model= is omitted', async () => {
       mockUnspeechVoices([])
       const configKV = createMockConfigKV({

server/apps/api/src/services/adapters/config-kv.test.ts

@@ -88,6 +88,24 @@ describe('configKVService', () => {
       })
   })
 
+  it('rejects unsupported native StepFun models at the ConfigKV boundary', async () => {
+    redis._store.set(configRedisKey('STEPFUN_STREAMING_TTS_UPSTREAM'), JSON.stringify({
+      rollout: 'available',
+      baseURL: 'wss://api.stepfun.com/v1/realtime/audio',
+      keys: [{ id: 'step-key', ciphertext: 'encrypted' }],
+      models: [{ id: 'stepfun/unsupported-model' }],
+      defaultModel: 'stepfun/unsupported-model',
+      voices: [{ id: 'lively-girl' }],
+    }))
+
+    await expect(service.getOptional('STEPFUN_STREAMING_TTS_UPSTREAM'))
+      .rejects
+      .toMatchObject({
+        statusCode: 503,
+        errorCode: 'CONFIG_INVALID',
+      })
+  })
+
   it('set should write value to Redis with prefix', async () => {
     await service.set('FLUX_PER_REQUEST', 10)
 

server/apps/api/src/services/adapters/config-kv.ts

@@ -164,27 +164,43 @@ export const streamingTtsUpstreamSchema = object({
   defaultModel: optional(string()),
 })
 
+/** Canonical public model ids supported by StepFun's native TTS websocket. */
+export const STEPFUN_STREAMING_TTS_MODEL_IDS = [
+  'stepfun/stepaudio-2.5-tts',
+  'stepfun/step-tts-2',
+  'stepfun/step-tts-mini',
+] as const
+
+/** Encryption context shared by the writer and runtime credential reader. */
+export const STEPFUN_STREAMING_TTS_KEY_CONTEXT = 'stepfun-streaming-tts'
+
+/**
+ * Controls whether StepFun is hidden, available for explicit model selection,
+ * or also owns the default streaming model.
+ */
+export const stepfunStreamingTtsRolloutSchema = picklist(['disabled', 'available', 'default'])
+
 /**
  * Dedicated StepFun streaming TTS configuration.
  *
  * StepFun's websocket protocol is not compatible with unSpeech's streaming
- * wire format, so it owns an explicit configuration entry and enable switch.
- * Keeping the switch in this record lets operators validate the provider
- * without changing the current streaming default before a measured rollout.
+ * wire format, so it owns an explicit configuration entry and rollout mode.
+ * `available` permits explicit model selection without changing the default;
+ * `default` performs the measured cutover.
  */
 export const stepfunStreamingTtsUpstreamSchema = pipe(object({
-  enabled: optional(boolean(), false),
+  rollout: optional(stepfunStreamingTtsRolloutSchema, 'disabled'),
   baseURL: pipe(string(), nonEmpty('STEPFUN_STREAMING_TTS_UPSTREAM.baseURL must not be empty')),
   keys: pipe(array(keyEntrySchema), check(v => v.length >= 1, 'STEPFUN_STREAMING_TTS_UPSTREAM.keys must contain at least 1 entry')),
   models: pipe(
     array(object({
-      id: pipe(string(), nonEmpty('STEPFUN_STREAMING_TTS_UPSTREAM.models[].id must not be empty')),
+      id: picklist(STEPFUN_STREAMING_TTS_MODEL_IDS, 'STEPFUN_STREAMING_TTS_UPSTREAM.models[].id is not supported'),
       name: optional(string()),
       description: optional(string()),
     })),
     check(v => v.length >= 1, 'STEPFUN_STREAMING_TTS_UPSTREAM.models must contain at least 1 entry'),
   ),
-  defaultModel: pipe(string(), nonEmpty('STEPFUN_STREAMING_TTS_UPSTREAM.defaultModel must not be empty')),
+  defaultModel: picklist(STEPFUN_STREAMING_TTS_MODEL_IDS, 'STEPFUN_STREAMING_TTS_UPSTREAM.defaultModel is not supported'),
   voices: pipe(
     array(object({
       id: pipe(string(), nonEmpty('STEPFUN_STREAMING_TTS_UPSTREAM.voices[].id must not be empty')),
@@ -196,7 +212,7 @@ export const stepfunStreamingTtsUpstreamSchema = pipe(object({
     check(v => v.length >= 1, 'STEPFUN_STREAMING_TTS_UPSTREAM.voices must contain at least 1 entry'),
   ),
   instruction: optional(pipe(string(), nonEmpty('STEPFUN_STREAMING_TTS_UPSTREAM.instruction must not be empty'))),
-}), check(config => new Set(config.models.map(model => model.id)).size === config.models.length, 'STEPFUN_STREAMING_TTS_UPSTREAM.models[].id must be unique'), check(config => config.models.some(model => model.id === config.defaultModel), 'STEPFUN_STREAMING_TTS_UPSTREAM.defaultModel must be present in models'))
+}), check(config => new Set(config.models.map(model => model.id)).size === config.models.length, 'STEPFUN_STREAMING_TTS_UPSTREAM.models[].id must be unique'), check(config => config.models.some(model => model.id === config.defaultModel), 'STEPFUN_STREAMING_TTS_UPSTREAM.defaultModel must be present in models'), check(config => new Set(config.voices.map(voice => voice.id)).size === config.voices.length, 'STEPFUN_STREAMING_TTS_UPSTREAM.voices[].id must be unique'))
 
 export const unspeechUpstreamSchema = object({
   restBaseURL: pipe(string(), nonEmpty('UNSPEECH_UPSTREAM.restBaseURL must not be empty')),
@@ -318,10 +334,9 @@ const ConfigEntrySchemas = {
   // carry the upstream-provider API key (Volcengine X-Api-Key), not an
   // unspeech tenant token (unspeech itself is unauthenticated).
   UNSPEECH_UPSTREAM: optional(unspeechUpstreamSchema),
-  // A dedicated native StepFun WebSocket provider. It is deliberately
-  // separate from UNSPEECH_UPSTREAM because their request/response protocols
-  // differ. `enabled: false` keeps the existing Volcengine path active until
-  // an operator completes a measured switch.
+  // Native StepFun WebSocket provider. `rollout: available` exposes its models
+  // for explicit selection while preserving the existing default; `default`
+  // performs the operator-controlled cutover.
   STEPFUN_STREAMING_TTS_UPSTREAM: optional(stepfunStreamingTtsUpstreamSchema),
 } as const
 

server/apps/api/src/services/domain/admin/router-config/index.ts

@@ -7,6 +7,7 @@ import type { asrModelSchema, ConfigKVService, llmModelSchema, llmRouterConfigSc
 import { useLogger } from '@guiiai/logg'
 
 import { createBadRequestError } from '../../../../utils/error'
+import { STEPFUN_STREAMING_TTS_KEY_CONTEXT } from '../../../adapters/config-kv'
 
 /**
  * AAD label used when encrypting/decrypting the streaming TTS upstream key.
@@ -16,7 +17,6 @@ import { createBadRequestError } from '../../../../utils/error'
  * `DECRYPT_FAILED` at session start.
  */
 const STREAMING_TTS_AAD_MODEL_NAME = 'streaming-tts'
-const STEPFUN_STREAMING_TTS_AAD_MODEL_NAME = 'stepfun-streaming-tts'
 
 /** Default key entry id per provider. Operator can override per request. */
 const DEFAULT_KEY_ENTRY_IDS = {
@@ -168,12 +168,12 @@ export interface StepfunSliceInput {
 /** Operator-owned native StepFun websocket TTS configuration. */
 export interface StepfunStreamingSliceInput {
   kind: 'stepfun-streaming'
-  /** Enables StepFun for the shared official streaming TTS provider. */
-  enabled: boolean
+  /** Controls explicit availability and ownership of the global streaming default. */
+  rollout: 'disabled' | 'available' | 'default'
   /** `wss://api.stepfun.com/v1/realtime/audio` or the Step Plan equivalent. */
   upstreamURL: string
-  models: Array<{ id: string, name?: string, description?: string }>
-  defaultModel: string
+  models: StepfunStreamingTtsUpstream['models']
+  defaultModel: StepfunStreamingTtsUpstream['defaultModel']
   voices: Array<{ id: string, name?: string, description?: string, labels?: Record<string, unknown>, languages?: Array<{ code: string, title: string }> }>
   instruction?: string
   plaintextKey?: string
@@ -514,7 +514,7 @@ export function buildUnspeechSlice(input: UnspeechSliceInput, envelope: Envelope
 export function buildStepfunStreamingSlice(input: StepfunStreamingSliceInput, envelope: EnvelopeCrypto): StepfunStreamingSlice {
   const keyEntryId = input.keyEntryId ?? DEFAULT_KEY_ENTRY_IDS['stepfun-streaming']
   const ciphertext = envelope.encryptKey(requiredPlaintextKey(input.plaintextKey, input.kind), {
-    modelName: STEPFUN_STREAMING_TTS_AAD_MODEL_NAME,
+    modelName: STEPFUN_STREAMING_TTS_KEY_CONTEXT,
     keyEntryId,
   })
   return {
@@ -522,7 +522,7 @@ export function buildStepfunStreamingSlice(input: StepfunStreamingSliceInput, en
     kind: input.kind,
     keyEntryId,
     value: {
-      enabled: input.enabled,
+      rollout: input.rollout,
       baseURL: input.upstreamURL,
       keys: [{ id: keyEntryId, ciphertext }],
       models: input.models,
@@ -733,7 +733,7 @@ function buildStepfunStreamingSlicePreservingKey(
     kind: input.kind,
     keyEntryId: key.id,
     value: {
-      enabled: input.enabled,
+      rollout: input.rollout,
       baseURL: input.upstreamURL,
       keys: [key],
       models: input.models,
@@ -1088,7 +1088,7 @@ function slicesFromStepfunStreaming(config: StepfunStreamingTtsUpstream | null):
   const key = config.keys[0]
   return [{
     kind: 'stepfun-streaming',
-    enabled: config.enabled,
+    rollout: config.rollout,
     upstreamURL: config.baseURL,
     models: config.models,
     defaultModel: config.defaultModel,
@@ -1204,7 +1204,6 @@ export function createAdminRouterConfigService(deps: AdminRouterConfigDeps) {
       missingKeys: [
         ...(routerConfig ? [] : ['LLM_ROUTER_CONFIG']),
         ...(unspeech ? [] : ['UNSPEECH_UPSTREAM']),
-        ...(stepfunStreaming ? [] : ['STEPFUN_STREAMING_TTS_UPSTREAM']),
         ...(chatModel ? [] : ['DEFAULT_CHAT_MODEL']),
         ...(ttsModel ? [] : ['DEFAULT_TTS_MODEL']),
       ],

server/apps/api/src/services/domain/admin/router-config/tests/admin-router-config.test.ts

@@ -292,7 +292,7 @@ describe('buildStepfunStreamingSlice', () => {
   it('encrypts the native websocket credential under its dedicated AAD label', () => {
     const built = buildStepfunStreamingSlice({
       kind: 'stepfun-streaming',
-      enabled: false,
+      rollout: 'disabled' as const,
       upstreamURL: 'wss://api.stepfun.com/v1/realtime/audio',
       models: [{ id: 'stepfun/step-tts-2', name: 'Step TTS 2' }],
       defaultModel: 'stepfun/step-tts-2',
@@ -302,7 +302,7 @@ describe('buildStepfunStreamingSlice', () => {
 
     expect(built.target).toBe('stepfun-streaming')
     expect(built.value).toMatchObject({
-      enabled: false,
+      rollout: 'disabled' as const,
       baseURL: 'wss://api.stepfun.com/v1/realtime/audio',
       defaultModel: 'stepfun/step-tts-2',
       voices: [{ id: 'lively-girl', labels: {}, languages: [] }],
@@ -584,10 +584,10 @@ describe('createAdminRouterConfigService', () => {
     const service = createAdminRouterConfigService({ configKV: kv.service, envelope, redis })
     const stepfunSlice = {
       kind: 'stepfun-streaming' as const,
-      enabled: false,
+      rollout: 'disabled' as const,
       upstreamURL: 'wss://api.stepfun.com/v1/realtime/audio',
-      models: [{ id: 'stepfun/step-tts-2' }],
-      defaultModel: 'stepfun/step-tts-2',
+      models: [{ id: 'stepfun/step-tts-2' as const }],
+      defaultModel: 'stepfun/step-tts-2' as const,
       voices: [{ id: 'lively-girl' }],
       plaintextKey: 'stepfun-secret',
     }
@@ -724,6 +724,7 @@ describe('createAdminRouterConfigService', () => {
     expect(current.request.defaults.chatModel).toBe('chat-live')
     expect(JSON.stringify(current.preview)).toContain('<ciphertext: 17 chars>')
     expect(JSON.stringify(current.preview)).not.toContain('secret-ciphertext')
+    expect(current.missingKeys).not.toContain('STEPFUN_STREAMING_TTS_UPSTREAM')
   })
 
   it('current classifies Bedrock and generic OpenAI-compatible LLM upstreams by baseURL', async () => {

server/apps/api/src/services/domain/product-events.ts

@@ -12,7 +12,7 @@ const logger = useLogger('product-events')
 
 export type ProductFeature = 'auth' | 'chat' | 'gen_ai_chat' | 'tts' | 'billing' | 'voice_pack'
 
-export type ProductEventStatus = 'started' | 'succeeded' | 'failed' | 'blocked'
+export type ProductEventStatus = 'started' | 'succeeded' | 'failed' | 'blocked' | 'cancelled'
 
 export type ProductAction
   = | 'user_signed_up'
@@ -25,6 +25,7 @@ export type ProductAction
     | 'speech_succeeded'
     | 'speech_failed'
     | 'speech_blocked'
+    | 'speech_cancelled'
     | 'voice_pack_created'
     | 'voice_pack_updated'
     | 'voice_pack_disabled'

server/apps/api/src/services/domain/streaming-tts-policy.ts

@@ -0,0 +1,30 @@
+/** Minimal model shape accepted by streaming TTS visibility policy. */
+export interface StreamingTtsModelSelection {
+  id: string
+}
+
+/**
+ * Decides whether an unSpeech streaming model can be selected.
+ *
+ * An empty curated list intentionally delegates filtering to unSpeech. A
+ * non-empty list is an operator allowlist. Catalog and websocket resolution
+ * must share this rule so visible models are always startable.
+ */
+export function isUnspeechStreamingModelEnabled(
+  models: StreamingTtsModelSelection[],
+  requestedModel: string,
+): boolean {
+  return models.length === 0 || models.some(model => model.id === requestedModel)
+}
+
+/**
+ * Converts a canonical public model id into the resource id expected by
+ * unSpeech's upstream voice catalog.
+ *
+ * Before: `volcengine/seed-tts-2.0`
+ * After: `seed-tts-2.0`
+ */
+export function streamingTtsModelResourceId(model: string): string {
+  const separator = model.indexOf('/')
+  return separator >= 0 ? model.slice(separator + 1) : model
+}