From 00193e53a12cc8fddb4d490de3797608ac4b31d0 Mon Sep 17 00:00:00 2001 From: Prajval Rasik Date: Mon, 10 Aug 2026 00:15:51 +0530 Subject: [PATCH 1/3] feat(voice): instrument judge STT pre-pass --- .../__tests__/judge-stt-telemetry.test.ts | 110 +++++++++++++++ javascript/src/voice/judge-stt.ts | 42 +++++- python/scenario/voice/_transcribe.py | 30 +++- .../tests/voice/test_judge_stt_telemetry.py | 128 ++++++++++++++++++ 4 files changed, 304 insertions(+), 6 deletions(-) create mode 100644 javascript/src/voice/__tests__/judge-stt-telemetry.test.ts create mode 100644 python/tests/voice/test_judge_stt_telemetry.py diff --git a/javascript/src/voice/__tests__/judge-stt-telemetry.test.ts b/javascript/src/voice/__tests__/judge-stt-telemetry.test.ts new file mode 100644 index 000000000..3cbca9790 --- /dev/null +++ b/javascript/src/voice/__tests__/judge-stt-telemetry.test.ts @@ -0,0 +1,110 @@ +/** Judge pre-pass STT span coverage for issue #785. */ + +import { trace, SpanStatusCode } from "@opentelemetry/api"; +import { InMemorySpanExporter, SimpleSpanProcessor } from "@opentelemetry/sdk-trace-base"; +import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node"; +import type { ModelMessage } from "ai"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { AudioChunk } from "../audio-chunk"; +import { prepareJudgeInput } from "../judge-stt"; +import { createAudioMessage } from "../messages"; +import type { STTProvider } from "../stt"; + +function tone(marker: number): AudioChunk { + const data = new Uint8Array(4800); + data.fill(marker); + return new AudioChunk({ data }); +} + +function textPart(message: ModelMessage): string | undefined { + const content = (message as { content?: unknown }).content; + if (!Array.isArray(content)) return undefined; + const part = content.find( + (item) => + item !== null && + typeof item === "object" && + (item as { type?: unknown }).type === "text", + ) as { text?: string } | undefined; + return part?.text; +} + +describe("judge pre-pass voice.stt.transcribe spans (#785)", () => { + let exporter: InMemorySpanExporter; + let provider: NodeTracerProvider; + + beforeEach(() => { + exporter = new InMemorySpanExporter(); + provider = new NodeTracerProvider({ + spanProcessors: [new SimpleSpanProcessor(exporter)], + }); + trace.setGlobalTracerProvider(provider); + }); + + afterEach(async () => { + await provider.shutdown(); + trace.disable(); + }); + + it("emits a judge-scoped span with success attributes", async () => { + const stt: STTProvider = { + transcribe: vi.fn(async () => "account restored"), + }; + await prepareJudgeInput({ + messages: [createAudioMessage(tone(1), "assistant") as ModelMessage], + stt, + }); + + const spans = exporter + .getFinishedSpans() + .filter((span) => span.name === "voice.stt.transcribe"); + expect(spans).toHaveLength(1); + expect(spans[0]!.attributes["voice.stt.scope"]).toBe("judge"); + expect(spans[0]!.attributes["voice.stt.speaker"]).toBe("assistant"); + expect(spans[0]!.attributes["voice.stt.audio_bytes"]).toBe(4800); + expect(spans[0]!.attributes["voice.stt.transcript_chars"]).toBe( + "account restored".length, + ); + expect(spans[0]!.attributes["langwatch.span.type"]).toBe("span"); + }); + + it("keeps a successful sibling when one message fails and exports only a sanitized error", async () => { + const rawError = "401 invalid key sk-secret body={provider response}"; + const stt: STTProvider = { + async transcribe(audio) { + if (audio.data[0] === 1) throw new Error(rawError); + return "successful sibling"; + }, + }; + const warn = vi.fn(); + const messages = [ + createAudioMessage(tone(1), "user") as ModelMessage, + createAudioMessage(tone(2), "assistant") as ModelMessage, + ]; + + const prepared = await prepareJudgeInput({ messages, stt, logWarn: warn }); + + expect(textPart(prepared.messages[0]!)).toBeUndefined(); + expect(textPart(prepared.messages[1]!)).toBe("successful sibling"); + const spans = exporter + .getFinishedSpans() + .filter((span) => span.name === "voice.stt.transcribe"); + expect(spans).toHaveLength(2); + const failed = spans.find((span) => span.status.code === SpanStatusCode.ERROR)!; + const succeeded = spans.find((span) => span.status.code !== SpanStatusCode.ERROR)!; + expect(failed.attributes["voice.stt.scope"]).toBe("judge"); + expect(failed.attributes["voice.stt.speaker"]).toBe("user"); + expect(failed.attributes["voice.stt.transcript_chars"]).toBeUndefined(); + expect(succeeded.attributes["voice.stt.transcript_chars"]).toBe( + "successful sibling".length, + ); + const recorded = JSON.stringify(failed.events); + expect(recorded).toContain("STT provider failed"); + expect(recorded).not.toContain("sk-secret"); + expect(recorded).not.toContain("401"); + expect(recorded).not.toContain("provider response"); + expect(warn).toHaveBeenCalledOnce(); + expect(warn.mock.calls[0]![0]).toContain("STT provider failed: Error"); + expect(warn.mock.calls[0]![0]).not.toContain(rawError); + }); +}); diff --git a/javascript/src/voice/judge-stt.ts b/javascript/src/voice/judge-stt.ts index ca757d7a6..5c7263ead 100644 --- a/javascript/src/voice/judge-stt.ts +++ b/javascript/src/voice/judge-stt.ts @@ -27,6 +27,7 @@ import type { ModelMessage } from "ai"; import { AudioChunk } from "./audio-chunk"; import { extractAudio } from "./messages"; import type { STTProvider } from "./stt"; +import { voiceSpan } from "./telemetry"; /** Judge audio knobs (PRD ยง4.3) โ€” resolved upstream, passed in here. */ export interface JudgeAudioOptions { @@ -79,6 +80,8 @@ export interface TranscribeAudioMessagesArgs { transcriptCache?: Map; /** Warning sink โ€” defaults to {@link console.warn}. */ logWarn?: (message: string) => void; + /** Internal telemetry scope. Set by the judge path, omitted by simulator fallback. */ + telemetryScope?: "judge"; } /** @@ -101,11 +104,11 @@ export interface TranscribeAudioMessagesArgs { export async function transcribeAudioMessages( args: TranscribeAudioMessagesArgs, ): Promise { - const { messages, stt, includeAudio, transcriptCache } = args; + const { messages, stt, includeAudio, transcriptCache, telemetryScope } = args; const warn = args.logWarn ?? ((m: string) => console.warn(m)); return Promise.all( messages.map((msg) => - transcribeMessage(msg, stt, includeAudio, warn, transcriptCache), + transcribeMessage(msg, stt, includeAudio, warn, transcriptCache, telemetryScope), ), ); } @@ -157,6 +160,7 @@ export async function prepareJudgeInput( stt: args.stt, includeAudio: args.options?.includeAudio ?? false, logWarn: args.logWarn, + telemetryScope: "judge", }); return { messages }; } @@ -190,6 +194,7 @@ async function transcribeMessage( includeAudio: boolean, warn: (message: string) => void, transcriptCache?: Map, + telemetryScope?: "judge", ): Promise { const content = (msg as { content?: unknown }).content; if (!Array.isArray(content)) return msg; @@ -211,7 +216,38 @@ async function transcribeMessage( const chunk = extractAudioChunk(msg); if (chunk) { try { - transcript = (await stt.transcribe(chunk)) || undefined; + const transcribe = async (): Promise => stt.transcribe(chunk); + if (telemetryScope === "judge") { + transcript = + (await voiceSpan( + "voice.stt.transcribe", + { + "voice.stt.scope": telemetryScope, + "voice.stt.speaker": String( + (msg as { role?: unknown }).role ?? "?", + ), + "voice.stt.audio_bytes": chunk.data.length, + }, + async (span) => { + let text: string; + try { + text = await transcribe(); + } catch (err) { + // Sanitize BEFORE voiceSpan records the exception. Provider + // SDK errors may include raw response bodies or key fragments. + throw new Error( + `STT provider failed: ${(err as Error)?.constructor?.name ?? "Error"}`, + ); + } + if (text) { + span.setAttribute("voice.stt.transcript_chars", text.length); + } + return text; + }, + )) || undefined; + } else { + transcript = (await transcribe()) || undefined; + } // Cache whenever STT actually RAN and RETURNED โ€” including an empty // result. `""` is the negative sentinel: the reuse branch above turns // a cached `""` back into `undefined` (no text part), so remembering diff --git a/python/scenario/voice/_transcribe.py b/python/scenario/voice/_transcribe.py index f8f2b4ad7..268834718 100644 --- a/python/scenario/voice/_transcribe.py +++ b/python/scenario/voice/_transcribe.py @@ -23,6 +23,7 @@ from .audio_chunk import AudioChunk from .recording import AudioSegment, VoiceRecording from .stt import STTProvider, get_stt_provider +from ._telemetry import voice_span logger = logging.getLogger("scenario.voice") @@ -55,7 +56,8 @@ async def transcribe_segments( if p is None: return # already warned targets = [ - s for s in recording.segments + s + for s in recording.segments if s.audio and (not only_missing or s.transcript is None) ] if not targets: @@ -78,8 +80,30 @@ def _try_get_provider() -> Optional[STTProvider]: async def _transcribe_one(provider: STTProvider, segment: AudioSegment) -> None: try: - text = await provider.transcribe(AudioChunk(data=segment.audio)) - segment.transcript = text or None + with voice_span( + "voice.stt.transcribe", + { + "voice.stt.scope": "judge", + "voice.stt.speaker": segment.speaker, + "voice.stt.audio_bytes": len(segment.audio), + }, + ) as stt_span: + try: + text = await provider.transcribe(AudioChunk(data=segment.audio)) + except Exception as exc: + # Provider SDK errors can include response bodies and key fragments. + # Keep the raw detail local and let telemetry record only a minimal + # provider-agnostic exception, matching the #783 STT guard. + logger.debug( + "scenario.voice.transcribe: STT provider error detail", + exc_info=True, + ) + raise RuntimeError( + f"STT provider failed: {type(exc).__name__}" + ) from None + segment.transcript = text or None + if text: + stt_span.set_attribute("voice.stt.transcript_chars", len(text)) except Exception as e: logger.warning( "scenario.voice.transcribe: STT failed for %s segment at %.2fs: %s", diff --git a/python/tests/voice/test_judge_stt_telemetry.py b/python/tests/voice/test_judge_stt_telemetry.py new file mode 100644 index 000000000..ce79bf1ce --- /dev/null +++ b/python/tests/voice/test_judge_stt_telemetry.py @@ -0,0 +1,128 @@ +"""Judge pre-pass STT span coverage for issue #785.""" + +from __future__ import annotations + +import logging + +import pytest +from opentelemetry import trace +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( + InMemorySpanExporter, +) +from opentelemetry.trace import StatusCode +from opentelemetry.util._once import Once + +from scenario.voice._transcribe import transcribe_segments +from scenario.voice.recording import AudioSegment, SpeakerRole, VoiceRecording +from scenario.voice.stt import STTProvider + +from ._span_assert import attrs + + +@pytest.fixture(autouse=True) +def reset_otel(): + """Reset the global provider around each telemetry test.""" + + def _reset() -> None: + trace._TRACER_PROVIDER = None + trace._TRACER_PROVIDER_SET_ONCE = Once() + + _reset() + yield + _reset() + + +def _install_exporter() -> InMemorySpanExporter: + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + trace.set_tracer_provider(provider) + return exporter + + +def _segment(speaker: SpeakerRole, marker: int) -> AudioSegment: + return AudioSegment( + speaker=speaker, + start_time=float(marker), + end_time=float(marker + 1), + audio=bytes([marker, 0]) * 100, + transcript=None, + ) + + +def _recorded_exception_text(span) -> str: + values: list[str] = [] + for event in span.events: + for key in ("exception.type", "exception.message", "exception.stacktrace"): + value = (event.attributes or {}).get(key) + if value: + values.append(str(value)) + return " ".join(values) + + +@pytest.mark.asyncio +async def test_judge_stt_emits_scoped_span_with_success_attributes(): + exporter = _install_exporter() + + class _STT(STTProvider): + async def transcribe(self, _audio): + return "account restored" + + recording = VoiceRecording(segments=[_segment("agent", 1)]) + await transcribe_segments(recording, provider=_STT()) + + spans = [ + s for s in exporter.get_finished_spans() if s.name == "voice.stt.transcribe" + ] + assert len(spans) == 1 + attributes = attrs(spans[0]) + assert attributes["voice.stt.scope"] == "judge" + assert attributes["voice.stt.speaker"] == "agent" + assert attributes["voice.stt.audio_bytes"] == 200 + assert attributes["voice.stt.transcript_chars"] == len("account restored") + assert attributes["langwatch.span.type"] == "span" + assert recording.segments[0].transcript == "account restored" + + +@pytest.mark.asyncio +async def test_judge_stt_mixed_batch_isolates_failure_and_sanitizes_span(caplog): + exporter = _install_exporter() + raw_error = "401 invalid key sk-secret body={provider response}" + + class _MixedSTT(STTProvider): + async def transcribe(self, audio): + if audio.data[0] == 1: + raise RuntimeError(raw_error) + return "successful sibling" + + recording = VoiceRecording(segments=[_segment("user", 1), _segment("agent", 2)]) + with caplog.at_level(logging.WARNING, logger="scenario.voice"): + await transcribe_segments(recording, provider=_MixedSTT()) + + assert [segment.transcript for segment in recording.segments] == [ + None, + "successful sibling", + ] + spans = [ + s for s in exporter.get_finished_spans() if s.name == "voice.stt.transcribe" + ] + assert len(spans) == 2 + failed = next(s for s in spans if s.status.status_code == StatusCode.ERROR) + succeeded = next(s for s in spans if s.status.status_code != StatusCode.ERROR) + failed_attributes = attrs(failed) + succeeded_attributes = attrs(succeeded) + assert failed_attributes["voice.stt.scope"] == "judge" + assert failed_attributes["voice.stt.speaker"] == "user" + assert "voice.stt.transcript_chars" not in failed_attributes + assert succeeded_attributes["voice.stt.transcript_chars"] == len( + "successful sibling" + ) + recorded = _recorded_exception_text(failed) + assert "STT provider failed" in recorded + assert "sk-secret" not in recorded + assert "401" not in recorded + assert "provider response" not in recorded + assert raw_error not in caplog.text + assert "STT provider failed: RuntimeError" in caplog.text From e28991d907daaac970b3d5c30d50954799ba1076 Mon Sep 17 00:00:00 2001 From: Prajval Rasik Date: Wed, 12 Aug 2026 22:29:29 +0530 Subject: [PATCH 2/3] fix(voice): gate Python judge STT telemetry --- python/scenario/judge_agent.py | 2 +- python/scenario/voice/_transcribe.py | 19 +++++++++++--- python/tests/test_judge_agent.py | 4 +-- .../tests/voice/test_judge_stt_telemetry.py | 25 +++++++++++++++++-- 4 files changed, 42 insertions(+), 8 deletions(-) diff --git a/python/scenario/judge_agent.py b/python/scenario/judge_agent.py index db7d2e17b..bb372b3dd 100644 --- a/python/scenario/judge_agent.py +++ b/python/scenario/judge_agent.py @@ -526,7 +526,7 @@ async def call( if conversation_has_audio and not self.effective_include_audio(conversation_has_audio): recording = self._extract_recording(input) if recording is not None: - await transcribe_segments(recording) + await transcribe_segments(recording, telemetry_scope="judge") working_messages = _enrich_messages_with_transcripts( input.messages, recording ) diff --git a/python/scenario/voice/_transcribe.py b/python/scenario/voice/_transcribe.py index 268834718..27495a0e1 100644 --- a/python/scenario/voice/_transcribe.py +++ b/python/scenario/voice/_transcribe.py @@ -32,6 +32,7 @@ async def transcribe_segments( recording: VoiceRecording, provider: Optional[STTProvider] = None, only_missing: bool = True, + telemetry_scope: Optional[str] = None, ) -> None: """ Run STT over recording.segments, mutating .transcript in place. @@ -43,6 +44,9 @@ async def transcribe_segments( only_missing: If True (default), skip segments whose transcript is already set. If False, re-transcribe everything (e.g. to overwrite adapter-side STT with a different provider). + telemetry_scope: When set, emit a ``voice.stt.transcribe`` span for + each provider call using this scope. Public callers are + uninstrumented by default; the judge path opts in with ``judge``. Concurrency: transcribes segments concurrently with asyncio.gather. Each segment's STT call is independent. Empty-data segments are skipped. @@ -62,7 +66,7 @@ async def transcribe_segments( ] if not targets: return - await asyncio.gather(*(_transcribe_one(p, s) for s in targets)) + await asyncio.gather(*(_transcribe_one(p, s, telemetry_scope) for s in targets)) def _try_get_provider() -> Optional[STTProvider]: @@ -78,12 +82,21 @@ def _try_get_provider() -> Optional[STTProvider]: return None -async def _transcribe_one(provider: STTProvider, segment: AudioSegment) -> None: +async def _transcribe_one( + provider: STTProvider, + segment: AudioSegment, + telemetry_scope: Optional[str], +) -> None: try: + if telemetry_scope is None: + text = await provider.transcribe(AudioChunk(data=segment.audio)) + segment.transcript = text or None + return + with voice_span( "voice.stt.transcribe", { - "voice.stt.scope": "judge", + "voice.stt.scope": telemetry_scope, "voice.stt.speaker": segment.speaker, "voice.stt.audio_bytes": len(segment.audio), }, diff --git a/python/tests/test_judge_agent.py b/python/tests/test_judge_agent.py index 86a25fe6d..ea9ea0acf 100644 --- a/python/tests/test_judge_agent.py +++ b/python/tests/test_judge_agent.py @@ -754,7 +754,7 @@ async def test_ac9_transcribe_segments_invoked_for_text_judge(): patch("scenario.judge_agent.litellm.completion", return_value=_make_llm_mock_response()): await judge.call(agent_input) - mock_ts.assert_called_once_with(recording) + mock_ts.assert_called_once_with(recording, telemetry_scope="judge") finally: context_scenario.reset(token) @@ -781,6 +781,6 @@ async def test_ac5b_stt_bridge_judge_invokes_transcribe_segments(): patch("scenario.judge_agent.litellm.completion", return_value=_make_llm_mock_response()): await judge.call(agent_input) - mock_ts.assert_called_once_with(recording) + mock_ts.assert_called_once_with(recording, telemetry_scope="judge") finally: context_scenario.reset(token) diff --git a/python/tests/voice/test_judge_stt_telemetry.py b/python/tests/voice/test_judge_stt_telemetry.py index ce79bf1ce..7aa0a83e4 100644 --- a/python/tests/voice/test_judge_stt_telemetry.py +++ b/python/tests/voice/test_judge_stt_telemetry.py @@ -71,7 +71,7 @@ async def transcribe(self, _audio): return "account restored" recording = VoiceRecording(segments=[_segment("agent", 1)]) - await transcribe_segments(recording, provider=_STT()) + await transcribe_segments(recording, provider=_STT(), telemetry_scope="judge") spans = [ s for s in exporter.get_finished_spans() if s.name == "voice.stt.transcribe" @@ -99,7 +99,9 @@ async def transcribe(self, audio): recording = VoiceRecording(segments=[_segment("user", 1), _segment("agent", 2)]) with caplog.at_level(logging.WARNING, logger="scenario.voice"): - await transcribe_segments(recording, provider=_MixedSTT()) + await transcribe_segments( + recording, provider=_MixedSTT(), telemetry_scope="judge" + ) assert [segment.transcript for segment in recording.segments] == [ None, @@ -126,3 +128,22 @@ async def transcribe(self, audio): assert "provider response" not in recorded assert raw_error not in caplog.text assert "STT provider failed: RuntimeError" in caplog.text + + +@pytest.mark.asyncio +async def test_direct_transcription_is_uninstrumented_by_default(): + exporter = _install_exporter() + + class _STT(STTProvider): + async def transcribe(self, _audio): + return "public caller transcript" + + recording = VoiceRecording(segments=[_segment("agent", 1)]) + await transcribe_segments(recording, provider=_STT()) + + assert recording.segments[0].transcript == "public caller transcript" + assert not [ + span + for span in exporter.get_finished_spans() + if span.name == "voice.stt.transcribe" + ] From fac25413a1562c1f16b4dcb415cc8df7a056a5a5 Mon Sep 17 00:00:00 2001 From: Prajval Rasik Date: Sat, 15 Aug 2026 18:55:16 +0530 Subject: [PATCH 3/3] fix(voice): keep STT provider failures sanitized --- .../__tests__/judge-stt-telemetry.test.ts | 19 ++++++++++++++++++- python/scenario/voice/_transcribe.py | 8 ++------ 2 files changed, 20 insertions(+), 7 deletions(-) diff --git a/javascript/src/voice/__tests__/judge-stt-telemetry.test.ts b/javascript/src/voice/__tests__/judge-stt-telemetry.test.ts index 3cbca9790..3484dbc88 100644 --- a/javascript/src/voice/__tests__/judge-stt-telemetry.test.ts +++ b/javascript/src/voice/__tests__/judge-stt-telemetry.test.ts @@ -7,7 +7,7 @@ import type { ModelMessage } from "ai"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { AudioChunk } from "../audio-chunk"; -import { prepareJudgeInput } from "../judge-stt"; +import { prepareJudgeInput, transcribeAudioMessages } from "../judge-stt"; import { createAudioMessage } from "../messages"; import type { STTProvider } from "../stt"; @@ -68,6 +68,23 @@ describe("judge pre-pass voice.stt.transcribe spans (#785)", () => { expect(spans[0]!.attributes["langwatch.span.type"]).toBe("span"); }); + it("leaves direct transcription uninstrumented by default", async () => { + const stt: STTProvider = { + transcribe: vi.fn(async () => "direct transcript"), + }; + + await transcribeAudioMessages({ + messages: [createAudioMessage(tone(1), "user") as ModelMessage], + stt, + includeAudio: false, + }); + + const spans = exporter + .getFinishedSpans() + .filter((span) => span.name === "voice.stt.transcribe"); + expect(spans).toHaveLength(0); + }); + it("keeps a successful sibling when one message fails and exports only a sanitized error", async () => { const rawError = "401 invalid key sk-secret body={provider response}"; const stt: STTProvider = { diff --git a/python/scenario/voice/_transcribe.py b/python/scenario/voice/_transcribe.py index 27495a0e1..59655fa78 100644 --- a/python/scenario/voice/_transcribe.py +++ b/python/scenario/voice/_transcribe.py @@ -105,12 +105,8 @@ async def _transcribe_one( text = await provider.transcribe(AudioChunk(data=segment.audio)) except Exception as exc: # Provider SDK errors can include response bodies and key fragments. - # Keep the raw detail local and let telemetry record only a minimal - # provider-agnostic exception, matching the #783 STT guard. - logger.debug( - "scenario.voice.transcribe: STT provider error detail", - exc_info=True, - ) + # Record only a provider-agnostic exception, matching the #783 + # STT guard, so logs and telemetry expose the same safe detail. raise RuntimeError( f"STT provider failed: {type(exc).__name__}" ) from None