fix(server): tts streaming issue

apps/server/src/utils/redis-keys.ts

@@ -23,10 +23,6 @@ export function userFluxRedisKey(userId: string): string {
   return createRedisKey('user', userId, 'flux')
 }
 
-export function ttsVoicesUpstreamCacheRedisKey(model: string): string {
-  return createRedisKey('tts', 'voices', 'upstream', model)
-}
-
 export function userFluxMeterDebtRedisKey(userId: string, meterName: string): string {
   return createRedisKey('user', userId, 'flux-meter', meterName, 'debt')
 }

packages/stage-ui/src/libs/chat-sync/ws-client.test.ts

@@ -1,6 +1,7 @@
 import { describe, expect, it } from 'vitest'
+import { ref } from 'vue'
 
-import { buildChatWsUrl, computeReconnectDelay, mapStatus, WS_CLOSE_UNAUTHORIZED } from './ws-client'
+import { buildChatWsUrl, computeReconnectDelay, createChatWsUrlRef, mapStatus, WS_CLOSE_UNAUTHORIZED } from './ws-client'
 
 describe('buildChatWsUrl', () => {
   /**
@@ -106,7 +107,8 @@ describe('mapStatus', () => {
     expect(mapStatus('CLOSED', false)).toBe('idle')
   })
 })
-describe('wS_CLOSE_UNAUTHORIZED', () => {
+
+describe('ws_CLOSE_UNAUTHORIZED', () => {
   // ROOT CAUSE:
   //
   // Browsers do not expose the HTTP 401 status to the WebSocket `close`
@@ -125,3 +127,56 @@ describe('wS_CLOSE_UNAUTHORIZED', () => {
     expect(WS_CLOSE_UNAUTHORIZED).toBe(4001)
   })
 })
+
+describe('createChatWsUrlRef', () => {
+  it('returns undefined when disabled regardless of token', () => {
+    const enabled = ref(false)
+    const url = createChatWsUrlRef(enabled, () => 'tok', 'https://api.example.com')
+    expect(url.value).toBeUndefined()
+  })
+
+  it('returns undefined when getToken yields null/empty', () => {
+    const enabled = ref(true)
+    const nullUrl = createChatWsUrlRef(enabled, () => null, 'https://api.example.com')
+    expect(nullUrl.value).toBeUndefined()
+  })
+
+  // ROOT CAUSE:
+  //
+  // Production wired `getToken: () => localStorage.getItem('auth/v1/token')`.
+  // The Vue `computed` cannot track non-reactive reads (DOM storage,
+  // module-level let, etc.), so the URL froze at first evaluation. After an
+  // OIDC `oauth2/token` refresh wrote a new access token into localStorage,
+  // `useWebSocket` kept reconnecting with the stale token in the query
+  // string, producing an infinite `/ws/chat?token=<old>` → 401 loop until
+  // the user reloaded the tab.
+  //
+  // Fix: callers MUST pass a closure that reads from a reactive source
+  // (Pinia store ref / Vue ref / computed). The two cases below pin the
+  // contract: reactive source rebuilds the URL on rotation; non-reactive
+  // source intentionally does NOT (so future regressions show up here).
+  it('rebuilds url when getToken reads a reactive ref (token rotation)', () => {
+    const enabled = ref(true)
+    const tokenRef = ref<string | null>('old-token')
+    const url = createChatWsUrlRef(enabled, () => tokenRef.value, 'https://api.example.com')
+
+    expect(url.value).toBe('wss://api.example.com/ws/chat?token=old-token')
+    tokenRef.value = 'new-token'
+    expect(url.value).toBe('wss://api.example.com/ws/chat?token=new-token')
+  })
+
+  it('freezes ws URL when getToken is non-reactive (regression guard)', () => {
+    const enabled = ref(true)
+    // Module-local let stands in for `localStorage.getItem` — neither is a
+    // Vue reactive dep, so the computed cannot observe mutations.
+    let storage: string | null = 'frozen-token'
+    const url = createChatWsUrlRef(enabled, () => storage, 'https://api.example.com')
+
+    expect(url.value).toBe('wss://api.example.com/ws/chat?token=frozen-token')
+    storage = 'rotated-token'
+    // Still the old value — this is what broke production. If this ever
+    // starts returning 'rotated-token' Vue's reactivity model changed and
+    // the contract comment on createChatWsUrlRef can be relaxed.
+    expect(url.value).toBe('wss://api.example.com/ws/chat?token=frozen-token')
+  })
+})

packages/stage-ui/src/libs/chat-sync/ws-client.ts

@@ -1,4 +1,5 @@
 import type { NewMessagesPayload, PullMessagesRequest, PullMessagesResponse, SendMessagesRequest, SendMessagesResponse } from '@proj-airi/server-sdk-shared'
+import type { ComputedRef, Ref } from 'vue'
 
 import { defineInvoke } from '@moeru/eventa'
 import { createContext as createWsContext, wsErrorEvent } from '@moeru/eventa/adapters/websocket/native'
@@ -179,19 +180,36 @@ export function mapStatus(vue: 'OPEN' | 'CONNECTING' | 'CLOSED', enabled: boolea
  *   indefinitely (eventa@0.3.0 does not flush its internal pending maps when
  *   the underlying context is disposed; we wrap each invoke in a race).
  */
+/**
+ * Build the reactive ws URL ref `useWebSocket` watches.
+ *
+ * `getToken` MUST read from a reactive source (Pinia store ref, Vue ref,
+ * computed). A non-reactive read (e.g. `localStorage.getItem`) freezes the
+ * URL at first evaluation and `useWebSocket` will reconnect forever with
+ * the stale token after the next OIDC refresh — verified by
+ * `freezes ws URL when getToken is non-reactive` in ws-client.test.ts.
+ */
+export function createChatWsUrlRef(
+  enabled: Ref<boolean>,
+  getToken: () => string | null,
+  serverUrl: string,
+): ComputedRef<string | undefined> {
+  return computed(() => {
+    if (!enabled.value)
+      return undefined
+    const token = getToken()
+    if (!token)
+      return undefined
+    return buildChatWsUrl(serverUrl, token)
+  })
+}
+
 export function createChatWsClient(options: CreateChatWsClientOptions): ChatWsClient {
   // `enabled` mirrors user intent: connect() flips on, disconnect() flips off.
   // The url ref returns `undefined` when disabled, which makes useWebSocket
   // close cleanly without firing the auto-reconnect loop.
   const enabled = ref(false)
-  const urlRef = computed<string | undefined>(() => {
-    if (!enabled.value)
-      return undefined
-    const token = options.getToken()
-    if (!token)
-      return undefined
-    return buildChatWsUrl(options.serverUrl, token)
-  })
+  const urlRef = createChatWsUrlRef(enabled, options.getToken, options.serverUrl)
 
   // The eventa context is rebuilt on every `onConnected` so RPC + push
   // listeners survive a reconnect by re-binding to the fresh ws.

packages/stage-ui/src/libs/speech/streaming-pipeline.ts

@@ -131,6 +131,21 @@ export function createStreamingTtsPipeline(options: StreamingTtsPipelineOptions)
   let chunks: ArrayBuffer[] = []
   let chunkBytes = 0
   let sentenceIndex = 0
+  /**
+   * Promise chain for serialized `flushAccumulatedAsSentence` invocations.
+   *
+   * Each `handleControlFrame` runs via `void handleControlFrame(...)`, so
+   * multiple control frames execute concurrently. Without serialization,
+   * `session.finished`'s synchronous `chunkBytes === 0` check fires
+   * immediately (the prior `sentence.end` already cleared the buffer
+   * synchronously before its `await decodeAudioData`), terminating the
+   * session before the last sentence's `decodeAudioData` resolves — its
+   * `onSentence` then arrives after `terminated = true` in tts-session.ts
+   * and gets dropped. Chaining all flushes through this single promise lets
+   * `requestTerminate` await the tail before tearing down.
+   */
+  let pendingFlush: Promise<void> = Promise.resolve()
+  let terminationRequested = false
   /**
    * FIFO of sentence texts seen via `sentence.start` events that haven't
    * been paired with a `sentence.end` yet. The protocol promises in-order
@@ -185,6 +200,27 @@ export function createStreamingTtsPipeline(options: StreamingTtsPipelineOptions)
     }
   }
 
+  function enqueueFlush(textOverride?: string): Promise<void> {
+    // `.catch(() => {})` keeps a single decode failure from poisoning the
+    // tail of the chain — failures already surface via `onError` inside
+    // `flushAccumulatedAsSentence`.
+    pendingFlush = pendingFlush.then(() => flushAccumulatedAsSentence(textOverride)).catch(() => {})
+    return pendingFlush
+  }
+
+  async function requestTerminate(err: Error | null) {
+    if (closed || terminationRequested)
+      return
+    terminationRequested = true
+    // Wait for every queued flush (and its `onSentence` dispatch) to drain
+    // before flipping `closed`. Without this await, late-resolving decodes
+    // would land after `onDone` has already set `terminated = true` in the
+    // consumer adapter and be dropped — that is the "last sentence missing"
+    // symptom observed in the wild.
+    await pendingFlush
+    terminate(err)
+  }
+
   ws.addEventListener('open', () => {
     const startFrame = {
       event: 'start',
@@ -234,7 +270,13 @@ export function createStreamingTtsPipeline(options: StreamingTtsPipelineOptions)
         if (bufferEntireSession)
           break
         const text = readSentenceText(evt.payload) ?? pendingSentenceTexts.shift() ?? ''
-        await flushAccumulatedAsSentence(text)
+        // Fire-and-forget into the serialized chain. We do NOT await here;
+        // awaiting from the message handler does not block sibling handlers
+        // (they run concurrently via `void handleControlFrame`), so an
+        // await would only delay this handler's own return without
+        // preventing the session.finished race. The chain itself is what
+        // enforces ordering.
+        void enqueueFlush(text)
         break
       }
       case 'subtitle': {
@@ -249,14 +291,14 @@ export function createStreamingTtsPipeline(options: StreamingTtsPipelineOptions)
       }
       case 'session.finished': {
         sawSessionFinished = true
-        await flushAccumulatedAsSentence()
-        terminate(null)
+        void enqueueFlush()
+        void requestTerminate(null)
         break
       }
       case 'error': {
         const code = evt.code ?? 'streaming_tts_error'
         const message = evt.message ?? code
-        terminate(new Error(`${code}: ${message}`))
+        void requestTerminate(new Error(`${code}: ${message}`))
         break
       }
     }
@@ -266,15 +308,17 @@ export function createStreamingTtsPipeline(options: StreamingTtsPipelineOptions)
     if (closed)
       return
     if (sawSessionFinished) {
-      // Normal end after `session.finished`; flushAccumulatedAsSentence
-      // already ran. Just mark closed and notify.
-      terminate(null)
+      // Normal end after `session.finished`; the session.finished handler
+      // already enqueued the tail flush and called requestTerminate. Just
+      // make sure termination happens even if that path somehow didn't
+      // (idempotent — requestTerminate guards against re-entry).
+      void requestTerminate(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}`))
+    void requestTerminate(new Error(`streaming_tts_closed: ${reason}`))
   })
 
   ws.addEventListener('error', () => {
@@ -312,12 +356,17 @@ export function createStreamingTtsPipeline(options: StreamingTtsPipelineOptions)
       safeSend(JSON.stringify({ event: 'finish' }))
     },
     cancel() {
-      if (closed)
+      if (closed || terminationRequested)
         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)
+      // Route through `requestTerminate` so any in-flight `decodeAudioData`
+      // can still resolve and emit `onSentence` before `onDone` flips the
+      // consumer's `terminated` flag. tts-session.ts then runs
+      // `stopByIntent` on the playback manager and drops whatever did
+      // schedule, so this does NOT prolong playback — it just keeps the
+      // termination semantics consistent across cancel / session.finished
+      // / error / close paths (codex review).
+      void requestTerminate(null)
     },
   }
 }

packages/stage-ui/src/stores/chat/session-store.ts

@@ -858,14 +858,7 @@ export const useChatSessionStore = defineStore('chat-session', () => {
     console.info('[chat-sync] creating WS client →', SERVER_URL)
     wsClient = createChatWsClient({
       serverUrl: SERVER_URL,
-      // NOTICE:
-      // `getAuthToken()` reads `localStorage` directly — that read is NOT
-      // reactive, so `ws-client`'s `urlRef = computed(() => ... getToken())`
-      // captures the initial token forever and ignores `oauth2/token`
-      // refreshes. The auto-reconnect loop then hammers `/ws/chat?token=<old>`
-      // and every upgrade returns 401 until the user reloads the tab.
-      // Read through the Pinia store ref so the computed actually tracks
-      // token rotation and rebuilds the URL.
+      // Reactive read — see `createChatWsUrlRef` contract.
       getToken: () => authToken.value,
     })