feat(stage-shared,stage-ui): add store and integrations for IO tracer

cspell.config.yaml

@@ -305,6 +305,7 @@ words:
   - tresjs
   - Triggerable
   - tsdown
+  - ttft
   - turborepo
   - unbird
   - unbundle

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

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<AudioBuffer>({
   playback: playbackManager,
 })
 
+initIOTracer()
+useIOTraceBridge(speechPipeline)
 void speechRuntimeStore.registerHost(speechPipeline)
 
 speechPipeline.on('onSpecial', (segment) => {

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<typeof createSpeechPipeline>) {
+  const cleanupFns: (() => void)[] = []
+
+  const segmentSpans = new Map<string, Span>()
+  const synthesisSpans = new Map<string, Span>()
+  const playbackSpans = new Map<string, Span>()
+  const segmentReasons = new Map<string, string>()
+
+  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()
+  })
+}

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
     }

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<string, any> {
+  const meta: Record<string, any> = {}
+  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<IOTurn[]>([])
+  const isRecording = ref(false)
+  const selectedSpanId = ref<string | null>(null)
+  const recordingStartTs = ref(0)
+  const revision = ref(0)
+
+  const turnsByTraceId = new Map<string, IOTurn>()
+
+  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<string, unknown> {
+    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 }),
+  }
+})

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)
           }