fix(speech): synchronize streaming catalog state (#2445)

2026-09-03 19:20:35 +08:00
parent 75bde7ee34
commit 2ca8339a49
14 changed files with 219 additions and 65 deletions
@@ -164,25 +164,17 @@ describe('official streaming speech provider settings', () => {
    const fetchMock = vi.fn<typeof fetch>(async input => responseFor(input))
    vi.stubGlobal('fetch', fetchMock)
    const pinia = createPinia()
-
    const { errors } = await renderPage(pinia)
-    const providerStore = useProviderStore(pinia)
-    const forceProviderConfigured = providerStore.forceProviderConfigured
-    let finishConfiguration: (() => void) | undefined
-    const configurationPending = new Promise<void>((resolve) => {
-      finishConfiguration = resolve
-    })
-    vi.spyOn(providerStore, 'forceProviderConfigured').mockImplementation(async (requestedProviderId) => {
-      await configurationPending
-      forceProviderConfigured(requestedProviderId)
+    const actionOrder: string[] = []
+    useProviderStore(pinia).$onAction(({ after, name }) => {
+      if (name === 'forceProviderConfigured')
+        after(() => actionOrder.push('configured'))
+      if (name === 'listProviderVoices')
+        actionOrder.push('voices')
    })

    useAuthStore(pinia).$patch({ user, session })
-    await expect.poll(() => fetchMock.mock.calls.some(([input]) => responseUrl(input).includes('/api/v1/audio/models/streaming'))).toBe(true)
-    expect(fetchMock.mock.calls.some(([input]) => responseUrl(input).includes('/api/v1/audio/voices/streaming'))).toBe(false)
-
-    finishConfiguration?.()
-    await expect.poll(() => fetchMock.mock.calls.some(([input]) => responseUrl(input).includes('/api/v1/audio/voices/streaming'))).toBe(true)
+    await expect.poll(() => actionOrder).toEqual(['configured', 'voices'])
    expect(errors.map(error => errorMessageFrom(error))).toEqual([])
  })

@@ -221,6 +213,7 @@ describe('official streaming speech provider settings', () => {
    expect(setProviderAvailabilityOverride).not.toHaveBeenCalledWith(providerId, false)
    expect(setProviderUnconfigured).not.toHaveBeenCalledWith(providerId)
    expect(providerConfigStore.providers[providerId]?.status).toBe('configured')
+    await expect.poll(() => fetchMock.mock.calls.some(([input]) => responseUrl(input).includes('/api/v1/audio/voices/streaming'))).toBe(true)
    expect(errors.map(error => errorMessageFrom(error))).toEqual([])
  })
 })
@@ -85,6 +85,8 @@ export interface ProviderModelCatalog {
  models: ModelInfo[]
  /** Whether the server exposes this catalog. Absent when discovery did not return an authoritative state. */
  available?: boolean
+  /** Last authoritative availability retained by the action owner after discovery fails. */
+  lastKnownAvailable?: boolean
  /** Server-selected model id, or null when the server has no default. */
  defaultModel?: string | null
 }
@@ -13,7 +13,7 @@ import { useProviderStore } from '@proj-airi/stage-ui/stores/providers/provider'
 import { Callout, ComboboxSelect } from '@proj-airi/ui'
 import { computedAsync } from '@vueuse/core'
 import { storeToRefs } from 'pinia'
-import { computed, ref, watch } from 'vue'
+import { computed, shallowRef, watch } from 'vue'
 import { useI18n } from 'vue-i18n'
 import { useRouter } from 'vue-router'

@@ -21,7 +21,7 @@ const router = useRouter()
 const { t } = useI18n()
 const authStore = useAuthStore()
 const providersStore = useProviderStore()
-const providerStore = useProviderConfigStore()
+const providerConfigStore = useProviderConfigStore()
 const speechStore = useSpeechStore()
 const { isAuthenticated, credits, needsLogin } = storeToRefs(authStore)

@@ -33,7 +33,7 @@ const providerMetadata = computedAsync(() => selectProviderMetadata(
 ))
 const fluxPurchaseDisabled = isFluxPurchaseDisabled()

-const providerConfig = computed(() => providerStore.getProviderConfig(providerId))
+const providerConfig = computed(() => providerConfigStore.getProviderConfig(providerId))

 // Model picker. The catalog and the default model id both come from the
 // server's `/api/v1/audio/models/streaming` response (operator-controlled
@@ -41,22 +41,19 @@ const providerConfig = computed(() => providerStore.getProviderConfig(providerId
 // adding ICL / other backends doesn't need a UI release.
 const providerModels = computed(() => providersStore.getModelsForProvider(providerId))
 const modelsLoading = computed(() => providersStore.isLoadingModels[providerId] || false)
-const serverDefaultModel = ref<string | null>(null)
-const streamingAvailable = ref(false)
-const model = computed({
-  get(): string {
-    return (providerConfig.value?.model as string | undefined) ?? serverDefaultModel.value ?? ''
-  },
-  set(val: string) {
-    const config = providerConfig.value
-    if (config)
-      config.model = val
-  },
-})
+const serverDefaultModel = shallowRef<string | null>(null)
+const streamingAvailable = shallowRef(false)
+const model = computed(() => (providerConfig.value?.model as string | undefined) ?? serverDefaultModel.value ?? '')
 const modelOptions = computed(() => providerModels.value.map(m => ({ label: m.name, value: m.id })))

 const availableVoices = computed(() => speechStore.availableVoices[providerId] || [])
-const voicesLoading = ref(false)
+const voicesLoading = shallowRef(false)
+
+async function setModel(value: string | number | undefined) {
+  if (typeof value !== 'string')
+    return
+  await providerConfigStore.setProviderModel(providerId, value)
+}

 async function loadVoices() {
  voicesLoading.value = true
@@ -71,10 +68,11 @@ async function loadVoices() {
 watch(isAuthenticated, async (authenticated, _, onCleanup) => {
  let active = true
  onCleanup(() => active = false)
+  if (!authenticated) {
    streamingAvailable.value = false
    serverDefaultModel.value = null
-  if (!authenticated)
    return
+  }

  await providersStore.initializeProvider(providerId)
  if (!active)
@@ -87,8 +85,13 @@ watch(isAuthenticated, async (authenticated, _, onCleanup) => {
  // An absent value means that discovery failed before the server returned an
  // authoritative state. Keep the last configured state and availability
  // override so a transient request failure cannot hide the provider.
-  if (catalog.available === undefined)
+  if (catalog.available === undefined) {
+    // Discovery did not produce a new authoritative state. Reuse the state
+    // returned by the leader so a late follower snapshot cannot disable
+    // model-scoped voice loading for the rest of this page mount.
+    streamingAvailable.value = catalog.lastKnownAvailable === true
    return
+  }

  const available = catalog.available
  await providersStore.setProviderAvailabilityOverride(providerId, available)
@@ -110,9 +113,8 @@ watch(isAuthenticated, async (authenticated, _, onCleanup) => {
  // first model in the same catalog response. Do not read synchronized model
  // state here because its follower snapshot can arrive after the action.
  serverDefaultModel.value = catalog.defaultModel ?? catalog.models[0]?.id ?? null
-  const config = providerConfig.value
-  if (config && !config.model && serverDefaultModel.value)
-    config.model = serverDefaultModel.value
+  if (serverDefaultModel.value)
+    await providerConfigStore.setProviderModelIfUnset(providerId, serverDefaultModel.value)
 }, { immediate: true })

 // Volcengine TTS 1.0 and 2.0 ship different voice catalogues (mars/moon/ICL
@@ -222,10 +224,11 @@ function handleLogin() {
            <p>Pick the streaming TTS model variant. All variants share the same voice catalogue today.</p>
          </Callout>
          <ComboboxSelect
-            v-model="model"
+            :model-value="model"
            :options="modelOptions"
            :disabled="modelsLoading || !providerConfig"
            placeholder="Choose a model..."
+            @update:model-value="setModel"
          />
        </div>

@@ -36,7 +36,7 @@ import { useIOTraceBridge } from '../../composables/use-io-trace-bridge'
 import { initIOTracer } from '../../composables/use-io-tracer'
 import { Emotion, EMOTION_EmotionMotionName_value, EMOTION_VRMExpressionName_value, EmotionThinkMotionName } from '../../constants/emotions'
 import { live2dMotionMagicProfiles, useLive2DMotionMagic, useLive2DMotionMagicSettings } from '../../features/motions/live2d'
-import { getDefaultStreamingModel, getDefinedProvider } from '../../libs/providers/providers'
+import { getDefinedProvider } from '../../libs/providers/providers'
 import { OFFICIAL_SPEECH_PROVIDER_ID, OFFICIAL_SPEECH_STREAMING_PROVIDER_ID } from '../../libs/providers/providers/official'
 import { bindSpeakingStateToPlaybackManager } from '../../libs/speech/playback-speaking-state'
 import { createStageTtsSession } from '../../libs/speech/tts-session'
@@ -729,7 +729,9 @@ function stopSpeechOutput(reason: string) {
 */
 function resolveStreamingSessionModel(): string | null {
  const activeModel = activeSpeechModel.value as string | undefined
-  const sessionModel = activeModel?.includes('/') ? activeModel : getDefaultStreamingModel()
+  const sessionModel = activeModel?.includes('/')
+    ? activeModel
+    : providersStore.getDefaultModelForProvider(OFFICIAL_SPEECH_STREAMING_PROVIDER_ID)
  if (!sessionModel?.includes('/'))
    return null
  return sessionModel
@@ -12,7 +12,6 @@ import './official'
 registerProviders(portableProviderDefinitions)

 export {
-  getDefaultStreamingModel,
  OFFICIAL_TRANSCRIPTION_PROVIDER_ID,
 } from './official'

@@ -33,15 +33,6 @@ export function getDefaultSpeechModel(): string | null {
  return defaultSpeechModelId
 }

-// Server-curated default streaming model id, populated by the streaming
-// provider's listModels(). Pages that need to seed an initial model selection
-// read this via getDefaultStreamingModel() instead of hardcoding an id.
-let defaultStreamingModelId: string | null = null
-
-export function getDefaultStreamingModel(): string | null {
-  return defaultStreamingModelId
-}
-
 const officialConfigSchema = z.object({})

 function authHeaders(): Record<string, string> {
@@ -57,22 +48,19 @@ async function listStreamingModelCatalog(): Promise<ProviderModelCatalog> {
  // (`UNSPEECH_UPSTREAM.streaming`). Wire shape uses `<backend>/<api_resource_id>`
  // (see `unspeech/docs/wire-protocols/audio-speech-stream-v1.md`); the
  // server returns whatever the operator put there, no client-side defaults.
-  // Reset the default up front so a failed probe cannot retain stale data.
-  defaultStreamingModelId = null
-
  const res = await globalThis.fetch(`${SERVER_URL}/api/v1/audio/models/streaming`, { headers: authHeaders() })
  if (!res.ok)
    throw new Error(`streaming models upstream ${res.status}: ${await res.text().catch(() => '')}`.slice(0, 256))

-  const data = await res.json() as { available?: boolean, models: { id: string, name?: string, description?: string }[], default?: string | null }
-  if (!Array.isArray(data.models))
-    throw new Error('streaming models upstream missing models[]')
-
-  defaultStreamingModelId = typeof data.default === 'string' && data.default.length > 0 ? data.default : null
+  const data = await res.json() as {
+    available: boolean
+    models: { id: string, name?: string, description?: string }[]
+    default: string | null
+  }

  return {
-    available: data.available === true,
-    defaultModel: defaultStreamingModelId,
+    available: data.available,
+    defaultModel: data.default ?? null,
    models: data.models.map(m => ({
      id: m.id,
      name: m.name ?? m.id,
@@ -219,6 +219,37 @@ describe('speech store helpers', () => {
    expect(listVoices).not.toHaveBeenCalled()
  })

+  // ROOT CAUSE:
+  //
+  // Streaming discovery kept its default model in one renderer's module
+  // variable. Another renderer received the synchronized models but selected
+  // the first entry instead of the operator default.
+  //
+  // Before: read getDefaultStreamingModel() from renderer-local memory.
+  //
+  // We fixed this by storing the default beside the synchronized model catalog.
+  it('selects the synchronized streaming default instead of the first model', async () => {
+    const providersStore = useProviderStore()
+    vi.spyOn(providersStore, 'listProviderVoices').mockResolvedValue([])
+    const speechStore = useSpeechStore()
+    await providersStore.initializeProvider(OFFICIAL_SPEECH_STREAMING_PROVIDER_ID)
+    providersStore.providerRuntimeState[OFFICIAL_SPEECH_STREAMING_PROVIDER_ID] = {
+      models: [
+        { id: 'volcengine/seed-tts-1.0', name: 'Seed TTS 1.0', provider: OFFICIAL_SPEECH_STREAMING_PROVIDER_ID },
+        { id: 'volcengine/seed-tts-2.0', name: 'Seed TTS 2.0', provider: OFFICIAL_SPEECH_STREAMING_PROVIDER_ID },
+      ],
+      defaultModel: 'volcengine/seed-tts-2.0',
+      modelStatus: 'ready',
+      modelError: null,
+    }
+    speechStore.activeSpeechProvider = OFFICIAL_SPEECH_STREAMING_PROVIDER_ID
+    speechStore.activeSpeechModel = ''
+
+    speechStore.ensureActiveSpeechModel()
+
+    expect(speechStore.activeSpeechModel).toBe('volcengine/seed-tts-2.0')
+  })
+
  /**
   * @example
   * speechStore.ensureActiveSpeechModel()
@@ -14,7 +14,7 @@ import { useI18n } from 'vue-i18n'
 import { toXml } from 'xast-util-to-xml'
 import { x } from 'xastscript'

-import { getDefaultSpeechModel, getDefaultStreamingModel, OFFICIAL_SPEECH_PROVIDER_ID, OFFICIAL_SPEECH_STREAMING_PROVIDER_ID, setupOfficialSpeechAutoPick } from '../../libs/providers/providers/official'
+import { getDefaultSpeechModel, OFFICIAL_SPEECH_PROVIDER_ID, OFFICIAL_SPEECH_STREAMING_PROVIDER_ID, setupOfficialSpeechAutoPick } from '../../libs/providers/providers/official'
 import { useProviderConfigStore } from '../providers/config'
 import { useProviderStore } from '../providers/provider'

@@ -174,7 +174,7 @@ export const useSpeechStore = defineStore('speech', () => {
    // When no default can be resolved yet (catalog not loaded), clear it to ''
    // so callers pass `undefined` (server returns the full streaming catalog)
    // rather than forwarding a stale non-streaming model id as `?model=`.
-    const nextModel = getDefaultStreamingModel() ?? streamingModels[0]?.id ?? ''
+    const nextModel = providersStore.getDefaultModelForProvider(OFFICIAL_SPEECH_STREAMING_PROVIDER_ID) ?? streamingModels[0]?.id ?? ''
    if (activeSpeechModel.value === nextModel)
      return
    activeSpeechModel.value = nextModel
@@ -153,6 +153,42 @@ export const useProviderConfigStore = defineStore('provider-config', () => {
      provider.status = status
  }

+  /**
+   * Updates the selected model in the leader-owned provider snapshot.
+   *
+   * Follower renderers must await this action instead of mutating replicated
+   * configuration directly, because `state: true` proposals contain the full
+   * store and can overwrite newer leader state.
+   */
+  async function setProviderModel(providerId: string, model: string) {
+    const provider = providers.value[providerId]
+    if (!provider)
+      return
+
+    providers.value[providerId] = {
+      ...provider,
+      config: { ...provider.config, model },
+    }
+  }
+
+  /**
+   * Seeds a discovered default without replacing a model selected by the user.
+   */
+  async function setProviderModelIfUnset(providerId: string, model: string) {
+    const provider = providers.value[providerId]
+    if (!provider)
+      return
+
+    const currentModel = provider.config.model
+    if (typeof currentModel === 'string' && currentModel.length > 0)
+      return
+
+    providers.value[providerId] = {
+      ...provider,
+      config: { ...provider.config, model },
+    }
+  }
+
  function mergeProviderSnapshot(snapshot: Record<string, InferenceServiceProvider>) {
    providers.value = { ...providers.value, ...snapshot }
    for (const providerId of Object.keys(snapshot))
@@ -252,6 +288,8 @@ export const useProviderConfigStore = defineStore('provider-config', () => {
    markProviderAdded,
    unmarkProviderAdded,
    setProviderStatus,
+    setProviderModel,
+    setProviderModelIfUnset,
    fetchProviders,
    addProvider,
    removeProvider,
@@ -266,6 +304,8 @@ export const useProviderConfigStore = defineStore('provider-config', () => {
      'markProviderAdded',
      'unmarkProviderAdded',
      'setProviderStatus',
+      'setProviderModel',
+      'setProviderModelIfUnset',
      'addProvider',
      'removeProvider',
      'updateProviderConfig',
@@ -105,5 +105,59 @@ describe('provider model catalog synchronization', () => {
        }),
      ],
    })
+    await vi.waitFor(() => expect(followerContext.providerStore.getDefaultModelForProvider(OFFICIAL_SPEECH_STREAMING_PROVIDER_ID)).toBe('volcengine/seed-tts-2.0'))
+
+    // https://github.com/moeru-ai/airi/pull/2445#discussion_r3913843853
+    // ROOT CAUSE:
+    //
+    // A function returned from a Pinia setup store becomes an action. The
+    // default-model lookup was a pure read, but it still ran action hooks.
+    //
+    // Before: getDefaultModelForProvider was a returned store function.
+    //
+    // We fixed this by exposing the parameterized lookup as a computed getter.
+    const actionNames: string[] = []
+    followerContext.providerStore.$onAction(({ name }) => actionNames.push(name))
+
+    expect(followerContext.providerStore.getDefaultModelForProvider(OFFICIAL_SPEECH_STREAMING_PROVIDER_ID)).toBe('volcengine/seed-tts-2.0')
+    expect(actionNames).not.toContain('getDefaultModelForProvider')
+  })
+
+  // https://github.com/moeru-ai/airi/pull/2440#discussion_r3912911639
+  // ROOT CAUSE:
+  //
+  // The settings renderer wrote the discovered default into its follower
+  // snapshot. The resulting full-state proposal could overwrite newer leader
+  // state, and the write was skipped when that snapshot arrived late.
+  //
+  // Before: mutate providerConfig.model in the follower page.
+  //
+  // We fixed this by routing model updates to awaited leader-owned actions.
+  it('applies defaults through the leader without replacing a user selection', async () => {
+    const namespace = `provider-model-default:${crypto.randomUUID()}`
+    const leaderContext = createSyncedContext(namespace, 'leader-only')
+    await vi.waitFor(() => expect(leaderContext.runtime.isLeader()).toBe(true))
+
+    const followerContext = createSyncedContext(namespace, 'follower-only')
+    await vi.waitFor(() => expect(followerContext.runtime.getLeaderId()).toBe(leaderContext.runtime.participantId))
+
+    await followerContext.providerStore.initializeProvider(OFFICIAL_SPEECH_STREAMING_PROVIDER_ID)
+    await followerContext.providerConfigStore.setProviderModelIfUnset(
+      OFFICIAL_SPEECH_STREAMING_PROVIDER_ID,
+      'volcengine/seed-tts-2.0',
+    )
+    await vi.waitFor(() => expect(followerContext.providerConfigStore.getProviderConfig(OFFICIAL_SPEECH_STREAMING_PROVIDER_ID)?.model).toBe('volcengine/seed-tts-2.0'))
+
+    await leaderContext.providerConfigStore.setProviderModel(
+      OFFICIAL_SPEECH_STREAMING_PROVIDER_ID,
+      'volcengine/seed-tts-1.0',
+    )
+    await followerContext.providerConfigStore.setProviderModelIfUnset(
+      OFFICIAL_SPEECH_STREAMING_PROVIDER_ID,
+      'volcengine/seed-tts-2.0',
+    )
+
+    expect(leaderContext.providerConfigStore.getProviderConfig(OFFICIAL_SPEECH_STREAMING_PROVIDER_ID)?.model).toBe('volcengine/seed-tts-1.0')
+    await vi.waitFor(() => expect(followerContext.providerConfigStore.getProviderConfig(OFFICIAL_SPEECH_STREAMING_PROVIDER_ID)?.model).toBe('volcengine/seed-tts-1.0'))
  })
 })
@@ -34,6 +34,7 @@ describe('provider store synchronization boundary', () => {
    const store = useProviderStore()
    const runtimeState = {
      models: [],
+      defaultModel: null,
      modelStatus: 'ready' as const,
      modelError: null,
    }
@@ -205,6 +206,7 @@ describe('provider store synchronization boundary', () => {

    store.providerRuntimeState['official-provider'] = {
      models: [],
+      defaultModel: null,
      modelStatus: 'loading',
      modelError: null,
    }
@@ -41,6 +41,7 @@ export type { ModelInfo, VoiceInfo } from '../../libs/providers/types'
 export interface ProviderRuntimeState {
  validatedCredentialHash?: string
  models: ModelInfo[]
+  defaultModel: string | null
  modelStatus: 'idle' | 'loading' | 'ready' | 'error'
  modelError: string | null
 }
@@ -365,6 +366,7 @@ export const useProviderStore = defineStore('provider', () => {
    if (!providerRuntimeState.value[providerId]) {
      providerRuntimeState.value[providerId] = {
        models: [],
+        defaultModel: null,
        modelStatus: 'idle',
        modelError: null,
      }
@@ -652,6 +654,7 @@ export const useProviderStore = defineStore('provider', () => {
          [providerId]: {
            ...currentRuntimeState,
            models: normalizedModels,
+            defaultModel: catalog.defaultModel ?? null,
            modelStatus: 'ready',
            modelError: null,
          },
@@ -679,7 +682,9 @@ export const useProviderStore = defineStore('provider', () => {
          },
        }
      }
-      return { models: [] }
+      const lastKnownAvailable = providerAvailabilityOverrides.value[providerId]
+        ?? (providerConfigStore.configuredProviders[providerId] ? true : undefined)
+      return { models: [], lastKnownAvailable }
    }
  }

@@ -688,6 +693,10 @@ export const useProviderStore = defineStore('provider', () => {
    return providerRuntimeState.value[providerId]?.models ?? emptyProviderModels
  }

+  const getDefaultModelForProvider = computed(() => (providerId: string) => {
+    return providerRuntimeState.value[providerId]?.defaultModel ?? null
+  })
+
  // Load models for all configured providers
  async function loadModelsForConfiguredProviders() {
    for (const providerId of availableProviders.value) {
@@ -1000,6 +1009,7 @@ export const useProviderStore = defineStore('provider', () => {
    modelLoadError,
    fetchModelsForProvider,
    getModelsForProvider,
+    getDefaultModelForProvider,
    listProviderVoices,
    loadProviderModel,
    loadModelsForConfiguredProviders,
@@ -156,7 +156,10 @@ export const streamingTtsUpstreamSchema = object({
    })),
    [],
  ),
-  defaultModel: optional(string()),
+  defaultModel: optional(pipe(
+    string(),
+    nonEmpty('UNSPEECH_UPSTREAM.streaming.defaultModel must not be empty'),
+  )),
 })

 export const unspeechUpstreamSchema = object({
@@ -92,6 +92,33 @@ describe('configKVService', () => {
      })
  })

+  // https://github.com/moeru-ai/airi/pull/2445#discussion_r3913931906
+  // ROOT CAUSE:
+  //
+  // The streaming TTS config accepted an empty default model. The catalog
+  // exposed that value as a present default, so clients skipped their fallback.
+  //
+  // Before: defaultModel used optional(string()).
+  //
+  // We fixed this by rejecting an empty configured default at the ConfigKV boundary.
+  it('rejects an empty streaming TTS default model', async () => {
+    store._store.set('UNSPEECH_UPSTREAM', JSON.stringify({
+      restBaseURL: 'http://unspeech.local:5933',
+      streaming: {
+        baseURL: 'wss://unspeech.local',
+        keys: [{ id: 'k1', ciphertext: 'enc' }],
+        defaultModel: '',
+      },
+    }))
+
+    await expect(service.getOptional('UNSPEECH_UPSTREAM'))
+      .rejects
+      .toMatchObject({
+        statusCode: 503,
+        errorCode: 'CONFIG_INVALID',
+      })
+  })
+
  it('wraps database failures as CONFIG_UNAVAILABLE', async () => {
    store.getRaw.mockRejectedValueOnce(new Error('database offline'))