feat(server,stage-ui): bidirectional streaming TTS + audio path refactor
Why:
- Add a real bidirectional streaming TTS path: raw LLM tokens are
forwarded to the upstream model (Volcengine v3 via the unspeech ws
bridge) without client-side segmentation, so the model owns sentence
splitting and audio chunks play as they arrive.
- Move audio endpoints out of /api/v1/openai/. `/audio/voices`,
`/audio/models`, `/audio/voices/streaming` are not real OpenAI public
APIs, and the streaming TTS surface has nothing to do with OpenAI —
keeping them under /openai/ mislabelled the contract.
- Introduce `capabilities.speech.transport` on ProviderDefinition so
future streaming providers (ElevenLabs / Cartesia / OpenAI Realtime)
opt in without touching Stage.vue or the session factory.
- Unify Stage.vue's TTS path through a single StageTtsSession so the
chat-orchestrator hooks no longer branch on provider id.
What:
- apps/server: new ws proxy /api/v1/audio/speech/ws bridges client ↔
unspeech with auth, pre-flight flux check, billing from upstream
session.finished.usage, OTel spans.
- apps/server: audio routes moved from /api/v1/openai/audio/* to
/api/v1/audio/* (hard cutover; 404 sentinel tests added).
- apps/server: new /api/v1/audio/voices/streaming proxy reads voices
from unspeech /api/voices?provider=volcengine.
- apps/server: new STREAMING_TTS_UPSTREAM configKV entry +
scripts/seed-streaming-tts.ts.
- stage-ui: new libs/speech/streaming-pipeline.ts opens one ws per LLM
intent (appendText / finish / cancel + onSentence / onError / onDone).
- stage-ui: new libs/speech/tts-session.ts — StageTtsSession interface
with segmenter and streaming adapters; factory dispatches by
capabilities.speech.transport instead of hard-coded provider id.
- stage-ui: providerOfficialSpeechStreaming with capabilities.speech =
{ transport: 'bidirectional-ws' }; settings page with model/voice
picker + ws-based preview.
- stage-ui: Stage.vue chat hooks collapsed to a single currentSession;
hot-swap watcher cancels mid-session on provider/voice/model change;
unmount cancels and drains playback.
Tests:
- 9 streaming-pipeline tests (happy path / buffered / error / cancel /
truncation)
- 11 tts-session tests (factory branch coverage + adapter contracts)
- 4 audio-speech-ws route tests (forwarding / billing / pre-flight /
config-missing)
- 3 legacy-path 404 sentinels in v1 route tests
- Verification doc updated to reflect automated coverage.
This commit is contained in:
RainbowBird
@@ -83,19 +83,22 @@ once it lands.
|
||||
|
||||
```bash
|
||||
cd apps/server
|
||||
LLM_ROUTER_MASTER_KEY="$LLM_ROUTER_MASTER_KEY" \
|
||||
STREAMING_TTS_UPSTREAM_URL="ws://airi-unspeech.railway.internal:5933/v1/audio/speech/stream" \
|
||||
VOLCENGINE_TTS_API_KEY="$VOLCENGINE_TTS_API_KEY" \
|
||||
pnpm exec dotenvx run --env-file=.env.local -- \
|
||||
tsx scripts/seed-router-config.ts \
|
||||
--streaming-tts-key "$VOLCENGINE_TTS_API_KEY"
|
||||
tsx scripts/seed-streaming-tts.ts
|
||||
```
|
||||
|
||||
> Note: `seed-router-config.ts` does not yet implement
|
||||
> `--streaming-tts-key`. Until then, write the configKV entry directly
|
||||
> from a one-off node REPL using `configKV.set('STREAMING_TTS_UPSTREAM',
|
||||
> { baseURL: 'ws://localhost:5933/v1/audio/speech/stream', keys: [{ id:
|
||||
> 'volcengine-prod-1', ciphertext: envelopeCrypto.encryptKey(<plaintext>,
|
||||
> { modelName: 'streaming-tts', keyEntryId: 'volcengine-prod-1' }) }] })`.
|
||||
The script reads `LLM_ROUTER_MASTER_KEY` and `REDIS_URL` from
|
||||
`.env.local`, envelope-encrypts the Volcengine key under AAD
|
||||
`{ modelName: 'streaming-tts', keyEntryId: 'volcengine-prod-1' }`, and
|
||||
writes the `STREAMING_TTS_UPSTREAM` configKV entry. Use `--dry-run` to
|
||||
preview the ciphertext length without committing.
|
||||
|
||||
To point at a different unspeech instance later, just re-run the script
|
||||
with a different `STREAMING_TTS_UPSTREAM_URL`. To rotate the upstream
|
||||
key, re-run with `--key-id volcengine-prod-N` (the audio-speech-ws
|
||||
route always reads `keys[0]`, so a write replaces the active key).
|
||||
|
||||
### Scenario L1: streaming session happy path
|
||||
|
||||
|
||||
@@ -0,0 +1,172 @@
|
||||
#!/usr/bin/env tsx
|
||||
/* eslint-disable no-console */
|
||||
/**
|
||||
* Seed the `STREAMING_TTS_UPSTREAM` configKV entry that powers
|
||||
* `/api/v1/audio/speech/ws`.
|
||||
*
|
||||
* Why a separate script from `seed-router-config.ts`:
|
||||
* - STREAMING_TTS_UPSTREAM is a sibling top-level configKV entry, not part
|
||||
* of `LLM_ROUTER_CONFIG.tts.models`. Mixing the two write paths would
|
||||
* make the merge semantics in `seed-router-config.ts` harder to follow.
|
||||
* - The streaming surface has only one upstream (one unspeech instance
|
||||
* per deployment), so a focused script avoids the multi-provider
|
||||
* merge logic entirely.
|
||||
*
|
||||
* What the script writes:
|
||||
* - `STREAMING_TTS_UPSTREAM.baseURL` — the unspeech ws endpoint, e.g.
|
||||
* `ws://airi-unspeech.railway.internal:5933/v1/audio/speech/stream`
|
||||
* or `wss://unspeech.example.com/v1/audio/speech/stream`.
|
||||
* - `STREAMING_TTS_UPSTREAM.keys[0]` — the **upstream provider** key
|
||||
* (Volcengine `X-Api-Key`) wrapped in an envelope ciphertext, not the
|
||||
* unspeech key — unspeech itself has no auth concept, it just
|
||||
* forwards the `Authorization` header verbatim to the upstream.
|
||||
*
|
||||
* Envelope AAD:
|
||||
* - `modelName: 'streaming-tts'` and `keyEntryId: 'volcengine-prod-1'`.
|
||||
* These must match the values `audio-speech-ws/index.ts` uses when
|
||||
* decrypting (`STREAM_MODEL_LABEL_FALLBACK = 'streaming-tts'`).
|
||||
*
|
||||
* Cross-instance invalidation:
|
||||
* - Published on `configkv:invalidate`. The audio-speech-ws route reads
|
||||
* STREAMING_TTS_UPSTREAM fresh on every connection (no in-memory
|
||||
* cache), so the publish is currently informational — it stays here
|
||||
* for forward compatibility if we add caching later.
|
||||
*
|
||||
* Usage:
|
||||
*
|
||||
* STREAMING_TTS_UPSTREAM_URL="ws://airi-unspeech.railway.internal:5933/v1/audio/speech/stream" \
|
||||
* VOLCENGINE_TTS_API_KEY="sk-..." \
|
||||
* pnpm exec dotenvx run --env-file=.env.local -- \
|
||||
* tsx scripts/seed-streaming-tts.ts
|
||||
*
|
||||
* # preview the ciphertext shape without writing:
|
||||
* pnpm exec dotenvx run --env-file=.env.local -- \
|
||||
* tsx scripts/seed-streaming-tts.ts --dry-run
|
||||
*
|
||||
* # rotate the key id (default 'volcengine-prod-1'):
|
||||
* tsx scripts/seed-streaming-tts.ts --key-id volcengine-prod-2
|
||||
*/
|
||||
import type { ConfigKVService } from '../src/services/config-kv'
|
||||
|
||||
import process, { env, exit } from 'node:process'
|
||||
|
||||
import Redis from 'ioredis'
|
||||
|
||||
import { parseEnv } from '../src/libs/env'
|
||||
import { createConfigKVService } from '../src/services/config-kv'
|
||||
import { createEnvelopeCrypto } from '../src/utils/envelope-crypto'
|
||||
|
||||
// Must match `STREAM_MODEL_LABEL_FALLBACK` in
|
||||
// apps/server/src/routes/audio-speech-ws/index.ts — the route decrypts
|
||||
// with this AAD, so seeding under a different label would surface as
|
||||
// `DECRYPT_FAILED` at session start.
|
||||
const STREAM_AAD_MODEL_NAME = 'streaming-tts'
|
||||
|
||||
interface Args {
|
||||
dryRun: boolean
|
||||
keyId: string
|
||||
}
|
||||
|
||||
function parseArgs(argv: string[]): Args {
|
||||
const flags = new Set<string>()
|
||||
const values: Record<string, string> = {}
|
||||
for (let i = 0; i < argv.length; i++) {
|
||||
const arg = argv[i]
|
||||
if (!arg.startsWith('--'))
|
||||
continue
|
||||
const key = arg.slice(2)
|
||||
const next = argv[i + 1]
|
||||
if (next == null || next.startsWith('--')) {
|
||||
flags.add(key)
|
||||
}
|
||||
else {
|
||||
values[key] = next
|
||||
i++
|
||||
}
|
||||
}
|
||||
return {
|
||||
dryRun: flags.has('dry-run'),
|
||||
keyId: values['key-id'] ?? 'volcengine-prod-1',
|
||||
}
|
||||
}
|
||||
|
||||
async function main() {
|
||||
const parsedEnv = parseEnv(env)
|
||||
if (!parsedEnv.LLM_ROUTER_MASTER_KEY) {
|
||||
console.error('error: LLM_ROUTER_MASTER_KEY env var is required (32 random bytes, base64)')
|
||||
exit(1)
|
||||
}
|
||||
|
||||
const args = parseArgs(process.argv.slice(2))
|
||||
|
||||
const upstreamURL = process.env.STREAMING_TTS_UPSTREAM_URL
|
||||
const volcKey = process.env.VOLCENGINE_TTS_API_KEY
|
||||
|
||||
if (!upstreamURL || !volcKey) {
|
||||
console.error('error: STREAMING_TTS_UPSTREAM_URL and VOLCENGINE_TTS_API_KEY env vars are required.')
|
||||
console.error(' STREAMING_TTS_UPSTREAM_URL=ws://<unspeech-host>:5933/v1/audio/speech/stream')
|
||||
console.error(' VOLCENGINE_TTS_API_KEY=<volcengine X-Api-Key>')
|
||||
console.error('keys are read from env so they never appear in shell history or `ps`.')
|
||||
exit(1)
|
||||
}
|
||||
|
||||
// Soft-validate the scheme. unspeech is reachable over either ws:// (Railway
|
||||
// internal networking) or wss:// (public TLS-terminated). Anything else
|
||||
// (http://, https://) is almost certainly a copy-paste error from the
|
||||
// unspeech REST endpoint URL.
|
||||
if (!upstreamURL.startsWith('ws://') && !upstreamURL.startsWith('wss://')) {
|
||||
console.error(`error: STREAMING_TTS_UPSTREAM_URL must start with ws:// or wss://, got: ${upstreamURL}`)
|
||||
console.error('hint: the apps/server proxy opens a WebSocket — http:// will fail at the new WebSocket() call.')
|
||||
exit(1)
|
||||
}
|
||||
|
||||
const envelope = createEnvelopeCrypto({
|
||||
masterKey: parsedEnv.LLM_ROUTER_MASTER_KEY,
|
||||
previousMasterKey: parsedEnv.LLM_ROUTER_MASTER_KEY_PREVIOUS,
|
||||
})
|
||||
|
||||
const ciphertext = envelope.encryptKey(volcKey, {
|
||||
modelName: STREAM_AAD_MODEL_NAME,
|
||||
keyEntryId: args.keyId,
|
||||
})
|
||||
|
||||
const value = {
|
||||
baseURL: upstreamURL,
|
||||
keys: [{ id: args.keyId, ciphertext }],
|
||||
adapterParams: {},
|
||||
}
|
||||
|
||||
console.log(`mode: seed${args.dryRun ? ' (dry-run)' : ''}`)
|
||||
console.log(`baseURL: ${upstreamURL}`)
|
||||
console.log(`keys[0].id: ${args.keyId}`)
|
||||
console.log(`ciphertext: <${ciphertext.length} chars>`)
|
||||
|
||||
if (args.dryRun) {
|
||||
console.log('dry-run: no writes, no publish.')
|
||||
return
|
||||
}
|
||||
|
||||
const redis = new Redis(parsedEnv.REDIS_URL)
|
||||
const configKV: ConfigKVService = createConfigKVService(redis)
|
||||
|
||||
// configKV.set runs the valibot validator (ttsUpstreamSchema) before
|
||||
// committing; a malformed shape fails here instead of at first request.
|
||||
await configKV.set('STREAMING_TTS_UPSTREAM', value as never)
|
||||
|
||||
const payload = JSON.stringify({
|
||||
key: 'STREAMING_TTS_UPSTREAM',
|
||||
version: Date.now(),
|
||||
publishedAt: Date.now(),
|
||||
})
|
||||
await redis.publish('configkv:invalidate', payload)
|
||||
|
||||
console.log('STREAMING_TTS_UPSTREAM written.')
|
||||
console.log('Published configkv:invalidate.')
|
||||
|
||||
await redis.quit()
|
||||
}
|
||||
|
||||
main().catch((err) => {
|
||||
console.error('seed-streaming-tts failed:', err)
|
||||
exit(1)
|
||||
})
|
||||
+12
-3
@@ -53,7 +53,7 @@ import { createCharacterRoutes } from './routes/characters'
|
||||
import { createChatWsHandlers } from './routes/chat-ws'
|
||||
import { createChatRoutes } from './routes/chats'
|
||||
import { createFluxRoutes } from './routes/flux'
|
||||
import { createV1CompletionsRoutes } from './routes/openai/v1'
|
||||
import { createV1Routes } from './routes/openai/v1'
|
||||
import { createProviderRoutes } from './routes/providers'
|
||||
import { createStripeRoutes } from './routes/stripe'
|
||||
import { createAdminFluxGrantsService } from './services/admin-flux-grants'
|
||||
@@ -204,6 +204,11 @@ export async function buildApp(deps: AppDeps) {
|
||||
logger: useLogger('config-sync').useGlobalConfig(),
|
||||
})
|
||||
|
||||
// Built once so the OpenAI-compat and audio routers share the same closure
|
||||
// (helpers like recordMetrics / recordRequestLog cross both surfaces) but
|
||||
// mount at different prefixes — see the `.route` calls below.
|
||||
const v1Routes = createV1Routes(deps.fluxService, deps.billingService, deps.configKV, deps.requestLogService, deps.ttsMeter, deps.llmRouter, deps.otel?.genAi, deps.otel?.revenue, deps.otel?.rateLimit)
|
||||
|
||||
const builtApp = app
|
||||
.use('*', sessionMiddleware(deps.auth, deps.env))
|
||||
.use('*', bodyLimit({ maxSize: 1024 * 1024 }))
|
||||
@@ -311,9 +316,13 @@ export async function buildApp(deps: AppDeps) {
|
||||
.route('/api/v1/chats', createChatRoutes(deps.chatService))
|
||||
|
||||
/**
|
||||
* V1 routes for official provider.
|
||||
* V1 OpenAI-compatible and audio routes. The factory returns two
|
||||
* sibling routers because the audio surface deliberately lives outside
|
||||
* `/openai/` — its `/voices`, `/voices/streaming`, and `/models`
|
||||
* extensions aren't OpenAI public APIs.
|
||||
*/
|
||||
.route('/api/v1/openai', createV1CompletionsRoutes(deps.fluxService, deps.billingService, deps.configKV, deps.requestLogService, deps.ttsMeter, deps.llmRouter, deps.otel?.genAi, deps.otel?.revenue, deps.otel?.rateLimit))
|
||||
.route('/api/v1/openai', v1Routes.openaiRoutes)
|
||||
.route('/api/v1/audio', v1Routes.audioRoutes)
|
||||
|
||||
/**
|
||||
* Flux routes.
|
||||
|
||||
@@ -67,7 +67,7 @@ function getLlmMetricAttributes(opts: { model: string, type: string, status: num
|
||||
}
|
||||
}
|
||||
|
||||
export function createV1CompletionsRoutes(
|
||||
export function createV1Routes(
|
||||
fluxService: FluxService,
|
||||
billingService: BillingService,
|
||||
configKV: ConfigKVService,
|
||||
@@ -635,6 +635,65 @@ export function createV1CompletionsRoutes(
|
||||
return Response.json({ voices, recommended })
|
||||
}
|
||||
|
||||
/**
|
||||
* Voice catalog for the streaming TTS provider (`/audio/speech/ws`).
|
||||
*
|
||||
* The HTTP `/audio/voices?model=…` endpoint above queries
|
||||
* `LLM_ROUTER_CONFIG.tts.models` and is unaware of the streaming
|
||||
* surface. Streaming uses `STREAMING_TTS_UPSTREAM` (a single unspeech
|
||||
* instance) instead, and unspeech ships an embed-time voice catalog
|
||||
* for Volcengine that doesn't require credentials — so we proxy
|
||||
* straight to it. Falls back to an empty list if streaming isn't
|
||||
* configured yet so the client can render "no voices" instead of
|
||||
* exploding.
|
||||
*/
|
||||
async function handleListStreamingVoices(_c: Context<HonoEnv>) {
|
||||
const upstream = await configKV.getOptional('STREAMING_TTS_UPSTREAM')
|
||||
if (!upstream || !upstream.baseURL)
|
||||
return Response.json({ voices: [], recommended: {} })
|
||||
|
||||
let voicesURL: string
|
||||
try {
|
||||
const u = new URL(upstream.baseURL)
|
||||
// ws:// → http://, wss:// → https://. unspeech serves both the WS
|
||||
// stream and the REST voices endpoint on the same listener.
|
||||
u.protocol = u.protocol === 'wss:' ? 'https:' : 'http:'
|
||||
u.pathname = '/api/voices'
|
||||
u.search = '?provider=volcengine'
|
||||
voicesURL = u.toString()
|
||||
}
|
||||
catch (err) {
|
||||
logger.withError(err).withFields({ baseURL: upstream.baseURL }).warn('streaming-voices: bad upstream URL')
|
||||
return Response.json({ voices: [], recommended: {} })
|
||||
}
|
||||
|
||||
let res: Response
|
||||
try {
|
||||
res = await globalThis.fetch(voicesURL, {
|
||||
signal: AbortSignal.timeout(5000),
|
||||
})
|
||||
}
|
||||
catch (err) {
|
||||
logger.withError(err).withFields({ voicesURL }).warn('streaming-voices: unspeech fetch failed')
|
||||
return Response.json({ voices: [], recommended: {} })
|
||||
}
|
||||
|
||||
if (!res.ok) {
|
||||
logger.withFields({ voicesURL, status: res.status }).warn('streaming-voices: unspeech non-2xx')
|
||||
return Response.json({ voices: [], recommended: {} })
|
||||
}
|
||||
|
||||
const data = await res.json().catch(() => ({})) as { voices?: unknown[] }
|
||||
return Response.json({
|
||||
voices: Array.isArray(data.voices) ? data.voices : [],
|
||||
// STREAMING_TTS_UPSTREAM is not part of DEFAULT_TTS_VOICES (which
|
||||
// keys on LLM_ROUTER_CONFIG model ids). Future enhancement: add a
|
||||
// separate streaming-recommended map if locale-based auto-pick
|
||||
// matters for streaming too.
|
||||
recommended: {},
|
||||
})
|
||||
}
|
||||
|
||||
async function handleListTTSModels(_c: Context<HonoEnv>) {
|
||||
// Mirror the chat provider: expose a single 'auto' routing alias instead
|
||||
// of the concrete DEFAULT_TTS_MODEL id. Keeps clients insulated from
|
||||
@@ -652,11 +711,27 @@ export function createV1CompletionsRoutes(
|
||||
// 60 requests per minute per user for LLM completions
|
||||
const completionsRateLimit = rateLimiter({ max: 60, windowSec: 60, metrics: rateLimitMetrics, routeLabel: 'openai.completions' })
|
||||
|
||||
return new Hono<HonoEnv>()
|
||||
// OpenAI-compatible surface (mounted at /api/v1/openai). Only routes that
|
||||
// mirror an actual OpenAI public endpoint belong here. Audio used to live
|
||||
// under this prefix too, but the `/audio/voices` listing endpoint isn't a
|
||||
// real OpenAI route and the streaming TTS protocol has nothing to do with
|
||||
// OpenAI — keeping them here mislabelled the surface, so audio now mounts
|
||||
// at /api/v1/audio (see `audioRoutes` below).
|
||||
const openaiRoutes = new Hono<HonoEnv>()
|
||||
.use('*', authGuard)
|
||||
.post('/chat/completions', completionsRateLimit, chatGuard, handleCompletion)
|
||||
.post('/chat/completion', completionsRateLimit, chatGuard, handleCompletion)
|
||||
.post('/audio/speech', ttsGuard, handleTTS)
|
||||
.get('/audio/voices', handleListVoices)
|
||||
.get('/audio/models', handleListTTSModels)
|
||||
|
||||
// AIRI audio surface (mounted at /api/v1/audio). Lives outside /openai/ so
|
||||
// the `/voices`, `/voices/streaming`, and `/models` extensions aren't
|
||||
// misread as OpenAI-compatible. `/audio/speech/ws` is registered
|
||||
// separately in app.ts because it needs the WebSocket upgrade middleware.
|
||||
const audioRoutes = new Hono<HonoEnv>()
|
||||
.use('*', authGuard)
|
||||
.post('/speech', ttsGuard, handleTTS)
|
||||
.get('/voices', handleListVoices)
|
||||
.get('/voices/streaming', handleListStreamingVoices)
|
||||
.get('/models', handleListTTSModels)
|
||||
|
||||
return { openaiRoutes, audioRoutes }
|
||||
}
|
||||
|
||||
@@ -8,7 +8,7 @@ import type { HonoEnv } from '../../../types/hono'
|
||||
import { Hono } from 'hono'
|
||||
import { afterAll, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
import { createV1CompletionsRoutes } from '.'
|
||||
import { createV1Routes } from '.'
|
||||
import { ApiError } from '../../../utils/error'
|
||||
|
||||
// --- Mock helpers ---
|
||||
@@ -122,7 +122,7 @@ function createTestApp(
|
||||
ttsMeter?: ReturnType<typeof createMockTtsMeter>,
|
||||
llmRouter?: LlmRouterService,
|
||||
) {
|
||||
const routes = createV1CompletionsRoutes(
|
||||
const { openaiRoutes, audioRoutes } = createV1Routes(
|
||||
fluxService,
|
||||
billingService ?? createMockBillingService(),
|
||||
configKV,
|
||||
@@ -153,7 +153,12 @@ function createTestApp(
|
||||
await next()
|
||||
})
|
||||
|
||||
app.route('/api/v1/openai', routes)
|
||||
// Mounting mirrors production (see app.ts): chat completions under
|
||||
// `/api/v1/openai`, audio under `/api/v1/audio`. Test request URLs were
|
||||
// batch-migrated from the legacy `/api/v1/openai/audio/*` prefix when the
|
||||
// audio surface was split out of the OpenAI-compat namespace.
|
||||
app.route('/api/v1/openai', openaiRoutes)
|
||||
app.route('/api/v1/audio', audioRoutes)
|
||||
return app
|
||||
}
|
||||
|
||||
@@ -492,7 +497,38 @@ describe('v1CompletionsRoutes', () => {
|
||||
})
|
||||
})
|
||||
|
||||
describe('pOST /api/v1/openai/audio/speech', () => {
|
||||
describe('legacy audio paths under /openai/', () => {
|
||||
// Audio used to live at /api/v1/openai/audio/*. After the refactor it
|
||||
// moved to /api/v1/audio/*; these are kept as 404 sentinels so a
|
||||
// future accidental re-mount under the old prefix is caught by tests.
|
||||
// Codex review LOW #6.
|
||||
it('returns 404 for /api/v1/openai/audio/speech (moved to /api/v1/audio/speech)', async () => {
|
||||
const app = createTestApp(createMockFluxService(), createMockConfigKV())
|
||||
const res = await app.fetch(
|
||||
new Request('http://localhost/api/v1/openai/audio/speech', { method: 'POST' }),
|
||||
{ user: testUser } as any,
|
||||
)
|
||||
expect(res.status).toBe(404)
|
||||
})
|
||||
it('returns 404 for /api/v1/openai/audio/voices', async () => {
|
||||
const app = createTestApp(createMockFluxService(), createMockConfigKV())
|
||||
const res = await app.fetch(
|
||||
new Request('http://localhost/api/v1/openai/audio/voices', { method: 'GET' }),
|
||||
{ user: testUser } as any,
|
||||
)
|
||||
expect(res.status).toBe(404)
|
||||
})
|
||||
it('returns 404 for /api/v1/openai/audio/models', async () => {
|
||||
const app = createTestApp(createMockFluxService(), createMockConfigKV())
|
||||
const res = await app.fetch(
|
||||
new Request('http://localhost/api/v1/openai/audio/models', { method: 'GET' }),
|
||||
{ user: testUser } as any,
|
||||
)
|
||||
expect(res.status).toBe(404)
|
||||
})
|
||||
})
|
||||
|
||||
describe('pOST /api/v1/audio/speech', () => {
|
||||
it('should proxy TTS request to upstream with resolved model', async () => {
|
||||
globalThis.fetch = vi.fn(async () => new Response(new Uint8Array([1]), {
|
||||
status: 200,
|
||||
@@ -505,7 +541,7 @@ describe('v1CompletionsRoutes', () => {
|
||||
)
|
||||
|
||||
await app.fetch(
|
||||
new Request('http://localhost/api/v1/openai/audio/speech', {
|
||||
new Request('http://localhost/api/v1/audio/speech', {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ model: 'auto', input: 'test', voice: 'alloy' }),
|
||||
@@ -532,7 +568,7 @@ describe('v1CompletionsRoutes', () => {
|
||||
const app = createTestApp(createMockFluxService(), createMockConfigKV(), billingService)
|
||||
|
||||
await app.fetch(
|
||||
new Request('http://localhost/api/v1/openai/audio/speech', {
|
||||
new Request('http://localhost/api/v1/audio/speech', {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ model: 'auto', input: 'hello', voice: 'alloy' }),
|
||||
@@ -554,7 +590,7 @@ describe('v1CompletionsRoutes', () => {
|
||||
const app = createTestApp(createMockFluxService(), createMockConfigKV(), billingService, undefined, undefined, llmRouter)
|
||||
|
||||
const res = await app.fetch(
|
||||
new Request('http://localhost/api/v1/openai/audio/speech', {
|
||||
new Request('http://localhost/api/v1/audio/speech', {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ model: 'auto', input: 'hello', voice: 'alloy' }),
|
||||
@@ -573,7 +609,7 @@ describe('v1CompletionsRoutes', () => {
|
||||
)
|
||||
|
||||
const res = await app.fetch(
|
||||
new Request('http://localhost/api/v1/openai/audio/speech', {
|
||||
new Request('http://localhost/api/v1/audio/speech', {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ model: 'auto', input: 'hello', voice: 'alloy' }),
|
||||
@@ -593,7 +629,7 @@ describe('v1CompletionsRoutes', () => {
|
||||
const app = createTestApp(createMockFluxService(), createMockConfigKV(), billingService)
|
||||
|
||||
await app.fetch(
|
||||
new Request('http://localhost/api/v1/openai/audio/speech', {
|
||||
new Request('http://localhost/api/v1/audio/speech', {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ model: 'auto', input: '', voice: 'alloy' }),
|
||||
@@ -618,7 +654,7 @@ describe('v1CompletionsRoutes', () => {
|
||||
const app = createTestApp(createMockFluxService(), createMockConfigKV(), billingService, undefined, ttsMeter)
|
||||
|
||||
await app.fetch(
|
||||
new Request('http://localhost/api/v1/openai/audio/speech', {
|
||||
new Request('http://localhost/api/v1/audio/speech', {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ model: 'auto', input: longInput, voice: 'alloy' }),
|
||||
@@ -634,7 +670,7 @@ describe('v1CompletionsRoutes', () => {
|
||||
it('should return 401 when unauthenticated', async () => {
|
||||
const app = createTestApp(createMockFluxService(), createMockConfigKV())
|
||||
|
||||
const res = await app.request('/api/v1/openai/audio/voices', { method: 'GET' })
|
||||
const res = await app.request('/api/v1/audio/voices', { method: 'GET' })
|
||||
expect(res.status).toBe(401)
|
||||
})
|
||||
|
||||
@@ -671,7 +707,7 @@ describe('v1CompletionsRoutes', () => {
|
||||
)
|
||||
|
||||
const res = await app.fetch(
|
||||
new Request('http://localhost/api/v1/openai/audio/speech', {
|
||||
new Request('http://localhost/api/v1/audio/speech', {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ model: 'auto', input: 'hi', voice: 'en-US-AvaMultilingualNeural' }),
|
||||
@@ -699,7 +735,7 @@ describe('v1CompletionsRoutes', () => {
|
||||
const app = createTestApp(createMockFluxService(), createMockConfigKV(), undefined, undefined, undefined, llmRouter)
|
||||
|
||||
const res = await app.fetch(
|
||||
new Request('http://localhost/api/v1/openai/audio/speech', {
|
||||
new Request('http://localhost/api/v1/audio/speech', {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ model: 'auto', input: 'hi', voice: 'alloy' }),
|
||||
@@ -710,7 +746,7 @@ describe('v1CompletionsRoutes', () => {
|
||||
})
|
||||
})
|
||||
|
||||
describe('gET /api/v1/openai/audio/models', () => {
|
||||
describe('gET /api/v1/audio/models', () => {
|
||||
it('exposes only the auto routing alias regardless of DEFAULT_TTS_MODEL', async () => {
|
||||
const app = createTestApp(
|
||||
createMockFluxService(),
|
||||
@@ -718,7 +754,7 @@ describe('v1CompletionsRoutes', () => {
|
||||
)
|
||||
|
||||
const res = await app.fetch(
|
||||
new Request('http://localhost/api/v1/openai/audio/models', { method: 'GET' }),
|
||||
new Request('http://localhost/api/v1/audio/models', { method: 'GET' }),
|
||||
{ user: testUser } as any,
|
||||
)
|
||||
|
||||
@@ -730,12 +766,12 @@ describe('v1CompletionsRoutes', () => {
|
||||
it('should return 401 when unauthenticated', async () => {
|
||||
const app = createTestApp(createMockFluxService(), createMockConfigKV())
|
||||
|
||||
const res = await app.request('/api/v1/openai/audio/models', { method: 'GET' })
|
||||
const res = await app.request('/api/v1/audio/models', { method: 'GET' })
|
||||
expect(res.status).toBe(401)
|
||||
})
|
||||
})
|
||||
|
||||
describe('gET /api/v1/openai/audio/voices', () => {
|
||||
describe('gET /api/v1/audio/voices', () => {
|
||||
it('returns the adapter voice catalog plus recommended map for the resolved model', async () => {
|
||||
const voices = [
|
||||
{ id: 'en-US-JennyNeural', name: 'Jenny', provider: 'azure', locale: 'en-US', gender: 'Female' },
|
||||
@@ -751,7 +787,7 @@ describe('v1CompletionsRoutes', () => {
|
||||
const app = createTestApp(createMockFluxService(), configKV, undefined, undefined, undefined, llmRouter)
|
||||
|
||||
const res = await app.fetch(
|
||||
new Request('http://localhost/api/v1/openai/audio/voices', { method: 'GET' }),
|
||||
new Request('http://localhost/api/v1/audio/voices', { method: 'GET' }),
|
||||
{ user: testUser } as any,
|
||||
)
|
||||
|
||||
@@ -769,7 +805,7 @@ describe('v1CompletionsRoutes', () => {
|
||||
|
||||
const app = createTestApp(createMockFluxService(), createMockConfigKV(), undefined, undefined, undefined, llmRouter)
|
||||
|
||||
await app.fetch(new Request('http://localhost/api/v1/openai/audio/voices?model=alibaba/cosyvoice-v1'), { user: testUser } as any)
|
||||
await app.fetch(new Request('http://localhost/api/v1/audio/voices?model=alibaba/cosyvoice-v1'), { user: testUser } as any)
|
||||
expect(llmRouter.listTtsVoices).toHaveBeenCalledWith('alibaba/cosyvoice-v1')
|
||||
})
|
||||
|
||||
@@ -781,7 +817,7 @@ describe('v1CompletionsRoutes', () => {
|
||||
|
||||
const app = createTestApp(createMockFluxService(), configKV, undefined, undefined, undefined, llmRouter)
|
||||
|
||||
await app.fetch(new Request('http://localhost/api/v1/openai/audio/voices?model=auto'), { user: testUser } as any)
|
||||
await app.fetch(new Request('http://localhost/api/v1/audio/voices?model=auto'), { user: testUser } as any)
|
||||
expect(llmRouter.listTtsVoices).toHaveBeenCalledWith('microsoft/v1')
|
||||
})
|
||||
})
|
||||
|
||||
+191
@@ -0,0 +1,191 @@
|
||||
<script setup lang="ts">
|
||||
import {
|
||||
ProviderSettingsContainer,
|
||||
ProviderSettingsLayout,
|
||||
SpeechPlayground,
|
||||
} from '@proj-airi/stage-ui/components'
|
||||
import { streamingSynthesize } from '@proj-airi/stage-ui/libs'
|
||||
import { useAuthStore } from '@proj-airi/stage-ui/stores/auth'
|
||||
import { useSpeechStore } from '@proj-airi/stage-ui/stores/modules/speech'
|
||||
import { useProvidersStore } from '@proj-airi/stage-ui/stores/providers'
|
||||
import { Callout, ComboboxSelect } from '@proj-airi/ui'
|
||||
import { storeToRefs } from 'pinia'
|
||||
import { computed, onMounted, ref, watch } from 'vue'
|
||||
import { useI18n } from 'vue-i18n'
|
||||
import { useRouter } from 'vue-router'
|
||||
|
||||
const router = useRouter()
|
||||
const { t } = useI18n()
|
||||
const authStore = useAuthStore()
|
||||
const providersStore = useProvidersStore()
|
||||
const speechStore = useSpeechStore()
|
||||
const { isAuthenticated, credits, needsLogin } = storeToRefs(authStore)
|
||||
|
||||
const providerId = 'official-provider-speech-streaming'
|
||||
const providerMetadata = computed(() => providersStore.getProviderMetadata(providerId))
|
||||
const defaultModel = 'volcengine/seed-tts-2.0'
|
||||
|
||||
const providerConfig = computed(() => providersStore.getProviderConfig(providerId))
|
||||
|
||||
// Model picker. Pulled from the provider's `extraMethods.listModels` —
|
||||
// today that's two hard-coded Volcengine variants, but the picker stays
|
||||
// generic so adding ICL / other backends later doesn't need UI changes.
|
||||
const providerModels = computed(() => providersStore.getModelsForProvider(providerId))
|
||||
const modelsLoading = computed(() => providersStore.isLoadingModels[providerId] || false)
|
||||
const model = computed({
|
||||
get(): string {
|
||||
return (providerConfig.value?.model as string | undefined) ?? defaultModel
|
||||
},
|
||||
set(val: string) {
|
||||
providerConfig.value.model = val
|
||||
},
|
||||
})
|
||||
const modelOptions = computed(() => providerModels.value.map(m => ({ label: m.name, value: m.id })))
|
||||
|
||||
const availableVoices = computed(() => speechStore.availableVoices[providerId] || [])
|
||||
const voicesLoading = ref(false)
|
||||
|
||||
async function loadVoices() {
|
||||
voicesLoading.value = true
|
||||
try {
|
||||
await speechStore.loadVoicesForProvider(providerId)
|
||||
}
|
||||
finally {
|
||||
voicesLoading.value = false
|
||||
}
|
||||
}
|
||||
|
||||
onMounted(async () => {
|
||||
await providersStore.fetchModelsForProvider(providerId)
|
||||
if (!providerConfig.value.model)
|
||||
providerConfig.value.model = defaultModel
|
||||
await loadVoices()
|
||||
})
|
||||
|
||||
// Reload voices when the model variant changes. The streaming provider
|
||||
// shares one voice catalogue across model variants (both Seed-TTS 2.0 and
|
||||
// 1.0 expose the same `zh_female_*` ids), so this is mostly a refresh —
|
||||
// but it keeps the flow consistent with provider pages where the variant
|
||||
// actually does change the catalogue.
|
||||
watch(model, async () => {
|
||||
await loadVoices()
|
||||
})
|
||||
|
||||
// Synthesize via the streaming session helper. The page uses the SAME
|
||||
// transport the runtime pipeline uses (ws → apps/server proxy → unspeech
|
||||
// bridge → Volcengine v3 bidirectional) so the preview faithfully
|
||||
// represents what the user hears in actual chat. The session is opened
|
||||
// per-preview because there's no LLM token stream here — we just send
|
||||
// one `text` frame containing the static preview prompt.
|
||||
async function handleGenerateSpeech(input: string, voiceId: string, _useSSML: boolean): Promise<ArrayBuffer> {
|
||||
const requestedModel = model.value || defaultModel
|
||||
// `model` looks like `volcengine/seed-tts-2.0`. The trailing path is
|
||||
// forwarded as Volcengine's `api_resource_id` so the upstream knows
|
||||
// which model variant to use; matches the wiring in `Stage.vue`.
|
||||
const apiResourceId = requestedModel.includes('/') ? requestedModel.split('/', 2)[1] : 'seed-tts-2.0'
|
||||
const result = await streamingSynthesize({
|
||||
model: requestedModel,
|
||||
voice: voiceId,
|
||||
input,
|
||||
extraBody: {
|
||||
api_resource_id: apiResourceId,
|
||||
audio: { sample_rate: 24000, bit_rate: 64000 },
|
||||
},
|
||||
})
|
||||
return result.audio
|
||||
}
|
||||
|
||||
function handleLogin() {
|
||||
needsLogin.value = true
|
||||
}
|
||||
</script>
|
||||
|
||||
<template>
|
||||
<ProviderSettingsLayout
|
||||
v-if="providerMetadata"
|
||||
:provider-name="providerMetadata?.localizedName"
|
||||
:provider-icon-color="providerMetadata?.iconColor"
|
||||
:on-back="() => router.back()"
|
||||
>
|
||||
<ProviderSettingsContainer>
|
||||
<div v-if="!isAuthenticated" flex flex-col gap-4>
|
||||
<Callout theme="primary">
|
||||
<template #label>
|
||||
{{ t('settings.pages.providers.provider.official.speech-streaming-title') }}
|
||||
</template>
|
||||
<div flex flex-col gap-3>
|
||||
<p>{{ t('settings.dialogs.onboarding.loginPrompt') }}</p>
|
||||
<button
|
||||
type="button"
|
||||
class="w-fit rounded-lg bg-primary-500 px-4 py-2 text-white transition-colors active:scale-95 hover:bg-primary-600"
|
||||
@click="handleLogin"
|
||||
>
|
||||
{{ t('settings.dialogs.onboarding.loginAction') }}
|
||||
</button>
|
||||
</div>
|
||||
</Callout>
|
||||
</div>
|
||||
|
||||
<div v-else flex flex-col gap-6>
|
||||
<div class="rounded-xl bg-neutral-100/50 p-6 backdrop-blur-sm dark:bg-neutral-800/50">
|
||||
<div flex items-center justify-between>
|
||||
<div flex flex-col gap-1>
|
||||
<span text="sm neutral-500 dark:neutral-400 font-medium uppercase tracking-wider">
|
||||
{{ t('settings.dialogs.onboarding.flux') }}
|
||||
</span>
|
||||
<span text="3xl font-bold text-primary-600 dark:text-primary-400">
|
||||
{{ credits }}
|
||||
</span>
|
||||
</div>
|
||||
<button
|
||||
type="button"
|
||||
class="rounded-full bg-primary-500/10 px-6 py-2 text-sm text-primary-600 font-semibold transition-all dark:bg-primary-400/10 hover:bg-primary-500 dark:text-primary-400 hover:text-white dark:hover:bg-primary-400 dark:hover:text-neutral-900"
|
||||
@click="router.push('/settings/flux')"
|
||||
>
|
||||
{{ t('settings.dialogs.onboarding.buyFlux') }}
|
||||
</button>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div class="border border-neutral-200/50 rounded-xl p-4 dark:border-neutral-700/50">
|
||||
<div flex items-center gap-3>
|
||||
<div class="h-2 w-2 animate-pulse rounded-full bg-green-500" />
|
||||
<span text="sm neutral-600 dark:neutral-300">
|
||||
{{ t('settings.pages.providers.provider.common.status.valid') }}
|
||||
</span>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div class="space-y-3">
|
||||
<Callout label="Model">
|
||||
<p>Pick the streaming TTS model variant. All variants share the same voice catalogue today.</p>
|
||||
</Callout>
|
||||
<ComboboxSelect
|
||||
v-model="model"
|
||||
:options="modelOptions"
|
||||
:disabled="modelsLoading"
|
||||
placeholder="Choose a model..."
|
||||
/>
|
||||
</div>
|
||||
|
||||
<SpeechPlayground
|
||||
:available-voices="availableVoices"
|
||||
:generate-speech="handleGenerateSpeech"
|
||||
:api-key-configured="true"
|
||||
:voices-loading="voicesLoading"
|
||||
default-text="你好，这是流式语音合成的试听样例。"
|
||||
/>
|
||||
</div>
|
||||
</ProviderSettingsContainer>
|
||||
</ProviderSettingsLayout>
|
||||
<div v-else class="p-8 text-center text-neutral-500">
|
||||
Provider is not available.
|
||||
</div>
|
||||
</template>
|
||||
|
||||
<route lang="yaml">
|
||||
meta:
|
||||
layout: settings
|
||||
stageTransition:
|
||||
name: slide
|
||||
</route>
|
||||
@@ -5,6 +5,7 @@ import type { SpeechProviderWithExtraOptions } from '@xsai-ext/providers/utils'
|
||||
import type { UnElevenLabsOptions } from 'unspeech'
|
||||
|
||||
import type { EmotionPayload } from '../../constants/emotions'
|
||||
import type { SpeechTransport, StageTtsSession, StreamingSessionSnapshot } from '../../libs/speech/tts-session'
|
||||
|
||||
import { sleep } from '@moeru/std'
|
||||
import { createLive2DLipSync } from '@proj-airi/model-driver-lipsync'
|
||||
@@ -28,9 +29,12 @@ import { useDuckDb } from '../../composables/use-duck-db'
|
||||
import { useIOTraceBridge } from '../../composables/use-io-trace-bridge'
|
||||
import { initIOTracer } from '../../composables/use-io-tracer'
|
||||
import { useSpeechPipelineAnalytics } from '../../composables/use-speech-pipeline-analytics'
|
||||
import { llmInferenceEndToken } from '../../constants'
|
||||
import { Emotion, EMOTION_EmotionMotionName_value, EMOTION_VRMExpressionName_value, EmotionThinkMotionName } from '../../constants/emotions'
|
||||
import { getDefinedProvider } from '../../libs/providers/providers'
|
||||
import { OFFICIAL_SPEECH_STREAMING_PROVIDER_ID } from '../../libs/providers/providers/official'
|
||||
import { streamingSynthesize } from '../../libs/speech/streaming-session'
|
||||
import { createStageTtsSession } from '../../libs/speech/tts-session'
|
||||
import { useAudioContext, useSpeakingStore } from '../../stores/audio'
|
||||
import { useBackgroundStore } from '../../stores/background'
|
||||
import { useChatOrchestratorStore } from '../../stores/chat'
|
||||
@@ -40,6 +44,7 @@ import { useSpeechStore } from '../../stores/modules/speech'
|
||||
import { useProvidersStore } from '../../stores/providers'
|
||||
import { useSettings } from '../../stores/settings'
|
||||
import { useSpeechRuntimeStore } from '../../stores/speech-runtime'
|
||||
|
||||
const props = withDefaults(defineProps<{
|
||||
paused?: boolean
|
||||
focusAt: { x: number, y: number }
|
||||
@@ -300,6 +305,23 @@ const speechPipeline = createSpeechPipeline<AudioBuffer>({
|
||||
if (!activeSpeechProvider.value)
|
||||
return null
|
||||
|
||||
// Streaming provider must NEVER reach this per-segment callback. The
|
||||
// streaming code path opens its own ws at `onBeforeMessageComposed`
|
||||
// and bypasses speech-pipeline entirely. If we got here while the
|
||||
// streaming provider is active, the open path failed (most often:
|
||||
// voice catalog hadn't finished loading when the user sent the
|
||||
// message). The old fallback would silently re-open a fresh ws per
|
||||
// segment — exactly the behavior the refactor is meant to delete.
|
||||
// Codex review MEDIUM #3: refuse loudly instead.
|
||||
if (resolveSpeechTransport(activeSpeechProvider.value) === 'bidirectional-ws') {
|
||||
console.warn('[Speech Pipeline] bidirectional-ws provider reached per-segment fallback', {
|
||||
reason: 'streaming session was not opened at intent start (voice unset?)',
|
||||
provider: activeSpeechProvider.value,
|
||||
segment: request.text?.slice(0, 40),
|
||||
})
|
||||
return null
|
||||
}
|
||||
|
||||
const provider = await providersStore.getProviderInstance(activeSpeechProvider.value) as SpeechProviderWithExtraOptions<string, UnElevenLabsOptions>
|
||||
if (!provider) {
|
||||
console.error('Failed to initialize speech provider')
|
||||
@@ -361,38 +383,14 @@ const speechPipeline = createSpeechPipeline<AudioBuffer>({
|
||||
: request.text
|
||||
|
||||
try {
|
||||
// Streaming TTS path: open a WebSocket to apps/server which proxies to
|
||||
// unspeech bidirectional. Server-side starts streaming audio as the
|
||||
// upstream synthesises, so first-byte latency is significantly lower
|
||||
// than the buffered REST round-trip. We still wait for `session.finished`
|
||||
// here and decode a single AudioBuffer per segment because the existing
|
||||
// playbackManager is AudioBuffer-shaped; a future refactor can chunk on
|
||||
// `sentence.end` events for true play-as-you-receive.
|
||||
let res: ArrayBuffer | null = null
|
||||
if (activeSpeechProvider.value === OFFICIAL_SPEECH_STREAMING_PROVIDER_ID) {
|
||||
// Derive the upstream resource id from the routed model. `model`
|
||||
// looks like `volcengine/seed-tts-2.0`; the bit after the slash
|
||||
// becomes `api_resource_id` for unspeech → Volcengine X-Api-Resource-Id.
|
||||
const apiResourceId = model.includes('/') ? model.split('/', 2)[1] : 'seed-tts-2.0'
|
||||
const result = await streamingSynthesize({
|
||||
model,
|
||||
voice: voice.id,
|
||||
input,
|
||||
signal,
|
||||
extraBody: {
|
||||
api_resource_id: apiResourceId,
|
||||
audio: { sample_rate: 24000, bit_rate: 64000 },
|
||||
},
|
||||
})
|
||||
res = result.audio
|
||||
}
|
||||
else {
|
||||
res = await generateSpeech({
|
||||
...provider.speech(model, providerConfig),
|
||||
input,
|
||||
voice: voice.id,
|
||||
})
|
||||
}
|
||||
// Non-streaming providers only: synth via REST. Streaming provider
|
||||
// was already early-returned above; it owns its own ws path opened
|
||||
// in `onBeforeMessageComposed`.
|
||||
const res = await generateSpeech({
|
||||
...provider.speech(model, providerConfig),
|
||||
input,
|
||||
voice: voice.id,
|
||||
})
|
||||
|
||||
if (signal.aborted || !res || res.byteLength === 0)
|
||||
return null
|
||||
@@ -547,7 +545,82 @@ function setupAnalyser() {
|
||||
}
|
||||
}
|
||||
|
||||
let currentChatIntent: ReturnType<typeof speechRuntimeStore.openIntent> | null = null
|
||||
// One TTS session per LLM intent. The active provider determines which
|
||||
// adapter `createStageTtsSession` returns: the segmenter-based adapter for
|
||||
// every non-streaming provider, or the bidirectional WebSocket adapter
|
||||
// for the official streaming provider. Stage.vue intentionally does NOT
|
||||
// branch on provider id anywhere below — the factory is the single
|
||||
// decision point. See `packages/stage-ui/src/libs/speech/tts-session.ts`.
|
||||
let currentSession: StageTtsSession | null = null
|
||||
|
||||
function buildStreamingSnapshot(): StreamingSessionSnapshot | null {
|
||||
// Snapshotted once per session, so a mid-session provider/voice swap
|
||||
// does not corrupt an in-flight session — the watcher below detects
|
||||
// changes and tears down explicitly. Returns `null` when streaming
|
||||
// can't be opened (no voice picked, no audioContext, no model);
|
||||
// `createStageTtsSession` falls back to the segmenter adapter in that
|
||||
// case, which is the right behaviour for the rest of the providers too.
|
||||
const voiceId = activeSpeechVoice.value?.id
|
||||
if (!voiceId)
|
||||
return null
|
||||
const sessionModel = (activeSpeechModel.value as string | undefined) || 'volcengine/seed-tts-2.0'
|
||||
const apiResourceId = sessionModel.includes('/') ? sessionModel.split('/', 2)[1] : 'seed-tts-2.0'
|
||||
// TTS 2.0 / ICL 2.0 ship subtitles asynchronously relative to audio
|
||||
// (per the wire spec), so chunk-on-sentence-end would drop frames.
|
||||
// Buffer the entire session and decode at session.finished instead.
|
||||
const bufferEntireSession = apiResourceId.startsWith('seed-tts-2.0') || apiResourceId.startsWith('seed-icl-2.0')
|
||||
return {
|
||||
model: sessionModel,
|
||||
voice: voiceId,
|
||||
bufferEntireSession,
|
||||
extraBody: {
|
||||
api_resource_id: apiResourceId,
|
||||
audio: { sample_rate: 24000, bit_rate: 64000 },
|
||||
},
|
||||
ownerId: activeCardId.value,
|
||||
onImmediateSpecial: playSpecialToken,
|
||||
}
|
||||
}
|
||||
|
||||
function resolveSpeechTransport(providerId: string | null | undefined): SpeechTransport | undefined {
|
||||
if (!providerId)
|
||||
return undefined
|
||||
// Read straight from the unified ProviderDefinition registry — keeps the
|
||||
// factory transport-agnostic and lets a new provider opt into streaming
|
||||
// by setting `capabilities.speech.transport: 'bidirectional-ws'` in its
|
||||
// own `defineProvider` call (no Stage / factory edits needed).
|
||||
return getDefinedProvider(providerId)?.capabilities?.speech?.transport
|
||||
}
|
||||
|
||||
function openTtsSession(): StageTtsSession {
|
||||
return createStageTtsSession<AudioBuffer>({
|
||||
transport: resolveSpeechTransport(activeSpeechProvider.value),
|
||||
streaming: buildStreamingSnapshot,
|
||||
audioContext,
|
||||
playbackManager,
|
||||
openIntent: opts => speechRuntimeStore.openIntent(opts),
|
||||
intentOptions: () => ({
|
||||
ownerId: activeCardId.value,
|
||||
priority: 'normal',
|
||||
behavior: 'queue',
|
||||
}),
|
||||
hooks: {
|
||||
onError: (err) => {
|
||||
console.error('[Speech Pipeline] streaming session error', {
|
||||
provider: activeSpeechProvider.value,
|
||||
model: activeSpeechModel.value,
|
||||
error: err,
|
||||
})
|
||||
if (currentSession?.intentId.startsWith('stream-'))
|
||||
currentSession = null
|
||||
},
|
||||
onDone: () => {
|
||||
if (currentSession?.intentId.startsWith('stream-'))
|
||||
currentSession = null
|
||||
},
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
chatHookCleanups.push(onBeforeMessageComposed(async () => {
|
||||
playbackManager.stopAll('new-message')
|
||||
@@ -571,16 +644,8 @@ chatHookCleanups.push(onBeforeMessageComposed(async () => {
|
||||
console.warn('[Stage] Failed to post present reset (channel may be closed)', { error })
|
||||
}
|
||||
|
||||
if (currentChatIntent) {
|
||||
currentChatIntent.cancel('new-message')
|
||||
currentChatIntent = null
|
||||
}
|
||||
|
||||
currentChatIntent = speechRuntimeStore.openIntent({
|
||||
ownerId: activeCardId.value,
|
||||
priority: 'normal',
|
||||
behavior: 'queue',
|
||||
})
|
||||
currentSession?.cancel('new-message')
|
||||
currentSession = openTtsSession()
|
||||
}))
|
||||
|
||||
chatHookCleanups.push(onBeforeSend(async () => {
|
||||
@@ -588,21 +653,26 @@ chatHookCleanups.push(onBeforeSend(async () => {
|
||||
}))
|
||||
|
||||
chatHookCleanups.push(onTokenLiteral(async (literal) => {
|
||||
currentChatIntent?.writeLiteral(literal)
|
||||
currentSession?.appendText(literal)
|
||||
}))
|
||||
|
||||
chatHookCleanups.push(onTokenSpecial(async (special) => {
|
||||
// console.debug('Stage received special token:', special)
|
||||
currentChatIntent?.writeSpecial(special)
|
||||
currentSession?.appendSpecial(special)
|
||||
}))
|
||||
|
||||
chatHookCleanups.push(onStreamEnd(async () => {
|
||||
currentChatIntent?.writeFlush()
|
||||
currentSession?.finishInput()
|
||||
}))
|
||||
|
||||
chatHookCleanups.push(onAssistantResponseEnd(async (_message) => {
|
||||
currentChatIntent?.end()
|
||||
currentChatIntent = null
|
||||
currentSession?.end()
|
||||
// Streaming sessions null-out via the onDone hook; segmenter sessions
|
||||
// stay around until the next `onBeforeMessageComposed` cancels them
|
||||
// (the segmenter pipeline's IntentHandle.end is idempotent and
|
||||
// ResourceMessages still arrive after end() — clearing here would
|
||||
// race with the pipeline's own cleanup). Keep the ref pointing at
|
||||
// the just-ended session; it costs nothing and the next message
|
||||
// replaces it.
|
||||
// const res = await embed({
|
||||
// ...transformersProvider.embed('Xenova/nomic-embed-text-v1'),
|
||||
// input: message,
|
||||
@@ -611,6 +681,32 @@ chatHookCleanups.push(onAssistantResponseEnd(async (_message) => {
|
||||
// await db.value?.execute(`INSERT INTO memory_test (vec) VALUES (${JSON.stringify(res.embedding)});`)
|
||||
}))
|
||||
|
||||
// Mid-session provider / voice / model swaps would otherwise keep feeding
|
||||
// tokens to the OLD adapter (segmenter for the new provider, or stale ws
|
||||
// for the streaming provider). Cancel the active session so the next LLM
|
||||
// token after the swap falls through `currentSession?.` cleanly (silent
|
||||
// drop is acceptable — we don't try to fork-replay text into a new
|
||||
// adapter with potentially different voice/model).
|
||||
watch(
|
||||
[activeSpeechProvider, () => activeSpeechVoice.value?.id, activeSpeechModel],
|
||||
([provider, voiceId, model], [prevProvider, prevVoiceId, prevModel]) => {
|
||||
if (!currentSession)
|
||||
return
|
||||
if (provider === prevProvider && voiceId === prevVoiceId && model === prevModel)
|
||||
return
|
||||
console.warn('[Speech Pipeline] provider/voice/model changed mid-session, tearing down', {
|
||||
provider,
|
||||
prevProvider,
|
||||
voiceId,
|
||||
prevVoiceId,
|
||||
model,
|
||||
prevModel,
|
||||
})
|
||||
currentSession.cancel('provider-or-voice-changed')
|
||||
currentSession = null
|
||||
},
|
||||
)
|
||||
|
||||
// Resume audio context on first user interaction (browser requirement)
|
||||
let audioContextResumed = false
|
||||
function resumeAudioContextOnInteraction() {
|
||||
@@ -716,6 +812,14 @@ onUnmounted(() => {
|
||||
resetLive2dLipSync()
|
||||
chatHookCleanups.forEach(dispose => dispose?.())
|
||||
viewUpdateCleanups.forEach(dispose => dispose?.())
|
||||
// Tear down any in-flight TTS session (segmenter or streaming) and
|
||||
// drain playback. Without this, a still-open streaming ws keeps
|
||||
// feeding sentences into a playbackManager whose listeners still
|
||||
// mutate component refs (caption / nowSpeaking). Codex review: HIGH
|
||||
// #1 + MEDIUM #5.
|
||||
currentSession?.cancel('unmount')
|
||||
currentSession = null
|
||||
playbackManager.stopAll('unmount')
|
||||
})
|
||||
|
||||
defineExpose({
|
||||
|
||||
@@ -1,4 +1,7 @@
|
||||
export * from './audio/manager'
|
||||
export * from './color-from-element'
|
||||
export * from './providers'
|
||||
export * from './speech/streaming-pipeline'
|
||||
export * from './speech/streaming-session'
|
||||
export * from './speech/tts-session'
|
||||
export * from './zod'
|
||||
|
||||
@@ -8,7 +8,7 @@ import { z } from 'zod'
|
||||
import { getAuthToken } from '../../../../libs/auth'
|
||||
import { SERVER_URL } from '../../../../libs/server'
|
||||
import { defineProvider } from '../registry'
|
||||
import { createOfficialOpenAIProvider, OFFICIAL_ICON, withCredentials } from './shared'
|
||||
import { createOfficialAudioProvider, createOfficialOpenAIProvider, OFFICIAL_ICON, withCredentials } from './shared'
|
||||
|
||||
export const OFFICIAL_SPEECH_PROVIDER_ID = 'official-provider-speech'
|
||||
export const OFFICIAL_SPEECH_STREAMING_PROVIDER_ID = 'official-provider-speech-streaming'
|
||||
@@ -78,7 +78,7 @@ export const providerOfficialSpeech = defineProvider({
|
||||
requiresCredentials: false,
|
||||
createProviderConfig: () => officialConfigSchema,
|
||||
createProvider(_config) {
|
||||
const provider = createOfficialOpenAIProvider()
|
||||
const provider = createOfficialAudioProvider()
|
||||
const originalSpeech = provider.speech.bind(provider)
|
||||
provider.speech = (model: string) => {
|
||||
const result = originalSpeech(model)
|
||||
@@ -90,7 +90,7 @@ export const providerOfficialSpeech = defineProvider({
|
||||
validationRequiredWhen: () => false,
|
||||
extraMethods: {
|
||||
listModels: async (): Promise<ModelInfo[]> => {
|
||||
const res = await globalThis.fetch(`${SERVER_URL}/api/v1/openai/audio/models`, { headers: authHeaders() })
|
||||
const res = await globalThis.fetch(`${SERVER_URL}/api/v1/audio/models`, { headers: authHeaders() })
|
||||
if (!res.ok)
|
||||
return []
|
||||
|
||||
@@ -105,7 +105,7 @@ export const providerOfficialSpeech = defineProvider({
|
||||
}))
|
||||
},
|
||||
listVoices: async (): Promise<VoiceInfo[]> => {
|
||||
const res = await globalThis.fetch(`${SERVER_URL}/api/v1/openai/audio/voices`, { headers: authHeaders() })
|
||||
const res = await globalThis.fetch(`${SERVER_URL}/api/v1/audio/voices`, { headers: authHeaders() })
|
||||
if (!res.ok)
|
||||
return []
|
||||
|
||||
@@ -180,9 +180,20 @@ export const providerOfficialSpeechStreaming = defineProvider({
|
||||
tasks: ['text-to-speech'],
|
||||
icon: OFFICIAL_ICON,
|
||||
requiresCredentials: false,
|
||||
// Mark this provider as speaking the bidirectional ws TTS protocol so the
|
||||
// session adapter (`tts-session.ts`) picks the streaming path without
|
||||
// hard-coding provider id. Default for every other provider is `'rest'`.
|
||||
capabilities: {
|
||||
speech: { transport: 'bidirectional-ws' },
|
||||
},
|
||||
createProviderConfig: () => officialConfigSchema,
|
||||
createProvider(_config) {
|
||||
const provider = createOfficialOpenAIProvider()
|
||||
// Same audio-scoped baseURL as the HTTP speech provider. The streaming
|
||||
// provider does not actually use `.speech()` for synthesis (it goes
|
||||
// through `streamingSynthesize` which opens its own WebSocket), but the
|
||||
// OpenAI-shaped provider instance is still returned so legacy fallback
|
||||
// and feature-detection helpers keep working.
|
||||
const provider = createOfficialAudioProvider()
|
||||
const originalSpeech = provider.speech.bind(provider)
|
||||
provider.speech = (model: string) => {
|
||||
const result = originalSpeech(model)
|
||||
@@ -213,11 +224,14 @@ export const providerOfficialSpeechStreaming = defineProvider({
|
||||
]
|
||||
},
|
||||
listVoices: async (): Promise<VoiceInfo[]> => {
|
||||
// Reuse the HTTP voices endpoint with an explicit `?model=` filter so
|
||||
// the catalog matches Volcengine's bidirectional upstream rather than
|
||||
// whatever DEFAULT_TTS_MODEL points at.
|
||||
// Streaming voices live behind a dedicated endpoint
|
||||
// (`/audio/voices/streaming`) because they come from a separate
|
||||
// configKV entry (`STREAMING_TTS_UPSTREAM`) than the HTTP TTS
|
||||
// `?model=...` lookup. The server proxies to unspeech's
|
||||
// `/api/voices?provider=volcengine`, which ships an embed-time
|
||||
// catalogue without requiring credentials.
|
||||
const res = await globalThis.fetch(
|
||||
`${SERVER_URL}/api/v1/openai/audio/voices?model=volcengine`,
|
||||
`${SERVER_URL}/api/v1/audio/voices/streaming`,
|
||||
{ headers: authHeaders() },
|
||||
)
|
||||
if (!res.ok)
|
||||
|
||||
@@ -23,3 +23,17 @@ export function withCredentials() {
|
||||
export function createOfficialOpenAIProvider() {
|
||||
return createOpenAI('', `${SERVER_URL}/api/v1/openai/`)
|
||||
}
|
||||
|
||||
/**
|
||||
* Provider scoped to the audio surface (`/api/v1/audio/`). The OpenAI helper
|
||||
* (`createOpenAI`) builds upstream URLs as `<baseURL><resource>`, where
|
||||
* `<resource>` is e.g. `audio/speech`. We point baseURL at `/api/v1/` so the
|
||||
* generated URL is `/api/v1/audio/speech` — matching the audio routes
|
||||
* mounted in app.ts after they were split out of `/api/v1/openai/`.
|
||||
*
|
||||
* Returned provider still exposes the OpenAI-shaped `.speech()` API so xsai's
|
||||
* `generateSpeech()` can consume it directly.
|
||||
*/
|
||||
export function createOfficialAudioProvider() {
|
||||
return createOpenAI('', `${SERVER_URL}/api/v1/`)
|
||||
}
|
||||
|
||||
@@ -190,6 +190,26 @@ export interface ProviderDefinition<TConfig extends any = any> {
|
||||
streamOutput: boolean
|
||||
streamInput: boolean
|
||||
}
|
||||
/**
|
||||
* Declares the TTS transport this provider speaks. Drives Stage's TTS
|
||||
* session adapter selection (`@proj-airi/stage-ui/libs/speech/tts-session`):
|
||||
*
|
||||
* - `rest` (default when this whole block is absent): the host opens
|
||||
* a `pipelines-audio` IntentHandle and the provider's `speech()` is
|
||||
* called per-segment by the speech-pipeline `tts()` callback. This
|
||||
* matches every OpenAI-shaped HTTP TTS provider.
|
||||
* - `bidirectional-ws`: the host opens one streaming TTS WebSocket
|
||||
* for the whole LLM intent and forwards raw token chunks without
|
||||
* client-side segmentation. The provider's `speech()` is unused
|
||||
* for synthesis on this path (kept only for legacy fallback).
|
||||
*
|
||||
* Designed so a future provider (ElevenLabs streaming, OpenAI Realtime
|
||||
* Voice, etc.) only needs to set this flag — Stage and the session
|
||||
* factory do not need to know each provider's id.
|
||||
*/
|
||||
speech?: {
|
||||
transport: 'rest' | 'bidirectional-ws'
|
||||
}
|
||||
}
|
||||
/**
|
||||
* When true, hides the "skip chat ping check" checkbox in the UI even
|
||||
|
||||
@@ -0,0 +1,297 @@
|
||||
import type { AddressInfo } from 'node:net'
|
||||
|
||||
import { Buffer } from 'node:buffer'
|
||||
import { createServer } from 'node:http'
|
||||
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import { WebSocketServer } from 'ws'
|
||||
|
||||
import { createStreamingTtsPipeline } from './streaming-pipeline'
|
||||
|
||||
vi.mock('../auth', () => ({
|
||||
getAuthToken: () => 'test-jwt',
|
||||
}))
|
||||
vi.mock('../server', () => ({
|
||||
SERVER_URL: 'http://placeholder',
|
||||
}))
|
||||
|
||||
interface MockServer {
|
||||
url: string
|
||||
receivedFrames: Array<{ kind: 'text' | 'binary', data: string | Buffer }>
|
||||
/** Resolves when the server has observed a `start` frame from the client. */
|
||||
startObserved: Promise<void>
|
||||
stop: () => Promise<void>
|
||||
}
|
||||
|
||||
async function startMockServer(handler: (ws: import('ws').WebSocket) => void): Promise<MockServer> {
|
||||
const receivedFrames: MockServer['receivedFrames'] = []
|
||||
const httpServer = createServer()
|
||||
const wss = new WebSocketServer({ server: httpServer })
|
||||
|
||||
let resolveStartObserved!: () => void
|
||||
const startObserved = new Promise<void>((res) => {
|
||||
resolveStartObserved = res
|
||||
})
|
||||
|
||||
wss.on('connection', (ws) => {
|
||||
ws.on('message', (data, isBinary) => {
|
||||
const buf = Buffer.isBuffer(data) ? data : Buffer.from(data as ArrayBuffer)
|
||||
const decoded = isBinary ? buf : buf.toString('utf8')
|
||||
receivedFrames.push({ kind: isBinary ? 'binary' : 'text', data: isBinary ? buf : (decoded as string) })
|
||||
if (!isBinary) {
|
||||
try {
|
||||
const ev = JSON.parse(decoded as string) as { event?: string }
|
||||
if (ev.event === 'start')
|
||||
resolveStartObserved()
|
||||
}
|
||||
catch {}
|
||||
}
|
||||
})
|
||||
handler(ws)
|
||||
})
|
||||
|
||||
await new Promise<void>(resolve => httpServer.listen(0, '127.0.0.1', resolve))
|
||||
const { port } = httpServer.address() as AddressInfo
|
||||
|
||||
return {
|
||||
url: `http://127.0.0.1:${port}`,
|
||||
receivedFrames,
|
||||
startObserved,
|
||||
async stop() {
|
||||
wss.close()
|
||||
await new Promise<void>(r => httpServer.close(() => r()))
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// jsdom-friendly stub AudioContext for `decodeAudioData`. The pipeline does
|
||||
// not introspect the AudioBuffer beyond passing it to consumers, so any
|
||||
// shape with the expected fields is fine.
|
||||
function makeStubAudioContext(): BaseAudioContext {
|
||||
let counter = 0
|
||||
const ctx = {
|
||||
sampleRate: 24000,
|
||||
decodeAudioData: vi.fn(async (buf: ArrayBuffer) => {
|
||||
// Return a fake AudioBuffer-like object identifiable by index/byteLength.
|
||||
counter += 1
|
||||
return {
|
||||
duration: buf.byteLength / 24000,
|
||||
length: buf.byteLength,
|
||||
numberOfChannels: 1,
|
||||
sampleRate: 24000,
|
||||
__index: counter,
|
||||
__byteLength: buf.byteLength,
|
||||
} as unknown as AudioBuffer
|
||||
}),
|
||||
}
|
||||
return ctx as unknown as BaseAudioContext
|
||||
}
|
||||
|
||||
describe('createStreamingTtsPipeline', () => {
|
||||
let server: MockServer | undefined
|
||||
|
||||
beforeEach(() => {
|
||||
server = undefined
|
||||
})
|
||||
afterEach(async () => {
|
||||
await server?.stop()
|
||||
})
|
||||
|
||||
it('forwards appendText / finish frames and chunks audio per sentence.end', async () => {
|
||||
const chunks = [Buffer.from([1, 2, 3, 4]), Buffer.from([5, 6, 7, 8]), Buffer.from([9, 10, 11, 12])]
|
||||
server = await startMockServer((ws) => {
|
||||
ws.on('message', async (data, isBinary) => {
|
||||
if (isBinary)
|
||||
return
|
||||
const ev = JSON.parse(data.toString()) as { event?: string }
|
||||
if (ev.event === 'finish') {
|
||||
// First sentence: chunk1 + chunk2 → sentence.end
|
||||
ws.send(JSON.stringify({ event: 'sentence.start', payload: { text: 'first one.' } }))
|
||||
ws.send(chunks[0], { binary: true })
|
||||
ws.send(chunks[1], { binary: true })
|
||||
ws.send(JSON.stringify({ event: 'sentence.end', payload: { text: 'first one.' } }))
|
||||
// Second sentence: chunk3 → sentence.end
|
||||
ws.send(JSON.stringify({ event: 'sentence.start', payload: { text: 'second sentence.' } }))
|
||||
ws.send(chunks[2], { binary: true })
|
||||
ws.send(JSON.stringify({ event: 'sentence.end', payload: { text: 'second sentence.' } }))
|
||||
ws.send(JSON.stringify({ event: 'session.finished', payload: { usage: { text_words: 4 } } }))
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
const onSentence = vi.fn()
|
||||
const onError = vi.fn()
|
||||
const onDone = vi.fn()
|
||||
|
||||
const handle = createStreamingTtsPipeline({
|
||||
serverUrl: server.url,
|
||||
model: 'volcengine/seed-tts-1.0',
|
||||
voice: 'mock',
|
||||
audioContext: makeStubAudioContext(),
|
||||
onSentence,
|
||||
onError,
|
||||
onDone,
|
||||
})
|
||||
|
||||
handle.appendText('hi ')
|
||||
handle.appendText('there')
|
||||
handle.finish()
|
||||
|
||||
// Wait for done.
|
||||
await new Promise<void>((resolve) => {
|
||||
onDone.mockImplementation(() => resolve())
|
||||
setTimeout(resolve, 1500)
|
||||
})
|
||||
|
||||
await server.startObserved
|
||||
const textFrames = server.receivedFrames.filter(f => f.kind === 'text').map(f => JSON.parse(f.data as string))
|
||||
expect(textFrames.map(f => f.event)).toEqual(['start', 'text', 'text', 'finish'])
|
||||
expect(textFrames[1]).toMatchObject({ event: 'text', text: 'hi ' })
|
||||
expect(textFrames[2]).toMatchObject({ event: 'text', text: 'there' })
|
||||
|
||||
expect(onError).not.toHaveBeenCalled()
|
||||
// Two `sentence.end` events → two AudioBuffers.
|
||||
expect(onSentence).toHaveBeenCalledTimes(2)
|
||||
const calls = onSentence.mock.calls.map(([s]) => s as { index: number, text: string, audio: { __byteLength: number } })
|
||||
expect(calls[0]).toMatchObject({ index: 0, text: 'first one.' })
|
||||
expect(calls[0].audio.__byteLength).toBe(chunks[0].length + chunks[1].length)
|
||||
expect(calls[1]).toMatchObject({ index: 1, text: 'second sentence.' })
|
||||
expect(calls[1].audio.__byteLength).toBe(chunks[2].length)
|
||||
})
|
||||
|
||||
it('buffers entire session when bufferEntireSession is true', async () => {
|
||||
const chunks = [Buffer.from([1, 2, 3, 4]), Buffer.from([5, 6, 7, 8])]
|
||||
server = await startMockServer((ws) => {
|
||||
ws.on('message', (data, isBinary) => {
|
||||
if (isBinary)
|
||||
return
|
||||
const ev = JSON.parse(data.toString()) as { event?: string }
|
||||
if (ev.event === 'finish') {
|
||||
// Two sentences with sentence.end events — but the pipeline should
|
||||
// IGNORE them in buffered mode (TTS 2.0 ships subtitles async).
|
||||
ws.send(chunks[0], { binary: true })
|
||||
ws.send(JSON.stringify({ event: 'sentence.end', payload: { text: 'sentence 1' } }))
|
||||
ws.send(chunks[1], { binary: true })
|
||||
ws.send(JSON.stringify({ event: 'sentence.end', payload: { text: 'sentence 2' } }))
|
||||
ws.send(JSON.stringify({ event: 'session.finished', payload: {} }))
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
const onSentence = vi.fn()
|
||||
const handle = createStreamingTtsPipeline({
|
||||
serverUrl: server.url,
|
||||
model: 'volcengine/seed-tts-2.0',
|
||||
voice: 'mock',
|
||||
audioContext: makeStubAudioContext(),
|
||||
bufferEntireSession: true,
|
||||
onSentence,
|
||||
})
|
||||
|
||||
handle.finish()
|
||||
|
||||
await new Promise<void>(resolve => setTimeout(resolve, 800))
|
||||
|
||||
expect(onSentence).toHaveBeenCalledTimes(1)
|
||||
const [sentence] = onSentence.mock.calls[0] as [{ index: number, audio: { __byteLength: number } }]
|
||||
expect(sentence.index).toBe(0)
|
||||
expect(sentence.audio.__byteLength).toBe(chunks[0].length + chunks[1].length)
|
||||
})
|
||||
|
||||
it('surfaces upstream error event then closes', async () => {
|
||||
server = await startMockServer((ws) => {
|
||||
ws.on('message', (data, isBinary) => {
|
||||
if (isBinary)
|
||||
return
|
||||
const ev = JSON.parse(data.toString()) as { event?: string }
|
||||
if (ev.event === 'start') {
|
||||
ws.send(JSON.stringify({ event: 'error', code: 'insufficient_flux', message: 'top up' }))
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
const onError = vi.fn()
|
||||
const onDone = vi.fn()
|
||||
createStreamingTtsPipeline({
|
||||
serverUrl: server.url,
|
||||
model: 'volcengine/seed-tts-1.0',
|
||||
voice: 'mock',
|
||||
audioContext: makeStubAudioContext(),
|
||||
onError,
|
||||
onDone,
|
||||
})
|
||||
|
||||
await new Promise<void>((resolve) => {
|
||||
onDone.mockImplementation(() => resolve())
|
||||
setTimeout(resolve, 1500)
|
||||
})
|
||||
|
||||
expect(onError).toHaveBeenCalledTimes(1)
|
||||
expect((onError.mock.calls[0][0] as Error).message).toMatch(/insufficient_flux.*top up/)
|
||||
})
|
||||
|
||||
it('surfaces close-before-finished as error', async () => {
|
||||
server = await startMockServer((ws) => {
|
||||
ws.on('message', (data, isBinary) => {
|
||||
if (isBinary)
|
||||
return
|
||||
const ev = JSON.parse(data.toString()) as { event?: string }
|
||||
if (ev.event === 'start') {
|
||||
// Drop ws without sending session.finished.
|
||||
setTimeout(() => ws.close(1011, 'simulated_truncation'), 10)
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
const onError = vi.fn()
|
||||
const onDone = vi.fn()
|
||||
createStreamingTtsPipeline({
|
||||
serverUrl: server.url,
|
||||
model: 'volcengine/seed-tts-1.0',
|
||||
voice: 'mock',
|
||||
audioContext: makeStubAudioContext(),
|
||||
onError,
|
||||
onDone,
|
||||
})
|
||||
|
||||
await new Promise<void>((resolve) => {
|
||||
onDone.mockImplementation(() => resolve())
|
||||
setTimeout(resolve, 1500)
|
||||
})
|
||||
|
||||
expect(onError).toHaveBeenCalledTimes(1)
|
||||
expect((onError.mock.calls[0][0] as Error).message).toMatch(/streaming_tts_closed/)
|
||||
})
|
||||
|
||||
it('cancel() sends cancel frame and terminates', async () => {
|
||||
let cancelObserved = false
|
||||
server = await startMockServer((ws) => {
|
||||
ws.on('message', (data, isBinary) => {
|
||||
if (isBinary)
|
||||
return
|
||||
const ev = JSON.parse(data.toString()) as { event?: string }
|
||||
if (ev.event === 'cancel')
|
||||
cancelObserved = true
|
||||
})
|
||||
})
|
||||
|
||||
const onDone = vi.fn()
|
||||
const handle = createStreamingTtsPipeline({
|
||||
serverUrl: server.url,
|
||||
model: 'volcengine/seed-tts-1.0',
|
||||
voice: 'mock',
|
||||
audioContext: makeStubAudioContext(),
|
||||
onDone,
|
||||
})
|
||||
|
||||
await server.startObserved
|
||||
handle.cancel()
|
||||
|
||||
await new Promise<void>((resolve) => {
|
||||
onDone.mockImplementation(() => resolve())
|
||||
setTimeout(resolve, 500)
|
||||
})
|
||||
|
||||
expect(cancelObserved).toBe(true)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,346 @@
|
||||
import { getAuthToken } from '../auth'
|
||||
import { SERVER_URL } from '../server'
|
||||
|
||||
/**
|
||||
* One synthesized sentence emitted by the streaming pipeline. The pipeline
|
||||
* delivers these in arrival order; consumers schedule them into their
|
||||
* playback manager directly.
|
||||
*/
|
||||
export interface StreamingPipelineSentence {
|
||||
/** 0-based sentence index within the session. */
|
||||
index: number
|
||||
/** Sentence text from the upstream `sentence.*` payload, when available. */
|
||||
text: string
|
||||
/** Decoded audio. Same `AudioContext` is used for every sentence in the session. */
|
||||
audio: AudioBuffer
|
||||
}
|
||||
|
||||
export interface StreamingTtsPipelineEvents {
|
||||
/**
|
||||
* Fires once per synthesized sentence (TTS 1.0) or once per session
|
||||
* (TTS 2.0 / `bufferEntireSession: true`). Schedule the audio into your
|
||||
* playback manager from this callback.
|
||||
*/
|
||||
onSentence?: (sentence: StreamingPipelineSentence) => void
|
||||
/**
|
||||
* Surfaced for any post-upgrade failure (server `error` event, ws close
|
||||
* without `session.finished`, decode failure). Consumers should treat the
|
||||
* session as terminated after this fires.
|
||||
*/
|
||||
onError?: (err: Error) => void
|
||||
/**
|
||||
* Fires after the ws closes for any reason. Always paired with either
|
||||
* `onSentence` (success path) or `onError` (failure path) preceding it.
|
||||
*/
|
||||
onDone?: () => void
|
||||
}
|
||||
|
||||
export interface StreamingTtsPipelineOptions extends StreamingTtsPipelineEvents {
|
||||
/** Server URL override. Defaults to {@link SERVER_URL}. */
|
||||
serverUrl?: string
|
||||
/** Override the auth token (Bearer). Defaults to {@link getAuthToken}. */
|
||||
token?: string
|
||||
/** unspeech-routed model id, e.g. `volcengine/seed-tts-2.0`. */
|
||||
model: string
|
||||
/** Upstream voice / speaker id. */
|
||||
voice: string
|
||||
/** OpenAI-style format. Default `mp3`. */
|
||||
responseFormat?: 'mp3' | 'opus' | 'aac' | 'flac' | 'pcm'
|
||||
/** Backend-specific knobs forwarded as the `extra_body` of the `start` frame. */
|
||||
extraBody?: Record<string, unknown>
|
||||
/**
|
||||
* Decoder context. The pipeline calls `decodeAudioData` on it for each
|
||||
* sentence (or once at session end in buffered mode). Reusing the page's
|
||||
* AudioContext is required so playback nodes that connect to its
|
||||
* destination see compatible sample rates.
|
||||
*/
|
||||
audioContext: BaseAudioContext
|
||||
/**
|
||||
* When `true`, accumulate every binary chunk until `session.finished` and
|
||||
* emit ONE `onSentence`. Use for models where per-sentence audio boundaries
|
||||
* are not synchronously aligned with `sentence.end` events (Volcengine
|
||||
* Seed-TTS 2.0 ships subtitles asynchronously; chunking on `sentence.end`
|
||||
* would drop frames). Default `false` (chunk per sentence — correct for
|
||||
* Seed-TTS 1.0 / ICL 1.0 where `sentence.end` arrives in-band with audio).
|
||||
*/
|
||||
bufferEntireSession?: boolean
|
||||
}
|
||||
|
||||
export interface StreamingTtsPipelineHandle {
|
||||
/**
|
||||
* Forward a chunk of LLM-generated text to the in-flight TTS session.
|
||||
* The text is sent verbatim — no client-side segmentation. The upstream
|
||||
* model decides where to split sentences and how to pace prosody.
|
||||
*
|
||||
* Safe to call before the ws is open; frames are queued and flushed
|
||||
* after the handshake completes.
|
||||
*/
|
||||
appendText: (text: string) => void
|
||||
/**
|
||||
* Signal end of the LLM text stream. The upstream emits any remaining
|
||||
* audio then `session.finished`; the pipeline closes the ws after.
|
||||
*/
|
||||
finish: () => void
|
||||
/**
|
||||
* Abort the in-flight session. Sends `cancel` upstream (best-effort, no
|
||||
* ack wait per protocol v1) and closes the ws.
|
||||
*/
|
||||
cancel: () => void
|
||||
}
|
||||
|
||||
/**
|
||||
* Drives a single bidirectional streaming TTS session for one LLM intent.
|
||||
*
|
||||
* Use when:
|
||||
* - You have a streaming LLM output you want voiced without client-side
|
||||
* sentence segmentation. The upstream model receives raw token chunks
|
||||
* and decides where to split.
|
||||
*
|
||||
* Expects:
|
||||
* - Authenticated user (or `options.token`).
|
||||
* - `STREAMING_TTS_UPSTREAM` configKV configured server-side.
|
||||
*
|
||||
* Returns:
|
||||
* - A handle with `appendText` / `finish` / `cancel`. Side-effect: audio
|
||||
* AudioBuffers are emitted via `options.onSentence` in arrival order.
|
||||
*/
|
||||
export function createStreamingTtsPipeline(options: StreamingTtsPipelineOptions): StreamingTtsPipelineHandle {
|
||||
const token = options.token ?? getAuthToken()
|
||||
if (!token) {
|
||||
const err = new Error('streaming-pipeline: not authenticated')
|
||||
queueMicrotask(() => {
|
||||
options.onError?.(err)
|
||||
options.onDone?.()
|
||||
})
|
||||
return noopHandle()
|
||||
}
|
||||
|
||||
const wsUrl = toWebSocketUrl(options.serverUrl ?? SERVER_URL, '/api/v1/audio/speech/ws', token)
|
||||
const ws = new WebSocket(wsUrl)
|
||||
ws.binaryType = 'arraybuffer'
|
||||
|
||||
let closed = false
|
||||
let sawSessionFinished = false
|
||||
/**
|
||||
* Queue for frames sent before the ws transitions to OPEN. Avoids
|
||||
* silently dropping early `appendText` calls (the caller doesn't know
|
||||
* the handshake hasn't completed yet).
|
||||
*/
|
||||
const beforeOpenQueue: string[] = []
|
||||
/** Binary chunks accumulated since the last sentence flush. */
|
||||
let chunks: ArrayBuffer[] = []
|
||||
let chunkBytes = 0
|
||||
let sentenceIndex = 0
|
||||
/**
|
||||
* FIFO of sentence texts seen via `sentence.start` events that haven't
|
||||
* been paired with a `sentence.end` yet. The protocol promises in-order
|
||||
* pairs, but a buggy upstream or re-ordered transport could send two
|
||||
* `sentence.start`s in a row; using a queue (instead of a single
|
||||
* `pendingSentenceText` variable) keeps each audio buffer labelled with
|
||||
* the right text instead of overwriting. Codex review MEDIUM #4.
|
||||
*/
|
||||
const pendingSentenceTexts: string[] = []
|
||||
const bufferEntireSession = options.bufferEntireSession ?? false
|
||||
|
||||
function safeSend(payload: string) {
|
||||
if (closed)
|
||||
return
|
||||
if (ws.readyState === WebSocket.OPEN) {
|
||||
ws.send(payload)
|
||||
return
|
||||
}
|
||||
if (ws.readyState === WebSocket.CONNECTING) {
|
||||
beforeOpenQueue.push(payload)
|
||||
}
|
||||
// CLOSING/CLOSED — drop silently; caller will see onDone shortly.
|
||||
}
|
||||
|
||||
async function flushAccumulatedAsSentence(textOverride?: string) {
|
||||
if (chunkBytes === 0)
|
||||
return
|
||||
const merged = new Uint8Array(chunkBytes)
|
||||
let offset = 0
|
||||
for (const c of chunks) {
|
||||
merged.set(new Uint8Array(c), offset)
|
||||
offset += c.byteLength
|
||||
}
|
||||
chunks = []
|
||||
chunkBytes = 0
|
||||
|
||||
// Prefer the explicit override (the `sentence.end` payload's own text)
|
||||
// over the queued `sentence.start` text — `sentence.end` is the
|
||||
// authoritative pairing. The queue covers the case where `sentence.end`
|
||||
// arrives without a text field.
|
||||
const text = textOverride ?? pendingSentenceTexts.shift() ?? ''
|
||||
|
||||
try {
|
||||
// decodeAudioData needs a transferable ArrayBuffer; pass the buffer
|
||||
// backing `merged`. Clone to a fresh buffer so subsequent flushes do
|
||||
// not race on a buffer the decoder may detach.
|
||||
const audio = await options.audioContext.decodeAudioData(merged.buffer.slice(0))
|
||||
options.onSentence?.({ index: sentenceIndex++, text, audio })
|
||||
}
|
||||
catch (err) {
|
||||
options.onError?.(err instanceof Error ? err : new Error(String(err)))
|
||||
}
|
||||
}
|
||||
|
||||
ws.addEventListener('open', () => {
|
||||
const startFrame = {
|
||||
event: 'start',
|
||||
model: options.model,
|
||||
voice: options.voice,
|
||||
response_format: options.responseFormat ?? 'mp3',
|
||||
...(options.extraBody ? { extra_body: options.extraBody } : {}),
|
||||
}
|
||||
ws.send(JSON.stringify(startFrame))
|
||||
for (const payload of beforeOpenQueue)
|
||||
ws.send(payload)
|
||||
beforeOpenQueue.length = 0
|
||||
})
|
||||
|
||||
ws.addEventListener('message', (e) => {
|
||||
if (typeof e.data === 'string') {
|
||||
void handleControlFrame(e.data)
|
||||
return
|
||||
}
|
||||
// binary audio chunk
|
||||
if (e.data instanceof ArrayBuffer) {
|
||||
chunks.push(e.data)
|
||||
chunkBytes += e.data.byteLength
|
||||
}
|
||||
})
|
||||
|
||||
async function handleControlFrame(raw: string) {
|
||||
let evt: { event?: string, payload?: Record<string, unknown>, text?: string, code?: string, message?: string }
|
||||
try {
|
||||
evt = JSON.parse(raw)
|
||||
}
|
||||
catch {
|
||||
return
|
||||
}
|
||||
|
||||
switch (evt.event) {
|
||||
case 'sentence.start': {
|
||||
// Append to the queue. `sentence.end` consumes from the head, so
|
||||
// back-to-back `sentence.start`s (which shouldn't happen but
|
||||
// codex MEDIUM #4 noted the race) don't clobber each other.
|
||||
const text = readSentenceText(evt.payload)
|
||||
if (text != null)
|
||||
pendingSentenceTexts.push(text)
|
||||
break
|
||||
}
|
||||
case 'sentence.end': {
|
||||
if (bufferEntireSession)
|
||||
break
|
||||
const text = readSentenceText(evt.payload) ?? pendingSentenceTexts.shift() ?? ''
|
||||
await flushAccumulatedAsSentence(text)
|
||||
break
|
||||
}
|
||||
case 'subtitle': {
|
||||
// TTS 2.0 emits subtitle events asynchronously (may arrive after
|
||||
// the next sentence's audio has already started). We surface the
|
||||
// text via the queue but do NOT flush audio here — buffered mode
|
||||
// flushes once at session.finished instead.
|
||||
const text = readSentenceText(evt.payload)
|
||||
if (text != null)
|
||||
pendingSentenceTexts.push(text)
|
||||
break
|
||||
}
|
||||
case 'session.finished': {
|
||||
sawSessionFinished = true
|
||||
await flushAccumulatedAsSentence()
|
||||
terminate(null)
|
||||
break
|
||||
}
|
||||
case 'error': {
|
||||
const code = evt.code ?? 'streaming_tts_error'
|
||||
const message = evt.message ?? code
|
||||
terminate(new Error(`${code}: ${message}`))
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
ws.addEventListener('close', (ev) => {
|
||||
if (closed)
|
||||
return
|
||||
if (sawSessionFinished) {
|
||||
// Normal end after `session.finished`; flushAccumulatedAsSentence
|
||||
// already ran. Just mark closed and notify.
|
||||
terminate(null)
|
||||
return
|
||||
}
|
||||
// Closed before completion: surface as an error so callers don't
|
||||
// mistake truncated audio for a successful (short) sentence.
|
||||
const reason = ev.reason || `closed_${ev.code}`
|
||||
terminate(new Error(`streaming_tts_closed: ${reason}`))
|
||||
})
|
||||
|
||||
ws.addEventListener('error', () => {
|
||||
// The `error` event carries no useful info per the WebSocket API; the
|
||||
// `close` event right after has the actual reason. Don't double-emit.
|
||||
})
|
||||
|
||||
function terminate(err: Error | null) {
|
||||
if (closed)
|
||||
return
|
||||
closed = true
|
||||
try {
|
||||
if (ws.readyState === WebSocket.OPEN || ws.readyState === WebSocket.CONNECTING)
|
||||
ws.close()
|
||||
}
|
||||
catch {}
|
||||
if (err != null)
|
||||
options.onError?.(err)
|
||||
options.onDone?.()
|
||||
}
|
||||
|
||||
return {
|
||||
appendText(text: string) {
|
||||
if (text.length === 0)
|
||||
return
|
||||
// Pure-whitespace chunks (e.g. the " " between two LLM tokens) ARE
|
||||
// forwarded verbatim. Dropping them would corrupt the text the
|
||||
// upstream model sees ("hello" + " " + "world" → "helloworld").
|
||||
// The per-character billing cost is negligible compared to the
|
||||
// semantic risk; codex review LOW #7 noted the wasted units but
|
||||
// accepted the trade-off.
|
||||
safeSend(JSON.stringify({ event: 'text', text }))
|
||||
},
|
||||
finish() {
|
||||
safeSend(JSON.stringify({ event: 'finish' }))
|
||||
},
|
||||
cancel() {
|
||||
if (closed)
|
||||
return
|
||||
safeSend(JSON.stringify({ event: 'cancel' }))
|
||||
// Surface cancel as a non-error termination; consumers already
|
||||
// initiated this so they don't need a synthetic error.
|
||||
terminate(null)
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
function noopHandle(): StreamingTtsPipelineHandle {
|
||||
return { appendText: () => {}, finish: () => {}, cancel: () => {} }
|
||||
}
|
||||
|
||||
function toWebSocketUrl(httpBase: string, path: string, token: string): string {
|
||||
const u = new URL(path, httpBase)
|
||||
u.protocol = u.protocol === 'https:' ? 'wss:' : 'ws:'
|
||||
u.searchParams.set('token', token)
|
||||
return u.toString()
|
||||
}
|
||||
|
||||
/**
|
||||
* Reads the sentence text from a `sentence.start` / `sentence.end` /
|
||||
* `subtitle` payload. Returns `null` when the payload doesn't carry one
|
||||
* (e.g. final upstream events with empty bodies).
|
||||
*/
|
||||
function readSentenceText(payload: Record<string, unknown> | undefined): string | null {
|
||||
if (!payload || typeof payload !== 'object')
|
||||
return null
|
||||
const text = (payload as { text?: unknown }).text
|
||||
return typeof text === 'string' ? text : null
|
||||
}
|
||||
@@ -0,0 +1,336 @@
|
||||
import type { IntentOptions, PlaybackItem } from '@proj-airi/pipelines-audio'
|
||||
|
||||
import type { PlaybackManagerSubset, StreamingSessionSnapshot } from './tts-session'
|
||||
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
|
||||
import { createStageTtsSession, createStreamingTtsSession } from './tts-session'
|
||||
|
||||
// Lightweight IntentHandle stub. We do not import the real one from
|
||||
// `@proj-airi/pipelines-audio` because the segmenter adapter only needs a
|
||||
// fixed subset, and constructing a full IntentHandle would drag in the
|
||||
// segmenter pipeline.
|
||||
function makeIntentStub(overrides: Partial<{ intentId: string }> = {}) {
|
||||
return {
|
||||
intentId: overrides.intentId ?? 'segmenter-intent-1',
|
||||
streamId: 'stream-id',
|
||||
priority: 0,
|
||||
writeLiteral: vi.fn<(text: string) => void>(),
|
||||
writeSpecial: vi.fn<(special: string) => void>(),
|
||||
writeFlush: vi.fn<() => void>(),
|
||||
end: vi.fn<() => void>(),
|
||||
cancel: vi.fn<(reason?: string) => void>(),
|
||||
// `IntentHandle` also carries a `stream: ReadableStream<TextToken>`
|
||||
// field; the adapter never touches it, so we stub it as never.
|
||||
stream: undefined as never,
|
||||
}
|
||||
}
|
||||
|
||||
function makePlaybackManagerStub<TAudio = AudioBuffer>(): PlaybackManagerSubset<TAudio> & {
|
||||
scheduled: Array<PlaybackItem<TAudio>>
|
||||
cancellations: Array<{ intentId: string, reason: string }>
|
||||
} {
|
||||
const scheduled: Array<PlaybackItem<TAudio>> = []
|
||||
const cancellations: Array<{ intentId: string, reason: string }> = []
|
||||
return {
|
||||
schedule: vi.fn((item: PlaybackItem<TAudio>) => {
|
||||
scheduled.push(item)
|
||||
}),
|
||||
stopByIntent: vi.fn((intentId: string, reason: string) => {
|
||||
cancellations.push({ intentId, reason })
|
||||
}),
|
||||
scheduled,
|
||||
cancellations,
|
||||
}
|
||||
}
|
||||
|
||||
function makeStreamingSnapshot(overrides: Partial<StreamingSessionSnapshot> = {}): StreamingSessionSnapshot {
|
||||
return {
|
||||
model: 'volcengine/seed-tts-2.0',
|
||||
voice: 'mock-voice',
|
||||
bufferEntireSession: false,
|
||||
extraBody: { api_resource_id: 'seed-tts-2.0' },
|
||||
ownerId: 'card-1',
|
||||
onImmediateSpecial: vi.fn(),
|
||||
...overrides,
|
||||
}
|
||||
}
|
||||
|
||||
// Stub for the streaming pipeline factory. Captures the callbacks so the
|
||||
// test can drive `onSentence` / `onError` / `onDone` directly. Tracks
|
||||
// `appendText` / `finish` / `cancel` invocations.
|
||||
function makePipelineStub() {
|
||||
const calls: { appendText: string[], finish: number, cancel: number } = {
|
||||
appendText: [],
|
||||
finish: 0,
|
||||
cancel: 0,
|
||||
}
|
||||
let captured: any
|
||||
const factory = vi.fn((options: any) => {
|
||||
captured = options
|
||||
return {
|
||||
appendText: (text: string) => {
|
||||
calls.appendText.push(text)
|
||||
},
|
||||
finish: () => {
|
||||
calls.finish += 1
|
||||
},
|
||||
cancel: () => {
|
||||
calls.cancel += 1
|
||||
},
|
||||
}
|
||||
})
|
||||
return {
|
||||
factory,
|
||||
calls,
|
||||
get options() { return captured },
|
||||
}
|
||||
}
|
||||
|
||||
const dummyAudioContext = { sampleRate: 24000 } as unknown as BaseAudioContext
|
||||
|
||||
describe('createStageTtsSession (factory)', () => {
|
||||
it('returns the segmenter adapter when transport is "rest"', () => {
|
||||
const intent = makeIntentStub({ intentId: 'segmenter-1' })
|
||||
const playback = makePlaybackManagerStub()
|
||||
const session = createStageTtsSession({
|
||||
transport: 'rest',
|
||||
streaming: () => null,
|
||||
audioContext: dummyAudioContext,
|
||||
playbackManager: playback,
|
||||
openIntent: vi.fn(() => intent),
|
||||
intentOptions: () => ({ ownerId: 'card-1', priority: 'normal', behavior: 'queue' } as IntentOptions),
|
||||
})
|
||||
|
||||
session.appendText('hello')
|
||||
session.appendSpecial('emotion:happy')
|
||||
session.finishInput()
|
||||
session.end()
|
||||
session.cancel('done')
|
||||
|
||||
expect(session.intentId).toBe('segmenter-1')
|
||||
expect(intent.writeLiteral).toHaveBeenCalledWith('hello')
|
||||
expect(intent.writeSpecial).toHaveBeenCalledWith('emotion:happy')
|
||||
expect(intent.writeFlush).toHaveBeenCalled()
|
||||
expect(intent.end).toHaveBeenCalled()
|
||||
expect(intent.cancel).toHaveBeenCalledWith('done')
|
||||
expect(playback.scheduled).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('returns the segmenter adapter when transport is undefined (REST default)', () => {
|
||||
const intent = makeIntentStub({ intentId: 'segmenter-default' })
|
||||
const session = createStageTtsSession({
|
||||
transport: undefined,
|
||||
streaming: () => makeStreamingSnapshot(),
|
||||
audioContext: dummyAudioContext,
|
||||
playbackManager: makePlaybackManagerStub(),
|
||||
openIntent: vi.fn(() => intent),
|
||||
intentOptions: () => ({ ownerId: 'card-1', priority: 'normal', behavior: 'queue' } as IntentOptions),
|
||||
})
|
||||
|
||||
expect(session.intentId).toBe('segmenter-default')
|
||||
})
|
||||
|
||||
it('falls back to segmenter when bidirectional-ws snapshot is missing', () => {
|
||||
const intent = makeIntentStub({ intentId: 'segmenter-fallback' })
|
||||
const playback = makePlaybackManagerStub()
|
||||
const session = createStageTtsSession({
|
||||
transport: 'bidirectional-ws',
|
||||
streaming: () => null, // No snapshot → fallback.
|
||||
audioContext: dummyAudioContext,
|
||||
playbackManager: playback,
|
||||
openIntent: vi.fn(() => intent),
|
||||
intentOptions: () => ({ ownerId: 'card-1', priority: 'normal', behavior: 'queue' } as IntentOptions),
|
||||
})
|
||||
|
||||
expect(session.intentId).toBe('segmenter-fallback')
|
||||
session.appendText('hi')
|
||||
expect(intent.writeLiteral).toHaveBeenCalledWith('hi')
|
||||
})
|
||||
|
||||
it('falls back to segmenter when audioContext is undefined', () => {
|
||||
const intent = makeIntentStub({ intentId: 'segmenter-no-ctx' })
|
||||
const session = createStageTtsSession({
|
||||
transport: 'bidirectional-ws',
|
||||
streaming: () => makeStreamingSnapshot(),
|
||||
audioContext: undefined,
|
||||
playbackManager: makePlaybackManagerStub(),
|
||||
openIntent: vi.fn(() => intent),
|
||||
intentOptions: () => ({ ownerId: 'card-1', priority: 'normal', behavior: 'queue' } as IntentOptions),
|
||||
})
|
||||
|
||||
expect(session.intentId).toBe('segmenter-no-ctx')
|
||||
})
|
||||
|
||||
it('falls back to segmenter when snapshot.voice is empty', () => {
|
||||
const intent = makeIntentStub({ intentId: 'segmenter-no-voice' })
|
||||
const session = createStageTtsSession({
|
||||
transport: 'bidirectional-ws',
|
||||
streaming: () => makeStreamingSnapshot({ voice: '' }),
|
||||
audioContext: dummyAudioContext,
|
||||
playbackManager: makePlaybackManagerStub(),
|
||||
openIntent: vi.fn(() => intent),
|
||||
intentOptions: () => ({ ownerId: 'card-1', priority: 'normal', behavior: 'queue' } as IntentOptions),
|
||||
})
|
||||
|
||||
expect(session.intentId).toBe('segmenter-no-voice')
|
||||
})
|
||||
})
|
||||
|
||||
describe('createStreamingTtsSession (adapter)', () => {
|
||||
it('schedules sentences into playbackManager with monotonic sequence', () => {
|
||||
const playback = makePlaybackManagerStub()
|
||||
const pipe = makePipelineStub()
|
||||
const snap = makeStreamingSnapshot()
|
||||
|
||||
createStreamingTtsSession({
|
||||
intentId: 'stream-abc',
|
||||
snapshot: snap,
|
||||
audioContext: dummyAudioContext,
|
||||
playbackManager: playback,
|
||||
pipelineFactory: pipe.factory as any,
|
||||
})
|
||||
|
||||
// Simulate the pipeline emitting two sentences.
|
||||
const audio0 = { __id: 0 } as unknown as AudioBuffer
|
||||
const audio1 = { __id: 1 } as unknown as AudioBuffer
|
||||
pipe.options.onSentence({ index: 0, text: 'first', audio: audio0 })
|
||||
pipe.options.onSentence({ index: 1, text: 'second', audio: audio1 })
|
||||
|
||||
expect(playback.scheduled).toHaveLength(2)
|
||||
expect(playback.scheduled[0]).toMatchObject({
|
||||
id: 'stream-abc-0',
|
||||
streamId: 'stream-abc',
|
||||
intentId: 'stream-abc',
|
||||
segmentId: 'stream-abc-0',
|
||||
sequence: 0,
|
||||
ownerId: 'card-1',
|
||||
text: 'first',
|
||||
special: null,
|
||||
})
|
||||
expect(playback.scheduled[0].audio).toBe(audio0)
|
||||
expect(playback.scheduled[1]).toMatchObject({
|
||||
sequence: 1,
|
||||
text: 'second',
|
||||
audio: audio1,
|
||||
})
|
||||
})
|
||||
|
||||
it('forwards appendText / finishInput to the pipeline', () => {
|
||||
const pipe = makePipelineStub()
|
||||
const session = createStreamingTtsSession({
|
||||
intentId: 'stream-x',
|
||||
snapshot: makeStreamingSnapshot(),
|
||||
audioContext: dummyAudioContext,
|
||||
playbackManager: makePlaybackManagerStub(),
|
||||
pipelineFactory: pipe.factory as any,
|
||||
})
|
||||
|
||||
session.appendText('hello')
|
||||
session.appendText('world')
|
||||
session.finishInput()
|
||||
|
||||
expect(pipe.calls.appendText).toEqual(['hello', 'world'])
|
||||
expect(pipe.calls.finish).toBe(1)
|
||||
})
|
||||
|
||||
it('appendSpecial fires the host immediate-special callback', () => {
|
||||
const pipe = makePipelineStub()
|
||||
const onSpecial = vi.fn()
|
||||
const session = createStreamingTtsSession({
|
||||
intentId: 'stream-special',
|
||||
snapshot: makeStreamingSnapshot({ onImmediateSpecial: onSpecial }),
|
||||
audioContext: dummyAudioContext,
|
||||
playbackManager: makePlaybackManagerStub(),
|
||||
pipelineFactory: pipe.factory as any,
|
||||
})
|
||||
|
||||
session.appendSpecial('emotion:angry')
|
||||
session.appendSpecial('delay:500')
|
||||
|
||||
expect(onSpecial).toHaveBeenCalledTimes(2)
|
||||
expect(onSpecial).toHaveBeenNthCalledWith(1, 'emotion:angry')
|
||||
expect(onSpecial).toHaveBeenNthCalledWith(2, 'delay:500')
|
||||
})
|
||||
|
||||
it('cancel sends pipeline cancel AND stops playback by intent', () => {
|
||||
const playback = makePlaybackManagerStub()
|
||||
const pipe = makePipelineStub()
|
||||
const session = createStreamingTtsSession({
|
||||
intentId: 'stream-cancel',
|
||||
snapshot: makeStreamingSnapshot(),
|
||||
audioContext: dummyAudioContext,
|
||||
playbackManager: playback,
|
||||
pipelineFactory: pipe.factory as any,
|
||||
})
|
||||
|
||||
session.cancel('user-aborted')
|
||||
|
||||
expect(pipe.calls.cancel).toBe(1)
|
||||
expect(playback.cancellations).toEqual([
|
||||
{ intentId: 'stream-cancel', reason: 'user-aborted' },
|
||||
])
|
||||
})
|
||||
|
||||
it('cancel after pipeline terminated still drains playback', () => {
|
||||
const playback = makePlaybackManagerStub()
|
||||
const pipe = makePipelineStub()
|
||||
const session = createStreamingTtsSession({
|
||||
intentId: 'stream-late',
|
||||
snapshot: makeStreamingSnapshot(),
|
||||
audioContext: dummyAudioContext,
|
||||
playbackManager: playback,
|
||||
pipelineFactory: pipe.factory as any,
|
||||
})
|
||||
|
||||
// Pipeline naturally completes first.
|
||||
pipe.options.onDone()
|
||||
// Then host cancels — pipeline.cancel should NOT be re-called, but
|
||||
// any straggler playback items must still be drained.
|
||||
session.cancel('post-done-cancel')
|
||||
|
||||
expect(pipe.calls.cancel).toBe(0)
|
||||
expect(playback.cancellations).toEqual([
|
||||
{ intentId: 'stream-late', reason: 'post-done-cancel' },
|
||||
])
|
||||
})
|
||||
|
||||
it('onSentence is dropped after pipeline terminated', () => {
|
||||
const playback = makePlaybackManagerStub()
|
||||
const pipe = makePipelineStub()
|
||||
createStreamingTtsSession({
|
||||
intentId: 'stream-after-done',
|
||||
snapshot: makeStreamingSnapshot(),
|
||||
audioContext: dummyAudioContext,
|
||||
playbackManager: playback,
|
||||
pipelineFactory: pipe.factory as any,
|
||||
})
|
||||
|
||||
// Mark terminated, then a straggler sentence arrives.
|
||||
pipe.options.onDone()
|
||||
pipe.options.onSentence({ index: 0, text: 'too late', audio: {} as AudioBuffer })
|
||||
|
||||
expect(playback.scheduled).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('hooks.onError fires on pipeline error', () => {
|
||||
const onError = vi.fn()
|
||||
const onDone = vi.fn()
|
||||
const pipe = makePipelineStub()
|
||||
createStreamingTtsSession({
|
||||
intentId: 'stream-err',
|
||||
snapshot: makeStreamingSnapshot(),
|
||||
audioContext: dummyAudioContext,
|
||||
playbackManager: makePlaybackManagerStub(),
|
||||
hooks: { onError, onDone },
|
||||
pipelineFactory: pipe.factory as any,
|
||||
})
|
||||
|
||||
const err = new Error('boom')
|
||||
pipe.options.onError(err)
|
||||
pipe.options.onDone()
|
||||
|
||||
expect(onError).toHaveBeenCalledWith(err)
|
||||
expect(onDone).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,318 @@
|
||||
import type { IntentHandle, IntentOptions, PlaybackItem } from '@proj-airi/pipelines-audio'
|
||||
|
||||
import type { StreamingTtsPipelineOptions } from './streaming-pipeline'
|
||||
|
||||
import { createStreamingTtsPipeline } from './streaming-pipeline'
|
||||
|
||||
/**
|
||||
* Stage-level TTS session abstraction.
|
||||
*
|
||||
* Both the segmenter-based path (every non-streaming provider — feeds tokens
|
||||
* through `pipelines-audio`'s `IntentHandle` → segmenter → per-segment
|
||||
* `tts()` callback) and the bidirectional WebSocket path (the official
|
||||
* streaming provider — forwards raw tokens upstream and lets the model do
|
||||
* its own sentence splitting) implement this surface.
|
||||
*
|
||||
* Stage.vue holds exactly one `StageTtsSession` at any moment and forwards
|
||||
* every chat-orchestrator hook into it without branching on provider id.
|
||||
* The decision of which adapter to construct lives once in
|
||||
* {@link createStageTtsSession}.
|
||||
*/
|
||||
export interface StageTtsSession {
|
||||
/** Stable id for this session. Used by playback to scope cancellation. */
|
||||
readonly intentId: string
|
||||
/** Forward an LLM token. Adapter decides whether it's segmented or raw. */
|
||||
appendText: (text: string) => void
|
||||
/**
|
||||
* Forward a special token (emotion / delay marker). Segmenter adapter
|
||||
* queues it in lockstep with audio; streaming adapter fires it
|
||||
* immediately (no segmenter queue to ride; see
|
||||
* {@link createStageTtsSession} comments).
|
||||
*/
|
||||
appendSpecial: (special: string) => void
|
||||
/** Signal end of LLM token stream. Caller still owes an `end()`. */
|
||||
finishInput: () => void
|
||||
/** Tear down on normal completion. */
|
||||
end: () => void
|
||||
/** Tear down on abort / new message / unmount. */
|
||||
cancel: (reason?: string) => void
|
||||
}
|
||||
|
||||
/**
|
||||
* Minimal `IntentHandle` shape the segmenter adapter needs. Quoting the
|
||||
* full `IntentHandle` would drag in extra fields (`stream`, `priority`,
|
||||
* etc.) that are not part of the session protocol; this typed subset keeps
|
||||
* the adapter honest about what it actually depends on.
|
||||
*/
|
||||
type IntentHandleSubset = Pick<IntentHandle, 'intentId' | 'writeLiteral' | 'writeSpecial' | 'writeFlush' | 'end' | 'cancel'>
|
||||
|
||||
/**
|
||||
* Direct adapter from a `pipelines-audio` `IntentHandle` to
|
||||
* {@link StageTtsSession}. Pure passthrough — the segmenter pipeline already
|
||||
* owns segmentation, special-token queueing, and playback scheduling.
|
||||
*/
|
||||
function fromIntent(intent: IntentHandleSubset): StageTtsSession {
|
||||
return {
|
||||
intentId: intent.intentId,
|
||||
appendText: intent.writeLiteral,
|
||||
appendSpecial: intent.writeSpecial,
|
||||
finishInput: intent.writeFlush,
|
||||
end: intent.end,
|
||||
cancel: intent.cancel,
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Per-session knobs the streaming adapter consumes. Snapshotted once at
|
||||
* session-open time so a mid-session provider/voice swap does not corrupt
|
||||
* an in-flight session — the hot-swap watcher in Stage.vue is responsible
|
||||
* for cancelling and re-opening.
|
||||
*/
|
||||
export interface StreamingSessionSnapshot {
|
||||
model: string
|
||||
voice: string
|
||||
bufferEntireSession: boolean
|
||||
extraBody: Record<string, unknown>
|
||||
/**
|
||||
* `ownerId` to stamp on each `PlaybackItem`. Mirrors the value the
|
||||
* segmenter-based intent uses (`activeCardId`) so playback manager
|
||||
* owner-quota policies treat both paths identically.
|
||||
*/
|
||||
ownerId?: string
|
||||
/**
|
||||
* Called when the host wants a special token (emotion / delay marker)
|
||||
* dispatched. Streaming has no in-band queue to ride; this callback
|
||||
* fires immediately on `appendSpecial`. Without it the streaming adapter
|
||||
* would silently drop emotion/delay tokens. The host wires this to
|
||||
* something like `playSpecialToken` (Stage.vue's existing helper).
|
||||
*/
|
||||
onImmediateSpecial: (special: string) => void
|
||||
}
|
||||
|
||||
/**
|
||||
* Minimal `PlaybackManager` shape the streaming adapter writes to. Same
|
||||
* `intentId` is passed to `stopByIntent` on cancel and used as
|
||||
* `streamId`/`intentId` on every scheduled item, so cancellation reliably
|
||||
* stops every audio buffer this session emitted.
|
||||
*/
|
||||
export interface PlaybackManagerSubset<TAudio> {
|
||||
schedule: (item: PlaybackItem<TAudio>) => void
|
||||
stopByIntent: (intentId: string, reason: string) => void
|
||||
}
|
||||
|
||||
/**
|
||||
* Internal helpers the streaming adapter calls out to. Lets the adapter
|
||||
* react to terminal events (error / done) by clearing whatever state the
|
||||
* host is holding — Stage.vue keeps a `currentSession` ref and needs to
|
||||
* null it when the underlying ws terminates on its own.
|
||||
*/
|
||||
export interface StreamingSessionHooks {
|
||||
/** Called once when the ws terminates with an error. */
|
||||
onError?: (err: Error) => void
|
||||
/** Called once when the ws terminates (success or error follows). */
|
||||
onDone?: () => void
|
||||
}
|
||||
|
||||
export interface CreateStreamingSessionOptions<TAudio = AudioBuffer> {
|
||||
intentId: string
|
||||
snapshot: StreamingSessionSnapshot
|
||||
audioContext: BaseAudioContext
|
||||
playbackManager: PlaybackManagerSubset<TAudio>
|
||||
hooks?: StreamingSessionHooks
|
||||
/**
|
||||
* Optional override for the underlying pipeline factory. Tests inject a
|
||||
* stub here; production wires the real `createStreamingTtsPipeline`.
|
||||
*
|
||||
* @default {@link createStreamingTtsPipeline}
|
||||
*/
|
||||
pipelineFactory?: (options: StreamingTtsPipelineOptions) => ReturnType<typeof createStreamingTtsPipeline>
|
||||
}
|
||||
|
||||
/**
|
||||
* Streaming adapter: opens ONE ws session for the whole intent and
|
||||
* schedules sentences into the playback manager as they arrive. Cancel
|
||||
* tells the pipeline to send `cancel` upstream AND drains any already-queued
|
||||
* playback items that belong to this intent.
|
||||
*
|
||||
* Use when:
|
||||
* - The active provider has the streaming surface enabled and a voice picked.
|
||||
*
|
||||
* Expects:
|
||||
* - `audioContext` is the same context the playback manager's `play`
|
||||
* callback will use to attach the buffer source. Mismatched contexts will
|
||||
* throw on decode.
|
||||
*
|
||||
* Returns:
|
||||
* - A {@link StageTtsSession} whose `appendSpecial` is intentionally
|
||||
* immediate (no segmenter to align with); see {@link createStageTtsSession}.
|
||||
*/
|
||||
export function createStreamingTtsSession<TAudio = AudioBuffer>(
|
||||
options: CreateStreamingSessionOptions<TAudio>,
|
||||
): StageTtsSession {
|
||||
const { intentId, snapshot, audioContext, playbackManager, hooks } = options
|
||||
const pipelineFactory = options.pipelineFactory ?? createStreamingTtsPipeline
|
||||
|
||||
let sequence = 0
|
||||
let terminated = false
|
||||
|
||||
const handle = pipelineFactory({
|
||||
model: snapshot.model,
|
||||
voice: snapshot.voice,
|
||||
audioContext,
|
||||
bufferEntireSession: snapshot.bufferEntireSession,
|
||||
extraBody: snapshot.extraBody,
|
||||
onSentence: ({ index, text, audio }) => {
|
||||
if (terminated)
|
||||
return
|
||||
playbackManager.schedule({
|
||||
id: `${intentId}-${index}`,
|
||||
streamId: intentId,
|
||||
intentId,
|
||||
segmentId: `${intentId}-${index}`,
|
||||
sequence: sequence++,
|
||||
ownerId: snapshot.ownerId,
|
||||
priority: 0,
|
||||
text: text ?? '',
|
||||
special: null,
|
||||
audio: audio as unknown as TAudio,
|
||||
createdAt: Date.now(),
|
||||
})
|
||||
},
|
||||
onError: (err) => {
|
||||
hooks?.onError?.(err)
|
||||
},
|
||||
onDone: () => {
|
||||
terminated = true
|
||||
hooks?.onDone?.()
|
||||
},
|
||||
})
|
||||
|
||||
function cancel(reason?: string) {
|
||||
if (terminated) {
|
||||
// Pipeline already closed itself; still drain any playback items it
|
||||
// managed to queue before terminating.
|
||||
playbackManager.stopByIntent(intentId, reason ?? 'session-already-terminated')
|
||||
return
|
||||
}
|
||||
terminated = true
|
||||
handle.cancel()
|
||||
playbackManager.stopByIntent(intentId, reason ?? 'session-cancelled')
|
||||
}
|
||||
|
||||
return {
|
||||
intentId,
|
||||
appendText: handle.appendText,
|
||||
// Streaming has no in-band queue to align audio with; fire the host's
|
||||
// immediate-special callback so emotion / delay tokens still reach the
|
||||
// queues. The perceptual mis-alignment vs audio is minor — codex
|
||||
// review and the segmenter parity comment in Stage.vue prior to the
|
||||
// refactor both accept it.
|
||||
appendSpecial: snapshot.onImmediateSpecial,
|
||||
finishInput: handle.finish,
|
||||
end: () => {
|
||||
// `finish()` already drove session.finished → onDone; nothing extra
|
||||
// to do here. Kept as a method for protocol symmetry with the
|
||||
// segmenter adapter.
|
||||
},
|
||||
cancel,
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Speech transport flavours the host knows how to drive. Mirrors the
|
||||
* `capabilities.speech.transport` enum on `ProviderDefinition`. Anything
|
||||
* the factory does not recognise is treated as `'rest'`.
|
||||
*/
|
||||
export type SpeechTransport = 'rest' | 'bidirectional-ws'
|
||||
|
||||
/**
|
||||
* Build context required by {@link createStageTtsSession}. Kept as a
|
||||
* single object so Stage.vue can construct it once per intent without
|
||||
* threading a dozen positional args.
|
||||
*/
|
||||
export interface StageTtsSessionContext<TAudio = AudioBuffer> {
|
||||
/**
|
||||
* Transport flavour of the active provider, read by the host from
|
||||
* `ProviderDefinition.capabilities.speech.transport`. `'rest'` (or any
|
||||
* other value) routes through the segmenter; `'bidirectional-ws'`
|
||||
* routes through the streaming WebSocket adapter.
|
||||
*/
|
||||
transport: SpeechTransport | undefined
|
||||
/**
|
||||
* Snapshot of the streaming provider's settings. Used only when
|
||||
* `transport === 'bidirectional-ws'` and the snapshot is non-null with
|
||||
* a voice picked. Returning `null` is a graceful "not ready yet" signal
|
||||
* and falls back to the segmenter path.
|
||||
*/
|
||||
streaming?: () => StreamingSessionSnapshot | null
|
||||
/** Host audio context. Required for the streaming path. */
|
||||
audioContext: BaseAudioContext | undefined
|
||||
/** Playback manager Stage uses for scheduling. */
|
||||
playbackManager: PlaybackManagerSubset<TAudio>
|
||||
/**
|
||||
* Factory for a fresh `IntentHandle` (segmenter path). Stage wires this
|
||||
* to `speechRuntimeStore.openIntent`.
|
||||
*/
|
||||
openIntent: (options: IntentOptions) => IntentHandleSubset
|
||||
/**
|
||||
* Default intent options for the segmenter path. Stage.vue's existing
|
||||
* call passed `{ownerId, priority:'normal', behavior:'queue'}`; we keep
|
||||
* the same defaults here.
|
||||
*/
|
||||
intentOptions: () => IntentOptions
|
||||
/** Lifecycle hooks shared by both paths. */
|
||||
hooks?: StreamingSessionHooks
|
||||
}
|
||||
|
||||
/**
|
||||
* One decision point: streaming path or segmenter path? Returns a fully
|
||||
* wired {@link StageTtsSession} so Stage.vue's chat-orchestrator hooks
|
||||
* never branch on provider id again.
|
||||
*
|
||||
* Use when:
|
||||
* - `onBeforeMessageComposed` fires and Stage needs a fresh session for
|
||||
* the next LLM intent.
|
||||
*
|
||||
* Expects:
|
||||
* - Caller has already cancelled / cleared any previous session ref.
|
||||
* - When `transport === 'bidirectional-ws'`, the snapshot's `voice` is
|
||||
* a real voice id and `audioContext` is set; otherwise the factory
|
||||
* silently falls back to the segmenter path (codex review MEDIUM #3
|
||||
* noted this fallback should not silently re-enter the legacy
|
||||
* per-segment path inside `tts()` — the segmenter adapter routes
|
||||
* through the normal segmenter+tts callback, which is the intended
|
||||
* behaviour for every REST provider).
|
||||
*
|
||||
* Returns:
|
||||
* - A `StageTtsSession`. Stage.vue stores it in a single `currentSession`
|
||||
* ref and calls `appendText` / `appendSpecial` / `finishInput` / `end`
|
||||
* / `cancel` on it from the hooks.
|
||||
*/
|
||||
export function createStageTtsSession<TAudio = AudioBuffer>(
|
||||
ctx: StageTtsSessionContext<TAudio>,
|
||||
): StageTtsSession {
|
||||
const wantsStreaming = ctx.transport === 'bidirectional-ws'
|
||||
const snapshot = wantsStreaming ? ctx.streaming?.() ?? null : null
|
||||
const canStream = wantsStreaming
|
||||
&& snapshot != null
|
||||
&& snapshot.voice.length > 0
|
||||
&& ctx.audioContext != null
|
||||
|
||||
if (!canStream) {
|
||||
// Segmenter path: open the existing IntentHandle and adapt 1:1.
|
||||
return fromIntent(ctx.openIntent(ctx.intentOptions()))
|
||||
}
|
||||
|
||||
const intentId = createStreamingIntentId()
|
||||
return createStreamingTtsSession<TAudio>({
|
||||
intentId,
|
||||
snapshot: snapshot!,
|
||||
audioContext: ctx.audioContext!,
|
||||
playbackManager: ctx.playbackManager,
|
||||
hooks: ctx.hooks,
|
||||
})
|
||||
}
|
||||
|
||||
function createStreamingIntentId(): string {
|
||||
return `stream-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`
|
||||
}
|
||||
Reference in New Issue
Block a user