From ba9247fb4788dd67f6993ac0126cb5135b8a74b4 Mon Sep 17 00:00:00 2001
From: RainbowBird <git@luoling.moe>
Date: Sun, 17 May 2026 01:35:40 +0800
Subject: [PATCH] feat(server,stage-ui): bidirectional streaming TTS + audio
 path refactor
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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.
---
 .../ai-context/verifications/streaming-tts.md |  21 +-
 apps/server/scripts/seed-streaming-tts.ts     | 172 +++++++++
 apps/server/src/app.ts                        |  15 +-
 apps/server/src/routes/openai/v1/index.ts     |  85 ++++-
 .../server/src/routes/openai/v1/route.test.ts |  76 +++-
 .../official-provider-speech-streaming.vue    | 191 ++++++++++
 .../stage-ui/src/components/scenes/Stage.vue  | 202 +++++++---
 packages/stage-ui/src/libs/index.ts           |   3 +
 .../providers/providers/official/index.ts     |  32 +-
 .../providers/providers/official/shared.ts    |  14 +
 packages/stage-ui/src/libs/providers/types.ts |  20 +
 .../libs/speech/streaming-pipeline.test.ts    | 297 +++++++++++++++
 .../src/libs/speech/streaming-pipeline.ts     | 346 ++++++++++++++++++
 .../src/libs/speech/tts-session.test.ts       | 336 +++++++++++++++++
 .../stage-ui/src/libs/speech/tts-session.ts   | 318 ++++++++++++++++
 15 files changed, 2033 insertions(+), 95 deletions(-)
 create mode 100644 apps/server/scripts/seed-streaming-tts.ts
 create mode 100644 packages/stage-pages/src/pages/settings/providers/speech/official-provider-speech-streaming.vue
 create mode 100644 packages/stage-ui/src/libs/speech/streaming-pipeline.test.ts
 create mode 100644 packages/stage-ui/src/libs/speech/streaming-pipeline.ts
 create mode 100644 packages/stage-ui/src/libs/speech/tts-session.test.ts
 create mode 100644 packages/stage-ui/src/libs/speech/tts-session.ts

diff --git a/apps/server/docs/ai-context/verifications/streaming-tts.md b/apps/server/docs/ai-context/verifications/streaming-tts.md
index 38287dbc5..570d8a3d1 100644
--- a/apps/server/docs/ai-context/verifications/streaming-tts.md
+++ b/apps/server/docs/ai-context/verifications/streaming-tts.md
@@ -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
 
diff --git a/apps/server/scripts/seed-streaming-tts.ts b/apps/server/scripts/seed-streaming-tts.ts
new file mode 100644
index 000000000..bd9ea6d6c
--- /dev/null
+++ b/apps/server/scripts/seed-streaming-tts.ts
@@ -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)
+})
diff --git a/apps/server/src/app.ts b/apps/server/src/app.ts
index 033657732..9d3a1ab22 100644
--- a/apps/server/src/app.ts
+++ b/apps/server/src/app.ts
@@ -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.
diff --git a/apps/server/src/routes/openai/v1/index.ts b/apps/server/src/routes/openai/v1/index.ts
index 34b67609e..22d23b607 100644
--- a/apps/server/src/routes/openai/v1/index.ts
+++ b/apps/server/src/routes/openai/v1/index.ts
@@ -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 }
 }
diff --git a/apps/server/src/routes/openai/v1/route.test.ts b/apps/server/src/routes/openai/v1/route.test.ts
index 642a09e67..e8b9d8f4e 100644
--- a/apps/server/src/routes/openai/v1/route.test.ts
+++ b/apps/server/src/routes/openai/v1/route.test.ts
@@ -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')
     })
   })
diff --git a/packages/stage-pages/src/pages/settings/providers/speech/official-provider-speech-streaming.vue b/packages/stage-pages/src/pages/settings/providers/speech/official-provider-speech-streaming.vue
new file mode 100644
index 000000000..c6f23c0e1
--- /dev/null
+++ b/packages/stage-pages/src/pages/settings/providers/speech/official-provider-speech-streaming.vue
@@ -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>
diff --git a/packages/stage-ui/src/components/scenes/Stage.vue b/packages/stage-ui/src/components/scenes/Stage.vue
index 8f4510569..526fccc39 100644
--- a/packages/stage-ui/src/components/scenes/Stage.vue
+++ b/packages/stage-ui/src/components/scenes/Stage.vue
@@ -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({
diff --git a/packages/stage-ui/src/libs/index.ts b/packages/stage-ui/src/libs/index.ts
index fd700b727..fe0d6f090 100644
--- a/packages/stage-ui/src/libs/index.ts
+++ b/packages/stage-ui/src/libs/index.ts
@@ -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'
diff --git a/packages/stage-ui/src/libs/providers/providers/official/index.ts b/packages/stage-ui/src/libs/providers/providers/official/index.ts
index 37b2c32fa..9f6d04cd2 100644
--- a/packages/stage-ui/src/libs/providers/providers/official/index.ts
+++ b/packages/stage-ui/src/libs/providers/providers/official/index.ts
@@ -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)
diff --git a/packages/stage-ui/src/libs/providers/providers/official/shared.ts b/packages/stage-ui/src/libs/providers/providers/official/shared.ts
index 0962b5002..c56f51995 100644
--- a/packages/stage-ui/src/libs/providers/providers/official/shared.ts
+++ b/packages/stage-ui/src/libs/providers/providers/official/shared.ts
@@ -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/`)
+}
diff --git a/packages/stage-ui/src/libs/providers/types.ts b/packages/stage-ui/src/libs/providers/types.ts
index 6f73da625..59b67411a 100644
--- a/packages/stage-ui/src/libs/providers/types.ts
+++ b/packages/stage-ui/src/libs/providers/types.ts
@@ -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
diff --git a/packages/stage-ui/src/libs/speech/streaming-pipeline.test.ts b/packages/stage-ui/src/libs/speech/streaming-pipeline.test.ts
new file mode 100644
index 000000000..cf9be0738
--- /dev/null
+++ b/packages/stage-ui/src/libs/speech/streaming-pipeline.test.ts
@@ -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)
+  })
+})
diff --git a/packages/stage-ui/src/libs/speech/streaming-pipeline.ts b/packages/stage-ui/src/libs/speech/streaming-pipeline.ts
new file mode 100644
index 000000000..56be4d0e7
--- /dev/null
+++ b/packages/stage-ui/src/libs/speech/streaming-pipeline.ts
@@ -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
+}
diff --git a/packages/stage-ui/src/libs/speech/tts-session.test.ts b/packages/stage-ui/src/libs/speech/tts-session.test.ts
new file mode 100644
index 000000000..e5e042f77
--- /dev/null
+++ b/packages/stage-ui/src/libs/speech/tts-session.test.ts
@@ -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)
+  })
+})
diff --git a/packages/stage-ui/src/libs/speech/tts-session.ts b/packages/stage-ui/src/libs/speech/tts-session.ts
new file mode 100644
index 000000000..f0652bff3
--- /dev/null
+++ b/packages/stage-ui/src/libs/speech/tts-session.ts
@@ -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)}`
+}