diff --git a/.agents/skills/analyze-io-traces/SKILL.md b/.agents/skills/analyze-io-traces/SKILL.md index ca8294a84b..e8c73b669c 100644 --- a/.agents/skills/analyze-io-traces/SKILL.md +++ b/.agents/skills/analyze-io-traces/SKILL.md @@ -20,7 +20,21 @@ The first path is the packaged app on macOS. The second is `pnpm dev` on macOS. ## Record shape -Each line is one span with `traceId`, `spanId`, `parentSpanId`, `name`, `startTimeNano`, `endTimeNano`, `status.code` (2 means error), `attributes`, and `events`. Time values are decimal strings in nanoseconds. A trace is one interaction turn. Subsystems are in `attributes["ai.moeru.airi.io.subsystem"]`. +Each line is one span with `traceId`, `spanId`, `parentSpanId`, `name`, `startTimeNano`, `endTimeNano`, `status.code` (2 means error), `attributes`, and `events`. Time values are decimal strings in nanoseconds. Subsystems are in `attributes["ai.moeru.airi.io.subsystem"]`. + +A trace is one turn. A typed turn starts at `Interaction turn`. A spoken turn starts earlier, and one trace covers the whole path: + +```text +Wake word (kws) only when a wake word started the input +└── Voice input (voice-input) trigger: control, speech, or wake-word. outcome: committed, drafted, cancelled, or failed + ├── Voice activity detection one span for each speech segment + ├── Speech recognition (asr) + ├── End of speech judgement one span for each pause that the chat model judged + ├── Transcript rephrase + └── Interaction turn LLM inference, streaming control, speech turn, TTS synthesis, audio playback +``` + +A span ends before its parent can end, so a file can hold children whose parent is missing. The parent is still open, or the app stopped first. A `Voice input` with the outcome `cancelled` or `drafted` has no `Interaction turn`. Attribute `ai.moeru.airi.io.cancelled` marks work that the caller aborted, such as a judgement after the speaker continued. ## Common queries @@ -36,6 +50,15 @@ jq -c 'select(.status.code == 2) | {name, traceId, message: .status.message}' "$ # All spans of one trace in start order jq -c 'select(.traceId == "TRACE_ID")' "$F" | jq -s 'sort_by(.startTimeNano|tonumber)' +# Spoken turns: trigger, outcome, and how long the input took +jq -c 'select(.name == "Voice input") | {traceId, trigger: .attributes["ai.moeru.airi.io.voice_input.trigger"], outcome: .attributes["ai.moeru.airi.io.voice_input.outcome"], reason: .attributes["ai.moeru.airi.io.voice_input.reason"], ms: (((.endTimeNano|tonumber)-(.startTimeNano|tonumber))/1e6)}' "$F" + +# Wake words that started an input, with the matched tokens +jq -c 'select(.name == "Wake word") | {traceId, text: .attributes["ai.moeru.airi.io.kws.text"], tokens: .attributes["ai.moeru.airi.io.kws.tokens"]}' "$F" + +# End of speech judgements, and judgements that a newer utterance cancelled +jq -c 'select(.name == "End of speech judgement") | {traceId, judgement: .attributes["ai.moeru.airi.io.end_judge.judgement"], cancelled: .attributes["ai.moeru.airi.io.cancelled"]}' "$F" + # Time to first token for each LLM span jq -c 'select(.events | any(.name | endswith("llm.first_token"))) | {traceId, events}' "$F" ``` diff --git a/apps/stage-tamagotchi/src/main/services/airi/io-trace-recording/register.ts b/apps/stage-tamagotchi/src/main/services/airi/io-trace-recording/register.ts index 1485c8ba9e..f6644fe4b1 100644 --- a/apps/stage-tamagotchi/src/main/services/airi/io-trace-recording/register.ts +++ b/apps/stage-tamagotchi/src/main/services/airi/io-trace-recording/register.ts @@ -1,4 +1,5 @@ import type { InvocableEventContext } from '@moeru/eventa' +import type { BrowserWindow } from 'electron' import type { IOTraceRecordingService } from '.' @@ -14,11 +15,21 @@ import { export function registerIOTraceRecording( context: InvocableEventContext, service: IOTraceRecordingService, + window: BrowserWindow, ): () => void { const disposers = [ defineInvokeHandler(context, ioTraceRecordingGet, () => service.getState()), defineInvokeHandler(context, ioTraceRecordingSetEnabled, ({ enabled }) => service.setEnabled(enabled)), - defineInvokeHandler(context, ioTraceRecordingRecordSpan, span => service.recordSpan(span)), + // NOTICE: + // Every window context listens on the shared `ipcMain`, so each registered window also receives the spans of + // the other windows. Without this check, an open settings window writes each span of the main window twice. + // Get and set stay unchecked: they are idempotent, and a handler that returns early would answer with undefined. + // Source: the eventa TODO about window-namespaced contexts in main/windows/*/rpc/index.electron.ts. + // Remove this check when eventa dispatches invokes only to the context of the sender window. + defineInvokeHandler(context, ioTraceRecordingRecordSpan, (span, options) => { + if (options?.raw.ipcMainEvent.sender.id === window.webContents.id) + return service.recordSpan(span) + }), service.onStateChange((state) => { void context.emit(ioTraceRecordingChanged, state) }), diff --git a/apps/stage-tamagotchi/src/main/windows/main/rpc/index.electron.ts b/apps/stage-tamagotchi/src/main/windows/main/rpc/index.electron.ts index 8dec3afc21..67c5c19c92 100644 --- a/apps/stage-tamagotchi/src/main/windows/main/rpc/index.electron.ts +++ b/apps/stage-tamagotchi/src/main/windows/main/rpc/index.electron.ts @@ -72,7 +72,7 @@ export async function setupMainWindowElectronInvokes(params: { createGodotStageService({ context, manager: params.godotStageManager, window: params.window }) createOnboardingService({ context, onboardingWindowManager: params.onboardingWindowManager, mainWindow: params.window }) createAuthService({ context, window: params.window }) - const stopIOTraceRecording = registerIOTraceRecording(context, params.ioTraceRecording) + const stopIOTraceRecording = registerIOTraceRecording(context, params.ioTraceRecording, params.window) params.window.once('closed', stopIOTraceRecording) defineInvokeHandler(context, electronCenterMainWindow, () => centerWindowOnDisplay(params.window)) diff --git a/apps/stage-tamagotchi/src/main/windows/settings/rpc/index.electron.ts b/apps/stage-tamagotchi/src/main/windows/settings/rpc/index.electron.ts index a30c147095..ceac3b3bd3 100644 --- a/apps/stage-tamagotchi/src/main/windows/settings/rpc/index.electron.ts +++ b/apps/stage-tamagotchi/src/main/windows/settings/rpc/index.electron.ts @@ -59,7 +59,7 @@ export async function setupSettingsWindowInvokes(params: { createMcpServersService({ context, manager: params.mcpManager }) createGodotStageService({ context, manager: params.godotStageManager, window: params.settingsWindow }) createAuthService({ context, window: params.settingsWindow }) - const stopIOTraceRecording = registerIOTraceRecording(context, params.ioTraceRecording) + const stopIOTraceRecording = registerIOTraceRecording(context, params.ioTraceRecording, params.settingsWindow) params.settingsWindow.once('closed', stopIOTraceRecording) // Register the global shortcut service for the settings window. diff --git a/packages/stage-pages/src/pages/devtools/io-tracer/components/io-tracer-chart.vue b/packages/stage-pages/src/pages/devtools/io-tracer/components/io-tracer-chart.vue index 4666cbbd70..5e972102a0 100644 --- a/packages/stage-pages/src/pages/devtools/io-tracer/components/io-tracer-chart.vue +++ b/packages/stage-pages/src/pages/devtools/io-tracer/components/io-tracer-chart.vue @@ -7,6 +7,7 @@ import { computed, ref, watch } from 'vue' import { GAP_WARN_THRESHOLD_MS, + INPUT_SUBSYSTEMS, LABEL_COL_WIDTH, MINIMAP_HEIGHT, ROW_HEIGHT, @@ -147,9 +148,10 @@ const layout = computed(() => { } isFirstTurn = false + const inputSpans = turnSpans.filter(s => INPUT_SUBSYSTEMS.has(s.subsystem)).sort((a, b) => a.startTs - b.startTs) const llmSpans = turnSpans.filter(s => s.subsystem === IOSubsystems.LLM).sort((a, b) => a.startTs - b.startTs) const auxiliarySpans = turnSpans - .filter(s => s.subsystem !== IOSubsystems.LLM && !ttsSubsystems.has(s.subsystem)) + .filter(s => s.subsystem !== IOSubsystems.LLM && !ttsSubsystems.has(s.subsystem) && !INPUT_SUBSYSTEMS.has(s.subsystem)) .sort((a, b) => a.startTs - b.startTs) const ttsSpanList = turnSpans.filter(s => ttsSubsystems.has(s.subsystem)) @@ -169,6 +171,11 @@ const layout = computed(() => { const sortedSegments = [...segmentGroups.values()] .sort((a, b) => a[0].startTs - b[0].startTs) + for (const span of inputSpans) { + rows.push({ type: 'span', span, turn, subsystem: span.subsystem, y }) + y += ROW_HEIGHT + } + for (const span of llmSpans) { rows.push({ type: 'span', span, turn, subsystem: IOSubsystems.LLM, y }) y += ROW_HEIGHT diff --git a/packages/stage-pages/src/pages/devtools/io-tracer/io-tracer-types.ts b/packages/stage-pages/src/pages/devtools/io-tracer/io-tracer-types.ts index fd3072a624..bfbd583bd4 100644 --- a/packages/stage-pages/src/pages/devtools/io-tracer/io-tracer-types.ts +++ b/packages/stage-pages/src/pages/devtools/io-tracer/io-tracer-types.ts @@ -11,7 +11,12 @@ export interface SubsystemConfig { } export const SUBSYSTEM_CONFIGS: SubsystemConfig[] = [ + { subsystem: IOSubsystems.KWS, label: 'Wake Word', color: '#ec4899', bgColor: '#ec489918', icon: 'i-lucide:bell-ring' }, + { subsystem: IOSubsystems.VoiceInput, label: 'Voice Input', color: '#6366f1', bgColor: '#6366f118', icon: 'i-lucide:audio-waveform' }, + { subsystem: IOSubsystems.VAD, label: 'VAD', color: '#f59e0b', bgColor: '#f59e0b18', icon: 'i-lucide:activity' }, { subsystem: IOSubsystems.ASR, label: 'ASR', color: '#3b82f6', bgColor: '#3b82f618', icon: 'i-lucide:mic' }, + { subsystem: IOSubsystems.EndJudge, label: 'End Judge', color: '#f97316', bgColor: '#f9731618', icon: 'i-lucide:scale' }, + { subsystem: IOSubsystems.Rephrase, label: 'Rephrase', color: '#14b8a6', bgColor: '#14b8a618', icon: 'i-lucide:pen-line' }, { subsystem: IOSubsystems.LLM, label: 'LLM', color: '#a855f7', bgColor: '#a855f718', icon: 'i-lucide:brain' }, { subsystem: IOSubsystems.StreamingControl, label: 'Streaming Control', color: '#06b6d4', bgColor: '#06b6d418', icon: 'i-lucide:radio-tower' }, { subsystem: IOSubsystems.TTS, label: 'TTS', color: '#22c55e', bgColor: '#22c55e18', icon: 'i-lucide:audio-lines' }, @@ -20,6 +25,16 @@ export const SUBSYSTEM_CONFIGS: SubsystemConfig[] = [ export const SUBSYSTEM_CONFIG_MAP = new Map(SUBSYSTEM_CONFIGS.map(c => [c.subsystem, c])) +/** Subsystems that run before the LLM in a spoken turn. The chart draws them above the LLM rows, in start order. */ +export const INPUT_SUBSYSTEMS: ReadonlySet = new Set([ + IOSubsystems.KWS, + IOSubsystems.VoiceInput, + IOSubsystems.VAD, + IOSubsystems.ASR, + IOSubsystems.EndJudge, + IOSubsystems.Rephrase, +]) + /** Height of one span row in pixels */ export const ROW_HEIGHT = 28 /** Height of subsystem group header */ diff --git a/packages/stage-shared/src/perf/io-trace.ts b/packages/stage-shared/src/perf/io-trace.ts index b7cd28d390..e7ab97e428 100644 --- a/packages/stage-shared/src/perf/io-trace.ts +++ b/packages/stage-shared/src/perf/io-trace.ts @@ -1,6 +1,10 @@ export const IOSubsystems = { + VoiceInput: 'voice-input', + KWS: 'kws', VAD: 'vad', ASR: 'asr', + EndJudge: 'end-judge', + Rephrase: 'rephrase', LLM: 'llm', StreamingControl: 'streaming-control', TTS: 'tts', @@ -8,7 +12,19 @@ export const IOSubsystems = { } as const export type IOSubsystem = (typeof IOSubsystems)[keyof typeof IOSubsystems] +/** + * Span names of the voice and chat pipeline. + * + * A spoken turn is one trace: `Wake word` (when a wake started the input) or `Voice input` is the root. + * `Voice input` holds VAD segments, speech recognition, end judgements, and the rephrase. + * The `Interaction turn` that the input submits is its child, and it holds the LLM, TTS, and playback spans. + * A typed turn starts its own trace at `Interaction turn`. + */ export const IOSpanNames = { + WakeWord: 'Wake word', + VoiceInput: 'Voice input', + EndOfSpeechJudgement: 'End of speech judgement', + TranscriptRephrase: 'Transcript rephrase', InteractionTurn: 'Interaction turn', SpeechTurn: 'Speech turn', VoiceActivityDetection: 'Voice activity detection', @@ -29,6 +45,19 @@ export const IOAttributes = { Subsystem: `${customPrefix}.subsystem`, TooltipKeys: `${customPrefix}.tooltip.keys`, LLM_TTFT: `${customPrefix}.llm.time_to_first_token`, + VoiceInputId: `${customPrefix}.voice_input.id`, + /** `control` (a voice button or Push to Talk), `speech` (VAD), or `wake-word`. */ + VoiceInputTrigger: `${customPrefix}.voice_input.trigger`, + /** `committed`, `drafted`, `cancelled`, or `failed`. */ + VoiceInputOutcome: `${customPrefix}.voice_input.outcome`, + VoiceInputReason: `${customPrefix}.voice_input.reason`, + WakeCharacterId: `${customPrefix}.kws.character_id`, + WakeText: `${customPrefix}.kws.text`, + WakeTokens: `${customPrefix}.kws.tokens`, + EndJudgement: `${customPrefix}.end_judge.judgement`, + EndJudgeTranscriptLength: `${customPrefix}.end_judge.transcript_length`, + RephraseSegmentCount: `${customPrefix}.rephrase.segment_count`, + RephraseChangedCount: `${customPrefix}.rephrase.changed_count`, VADAudioDurationMs: `${customPrefix}.vad.audio_duration_ms`, VADAborted: `${customPrefix}.vad.aborted`, ASRText: `${customPrefix}.asr.text`, @@ -58,6 +87,8 @@ export const IOAttributes = { TTSInterruptReason: `${customPrefix}.tts.interrupt_reason`, TTSCanceled: `${customPrefix}.tts.canceled`, TurnId: `${customPrefix}.turn_id`, + /** The caller aborted the work. The span has no error status then. */ + Cancelled: `${customPrefix}.cancelled`, } as const export const IOEvents = { diff --git a/packages/stage-ui/README.md b/packages/stage-ui/README.md index ff5d86790b..b27d554ed9 100644 --- a/packages/stage-ui/README.md +++ b/packages/stage-ui/README.md @@ -204,6 +204,21 @@ Tool and content-array compatibility failures belong to the model of the failing Session resets and autonomous artistry hooks also use the session character. Each concurrent turn owns its IO trace. `activeTurnSpan` points to the most recently started turn that is still running. +A spoken turn is one IO trace, from the wake word to playback: + +- `Wake word` (KWS) is the root when a wake started the input. It covers the match and the session selection. +- `Voice input` is the root otherwise. It records the trigger (`control`, `speech`, or `wake-word`) and the outcome. + `activeVoiceInputSpan` points to it while the input runs. +- Its children are `Voice activity detection` (one span for each speech segment), `Speech recognition`, + `End of speech judgement`, and `Transcript rephrase`. +- The `Interaction turn` of the submission is also its child. The input ID is the turn ID, and `takeVoiceTurnParent` + connects them. The turn holds the LLM, streaming control, TTS, and playback spans. +- A composer dictation returns its text to the composer, which sends it later under a new ID. The next turn of the + same session within 60 seconds continues the dictation trace through `takeComposerVoiceParent`. This link is by + session, so a typed turn that replaces discarded dictated text in that time also joins the dictation trace. + +A typed turn starts its trace at `Interaction turn`. + ## Button analytics Register the shared plugin once in each Vue application: diff --git a/packages/stage-ui/src/composables/use-io-tracer.ts b/packages/stage-ui/src/composables/use-io-tracer.ts index 356a53b88f..be4bf21485 100644 --- a/packages/stage-ui/src/composables/use-io-tracer.ts +++ b/packages/stage-ui/src/composables/use-io-tracer.ts @@ -1,10 +1,12 @@ -import type { Span, SpanContext, SpanStatusCode } from '@opentelemetry/api' +import type { Span, SpanContext } from '@opentelemetry/api' import type { ReadableSpan, SpanExporter } from '@opentelemetry/sdk-trace-base' import type { SerializedIOSpan } from '@proj-airi/stage-shared/types/io-trace' -import { context, trace } from '@opentelemetry/api' +import { errorMessageFrom } from '@moeru/std' +import { context, SpanStatusCode, trace } from '@opentelemetry/api' import { hrTimeToNanoseconds } from '@opentelemetry/core' import { BasicTracerProvider, SimpleSpanProcessor } from '@opentelemetry/sdk-trace-base' +import { IOAttributes } from '@proj-airi/stage-shared' import { shallowRef } from 'vue' export type { ReadableSpan } from '@opentelemetry/sdk-trace-base' @@ -153,3 +155,82 @@ export function startSpan(name: string, parent?: Span, attrs?: Record() + +/** + * Runs `work` and ends `span` when it settles. + * + * A rejection after `signal` aborted is a cancellation that the caller chose, so the span gets the `Cancelled` + * attribute and no error status. Any other rejection sets the error status. The rejection still reaches the caller. + */ +export async function runInSpan(span: Span, signal: AbortSignal, work: () => Promise): Promise { + try { + return await work() + } + catch (error) { + if (signal.aborted) + span.setAttribute(IOAttributes.Cancelled, true) + else + span.setStatus({ code: SpanStatusCode.ERROR, message: errorMessageFrom(error) ?? 'Operation failed' }) + throw error + } + finally { + span.end() + } +} + +/** + * The span of the speech input that the voice host captures now. + * Speech recognition, end judgements, and the rephrase of that input are its children. + * The voice host sets it when an input starts and clears it when the input settles. + */ +export const activeVoiceInputSpan = shallowRef() + +/** + * Voice input spans by the chat turn ID that their submission uses. + * A submission uses the input ID as the message ID, and the chat runtime uses the message ID as the turn ID. + * The chat store takes the parent when it starts the `Interaction turn` span, so a spoken turn is one trace. + */ +const voiceTurnParents = new Map() + +/** Recent submissions only. A draft that is never sent leaves its entry until newer inputs replace it. */ +const MAX_VOICE_TURN_PARENTS = 32 + +export function linkVoiceTurn(turnId: string, span: Span) { + voiceTurnParents.delete(turnId) + voiceTurnParents.set(turnId, span) + while (voiceTurnParents.size > MAX_VOICE_TURN_PARENTS) + voiceTurnParents.delete(voiceTurnParents.keys().next().value!) +} + +/** Returns the voice input span of a turn once. A typed turn has none. */ +export function takeVoiceTurnParent(turnId: string): Span | undefined { + const span = voiceTurnParents.get(turnId) + voiceTurnParents.delete(turnId) + return span +} + +/** + * Time in which the next turn of a session can continue the trace of a composer dictation. + * A dictation returns text to the composer, and the composer sends it later as a new message with a new turn ID. + */ +const COMPOSER_VOICE_LINK_MS = 60_000 + +/** + * The newest finished composer dictation of each session. + * + * The link is by session, not by message, so it is a heuristic: the next turn of the session within + * {@link COMPOSER_VOICE_LINK_MS} takes it once. If the user discards the dictated text and sends typed text in that + * time, the typed turn also joins the dictation trace. The composer does not carry the input ID to its send. + */ +const composerVoiceParents = new Map() + +export function linkComposerVoice(sessionId: string, span: Span) { + composerVoiceParents.set(sessionId, { span, finishedAt: Date.now() }) +} + +/** Returns the composer dictation span of a session once, if it finished recently. */ +export function takeComposerVoiceParent(sessionId: string): Span | undefined { + const link = composerVoiceParents.get(sessionId) + composerVoiceParents.delete(sessionId) + return link && Date.now() - link.finishedAt < COMPOSER_VOICE_LINK_MS ? link.span : undefined +} diff --git a/packages/stage-ui/src/composables/voice-end-judge.ts b/packages/stage-ui/src/composables/voice-end-judge.ts index 3bea933521..56733021e4 100644 --- a/packages/stage-ui/src/composables/voice-end-judge.ts +++ b/packages/stage-ui/src/composables/voice-end-judge.ts @@ -2,12 +2,14 @@ import type { Conversation, TurnEvidence } from '@proj-airi/core-agent' import type { EndJudgement } from '../libs/voice/voice-end-judge-plugin' +import { IOAttributes, IOSpanNames, IOSubsystems } from '@proj-airi/stage-shared' import { boolean, object, parse } from 'valibot' import { useLLM } from '../stores/ai/chat-llm/llm' import { useConsciousnessStore } from '../stores/modules/consciousness' import { useHearingStore } from '../stores/modules/hearing' import { useProviderStore } from '../stores/providers/provider' +import { activeVoiceInputSpan, runInSpan, startSpan } from './use-io-tracer' const END_JUDGE_INSTRUCTIONS = [ 'You decide if a speaker finished a spoken message, from the speech recognition text of it so far.', @@ -54,6 +56,20 @@ export function useVoiceEndJudge() { if (!providerId || !model) throw new Error('No chat model is configured for voice end judgement') + const span = startSpan(IOSpanNames.EndOfSpeechJudgement, activeVoiceInputSpan.value, { + [IOAttributes.Subsystem]: IOSubsystems.EndJudge, + [IOAttributes.GenAIRequestModel]: model, + [IOAttributes.GenAIProviderName]: providerId, + [IOAttributes.EndJudgeTranscriptLength]: transcript.length, + }) + return runInSpan(span, signal, async () => { + const judgement = await requestJudgement(providerId, model, transcript, context, signal) + span.setAttribute(IOAttributes.EndJudgement, judgement) + return judgement + }) + } + + async function requestJudgement(providerId: string, model: string, transcript: string, context: TurnEvidence['context'], signal: AbortSignal) { const provider = await providers.getChatProviderInstance(providerId) const conversation: Conversation = { turns: [ { id: 'voice-end-judge-instructions', type: 'system', authority: 'system', content: [{ type: 'text', text: END_JUDGE_INSTRUCTIONS }] }, diff --git a/packages/stage-ui/src/composables/voice-input-trace.test.ts b/packages/stage-ui/src/composables/voice-input-trace.test.ts new file mode 100644 index 0000000000..f1492dfb53 --- /dev/null +++ b/packages/stage-ui/src/composables/voice-input-trace.test.ts @@ -0,0 +1,167 @@ +import type { SpeechInputAttemptOutcome } from '@proj-airi/core-agent' + +import type { ReadableSpan } from './use-io-tracer' + +import { SpanStatusCode } from '@opentelemetry/api' +import { IOAttributes, IOSpanNames } from '@proj-airi/stage-shared' +import { afterEach, describe, expect, it, vi } from 'vitest' + +import { activeVoiceInputSpan, onIOSpan, runInSpan, startSpan, takeComposerVoiceParent, takeVoiceTurnParent } from './use-io-tracer' +import { createVoiceInputTracer } from './voice-input-trace' + +const spans: ReadableSpan[] = [] +onIOSpan(span => spans.push(span)) + +afterEach(() => { + spans.length = 0 +}) + +function spansNamed(name: string) { + return spans.filter(span => span.name === name) +} + +/** An input whose outcome the test settles. Only the ID and the outcome reach the tracer. */ +function attempt(id: string) { + const done = Promise.withResolvers() + return { id, done: done.promise, settle: done.resolve } +} + +describe('createVoiceInputTracer', () => { + it('traces an input from start to outcome and makes it the parent of its recognition', async () => { + const tracer = createVoiceInputTracer() + const input = attempt('input-1') + + tracer.start(input, 'speech') + const recognition = startSpan(IOSpanNames.SpeechRecognition, activeVoiceInputSpan.value) + recognition.end() + input.settle({ status: 'committed', messageId: 'input-1' }) + + await vi.waitFor(() => expect(spansNamed(IOSpanNames.VoiceInput)).toHaveLength(1)) + const span = spansNamed(IOSpanNames.VoiceInput)[0]! + expect(span.attributes[IOAttributes.VoiceInputTrigger]).toBe('speech') + expect(span.attributes[IOAttributes.VoiceInputOutcome]).toBe('committed') + expect(spansNamed(IOSpanNames.SpeechRecognition)[0]!.parentSpanContext?.spanId).toBe(span.spanContext().spanId) + expect(activeVoiceInputSpan.value).toBeUndefined() + }) + + it('continues the trace of the wake that started the input', async () => { + const tracer = createVoiceInputTracer() + const wake = startSpan(IOSpanNames.WakeWord) + wake.end() + const input = attempt('input-2') + + tracer.start(input, 'wake-word', wake) + input.settle({ status: 'cancelled', reason: 'Speech segment was too short' }) + + await vi.waitFor(() => expect(spansNamed(IOSpanNames.VoiceInput)).toHaveLength(1)) + const span = spansNamed(IOSpanNames.VoiceInput)[0]! + expect(span.spanContext().traceId).toBe(wake.spanContext().traceId) + expect(span.parentSpanContext?.spanId).toBe(wake.spanContext().spanId) + expect(span.attributes[IOAttributes.VoiceInputReason]).toBe('Speech segment was too short') + }) + + it('links the chat turn of the submission, and of a draft that the input merged into', async () => { + const tracer = createVoiceInputTracer() + const input = attempt('input-3') + + tracer.start(input, 'control') + const span = activeVoiceInputSpan.value + input.settle({ status: 'drafted', draftId: 'earlier-draft' }) + await vi.waitFor(() => expect(spansNamed(IOSpanNames.VoiceInput)).toHaveLength(1)) + + expect(takeVoiceTurnParent('input-3')).toBe(span) + expect(takeVoiceTurnParent('earlier-draft')).toBe(span) + // A turn takes its parent once. + expect(takeVoiceTurnParent('earlier-draft')).toBeUndefined() + }) + + // Codex review on PR #2939: the composer sends dictated text under a new message ID, so the input ID never matches. + it('links a composer dictation to the next turn of its session, once and only for a short time', async () => { + const tracer = createVoiceInputTracer() + const first = attempt('dictation-1') + tracer.start(first, 'control') + const span = activeVoiceInputSpan.value + tracer.markComposer('dictation-1', 'alice') + first.settle({ status: 'drafted', draftId: 'dictation-1' }) + await vi.waitFor(() => expect(spansNamed(IOSpanNames.VoiceInput)).toHaveLength(1)) + + expect(takeComposerVoiceParent('bob')).toBeUndefined() + expect(takeComposerVoiceParent('alice')).toBe(span) + expect(takeComposerVoiceParent('alice')).toBeUndefined() + + vi.useFakeTimers() + try { + const second = attempt('dictation-2') + tracer.start(second, 'control') + tracer.markComposer('dictation-2', 'alice') + second.settle({ status: 'drafted', draftId: 'dictation-2' }) + await vi.waitFor(() => expect(spansNamed(IOSpanNames.VoiceInput)).toHaveLength(2)) + vi.advanceTimersByTime(61_000) + + expect(takeComposerVoiceParent('alice')).toBeUndefined() + } + finally { + vi.useRealTimers() + } + }) + + it('records one VAD span for each speech segment, and closes an open segment with the input', async () => { + const tracer = createVoiceInputTracer() + const input = attempt('input-4') + tracer.start(input, 'speech') + const window = (speech: boolean) => tracer.activity({ inputId: 'input-4', speech, durationMs: 100 }) + + // A short breath stays in one segment. A 300 ms pause closes it. + window(true) + window(true) + window(false) + window(true) + window(false) + window(false) + window(false) + window(true) + input.settle({ status: 'committed', messageId: 'input-4' }) + + await vi.waitFor(() => expect(spansNamed(IOSpanNames.VoiceInput)).toHaveLength(1)) + const segments = spansNamed(IOSpanNames.VoiceActivityDetection) + expect(segments.map(segment => segment.attributes[IOAttributes.VADAudioDurationMs])).toEqual([300, 100]) + expect(segments.every(segment => segment.parentSpanContext?.spanId === spansNamed(IOSpanNames.VoiceInput)[0]!.spanContext().spanId)).toBe(true) + }) + + it('marks a failed input as an error', async () => { + const tracer = createVoiceInputTracer() + const input = attempt('input-5') + + tracer.start(input, 'speech') + input.settle({ status: 'failed', stage: 'transcription', error: new Error('provider down') }) + + await vi.waitFor(() => expect(spansNamed(IOSpanNames.VoiceInput)).toHaveLength(1)) + const span = spansNamed(IOSpanNames.VoiceInput)[0]! + expect(span.status).toEqual({ code: SpanStatusCode.ERROR, message: 'provider down' }) + expect(span.attributes[IOAttributes.VoiceInputReason]).toBe('transcription') + }) +}) + +describe('runInSpan', () => { + it('ends the span with the result, an error status, or a cancellation', async () => { + const ok = startSpan('ok') + const failed = startSpan('failed') + const cancelled = startSpan('cancelled') + const abort = new AbortController() + + await expect(runInSpan(ok, abort.signal, async () => 'done')).resolves.toBe('done') + await expect(runInSpan(failed, abort.signal, async () => { + throw new Error('model unavailable') + })).rejects.toThrow('model unavailable') + abort.abort('Speech evidence changed') + await expect(runInSpan(cancelled, abort.signal, async () => { + throw abort.signal.reason + })).rejects.toBe('Speech evidence changed') + + await vi.waitFor(() => expect(spans).toHaveLength(3)) + expect(spansNamed('ok')[0]!.status.code).toBe(SpanStatusCode.UNSET) + expect(spansNamed('failed')[0]!.status).toEqual({ code: SpanStatusCode.ERROR, message: 'model unavailable' }) + expect(spansNamed('cancelled')[0]!.status.code).toBe(SpanStatusCode.UNSET) + expect(spansNamed('cancelled')[0]!.attributes[IOAttributes.Cancelled]).toBe(true) + }) +}) diff --git a/packages/stage-ui/src/composables/voice-input-trace.ts b/packages/stage-ui/src/composables/voice-input-trace.ts new file mode 100644 index 0000000000..2eb1261e55 --- /dev/null +++ b/packages/stage-ui/src/composables/voice-input-trace.ts @@ -0,0 +1,133 @@ +import type { Span } from '@opentelemetry/api' +import type { SpeechInputAttempt, SpeechInputAttemptOutcome } from '@proj-airi/core-agent' + +import { SpanStatusCode } from '@opentelemetry/api' +import { IOAttributes, IOSpanNames, IOSubsystems } from '@proj-airi/stage-shared' + +import { activeVoiceInputSpan, linkComposerVoice, linkVoiceTurn, startSpan } from './use-io-tracer' + +/** + * What started a speech input. + * + * - `control`: a voice button, the inlay, or the Push to Talk key. + * - `speech`: voice activity detection found speech. + * - `wake-word`: a wake word selected a character. + */ +export type VoiceInputTrigger = 'control' | 'speech' | 'wake-word' + +/** + * Non-speech that ends a traced speech segment. Shorter pauses are breaths inside one segment. + * The value only shapes the trace. It does not affect when an input ends. + */ +const SEGMENT_PAUSE_MS = 300 + +interface TracedInput { + readonly span: Span + segment?: { span: Span, speechMs: number, silenceMs: number, lastSpeechAt: number } + /** The session of a composer dictation. Its text returns to the composer instead of a draft. */ + composerSessionId?: string +} + +/** + * Traces the speech inputs of one voice host. + * + * Each input gets a `Voice input` span. It is the trace root, or the child of the `Wake word` span that started it. + * While the input runs, `activeVoiceInputSpan` points to it, so speech recognition, end judgements, and the rephrase + * become its children. VAD results become `Voice activity detection` child spans, one for each speech segment. + * The input ID is also the turn ID of the chat turn that its submission starts, so that turn continues the trace. + * + * State: `inputs` holds the open inputs by ID. The input outcome ends the input span and its open segment. + */ +export function createVoiceInputTracer() { + const inputs = new Map() + + /** Starts the trace of an input that the controller has just begun. */ + function start(attempt: Pick, trigger: VoiceInputTrigger, parent?: Span) { + const span = startSpan(IOSpanNames.VoiceInput, parent, { + [IOAttributes.Subsystem]: IOSubsystems.VoiceInput, + [IOAttributes.VoiceInputId]: attempt.id, + [IOAttributes.VoiceInputTrigger]: trigger, + }) + inputs.set(attempt.id, { span }) + activeVoiceInputSpan.value = span + linkVoiceTurn(attempt.id, span) + void attempt.done.then(outcome => settle(attempt.id, outcome)) + } + + /** + * Marks an input as a composer dictation. Its trace then waits for the next turn of the session, because the + * composer sends the dictated text later under a new message ID. + */ + function markComposer(inputId: string, sessionId: string) { + const input = inputs.get(inputId) + if (input) + input.composerSessionId = sessionId + } + + /** + * Records the VAD result of one window of an input. + * The first speech window opens a segment. A pause of `SEGMENT_PAUSE_MS` closes it at the time of its last speech window. + */ + function activity(event: { inputId: string, speech: boolean, durationMs: number }) { + const input = inputs.get(event.inputId) + if (!input) + return + + const now = performance.now() + if (event.speech) { + input.segment ??= { + span: startSpan(IOSpanNames.VoiceActivityDetection, input.span, { [IOAttributes.Subsystem]: IOSubsystems.VAD }), + speechMs: 0, + silenceMs: 0, + lastSpeechAt: now, + } + input.segment.speechMs += event.durationMs + input.segment.silenceMs = 0 + input.segment.lastSpeechAt = now + return + } + + if (!input.segment) + return + input.segment.silenceMs += event.durationMs + if (input.segment.silenceMs >= SEGMENT_PAUSE_MS) + closeSegment(input) + } + + function closeSegment(input: TracedInput) { + const segment = input.segment + if (!segment) + return + input.segment = undefined + segment.span.setAttribute(IOAttributes.VADAudioDurationMs, Math.round(segment.speechMs)) + segment.span.end(segment.lastSpeechAt) + } + + function settle(inputId: string, outcome: SpeechInputAttemptOutcome) { + const input = inputs.get(inputId) + if (!input) + return + inputs.delete(inputId) + closeSegment(input) + + input.span.setAttribute(IOAttributes.VoiceInputOutcome, outcome.status) + if (outcome.status === 'cancelled') + input.span.setAttribute(IOAttributes.VoiceInputReason, outcome.reason) + if (outcome.status === 'failed') { + input.span.setAttribute(IOAttributes.VoiceInputReason, outcome.stage) + input.span.setStatus({ code: SpanStatusCode.ERROR, message: outcome.error.message }) + } + // A dictation that produced text continues in the next composer turn of its session. + if (input.composerSessionId && outcome.status === 'drafted') + linkComposerVoice(input.composerSessionId, input.span) + // A draft that merged into an earlier draft is sent later under that draft's ID. + else if (outcome.status === 'drafted' && outcome.draftId !== inputId) + linkVoiceTurn(outcome.draftId, input.span) + + input.span.end() + if (activeVoiceInputSpan.value === input.span) + activeVoiceInputSpan.value = undefined + } + + return { start, markComposer, activity } +} diff --git a/packages/stage-ui/src/composables/voice-rephrase.ts b/packages/stage-ui/src/composables/voice-rephrase.ts index 892041d698..c489a787ce 100644 --- a/packages/stage-ui/src/composables/voice-rephrase.ts +++ b/packages/stage-ui/src/composables/voice-rephrase.ts @@ -1,10 +1,13 @@ import type { Conversation } from '@proj-airi/core-agent' +import { IOAttributes, IOSpanNames, IOSubsystems } from '@proj-airi/stage-shared' + import { parseRephrasedSegments } from '../libs/voice/voice-rephrase-plugin' import { useLLM } from '../stores/ai/chat-llm/llm' import { useConsciousnessStore } from '../stores/modules/consciousness' import { useHearingStore } from '../stores/modules/hearing' import { useProviderStore } from '../stores/providers/provider' +import { activeVoiceInputSpan, runInSpan, startSpan } from './use-io-tracer' const REPHRASE_INSTRUCTIONS = [ 'You clean up text that speech recognition produced from one spoken message.', @@ -37,6 +40,20 @@ export function useVoiceRephrase() { if (!providerId || !model) throw new Error('No chat model is configured for voice rephrasing') + const span = startSpan(IOSpanNames.TranscriptRephrase, activeVoiceInputSpan.value, { + [IOAttributes.Subsystem]: IOSubsystems.Rephrase, + [IOAttributes.GenAIRequestModel]: model, + [IOAttributes.GenAIProviderName]: providerId, + [IOAttributes.RephraseSegmentCount]: segments.length, + }) + return runInSpan(span, signal, async () => { + const rewritten = await requestRephrase(providerId, model, segments, signal) + span.setAttribute(IOAttributes.RephraseChangedCount, rewritten.filter((text, index) => text.trim() !== segments[index]?.trim()).length) + return rewritten + }) + } + + async function requestRephrase(providerId: string, model: string, segments: readonly string[], signal: AbortSignal) { const provider = await providers.getChatProviderInstance(providerId) const conversation: Conversation = { turns: [ { id: 'voice-rephrase-instructions', type: 'system', authority: 'system', content: [{ type: 'text', text: REPHRASE_INSTRUCTIONS }] }, diff --git a/packages/stage-ui/src/libs/audio/hearing-transcriber.ts b/packages/stage-ui/src/libs/audio/hearing-transcriber.ts index 56f3d6985a..5f64736607 100644 --- a/packages/stage-ui/src/libs/audio/hearing-transcriber.ts +++ b/packages/stage-ui/src/libs/audio/hearing-transcriber.ts @@ -7,7 +7,7 @@ import { errorMessageFrom } from '@moeru/std' import { SpanStatusCode } from '@opentelemetry/api' import { IOAttributes, IOSpanNames, IOSubsystems } from '@proj-airi/stage-shared' -import { activeTurnSpan, startSpan } from '../../composables/use-io-tracer' +import { activeTurnSpan, activeVoiceInputSpan, startSpan } from '../../composables/use-io-tracer' /** * Converts provider output into complete transcript revisions. Each adapter owns its segmenters and cancellation. @@ -26,7 +26,9 @@ export function createHearingTranscriber( const signal = AbortSignal.any([request.signal, cancelled.signal]) let reader: ReadableStreamDefaultReader | undefined let closed = false - const span = startSpan(IOSpanNames.SpeechRecognition, activeTurnSpan.value, { [IOAttributes.Subsystem]: IOSubsystems.ASR }) + // Recognition belongs to the input that it transcribes. A transcription outside a voice input, such as a voice + // message, attaches to the most recent turn. + const span = startSpan(IOSpanNames.SpeechRecognition, activeVoiceInputSpan.value ?? activeTurnSpan.value, { [IOAttributes.Subsystem]: IOSubsystems.ASR }) return new ReadableStream({ start: (output) => { diff --git a/packages/stage-ui/src/libs/voice/voice-activity-plugin.test.ts b/packages/stage-ui/src/libs/voice/voice-activity-plugin.test.ts index b9c2379cb5..0588a57c55 100644 --- a/packages/stage-ui/src/libs/voice/voice-activity-plugin.test.ts +++ b/packages/stage-ui/src/libs/voice/voice-activity-plugin.test.ts @@ -15,6 +15,8 @@ it('uses ordered VAD signals to capture, stream transcription, and submit to the const submit = vi.fn(async () => ({ status: 'committed' as const, messageId: 'accepted' })) let sessionId = 'alice' let attempt: SpeechInputAttempt | undefined + const activity = vi.fn() + const causes = vi.fn() const controller = new VoiceController({ audio, submit, transcriber: () => ({ transcribe: (request) => { const media = request.audio @@ -31,6 +33,8 @@ it('uses ordered VAD signals to capture, stream transcription, and submit to the detect: async window => window.channels[0][0], target: () => ({ sessionId, interruptTurns: [] }), onInput: () => { attempt = controller.activeInput }, + onActivity: activity, + beforeBeginInput: causes, minSpeechMs: 32, silenceMs: 64, }), { grants: ['input-control', 'cancel-input'] }) @@ -42,6 +46,13 @@ it('uses ordered VAD signals to capture, stream transcription, and submit to the expect(await attempt!.done).toEqual({ status: 'committed', messageId: 'accepted' }) expect(received).toHaveLength(96) expect(submit).toHaveBeenCalledWith(expect.objectContaining({ sessionId: 'alice', text: 'Hello' }), expect.any(AbortSignal)) + // Each noted window reaches the host trace in order. + expect(activity.mock.calls.map(([event]) => [event.inputId, event.speech, event.durationMs])).toEqual([ + [attempt!.id, true, 32], + [attempt!.id, false, 32], + [attempt!.id, false, 32], + ]) + expect(causes.mock.calls).toEqual([['speech']]) await controller.close() }) @@ -77,10 +88,12 @@ it('rejects playback echo and opens the matched character input only after playb let accepted = false const detected = vi.fn(async () => ({ sessionId: 'alice', interruptTurns: [turn] })) const gate = vi.fn(async () => accepted) + const causes = vi.fn() controller.use(createVoiceActivityPlugin({ detect: async window => window.channels[0][0], detectWakeWord: async window => window.range.startFrame < 96 ? detected() : undefined, acceptSpeech: gate, + beforeBeginInput: causes, target: () => ({ sessionId: 'other', interruptTurns: [] }), minSpeechMs: 32, silenceMs: 32, @@ -93,9 +106,12 @@ it('rejects playback echo and opens the matched character input only after playb await expect.poll(() => gate).toHaveBeenCalledOnce() expect(controller.activeInput).toBeUndefined() expect(fade).not.toHaveBeenCalled() + // Codex review on PR #2939: a wake that echo rejection dropped must not report a wake-word input. + expect(causes).not.toHaveBeenCalled() accepted = true push(32, 1) await expect.poll(() => fade).toHaveBeenCalledOnce() + expect(causes.mock.calls).toEqual([['wake-word']]) const attempt = controller.activeInput! expect(attempt.state).toEqual({ phase: 'pending', waitingFor: 'silence' }) push(64, 1) diff --git a/packages/stage-ui/src/libs/voice/voice-activity-plugin.ts b/packages/stage-ui/src/libs/voice/voice-activity-plugin.ts index 5a94608c86..b1be004bf2 100644 --- a/packages/stage-ui/src/libs/voice/voice-activity-plugin.ts +++ b/packages/stage-ui/src/libs/voice/voice-activity-plugin.ts @@ -15,6 +15,17 @@ export interface VoiceActivityOptions { readonly acceptSpeech?: (window: AudioWindow, signal: AbortSignal) => Promise readonly enabled?: () => boolean readonly onInput?: () => void + /** + * Runs right before the plugin begins an input, with the reason. The controller announces the input + * synchronously inside that call, so the host knows the cause of the input that it is told about next. + * A wake that does not lead to a begin, for example after echo rejection, never reaches this callback. + */ + readonly beforeBeginInput?: (cause: 'wake-word' | 'speech') => void + /** + * Receives the VAD result of each window that the plugin noted on its input, in window order. + * The voice host traces speech segments with it. It never changes how the input starts or ends. + */ + readonly onActivity?: (activity: { inputId: string, speech: boolean, durationMs: number }) => void /** @default 0.6. A live input uses 30 percent of this score to keep speech active. */ readonly threshold?: number /** @default 300. Shorter speech is discarded when silence ends an input. */ @@ -74,6 +85,7 @@ export function createVoiceActivityPlugin(options: VoiceActivityOptions): VoiceP if (!target) return + options.beforeBeginInput?.(wake ? 'wake-word' : 'speech') const result = ctx.controls.beginInput({ ...target, start: wake ? { kind: 'after-silence' } : { kind: 'speech-onset', at: { sourceId: ctx.window.range.sourceId, frame: ctx.window.range.startFrame }, preRollMs: options.preRollMs ?? 360 } }) @@ -92,6 +104,7 @@ export function createVoiceActivityPlugin(options: VoiceActivityOptions): VoiceP if (!input.noteActivity({ range: ctx.window.range, speech })) return + options.onActivity?.({ inputId: input.id, speech, durationMs: duration }) speechMs += speech ? duration : 0 silenceMs = speech ? 0 : silenceMs + duration diff --git a/packages/stage-ui/src/libs/voice/wake-word-detector.test.ts b/packages/stage-ui/src/libs/voice/wake-word-detector.test.ts index 62065ac254..6f059d4026 100644 --- a/packages/stage-ui/src/libs/voice/wake-word-detector.test.ts +++ b/packages/stage-ui/src/libs/voice/wake-word-detector.test.ts @@ -96,7 +96,7 @@ describe('wakeWordDetector', () => { ]) expect(harness.spotter.processAudio.mock.calls[0]![0]).toHaveLength(BATCH) expect(harness.spotter.processAudio.mock.calls[0]![1]).toBe(16_000) - expect(harness.resolveTarget).toHaveBeenCalledWith('bob', expect.any(AbortSignal)) + expect(harness.resolveTarget).toHaveBeenCalledWith('bob', expect.any(AbortSignal), bob) expect(target).toEqual({ sessionId: 'session-bob', interruptTurns: [] }) expect(harness.detector.take(new AbortController().signal)).toBeUndefined() expect(harness.onPreparationChange).toHaveBeenLastCalledWith('ready') diff --git a/packages/stage-ui/src/libs/voice/wake-word-detector.ts b/packages/stage-ui/src/libs/voice/wake-word-detector.ts index bae9052c92..40d2faa5d1 100644 --- a/packages/stage-ui/src/libs/voice/wake-word-detector.ts +++ b/packages/stage-ui/src/libs/voice/wake-word-detector.ts @@ -28,8 +28,11 @@ export interface WakeWordDetectorOptions { readonly modelId: string /** Creates a spotter Worker. Initialization requires at least one keyword. The detector owns and disposes the result. */ readonly createSpotter: (keywords: readonly KeywordEntry[]) => Promise - /** Selects the character session after a match. The signal aborts when the observed audio window is abandoned. */ - readonly resolveTarget: (characterId: string, signal: AbortSignal) => Promise + /** + * Selects the character session after a match. The signal aborts when the observed audio window is abandoned. + * `pronunciation` is the matched catalog entry, so the caller can trace which word woke the character. + */ + readonly resolveTarget: (characterId: string, signal: AbortSignal, pronunciation: WakePronunciation) => Promise readonly onPreparationChange?: (preparation: WakeWordPreparation, error?: string) => void /** Receives a failure to select the target session. Recognition continues with the next window. */ readonly onError?: (error: unknown) => void @@ -378,7 +381,7 @@ export class WakeWordDetector { private async resolveWake(pronunciation: WakePronunciation, generation: number) { const signal = this.lifetime.signal try { - const target = await this.options.resolveTarget(pronunciation.characterId, signal) + const target = await this.options.resolveTarget(pronunciation.characterId, signal, pronunciation) if (generation === this.generation && !signal.aborted) this.wake = target } diff --git a/packages/stage-ui/src/stores/chat.contract.test.ts b/packages/stage-ui/src/stores/chat.contract.test.ts index ff84428d08..01da5bb207 100644 --- a/packages/stage-ui/src/stores/chat.contract.test.ts +++ b/packages/stage-ui/src/stores/chat.contract.test.ts @@ -47,6 +47,10 @@ const ioTracerMocks = vi.hoisted(() => { activeTurnSpan, spans, startSpanMock, + /** Voice input spans by turn ID. A typed turn has none. */ + takeVoiceTurnParent: vi.fn((_turnId: string): unknown => undefined), + /** The composer dictation span of a session. */ + takeComposerVoiceParent: vi.fn((_sessionId: string): unknown => undefined), } }) @@ -132,6 +136,8 @@ vi.mock('../libs/product-signals', () => ({ vi.mock('../composables/use-io-tracer', () => ({ activeTurnSpan: ioTracerMocks.activeTurnSpan, startSpan: ioTracerMocks.startSpanMock, + takeVoiceTurnParent: ioTracerMocks.takeVoiceTurnParent, + takeComposerVoiceParent: ioTracerMocks.takeComposerVoiceParent, })) vi.mock('./chat/context-providers', () => ({ @@ -365,6 +371,8 @@ describe('chat store contract', () => { ioTracerMocks.activeTurnSpan.value = undefined ioTracerMocks.spans.length = 0 ioTracerMocks.startSpanMock.mockClear() + ioTracerMocks.takeVoiceTurnParent.mockReset() + ioTracerMocks.takeComposerVoiceParent.mockReset() activeSessionIdRef.value = 'session-1' activeProviderRef.value = 'mock-provider' cardSelections.clear() @@ -1572,6 +1580,30 @@ describe('chat store contract', () => { expect(store.$state).not.toHaveProperty('activeStreamingMessage') }) + it('continues the trace of the voice input that submitted the turn', async () => { + const voiceInput = { name: IOSpanNames.VoiceInput } + ioTracerMocks.takeVoiceTurnParent.mockReturnValueOnce(voiceInput) + + const store = useChatStore() + await store.ingest('spoken text', { model: 'gpt-test', chatProvider: provider }) + + const [turnId] = ioTracerMocks.takeVoiceTurnParent.mock.calls[0]! + expect(turnId).toEqual(expect.any(String)) + expect(turnId).not.toBe('') + expect(ioTracerMocks.startSpanMock).toHaveBeenCalledWith(IOSpanNames.InteractionTurn, voiceInput) + }) + + it('continues a composer dictation trace when the turn has no voice input of its own', async () => { + const dictation = { name: IOSpanNames.VoiceInput } + ioTracerMocks.takeComposerVoiceParent.mockReturnValueOnce(dictation) + + const store = useChatStore() + await store.ingest('dictated text', { model: 'gpt-test', chatProvider: provider }) + + expect(ioTracerMocks.takeComposerVoiceParent).toHaveBeenCalledOnce() + expect(ioTracerMocks.startSpanMock).toHaveBeenCalledWith(IOSpanNames.InteractionTurn, dictation) + }) + it('does not end the owned IO turn span when external sending mirror is cleared mid-send', async () => { let releaseStream: (() => void) | undefined llmStreamMock.mockImplementationOnce(async () => { diff --git a/packages/stage-ui/src/stores/chat.ts b/packages/stage-ui/src/stores/chat.ts index 16c87352e6..cd3ac26cc2 100644 --- a/packages/stage-ui/src/stores/chat.ts +++ b/packages/stage-ui/src/stores/chat.ts @@ -20,7 +20,7 @@ import { useI18n } from 'vue-i18n' import { getConversationAnalyticsSurface } from '../composables' import { useAiriRuntimePrompt } from '../composables/use-airi-runtime-prompt' -import { activeTurnSpan, startSpan } from '../composables/use-io-tracer' +import { activeTurnSpan, startSpan, takeComposerVoiceParent, takeVoiceTurnParent } from '../composables/use-io-tracer' import { useChatVision } from '../composables/vision/use-chat-vision' import { useVisionInference } from '../composables/vision/use-vision-inference' import { chatAssetIdFrom, inlineConversationAssets, storeChatAttachments } from '../libs/chat-assets' @@ -342,10 +342,16 @@ export const useChatStore = defineStore('chat', () => { : undefined let turnSpan = turnKey ? ownedTurnSpans.get(turnKey) : activeTurnSpan.value if (!turnSpan && turnKey) { - turnSpan = startSpan(IOSpanNames.InteractionTurn) + // A spoken turn continues the trace of its voice input. A turn after a composer dictation continues that + // dictation, because the composer sends the dictated text under a new ID. Any other turn starts a new trace. + const correlation = options?.requestCorrelation + const voiceParent = correlation + ? takeVoiceTurnParent(correlation.turnId) ?? takeComposerVoiceParent(correlation.conversationId) + : undefined + turnSpan = startSpan(IOSpanNames.InteractionTurn, voiceParent) ownedTurnSpans.set(turnKey, turnSpan) } - // Speech recognition, speech output, and streaming control attach to the most recently started turn. + // Speech output and streaming control attach to the most recently started turn. if (turnSpan) activeTurnSpan.value = turnSpan diff --git a/packages/stage-ui/src/stores/devtools/io-tracer.test.ts b/packages/stage-ui/src/stores/devtools/io-tracer.test.ts new file mode 100644 index 0000000000..4d59c11995 --- /dev/null +++ b/packages/stage-ui/src/stores/devtools/io-tracer.test.ts @@ -0,0 +1,33 @@ +import { IOAttributes, IOSpanNames, IOSubsystems } from '@proj-airi/stage-shared' +import { createPinia, setActivePinia } from 'pinia' +import { afterEach, describe, expect, it } from 'vitest' +import { computed } from 'vue' + +import { startSpan } from '../../composables/use-io-tracer' +import { useIOTracerStore } from './io-tracer' + +describe('useIOTracerStore', () => { + afterEach(() => { + useIOTracerStore().stopRecording() + }) + + it('updates views of a turn when more spans of its trace end', () => { + setActivePinia(createPinia()) + const store = useIOTracerStore() + store.startRecording() + // The chart receives the turn list as a prop, so it keeps one array reference and reads spans through it. + const turns = store.turns + const spanNames = computed(() => turns.map(turn => turn.spans.map(span => span.name))) + + const voiceInput = startSpan(IOSpanNames.VoiceInput, undefined, { [IOAttributes.Subsystem]: IOSubsystems.VoiceInput }) + const recognition = startSpan(IOSpanNames.SpeechRecognition, voiceInput, { [IOAttributes.Subsystem]: IOSubsystems.ASR }) + recognition.end() + expect(spanNames.value).toEqual([[IOSpanNames.SpeechRecognition]]) + + voiceInput.end() + + expect(spanNames.value).toEqual([[IOSpanNames.SpeechRecognition, IOSpanNames.VoiceInput]]) + // The child ended first, but the turn starts where its parent started. + expect(store.turns[0]!.startTs).toBe(store.turns[0]!.spans[1]!.startTs) + }) +}) diff --git a/packages/stage-ui/src/stores/devtools/io-tracer.ts b/packages/stage-ui/src/stores/devtools/io-tracer.ts index ca9fe0489f..b5c0b74ca7 100644 --- a/packages/stage-ui/src/stores/devtools/io-tracer.ts +++ b/packages/stage-ui/src/stores/devtools/io-tracer.ts @@ -83,13 +83,15 @@ export const useIOTracerStore = defineStore('devtools:io-tracer', () => { function getOrCreateTurn(): IOTurn { let turn = turnsByTraceId.get(traceId) if (!turn) { - turn = { + turns.value.push({ id: traceId, startTs: startMs, spans: [], - } + }) + // The reactive copy, not the raw object. Later spans of this trace mutate it, and views that hold the turn + // list, such as the chart through its prop, only see mutations through the proxy. + turn = turns.value.at(-1)! turnsByTraceId.set(traceId, turn) - turns.value.push(turn) while (turns.value.length > MAX_TURNS) { const evicted = turns.value.shift() @@ -97,6 +99,9 @@ export const useIOTracerStore = defineStore('devtools:io-tracer', () => { turnsByTraceId.delete(evicted.id) } } + // Spans arrive when they end, so a child, such as a VAD segment, can arrive before the span that starts the turn. + if (startMs < turn.startTs) + turn.startTs = startMs return turn } diff --git a/packages/stage-ui/src/stores/voice.ts b/packages/stage-ui/src/stores/voice.ts index 0ab9c9e832..f86ce1c123 100644 --- a/packages/stage-ui/src/stores/voice.ts +++ b/packages/stage-ui/src/stores/voice.ts @@ -13,6 +13,7 @@ import { useVoiceController } from '../composables/audio/voice-controller' import { traceSpeechOutput } from '../composables/speech-output-trace' import { useVoiceDrafts } from '../composables/voice-drafts' import { useVoiceEndJudge } from '../composables/voice-end-judge' +import { createVoiceInputTracer } from '../composables/voice-input-trace' import { useVoiceRephrase } from '../composables/voice-rephrase' import { observeInputLevel } from '../libs/voice/input-level' import { createVoiceActivityPlugin } from '../libs/voice/voice-activity-plugin' @@ -69,6 +70,7 @@ export const useVoiceStore = defineStore('voice', () => { const composerInputs = new Map() let presentedInput: { requestId: string, attempt: SpeechInputAttempt } | undefined + const inputTracer = createVoiceInputTracer() const { controller, state, snapshot: transcript, error } = useVoiceController({ transcriber: () => hearing.createTranscriber(), speech: (turn) => { @@ -85,6 +87,9 @@ export const useVoiceStore = defineStore('voice', () => { // The composer owns this text as its own draft. No voice draft or message is created. composerInputs.set(submission.submissionId, submission.text) + // The composer sends dictated text later. Empty text gives the composer nothing to send. + if (submission.text.trim()) + inputTracer.markComposer(submission.submissionId, submission.sessionId) return { status: 'drafted', draftId: submission.submissionId } }, recordInterruption: async (event) => { @@ -111,8 +116,18 @@ export const useVoiceStore = defineStore('voice', () => { console.error('[voice]', cause) } + /** + * The cause of the input that begins next. A control sets `control` around its begin call, and the voice activity + * plugin reports `speech` or `wake-word` right before its begin call. `onInput` runs inside that call and reads it. + */ + let beginCause: 'control' | 'speech' | 'wake-word' | undefined + controller.onInput((attempt) => { presentedInput = { requestId: attempt.id, attempt } + const cause = beginCause ?? 'speech' + beginCause = undefined + // Only an input that a wake began takes the wake span, so a wake that started nothing cannot join a later input. + inputTracer.start(attempt, cause, cause === 'wake-word' ? wakeWordDetection.claimWakeSpan() : undefined) }) function publishSnapshot() { @@ -149,7 +164,13 @@ export const useVoiceStore = defineStore('voice', () => { } function beginManual(sessionId: string) { - return beginInput({ sessionId, interruptTurns: activeTurns.value.filter(turn => turn.sessionId === sessionId), start: { kind: 'after-silence' } }) + beginCause = 'control' + try { + return beginInput({ sessionId, interruptTurns: activeTurns.value.filter(turn => turn.sessionId === sessionId), start: { kind: 'after-silence' } }) + } + finally { + beginCause = undefined + } } function endInput() { @@ -442,6 +463,10 @@ export const useVoiceStore = defineStore('voice', () => { const wakeWords = options.detectWakeWord ? undefined : wakeWordDetection.start({ resolveTarget: resolveWakeTarget, onError: report }) const plugin = createVoiceActivityPlugin({ detect: (window, signal) => model.score(window, signal), + onActivity: inputTracer.activity, + beforeBeginInput: (cause) => { + beginCause = cause + }, detectWakeWord: options.detectWakeWord ?? wakeWords?.detect, enabled: () => !voiceMessages.isRecording, silenceMs: () => hearing.endJudgeEnabled ? VOICE_END_SILENCE.pauseMs + VOICE_END_SILENCE.graceMs : VOICE_END_SILENCE.pauseMs, diff --git a/packages/stage-ui/src/stores/wake-word-detection.ts b/packages/stage-ui/src/stores/wake-word-detection.ts index a699bddc04..683375736c 100644 --- a/packages/stage-ui/src/stores/wake-word-detection.ts +++ b/packages/stage-ui/src/stores/wake-word-detection.ts @@ -1,10 +1,13 @@ +import type { Span } from '@opentelemetry/api' import type { AudioWindow } from '@proj-airi/pipelines-audio' import type { WakeActivity, WakeTarget, WakeWordPreparation } from '../libs/voice/wake-word-detector' +import { IOAttributes, IOSpanNames, IOSubsystems } from '@proj-airi/stage-shared' import { defineStore } from 'pinia' import { computed, shallowRef, watch } from 'vue' +import { runInSpan, startSpan } from '../composables/use-io-tracer' import { createKwsSpotter, kwsModel, kwsSearch, kwsToneVariants } from '../libs/voice/kws-model' import { WakeWordDetector } from '../libs/voice/wake-word-detector' import { useSettingsAudioDevice } from './settings/audio-device' @@ -31,6 +34,11 @@ export const useWakeWordDetectionStore = defineStore('wake-word-detection', () = /** Pronunciations of other models are not in the pinned vocabulary. They stay on their cards but do not run. */ const pronunciations = computed(() => wakeWords.catalog.active.filter(pronunciation => pronunciation.modelId === kwsModel.id)) let running: { detector: WakeWordDetector, stopWatches: readonly (() => void)[] } | undefined + /** + * The `Wake word` span of the newest resolved wake, until an input claims it as its trace root. + * A wake that starts no input, for example because that session already listens, leaves a span that nothing claims. + */ + let unclaimedWake: { span: Span, resolvedAt: number } | undefined function stop() { const current = running @@ -55,7 +63,20 @@ export const useWakeWordDetectionStore = defineStore('wake-word-detection', () = createSpotter: createKwsSpotter, search: { score: kwsSearch.score, threshold: kwsSearch.threshold }, variants: kwsToneVariants, - resolveTarget: options.resolveTarget, + // The span covers the match and the session selection. The voice input that the wake starts becomes its child. + resolveTarget: (characterId, signal, pronunciation) => { + const span = startSpan(IOSpanNames.WakeWord, undefined, { + [IOAttributes.Subsystem]: IOSubsystems.KWS, + [IOAttributes.WakeCharacterId]: characterId, + [IOAttributes.WakeText]: pronunciation.text, + [IOAttributes.WakeTokens]: pronunciation.tokens.join(' '), + }) + return runInSpan(span, signal, async () => { + const target = await options.resolveTarget(characterId, signal) + unclaimedWake = { span, resolvedAt: performance.now() } + return target + }) + }, onError: options.onError, onPreparationChange: (value, message) => { // A replaced detection can still settle a model load. Only the running detector updates the UI state. @@ -91,5 +112,16 @@ export const useWakeWordDetectionStore = defineStore('wake-word-detection', () = } } - return { preparation, error, pronunciations, start, stop } + /** + * Returns the span of the newest resolved wake, once. Call it only for an input that a wake began. + * A wake that started no input stays unclaimed until a newer wake replaces it. + * The second limit drops a wake that resolved long before, for example while the plugin was paused. + */ + function claimWakeSpan(): Span | undefined { + const wake = unclaimedWake + unclaimedWake = undefined + return wake && performance.now() - wake.resolvedAt < 1000 ? wake.span : undefined + } + + return { preparation, error, pronunciations, start, stop, claimWakeSpan } })