diff --git a/cspell.config.yaml b/cspell.config.yaml index 1057461f2..0e69379d7 100644 --- a/cspell.config.yaml +++ b/cspell.config.yaml @@ -305,6 +305,7 @@ words: - tresjs - Triggerable - tsdown + - ttft - turborepo - unbird - unbundle diff --git a/packages/stage-shared/src/perf/io-trace.ts b/packages/stage-shared/src/perf/io-trace.ts index 6b65beca7..f62edcbc3 100644 --- a/packages/stage-shared/src/perf/io-trace.ts +++ b/packages/stage-shared/src/perf/io-trace.ts @@ -1,13 +1,41 @@ export const IOSubsystems = { ASR: 'asr', LLM: 'llm', - TTSChunking: 'tts-chunking', - TTSSynthesis: 'tts-synthesis', + TTS: 'tts', Playback: 'playback', } as const - export type IOSubsystem = (typeof IOSubsystems)[keyof typeof IOSubsystems] +export const IOSpanNames = { + InteractionTurn: 'Interaction turn', + SpeechRecognition: 'Speech recognition', + LLMInference: 'LLM inference', + TTSSegment: 'TTS segment', + TTSSynthesis: 'TTS synthesis', + AudioPlayback: 'Audio playback', +} as const + +export const IOAttrs = { + Subsystem: 'io.subsystem', + ASRProvider: 'asr.provider', + ASRText: 'asr.text', + ASRAbort: 'asr.abort', + LLMModel: 'llm.model', + LLM_TTFT: 'llm.ttft_ms', + LLMTextLength: 'llm.text_length', + TTSSegmentId: 'tts.segment_id', + TTSText: 'tts.text', + TTSChunkReason: 'tts.chunk_reason', + TTSInterrupted: 'tts.interrupted', + TTSInterruptReason: 'tts.interrupt_reason', + TTSCanceled: 'tts.canceled', +} as const + +export const IOEvents = { + FirstToken: 'ai.moeru.airi.io.first_token', // Non-standard + SentenceEnd: 'ai.moeru.airi.io.sentence_end', // Non-standard +} as const + export interface IOSpan { id: string traceId: string diff --git a/packages/stage-ui/src/components/scenes/Stage.vue b/packages/stage-ui/src/components/scenes/Stage.vue index 0fb3e65de..652dd7137 100644 --- a/packages/stage-ui/src/components/scenes/Stage.vue +++ b/packages/stage-ui/src/components/scenes/Stage.vue @@ -26,6 +26,8 @@ import { computed, onMounted, onUnmounted, ref, watch } from 'vue' import { useDelayMessageQueue, useEmotionsMessageQueue } from '../../composables/queues' import { useAuthProviderSync } from '../../composables/use-auth-provider-sync' +import { useIOTraceBridge } from '../../composables/use-io-trace-bridge' +import { initIOTracer } from '../../composables/use-io-tracer' import { llmInferenceEndToken } from '../../constants' import { EMOTION_EmotionMotionName_value, EMOTION_VRMExpressionName_value, EmotionThinkMotionName } from '../../constants/emotions' import { useAudioContext, useSpeakingStore } from '../../stores/audio' @@ -321,6 +323,8 @@ const speechPipeline = createSpeechPipeline({ playback: playbackManager, }) +initIOTracer() +useIOTraceBridge(speechPipeline) void speechRuntimeStore.registerHost(speechPipeline) speechPipeline.on('onSpecial', (segment) => { diff --git a/packages/stage-ui/src/composables/use-io-trace-bridge.ts b/packages/stage-ui/src/composables/use-io-trace-bridge.ts new file mode 100644 index 000000000..c19e28355 --- /dev/null +++ b/packages/stage-ui/src/composables/use-io-trace-bridge.ts @@ -0,0 +1,135 @@ +import type { Span } from '@opentelemetry/api' +import type { createSpeechPipeline } from '@proj-airi/pipelines-audio' + +import { IOAttrs, IOSpanNames, IOSubsystems } from '@proj-airi/stage-shared' +import { onScopeDispose, watch } from 'vue' + +import { activeTurnSpan, startSpan } from './use-io-tracer' + +export function useIOTraceBridge(pipeline: ReturnType) { + const cleanupFns: (() => void)[] = [] + + const segmentSpans = new Map() + const synthesisSpans = new Map() + const playbackSpans = new Map() + const segmentReasons = new Map() + + let currentParent: Span | undefined + let hadSegments = false + + const stopWatch = watch(activeTurnSpan, (newSpan) => { + if (newSpan) { + currentParent = newSpan + hadSegments = false + } + }, { immediate: true }) + + function tryCloseTurn() { + if (hadSegments && segmentSpans.size === 0) { + activeTurnSpan.value?.end() + activeTurnSpan.value = undefined + } + } + + function closeSegment(segmentId: string) { + const segSpan = segmentSpans.get(segmentId) + if (segSpan) { + segSpan.end() + segmentSpans.delete(segmentId) + } + } + + cleanupFns.push(pipeline.on('onSegment', (segment) => { + segmentReasons.set(segment.segmentId, segment.reason) + })) + + cleanupFns.push(pipeline.on('onTtsRequest', (request) => { + let ttsSegmentSpan = segmentSpans.get(request.segmentId) + if (!ttsSegmentSpan) { + hadSegments = true + ttsSegmentSpan = startSpan(IOSpanNames.TTSSegment, currentParent, { + [IOAttrs.Subsystem]: IOSubsystems.TTS, + [IOAttrs.TTSSegmentId]: request.segmentId, + [IOAttrs.TTSText]: request.text, + [IOAttrs.TTSChunkReason]: segmentReasons.get(request.segmentId) ?? '', + }) + segmentReasons.delete(request.segmentId) + segmentSpans.set(request.segmentId, ttsSegmentSpan) + } + + const ttsSynthesisSpan = startSpan(IOSpanNames.TTSSynthesis, ttsSegmentSpan, { + [IOAttrs.Subsystem]: IOSubsystems.TTS, + [IOAttrs.TTSSegmentId]: request.segmentId, + [IOAttrs.TTSText]: request.text, + }) + synthesisSpans.set(request.segmentId, ttsSynthesisSpan) + })) + + cleanupFns.push(pipeline.on('onTtsResult', (result) => { + const span = synthesisSpans.get(result.segmentId) + if (span) { + span.end() + synthesisSpans.delete(result.segmentId) + } + })) + + cleanupFns.push(pipeline.on('onPlaybackStart', (event) => { + const segSpan = segmentSpans.get(event.item.segmentId) + const playbackSpan = startSpan(IOSpanNames.AudioPlayback, segSpan, { + [IOAttrs.Subsystem]: IOSubsystems.Playback, + [IOAttrs.TTSSegmentId]: event.item.segmentId, + [IOAttrs.TTSText]: event.item.text, + }) + playbackSpans.set(event.item.segmentId, playbackSpan) + })) + + cleanupFns.push(pipeline.on('onPlaybackEnd', (event) => { + const playbackSpan = playbackSpans.get(event.item.segmentId) + if (playbackSpan) { + playbackSpan.end() + playbackSpans.delete(event.item.segmentId) + } + closeSegment(event.item.segmentId) + tryCloseTurn() + })) + + cleanupFns.push(pipeline.on('onPlaybackInterrupt', (event) => { + const playbackSpan = playbackSpans.get(event.item.segmentId) + if (playbackSpan) { + playbackSpan.setAttribute(IOAttrs.TTSInterrupted, true) + playbackSpan.setAttribute(IOAttrs.TTSInterruptReason, event.reason) + playbackSpan.end() + playbackSpans.delete(event.item.segmentId) + } + closeSegment(event.item.segmentId) + tryCloseTurn() + })) + + cleanupFns.push(pipeline.on('onPlaybackReject', (event) => { + closeSegment(event.item.segmentId) + tryCloseTurn() + })) + + cleanupFns.push(pipeline.on('onIntentCancel', () => { + for (const [segmentId, span] of segmentSpans) { + span.setAttribute(IOAttrs.TTSCanceled, true) + span.end() + segmentSpans.delete(segmentId) + } + for (const [segmentId, span] of synthesisSpans) { + span.end() + synthesisSpans.delete(segmentId) + } + for (const [segmentId, span] of playbackSpans) { + span.end() + playbackSpans.delete(segmentId) + } + tryCloseTurn() + })) + + onScopeDispose(() => { + stopWatch() + for (const cleanup of cleanupFns) + cleanup() + }) +} diff --git a/packages/stage-ui/src/stores/chat.ts b/packages/stage-ui/src/stores/chat.ts index 6f1f7a0df..87f812622 100644 --- a/packages/stage-ui/src/stores/chat.ts +++ b/packages/stage-ui/src/stores/chat.ts @@ -5,6 +5,7 @@ import type { CommonContentPart, Message, ToolMessage } from '@xsai/shared-chat' import type { ChatAssistantMessage, ChatSlices, ChatStreamEventContext, StreamingAssistantMessage } from '../types/chat' import type { StreamEvent, StreamOptions } from './llm' +import { IOAttrs, IOEvents, IOSpanNames, IOSubsystems } from '@proj-airi/stage-shared' import { createQueue } from '@proj-airi/stream-kit' import { nanoid } from 'nanoid' import { defineStore, storeToRefs } from 'pinia' @@ -13,6 +14,7 @@ import { computed, ref, toRaw } from 'vue' import { useAnalytics } from '../composables' import { useLlmmarkerParser } from '../composables/llm-marker-parser' import { categorizeResponse, createStreamingCategorizer } from '../composables/response-categoriser' +import { activeTurnSpan, startSpan } from '../composables/use-io-tracer' import { formatContextPromptText } from './chat/context-prompt' import { createDatetimeContext, createMinecraftContext } from './chat/context-providers' import { useChatContextStore } from './chat/context-store' @@ -162,6 +164,7 @@ export const useChatOrchestratorStore = defineStore('chat-orchestrator', () => { return sending.value = true + let hadExistingTurn = false const isForegroundSession = () => sessionId === activeSessionId.value @@ -360,49 +363,74 @@ export const useChatOrchestratorStore = defineStore('chat-orchestrator', () => { if (shouldAbort()) return - await llmStore.stream(options.model, options.chatProvider, newMessages as Message[], { - headers, - tools: options.tools, - // NOTICE: xsai stream may emit `finish` before tool steps continue, so keep waiting until - // the final non-tool finish to avoid ending the chat turn with no assistant reply. - waitForTools: true, - onStreamEvent: async (event: StreamEvent) => { - switch (event.type) { - case 'tool-call': - toolCallQueue.enqueue({ - type: 'tool-call', - toolCall: event, - }) + hadExistingTurn = !!activeTurnSpan.value + if (!hadExistingTurn) + activeTurnSpan.value = startSpan(IOSpanNames.InteractionTurn) - break - case 'tool-result': - toolCallQueue.enqueue({ - type: 'tool-call-result', - id: event.toolCallId, - result: event.result, - }) - - break - case 'tool-error': - toolCallQueue.enqueue({ - type: 'tool-call-result', - id: event.toolCallId, - isError: true, - result: event.result, - }) - - break - case 'text-delta': - fullText += event.text - await parser.consume(event.text) - break - case 'finish': - break - case 'error': - throw event.error ?? new Error('Stream error') - } - }, + const llmSpan = startSpan(IOSpanNames.LLMInference, activeTurnSpan.value, { + [IOAttrs.Subsystem]: IOSubsystems.LLM, + [IOAttrs.LLMModel]: options.model, }) + const llmRequestTs = performance.now() + let llmFirstTokenEmitted = false + + try { + await llmStore.stream(options.model, options.chatProvider, newMessages as Message[], { + headers, + tools: options.tools, + // NOTICE: xsai stream may emit `finish` before tool steps continue, so keep waiting until + // the final non-tool finish to avoid ending the chat turn with no assistant reply. + waitForTools: true, + onStreamEvent: async (event: StreamEvent) => { + switch (event.type) { + case 'tool-call': + toolCallQueue.enqueue({ + type: 'tool-call', + toolCall: event, + }) + + break + case 'tool-result': + toolCallQueue.enqueue({ + type: 'tool-call-result', + id: event.toolCallId, + result: event.result, + }) + + break + case 'tool-error': + toolCallQueue.enqueue({ + type: 'tool-call-result', + id: event.toolCallId, + isError: true, + result: event.result, + }) + + break + case 'text-delta': + if (!llmFirstTokenEmitted) { + llmFirstTokenEmitted = true + llmSpan.addEvent(IOEvents.FirstToken, { + [IOAttrs.LLM_TTFT]: performance.now() - llmRequestTs, + }) + } + fullText += event.text + await parser.consume(event.text) + break + case 'finish': + break + case 'error': + throw event.error ?? new Error('Stream error') + } + }, + }) + + llmSpan.setAttribute(IOAttrs.LLMTextLength, fullText.length) + } + finally { + // TODO: Record errors on llmSpan + llmSpan.end() + } await parser.end() @@ -421,11 +449,19 @@ export const useChatOrchestratorStore = defineStore('chat-orchestrator', () => { toolCalls: sessionMessagesForSend.filter(msg => msg.role === 'tool') as ToolMessage[], }, streamingMessageContext) + // TODO: Close turn span for zero-segment responses (tool-only, empty). + // Currently the bridge's tryCloseTurn() only fires after TTS playback, + // so turns without segments stay open until the next interaction. + if (isForegroundSession()) { streamingMessage.value = { role: 'assistant', content: '', slices: [], tool_results: [] } } } catch (error) { + if (!hadExistingTurn && activeTurnSpan.value) { + activeTurnSpan.value.end() + activeTurnSpan.value = undefined + } console.error('Error sending message:', error) throw error } diff --git a/packages/stage-ui/src/stores/devtools/io-tracer.ts b/packages/stage-ui/src/stores/devtools/io-tracer.ts new file mode 100644 index 000000000..ebc7b2625 --- /dev/null +++ b/packages/stage-ui/src/stores/devtools/io-tracer.ts @@ -0,0 +1,275 @@ +import type { Attributes } from '@opentelemetry/api' +import type { ReadableSpan } from '@opentelemetry/sdk-trace-base' +import type { IOSpan, IOSubsystem, IOTurn } from '@proj-airi/stage-shared' + +import { getTimeOrigin, hrTimeToMilliseconds, hrTimeToNanoseconds } from '@opentelemetry/core' +import { IOAttrs, IOEvents, IOSpanNames } from '@proj-airi/stage-shared' +import { defineStore } from 'pinia' +import { computed, ref, triggerRef } from 'vue' + +import { activeTurnSpan, initIOTracer, onIOSpan, onRemoteIOSpan } from '../../composables/use-io-tracer' + +const MAX_TURNS = 50 + +function attrsToMeta(attrs: Attributes): Record { + const meta: Record = {} + for (const [key, value] of Object.entries(attrs)) { + const shortKey = key.includes('.') ? key.split('.').at(-1)! : key + meta[shortKey] = value + } + return meta +} + +export const useIOTracerStore = defineStore('devtools:io-tracer', () => { + const turns = ref([]) + const isRecording = ref(false) + const selectedSpanId = ref(null) + const recordingStartTs = ref(0) + const revision = ref(0) + + const turnsByTraceId = new Map() + + const rawSpans: ReadableSpan[] = [] + let unsubRemote: (() => void) | undefined + + function notifyUpdate() { + triggerRef(turns) + revision.value++ + } + + const activeTurn = computed(() => { + if (turns.value.length === 0) + return undefined + const last = turns.value.at(-1) + return last?.endTs == null ? last : undefined + }) + + const selectedSpan = computed(() => { + if (!selectedSpanId.value) + return undefined + for (const turn of turns.value) { + const span = turn.spans.find(s => s.id === selectedSpanId.value) + if (span) + return { span, turn } + } + return undefined + }) + + function formatOtlpValue(value: unknown): Record { + if (typeof value === 'string') + return { stringValue: value } + if (typeof value === 'number') + return Number.isInteger(value) ? { intValue: String(value) } : { doubleValue: value } + if (typeof value === 'boolean') + return { boolValue: value } + if (Array.isArray(value)) + return { arrayValue: { values: value.map(v => formatOtlpValue(v)) } } + return { stringValue: String(value) } + } + + function handleSpan(readable: ReadableSpan) { + rawSpans.push(readable) + + const spanCtx = readable.spanContext() + const traceId = spanCtx.traceId + const spanId = spanCtx.spanId + const startMs = hrTimeToMilliseconds(readable.startTime) + const endMs = readable.ended ? hrTimeToMilliseconds(readable.endTime) : undefined + + function getOrCreateTurn(): IOTurn { + let turn = turnsByTraceId.get(traceId) + if (!turn) { + turn = { + id: traceId, + startTs: startMs, + spans: [], + } + turnsByTraceId.set(traceId, turn) + turns.value.push(turn) + + while (turns.value.length > MAX_TURNS) { + const evicted = turns.value.shift() + if (evicted) + turnsByTraceId.delete(evicted.id) + } + } + return turn + } + + if (readable.name === IOSpanNames.InteractionTurn) { + const turn = getOrCreateTurn() + if (endMs) + turn.endTs = endMs + const text = readable.attributes[IOAttrs.ASRText] + if (typeof text === 'string') + turn.inputText = text + + notifyUpdate() + return + } + + const subsystem = readable.attributes[IOAttrs.Subsystem] as IOSubsystem | undefined + + if (!subsystem) { + if (readable.name === IOSpanNames.TTSSegment) { + const turn = getOrCreateTurn() + const text = readable.attributes[IOAttrs.TTSText] + if (typeof text === 'string' && !turn.outputText) + turn.outputText = text + } + notifyUpdate() + return + } + + const turn = getOrCreateTurn() + const meta = attrsToMeta(readable.attributes) + + for (const event of readable.events) { + const eventAttrs = event.attributes ?? {} + for (const [key, value] of Object.entries(eventAttrs)) { + const shortKey = key.includes('.') ? key.split('.').at(-1)! : key + meta[shortKey] = value + } + if (event.name === IOEvents.FirstToken) { + meta.firstTokenTs = hrTimeToMilliseconds(event.time) + } + } + + if (subsystem === 'asr' && typeof readable.attributes[IOAttrs.ASRText] === 'string') + turn.inputText = (turn.inputText ?? '') + (readable.attributes[IOAttrs.ASRText] as string) + if (subsystem === 'llm' && typeof meta.text_length === 'number') + turn.outputText = `(${meta.text_length} chars)` + + const segmentId = readable.attributes[IOAttrs.TTSSegmentId] + + const ioSpan: IOSpan = { + id: spanId, + traceId, + parentSpanId: readable.parentSpanContext?.spanId, + ttsCorrelationId: typeof segmentId === 'string' ? segmentId : undefined, + subsystem, + name: readable.name.split(':').at(-1) ?? readable.name, + startTs: startMs, + endTs: endMs, + meta, + } + + turn.spans.push(ioSpan) + notifyUpdate() + } + + function startRecording() { + if (isRecording.value) + return + + initIOTracer() + onIOSpan(handleSpan) + unsubRemote = onRemoteIOSpan(handleSpan) + recordingStartTs.value = getTimeOrigin() + performance.now() + isRecording.value = true + + console.info('[IOTracer] Recording started (OTel mode, local + remote)') + } + + function stopRecording() { + if (!isRecording.value) + return + + activeTurnSpan.value?.end() + activeTurnSpan.value = undefined + + onIOSpan(undefined) + unsubRemote?.() + unsubRemote = undefined + isRecording.value = false + + console.info('[IOTracer] Recording stopped') + } + + function clear() { + turns.value = [] + turnsByTraceId.clear() + rawSpans.length = 0 + selectedSpanId.value = null + recordingStartTs.value = getTimeOrigin() + performance.now() + } + + function selectSpan(spanId: string | null) { + selectedSpanId.value = spanId + } + + function exportOtlpJson() { + if (rawSpans.length === 0) + return + + const spanJsons = rawSpans.map((span) => { + const ctx = span.spanContext() + const parentCtx = span.parentSpanContext + + return { + traceId: ctx.traceId, + spanId: ctx.spanId, + parentSpanId: parentCtx?.spanId ?? '', + name: span.name, + kind: span.kind, + startTimeUnixNano: String(hrTimeToNanoseconds(span.startTime)), + endTimeUnixNano: span.ended ? String(hrTimeToNanoseconds(span.endTime)) : '0', + attributes: Object.entries(span.attributes).map(([key, value]) => ({ + key, + value: formatOtlpValue(value), + })), + events: span.events.map(event => ({ + timeUnixNano: String(hrTimeToNanoseconds(event.time)), + name: event.name, + attributes: Object.entries(event.attributes ?? {}).map(([key, value]) => ({ + key, + value: formatOtlpValue(value), + })), + })), + status: { + code: span.status.code, + message: span.status.message ?? '', + }, + } + }) + + const otlpPayload = { + resourceSpans: [{ + resource: { + attributes: [ + { key: 'service.name', value: { stringValue: 'airi-io' } }, + ], + }, + scopeSpans: [{ + scope: { name: 'io' }, + spans: spanJsons, + }], + }], + } + + const json = JSON.stringify(otlpPayload, null, 2) + const blob = new Blob([json], { type: 'application/json' }) + const url = URL.createObjectURL(blob) + const a = document.createElement('a') + a.href = url + a.download = `trace_${Date.now()}.json` + a.click() + URL.revokeObjectURL(url) + } + + return { + turns, + activeTurn, + isRecording, + recordingStartTs, + revision, + selectedSpanId, + selectedSpan, + startRecording, + stopRecording, + clear, + selectSpan, + exportOtlpJson, + rawSpanCount: computed(() => { revision.value; return rawSpans.length }), + } +}) diff --git a/packages/stage-ui/src/stores/modules/hearing.ts b/packages/stage-ui/src/stores/modules/hearing.ts index 554c7999d..24a0acd2e 100644 --- a/packages/stage-ui/src/stores/modules/hearing.ts +++ b/packages/stage-ui/src/stores/modules/hearing.ts @@ -1,8 +1,10 @@ +import type { Span } from '@opentelemetry/api' import type { TranscriptionProviderWithExtraOptions } from '@xsai-ext/providers/utils' import type { WithUnknown } from '@xsai/shared' import type { StreamTranscriptionResult, StreamTranscriptionOptions as XSAIStreamTranscriptionOptions } from '@xsai/stream-transcription' import { errorMessageFrom, tryCatch } from '@moeru/std' +import { IOAttrs, IOEvents, IOSpanNames, IOSubsystems } from '@proj-airi/stage-shared' import { useLocalStorageManualReset } from '@proj-airi/stage-shared/composables' import { refManualReset } from '@vueuse/core' import { generateTranscription } from '@xsai/generate-transcription' @@ -11,6 +13,7 @@ import { computed, ref, shallowRef, watch } from 'vue' import vadWorkletUrl from '../../workers/vad/process.worklet?worker&url' +import { activeTurnSpan, startSpan } from '../../composables/use-io-tracer' import { useProvidersStore } from '../providers' import { streamAliyunTranscription } from '../providers/aliyun/stream-transcription' import { streamWebSpeechAPITranscription } from '../providers/web-speech-api' @@ -310,6 +313,8 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech } }>() + let asrSpan: Span | undefined + const supportsStreamInput = computed(() => { const providerId = activeTranscriptionProvider.value if (!providerId) @@ -388,6 +393,12 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech if (!session) return + if (asrSpan) { + asrSpan.setAttribute(IOAttrs.ASRAbort, !!abort) + asrSpan.end() + asrSpan = undefined + } + // Special handling for Web Speech API if (session.providerId === 'browser-web-speech-api') { try { @@ -492,6 +503,14 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech onSentenceEnd?: (delta: string) => void onSpeechEnd?: (text: string) => void }) { + activeTurnSpan.value?.end() + const turnSpan = startSpan(IOSpanNames.InteractionTurn) + activeTurnSpan.value = turnSpan + asrSpan = startSpan(IOSpanNames.SpeechRecognition, turnSpan, { + [IOAttrs.Subsystem]: IOSubsystems.ASR, + [IOAttrs.ASRProvider]: activeTranscriptionProvider.value ?? '', + }) + console.info('[Hearing Pipeline] transcribeForMediaStream called', { supportsStreamInput: supportsStreamInput.value, hasStream: !!stream, @@ -605,10 +624,17 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech abortSignal: abortController.signal, onSentenceEnd: (delta) => { bumpIdle() // Bump idle timer on activity (only if enabled) + if (asrSpan) + asrSpan.addEvent(IOEvents.SentenceEnd, { [IOAttrs.ASRText]: delta }) // Call the options callback options?.onSentenceEnd?.(delta) }, onSpeechEnd: (text) => { + if (asrSpan) { + asrSpan.setAttribute(IOAttrs.ASRText, text) + asrSpan.end() + asrSpan = undefined + } // Call the options callback options?.onSpeechEnd?.(text) }, @@ -766,6 +792,8 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech break if (value) { fullText += value + if (asrSpan) + asrSpan.addEvent(IOEvents.SentenceEnd, { [IOAttrs.ASRText]: value }) // Use captured callbacks to avoid cross-session leakage sessionCallbacks.onSentenceEnd?.(value) } @@ -776,6 +804,11 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech console.error('Error reading text stream:', err) } finally { + if (asrSpan) { + asrSpan.setAttribute(IOAttrs.ASRText, fullText) + asrSpan.end() + asrSpan = undefined + } // Use captured callbacks to avoid cross-session leakage sessionCallbacks.onSpeechEnd?.(fullText) }