feat(voice): instrument judge STT pre-pass

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

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<string, string>;
   /** 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<ModelMessage[]> {
-  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<string, string>,
+  telemetryScope?: "judge",
 ): Promise<ModelMessage> {
   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<string> => 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

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
                 )

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")
 
@@ -31,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.
@@ -42,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.
@@ -55,12 +60,13 @@ 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:
         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]:
@@ -76,10 +82,41 @@ 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:
-        text = await provider.transcribe(AudioChunk(data=segment.audio))
-        segment.transcript = text or None
+        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": telemetry_scope,
+                "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",

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)