From 2854e0accccdb0bea2b81e3d0f48665721983bdd Mon Sep 17 00:00:00 2001 From: RainbowBird Date: Sun, 17 May 2026 13:42:41 +0800 Subject: [PATCH] fix(server): tts streaming issue --- apps/server/src/utils/redis-keys.ts | 4 - .../src/libs/chat-sync/ws-client.test.ts | 59 ++++++++++++++- .../stage-ui/src/libs/chat-sync/ws-client.ts | 34 +++++++-- .../src/libs/speech/streaming-pipeline.ts | 73 ++++++++++++++++--- .../stage-ui/src/stores/chat/session-store.ts | 9 +-- 5 files changed, 145 insertions(+), 34 deletions(-) diff --git a/apps/server/src/utils/redis-keys.ts b/apps/server/src/utils/redis-keys.ts index e462391b1..bfc220eee 100644 --- a/apps/server/src/utils/redis-keys.ts +++ b/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') } diff --git a/packages/stage-ui/src/libs/chat-sync/ws-client.test.ts b/packages/stage-ui/src/libs/chat-sync/ws-client.test.ts index f7cdf0b1a..18a58be78 100644 --- a/packages/stage-ui/src/libs/chat-sync/ws-client.test.ts +++ b/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=` → 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('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') + }) +}) diff --git a/packages/stage-ui/src/libs/chat-sync/ws-client.ts b/packages/stage-ui/src/libs/chat-sync/ws-client.ts index b46954302..77735124c 100644 --- a/packages/stage-ui/src/libs/chat-sync/ws-client.ts +++ b/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, + getToken: () => string | null, + serverUrl: string, +): ComputedRef { + 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(() => { - 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. diff --git a/packages/stage-ui/src/libs/speech/streaming-pipeline.ts b/packages/stage-ui/src/libs/speech/streaming-pipeline.ts index 56be4d0e7..31cad5ae9 100644 --- a/packages/stage-ui/src/libs/speech/streaming-pipeline.ts +++ b/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 = 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 { + // `.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) }, } } diff --git a/packages/stage-ui/src/stores/chat/session-store.ts b/packages/stage-ui/src/stores/chat/session-store.ts index aaba08c42..b1ecc1c49 100644 --- a/packages/stage-ui/src/stores/chat/session-store.ts +++ b/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=` - // 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, })