Module scenario.voice.adapter
VoiceAgentAdapter — base class for voice-capable agents.
Extends AgentAdapter (text-based) with audio send/receive primitives and a
capability matrix. Concrete subclasses live under
scenario.voice.adapters (PipecatAgentAdapter, LiveKitAgentAdapter, etc.).
The scenario executor calls connect() automatically at scenario start and
disconnect() at end — users do not manage lifecycle.
The default call() implementation records the audio it sends and receives
into the executor's VoiceRecording so result.audio is populated without
each adapter needing its own bookkeeping.
Expand source code
"""
VoiceAgentAdapter — base class for voice-capable agents.
Extends AgentAdapter (text-based) with audio send/receive primitives and a
capability matrix. Concrete subclasses live under
``scenario.voice.adapters`` (PipecatAgentAdapter, LiveKitAgentAdapter, etc.).
The scenario executor calls ``connect()`` automatically at scenario start and
``disconnect()`` at end — users do not manage lifecycle.
The default ``call()`` implementation records the audio it sends and receives
into the executor's ``VoiceRecording`` so ``result.audio`` is populated without
each adapter needing its own bookkeeping.
"""
from __future__ import annotations
import asyncio
import contextlib
import dataclasses
import logging
import time
from abc import abstractmethod
from typing import Any, Callable, ClassVar, Iterator, List, Optional
from opentelemetry import context as otel_context
from opentelemetry.context import Context
logger = logging.getLogger("scenario.voice")
from ..agent_adapter import AgentAdapter
from ..types import AgentInput, AgentReturnTypes, AgentRole
from .audio_chunk import AudioChunk
from .capabilities import AdapterCapabilities
from .messages import create_audio_message, extract_audio
from .stt import transcribe
from .recording import AudioSegment, VoiceEvent
from ._telemetry import voice_span
_FIRST_CHUNK_PHASE = "first-chunk"
"""Phase marker for the first-chunk recv timeout (used in FirstChunkTimeoutError)."""
class AgentStreamEndedError(Exception):
"""An adapter's recv_audio raises this when the agent's audio transport has
TERMINATED — a background read-loop crash or a clean close by the peer — so
no further audio can arrive on this connection.
WHY distinct from asyncio.TimeoutError: a timeout is TRANSIENT (the agent may
still be mid-think; audio could still arrive), a stream-ended is TERMINAL (the
connection is done). _drain_agent_response treats them differently: on the
FIRST chunk it propagates this unchanged (it already names the real cause —
this is the #498 diagnostic fix), on TAIL chunks it ends the turn normally
(the peer closed after the agent finished speaking). Subclasses (e.g.
PipecatRecvError) carry a transport-specific message and chain the underlying
cause via __cause__.
"""
class FirstChunkTimeoutError(asyncio.TimeoutError):
"""Raised when the agent fails to send its first audio chunk within ``response_timeout``.
WHY this subclass exists: operators could not distinguish a first-chunk hang
(agent never spoke — wrong endpoint, VAD never fired, response_timeout too
short) from a tail-silence cutoff (agent finished speaking normally). The
bare ``asyncio.TimeoutError`` that escaped previously had an empty ``str()``
and no structured attributes, so log aggregators and re-raise chains had no
signal. This class embeds the phase marker (``_FIRST_CHUNK_PHASE``) in its
message, a machine-readable ``.timeout`` attribute, and chains the original
transport error via ``__cause__``.
"""
def __init__(self, *, timeout: float) -> None:
self.timeout = timeout
self.phase = _FIRST_CHUNK_PHASE
super().__init__(
f"agent did not send its first audio chunk within {timeout}s "
f"(phase={_FIRST_CHUNK_PHASE})"
)
class VoiceAgentAdapter(AgentAdapter):
"""
Abstract base for voice agents that exchange audio with the agent under test.
Subclasses implement ``connect``, ``disconnect``, ``send_audio``, and
``recv_audio``. The default ``call`` implementation threads audio extracted
from the last incoming message through the transport and wraps the response
back into an assistant message.
Attributes:
capabilities: Declaration of what the adapter can and cannot do. Each
concrete subclass must set this as a class attribute.
response_timeout: Seconds to wait for agent audio after sending user
audio. Defaults to 60 seconds.
60 seconds covers a typical real-world STT → LLM → TTS round-trip
including backoff/retry inside each provider, tool calls, and RAG
lookups. If you see TimeoutError flakes against a fast LLM-only
chain, you can lower this; if your agent does heavy processing
(MCP roundtrips, multi-step tool chains), consider raising it.
Override per-adapter at construction time::
adapter = MyVoiceAdapter()
adapter.response_timeout = 90.0 # slow tool-call chain
"""
role: ClassVar[AgentRole] = AgentRole.AGENT
capabilities: ClassVar[AdapterCapabilities] = AdapterCapabilities()
response_timeout: float = 60.0 # 60s: STT + LLM + TTS budget (see docstring)
# Tail silence: once the first agent chunk arrives, keep draining recv_audio
# until no chunk shows up within this many seconds — that's how we detect the
# agent finished talking. Without this, demos record only the first ~100ms.
response_tail_silence: float = 0.6
# Hard cap on a single agent turn's audio. Prevents runaway loops if a
# transport never signals end-of-stream. 30s = a long sentence.
response_max_duration: float = 30.0
# Live OTel context of the CURRENT ``voice.turn``, published by ``call()``
# for background-receive-loop adapters (Pipecat/Twilio) to parent their
# detached-task recv spans under the turn (#774). ``None`` between turns.
# Class-level default so it exists even for a subclass that skips
# ``super().__init__()`` (mirrors the ``_agent_speaking`` safety-net).
_voice_turn_context: Optional[Context] = None
#: SET when the AGENT deliberately ended the call (e.g. an ElevenLabs hosted
#: agent invoking the ``end_call`` system tool), as opposed to the transport
#: dropping. A scripted turn that arrives after this concludes the
#: conversation instead of failing the run — the agent behaved as designed.
#: Assertions and judges can read it to reason about WHO ended the call.
#:
#: Class-level default for the same reason as ``_voice_turn_context``: a
#: subclass that skips ``super().__init__()`` would otherwise raise
#: AttributeError from the ``call()`` gate instead of the intended
#: TransportNotConnectedError. Matches the TS field default (``agentHungUp``).
agent_hung_up: bool = False
def __init__(self) -> None:
# Per-instance event used by the interruption path to wait until
# the agent is actually speaking before firing an interrupt — so
# we don't fire ``clear`` at a silent SUT. Subclasses that
# override ``__init__`` must call ``super().__init__()``.
self._agent_speaking = asyncio.Event()
@property
def _agent_speaking_event(self) -> asyncio.Event:
"""Event set when the agent emits its first chunk of the current turn."""
# Safety net for subclasses that pre-date this base ``__init__``
# contract and didn't call ``super().__init__()``. They get a
# one-shot lazy event so the interruption path doesn't crash.
# We emit a single warning per subclass — silent fallback masks
# bugs, but a warning per call would spam the timing-critical
# interruption path. New adapters must call super().__init__().
ev = getattr(self, "_agent_speaking", None)
if ev is None:
cls = type(self)
if not getattr(cls, "_agent_speaking_lazy_warned", False):
logger.warning(
"%s.__init__() did not call super().__init__(); "
"lazily initialising _agent_speaking event. "
"Add super().__init__() to silence this warning.",
cls.__name__,
)
# setattr() form: pyright won't infer this dynamic class attr
# otherwise (reportAttributeAccessIssue). Functionally identical
# to cls._agent_speaking_lazy_warned = True.
setattr(cls, "_agent_speaking_lazy_warned", True)
ev = asyncio.Event()
self._agent_speaking = ev
return ev
def is_connected(self) -> bool:
"""Whether the transport is open and ready to exchange audio.
Base default is ``True``: adapters without a persistent socket (or
that manage liveness elsewhere) are always considered ready, so the
pre-turn guard in :meth:`call` never blocks them. Transports with a
real socket override this — e.g. :class:`ElevenLabsAgentAdapter`
returns ``self._ws is not None and not self._ws.closed`` (parity with
the TS ``isConnected()`` override, ``adapters/elevenlabs.ts:531-534``).
"""
return True
@abstractmethod
async def connect(self) -> None:
"""Open the transport and prepare to exchange audio."""
@abstractmethod
async def disconnect(self) -> None:
"""Close the transport and release resources."""
@abstractmethod
async def send_audio(self, chunk: AudioChunk) -> None:
"""Transmit an AudioChunk to the agent under test."""
@abstractmethod
async def recv_audio(self, timeout: float) -> AudioChunk:
"""Receive the next AudioChunk from the agent."""
async def __aenter__(self):
# Default async context manager: subclasses don't need to
# reimplement this — they get connect/disconnect sandwiching
# for free. Override only if a transport needs extra setup
# ordering around connect.
await self.connect()
return self
async def __aexit__(self, *exc_info: Any) -> None:
await self.disconnect()
async def interrupt(self) -> None:
"""Send a first-class interrupt signal to the agent under test.
Adapters that advertise ``capabilities.interruption=True`` override
this to send the transport-native interrupt (e.g., Twilio ``clear``,
OpenAI Realtime ``response.cancel``). The agent stops generating
audio immediately — much more deterministic than racing VAD against
a wall-clock sleep.
The default raises ``UnsupportedCapabilityError``. Callers
(``scenario.interrupt()``) check ``capabilities.interruption`` and
fall back to timing-based barge-in (sending audio while the agent
is speaking) when this returns False.
"""
from .capabilities import UnsupportedCapabilityError
raise UnsupportedCapabilityError(
type(self).__name__,
"interruption",
hint=(
"This adapter has no native interrupt signal. Use the "
"timing-based barge-in pattern instead: "
"agent(wait=False) + sleep(N) + user(content), where the "
"user audio overlaps with the agent's TTS and the SUT's "
"VAD detects it."
),
)
@contextlib.contextmanager
def _voice_turn_context_scope(self) -> Iterator[None]:
"""Publish the live ``voice.turn`` OTel context for background-loop adapters.
Pipecat/Twilio run their real receive in a background task/callback whose
OTel context was frozen at ``connect()`` (a now-closed span). They read
:attr:`_voice_turn_context` to parent those detached recv spans under the
CURRENT ``voice.turn`` (#774 — the reusable pattern Twilio PR5 inherits).
Entered INSIDE the ``voice.turn`` span so ``get_current()`` captures it;
cleared on exit so a late background frame BETWEEN turns finds ``None`` and
skips its span rather than parenting under a closed turn.
"""
self._voice_turn_context = otel_context.get_current()
try:
yield
finally:
self._voice_turn_context = None
async def call(self, input: AgentInput) -> AgentReturnTypes:
"""
Default implementation: extract audio from the latest user message,
send it, drain the agent's full response (multiple recv_audio chunks
until tail silence), record once, return as one assistant audio message.
Why drain instead of taking one chunk: TTS and realtime APIs stream
their response in many small chunks. A single recv_audio() returns the
first one only — the recorder would log ~100ms of agent audio per turn
and the judge would receive a truncated response. Draining until
tail-silence (no new chunk for ``response_tail_silence`` seconds) gives
the natural "agent finished talking" signal that works across
adapters without each one needing to know its transport's done event.
Subclasses may override this for specialised flows but will usually
inherit it.
"""
# Uniform pre-turn connected-state gate (mirror TS
# ``adapter.runtime.ts:249-254``): a call() issued before the
# executor's connect() — or after a dropped transport — fails once with
# a clear error naming the adapter, rather than a transport-specific
# null-deref or a silent hang. Checked ONCE, BEFORE send_audio/
# recv_audio. It does NOT suppress ``FirstChunkTimeoutError``: a
# connected adapter whose first chunk never arrives still surfaces that
# timeout from the drain below.
#
# We raise ``TransportNotConnectedError`` (a subclass of
# ``PendingTransportError``, so the TS-parity ``except
# PendingTransportError`` gate still catches it) whose message is
# actionable for a real, implemented adapter — "call connect()/reconnect"
# — rather than the base "implement your transport" guidance meant for
# unshipped stubs.
if not self.is_connected():
if self.agent_hung_up:
# The AGENT ended the call on purpose (issue #839) — hosted
# agents routinely invoke a hangup tool right after their
# farewell, which closes the transport. Any scripted turn left
# in the script has nobody to talk to, but the agent did
# exactly what it was designed to do, so concluding here and
# letting the script fall through to the judge is the correct
# outcome. Failing the run would punish correct behaviour.
# Returning no messages leaves the transcript ending on the
# agent's farewell, which is what the judge should assess.
logger.info(
"%s: agent ended the call; concluding the conversation "
"instead of failing the remaining scripted turn(s)",
type(self).__name__,
)
return []
from .adapters._stub import TransportNotConnectedError
raise TransportNotConnectedError(type(self).__name__)
# One ``voice.turn`` span per call(), nesting under the executor's
# existing ``{cls}.call`` agent span (ambient OTel context — no parent
# passed). The transport spans below (send/receive) nest under it.
turn_index = getattr(
getattr(input, "scenario_state", None), "current_turn", None
)
with voice_span(
"voice.turn",
{
"voice.adapter.class": type(self).__name__,
"voice.turn.index": turn_index,
},
) as _turn_span, self._voice_turn_context_scope():
_turn_started = time.monotonic()
# Clear the speaking-event for this turn — set in _drain on first chunk.
self._agent_speaking_event.clear()
recorder = _AdapterRecorder(input)
incoming = (
extract_audio(input.new_messages[-1]) if input.new_messages else None
)
if incoming is not None:
# BEFORE the user's audio goes out (and before the user segment
# is written, so an agent segment is still last on the cursor):
# sweep up any agent audio an early turn close stranded in
# flight, so it lands on the turn that produced it instead of
# bleeding out as the next turn's opening audio (#749).
await reconcile_prior_agent_audio(
self, recorder._executor, recorder._offset()
)
# The sweep is cleanup of the PREVIOUS turn, so re-stamp the
# start: leaving it would bill this turn for time spent draining
# the last one and inflate the reported turn latency.
_turn_started = time.monotonic()
# Wrap send_audio so user.start = "we began transmitting" and
# user.end = "we finished transmitting" — both real flow points.
recorder.mark_user_start()
with voice_span(
"voice.audio.send", {"voice.audio.bytes": len(incoming.data)}
):
await self.send_audio(incoming)
recorder.record_user(incoming)
# Drain. Recorder grabs agent.start at first chunk via
# mark_agent_start, so agent.start is "first chunk on the wire,"
# not "now minus merged.duration."
merged = await self._drain_agent_response(
on_first_chunk=recorder.mark_agent_start
)
# Mark agent.end BEFORE the STT round-trip below — the agent stopped
# speaking when drain settled, not after transcription returned.
recorder.mark_agent_end()
_turn_span.set_attribute(
"voice.turn.latency_ms",
round((time.monotonic() - _turn_started) * 1000),
)
if incoming is not None and incoming.data:
_turn_span.set_attribute(
"voice.turn.user_audio_bytes", len(incoming.data)
)
if merged.data:
_turn_span.set_attribute(
"voice.turn.agent_audio_bytes", len(merged.data)
)
merged = await self._ensure_transcript(merged)
recorder.record_agent(merged)
return create_audio_message(merged, role="assistant")
async def _ensure_transcript(self, merged: AudioChunk) -> AudioChunk:
"""Best-effort runtime STT for adapters whose transport carries no text.
The assistant message built from this chunk feeds the conversation
history that the user simulator (a text-only LLM) reads. Without a
text part the simulator sees the ``[audio message]`` placeholder for
every agent turn and replies blind — both sides talk past each other
while the judge, which transcribes the recording post-hoc, still
renders a perfectly readable transcript. Transcribing here closes
that gap; it also fills the recording segment's transcript, so the
judge fallback (``transcribe_segments(only_missing=True)``) skips
these turns.
Adapters that already ship transcripts on their chunks (realtime
APIs) return them merged — ``merged.transcript`` is set and this is
a no-op. STT failures are logged and the audio-only chunk is
returned unchanged, same contract as ``transcribe_segments``.
"""
if not merged.data or merged.transcript:
return merged
try:
# ``voice.stt.transcribe`` — one span per per-turn STT provider call,
# nesting under this turn's ``voice.turn`` span (#776). Python runs
# STT per-turn (here); the TS mirror is a per-RUN back-fill batch
# (``voice.stt.backfill`` → ``voice.stt.transcribe``). Shared
# attributes across BOTH languages: ``voice.stt.scope`` /
# ``.speaker`` / ``.audio_bytes`` / ``.transcript_chars``.
# ``voice.adapter.class`` is Python-ONLY here: the per-turn STT always
# transcribes THIS adapter's own agent output, whereas the TS per-run
# back-fill is adapter-agnostic over recorded segments. Disambiguate
# cross-language by ``voice.stt.scope`` (turn vs run).
with voice_span(
"voice.stt.transcribe",
{
"voice.adapter.class": type(self).__name__,
"voice.stt.scope": "turn",
"voice.stt.speaker": "agent",
"voice.stt.audio_bytes": len(merged.data),
},
) as stt_span:
try:
text = await transcribe(merged)
except Exception as exc:
# Sanitize BEFORE the span records it: provider SDK errors
# (OpenAI/ElevenLabs) can embed the raw response body — and a
# key fragment on a 401 — in the message, which
# ``voice_span``'s ``record_exception`` would EXPORT to
# telemetry. Log the detail locally at DEBUG (non-exported);
# raise a minimal, provider-agnostic error (``from None`` so
# the chained original is not formatted into the recorded
# traceback) so the span still marks ERROR without leaking.
# Mirrors ``ElevenLabsSTTProvider`` in ``voice/stt.py``.
logger.debug(
"voice: STT provider error detail", exc_info=True
)
raise RuntimeError(
f"STT provider failed: {type(exc).__name__}"
) from None
if text:
stt_span.set_attribute("voice.stt.transcript_chars", len(text))
except Exception:
logger.warning(
"voice: agent-turn STT failed; the user simulator will see "
"'[audio message]' instead of this turn's words",
)
logger.debug("voice: agent-turn STT failure detail", exc_info=True)
return merged
if not text:
return merged
return dataclasses.replace(merged, transcript=text)
async def _drain_agent_response(
self, on_first_chunk: Optional[Callable[[], None]] = None
) -> AudioChunk:
"""Loop ``recv_audio`` until tail silence or max duration; merge result.
``on_first_chunk`` is invoked synchronously the moment the first
non-empty audio chunk arrives — used by the recorder to capture
agent.start at a real flow point rather than back-computing from
the merged-chunk duration.
"""
# ``voice.audio.receive`` — THE user-story span: a receiveAudio timeout
# surfaces as ERROR here, so the trace shows WHY a run failed.
with voice_span("voice.audio.receive") as _recv_span:
_recv_started = time.monotonic()
try:
first = await self.recv_audio(timeout=self.response_timeout)
except asyncio.TimeoutError as err:
_recv_span.set_attribute(
"voice.audio.terminated_reason", "first_chunk_timeout"
)
raise FirstChunkTimeoutError(timeout=self.response_timeout) from err
# An AgentStreamEndedError is intentionally NOT caught here: it is not a
# TimeoutError, so it propagates past this handler unchanged. That
# preserves the real terminal cause (recv-loop crash or clean peer close)
# for the #498 diagnostic fix instead of masking it as a first-chunk
# timeout. Do NOT add a bare ``except Exception`` here. It exits the
# span as ERROR, which is exactly what the trace should show.
_recv_span.set_attribute(
"voice.audio.first_chunk_latency_ms",
round((time.monotonic() - _recv_started) * 1000),
)
# First chunk arrived → agent is now speaking. Wakes anyone awaiting
# _agent_speaking_event (the interruption path).
if first.data and on_first_chunk is not None:
on_first_chunk()
self._agent_speaking_event.set()
chunks: List[AudioChunk] = [first]
accumulated = first.duration_seconds
# Default reason: the loop condition fell through, i.e. the runaway
# backstop (Python-only — TS terminates on a hard ceiling instead).
terminated_reason = "max_duration"
while accumulated < self.response_max_duration:
try:
nxt = await self.recv_audio(timeout=self.response_tail_silence)
except asyncio.TimeoutError:
terminated_reason = "tail_silence"
break
except AgentStreamEndedError:
# The stream ended after the agent already spoke — a normal
# end-of-turn (the peer closed once it finished). Return the
# audio collected so far instead of surfacing the terminal cause.
terminated_reason = "stream_ended"
break
if not nxt.data:
terminated_reason = "terminal_chunk"
break
chunks.append(nxt)
accumulated += nxt.duration_seconds
merged = _merge_chunks(chunks)
_recv_span.set_attribute("voice.audio.terminated_reason", terminated_reason)
_recv_span.set_attribute("voice.audio.chunk_count", len(chunks))
_recv_span.set_attribute(
"voice.audio.bytes", len(merged.data) if merged.data else 0
)
return merged
class _AdapterRecorder:
"""Bridges a single call() turn's audio and timing into the executor state.
Kept as a private helper so the default ``VoiceAgentAdapter.call`` stays
short and each subclass can opt-out by overriding ``call()``.
Timing model: every segment's start/end is captured at a real audio
flow point — when transmission begins, when it ends, when the first
chunk arrives. Nothing is back-computed from chunk byte length, so
user and agent segments share a single timeline and do not overlap.
"""
def __init__(self, input: AgentInput) -> None:
# ``scenario_state`` is declared on AgentInput, but tests use lightweight
# _FakeInput stubs that don't carry it. Guard so the recorder
# degrades to a no-op (segments unwritten) instead of crashing the
# call(), matching the established test-double seam.
state = getattr(input, "scenario_state", None)
executor = getattr(state, "_executor", None) if state is not None else None
self._executor = executor
self._user_start: Optional[float] = None
self._user_end: Optional[float] = None
self._agent_start: Optional[float] = None
self._agent_end: Optional[float] = None
def _offset(self) -> float:
anchor = getattr(self._executor, "_voice_recording_started_at", None)
if anchor is None:
return 0.0
return time.monotonic() - anchor
def mark_user_start(self) -> None:
"""Capture the moment send_audio begins transmitting user audio."""
self._user_start = self._offset()
def record_user(self, chunk: AudioChunk) -> None:
"""Finalise the user segment after send_audio returns.
Uses real flow timestamps: start = when transmission began,
end = now (transmission complete). The chunk's intrinsic
duration is metadata only, not used to compute timestamps.
"""
if self._executor is None or not chunk.data:
return
end = self._offset()
start = self._user_start if self._user_start is not None else end
self._user_end = end
write_user_segment(self._executor, chunk, start, end)
def mark_agent_start(self) -> None:
"""Capture the moment the first agent chunk arrives.
Called by ``_drain_agent_response`` synchronously when its first
non-empty chunk lands, so the agent segment's start reflects when
audio actually started flowing back from the AUT — not when drain
eventually returns.
"""
self._agent_start = self._offset()
def mark_agent_end(self) -> None:
"""Capture the moment drain settles — the agent has stopped speaking.
Called before any post-drain processing (runtime STT) so the agent
segment's end and the agent_stop_speaking event reflect when audio
stopped flowing, not when transcription returned.
"""
self._agent_end = self._offset()
def record_agent(self, chunk: AudioChunk) -> None:
"""Finalise the agent segment after drain completes.
start = when first chunk arrived (captured by mark_agent_start).
end = when drain settled (captured by mark_agent_end).
latency = agent.start - user.end. Real measurement; no clamp.
"""
if self._executor is None or not chunk.data:
return
_fire_audio_chunk(self._executor, chunk)
end = self._agent_end if self._agent_end is not None else self._offset()
start = self._agent_start if self._agent_start is not None else end
_append_segment(self._executor, "agent", start, end, chunk)
latency = None
if self._user_end is not None:
latency = start - self._user_end
# Negative latency means the agent began emitting audio before
# the user audio finished transmitting — which the wire model
# forbids on serial adapters. Treat as a measurement artefact
# and skip the record so p50/p95 aren't poisoned.
if latency >= 0:
_record_latency(self._executor, latency)
else:
latency = None
_append_event(
self._executor,
VoiceEvent(time=start, type="agent_start_speaking", latency=latency),
)
_append_event(self._executor, VoiceEvent(time=end, type="agent_stop_speaking"))
def _merge_chunks(chunks: List[AudioChunk]) -> AudioChunk:
"""Concatenate PCM bytes from drained agent chunks into one AudioChunk.
Transcripts: each adapter populates ``chunk.transcript`` differently —
some on the last chunk (after STT settles), some incrementally. Joining
non-empty transcripts with a space preserves whatever the adapter shipped
without forcing adapters to coordinate.
"""
if len(chunks) == 1:
return chunks[0]
data = b"".join(c.data for c in chunks)
parts = [c.transcript for c in chunks if c.transcript]
transcript = " ".join(parts) if parts else None
return AudioChunk(data=data, transcript=transcript)
def write_user_segment(executor, chunk: AudioChunk, start: float, end: float) -> None:
"""Append a finalised user segment + start/stop timeline events.
Single path that both ``_AdapterRecorder.record_user`` (the default
``call()`` flow) and ``ScenarioExecutor._record_interrupt_user_segment``
(the barge-in flow that bypasses the recorder) call into. Previously
those two paths each open-coded the same four-step sequence
(``_fire_audio_chunk`` + ``_append_segment`` + two ``_append_event``s),
drifting apart as the timing model evolved.
"""
if executor is None or not chunk.data:
return
_fire_audio_chunk(executor, chunk)
_append_segment(executor, "user", start, end, chunk)
_append_event(executor, VoiceEvent(time=start, type="user_start_speaking"))
_append_event(executor, VoiceEvent(time=end, type="user_stop_speaking"))
async def reconcile_prior_agent_audio(
adapter: Any, executor: Any, now: float
) -> None:
"""Sweep up agent audio stranded by an early turn close and give it back to
the utterance that produced it (issue #749; TypeScript parity with #748).
The drain ends a turn on ``response_tail_silence``. A delivery gap longer
than that leaves the rest of the agent's utterance in flight, and the next
drain shifts it out as the opening audio of the NEXT agent turn — so turn
N+1 appears to answer question N. Run at the pre-user-``send_audio``
boundary, where the agent cannot yet have begun its next reply, so anything
still arriving is unambiguously the prior turn's tail.
Attribution is position-gated, mirroring the TypeScript reconcile:
- Cursor-safe (an AGENT segment is still last — the user segment for this
turn has not been written yet): grow it. Extending its ``end_time`` and
audio cannot overlap a later segment. The transcript is left ALONE: the
turn was cut short in AUDIO only — the provider's own ``agent_response``
text already covers the whole utterance — so appending the tail makes the
two consistent. Clearing it would discard a correct transcript and, for an
audio-capable judge (which never runs the STT back-fill), leave the
segment with none at all.
- Cursor-unsafe (no recording, the opening greeting with no prior agent
segment, or a barge-in where a user segment is last): the audio is already
off the wire, so the bleed is prevented either way; it is dropped with a
warning rather than corrupting the append-only cursor.
Adapters that expose no ``reconcile_pending_audio`` are untouched.
``adapter`` and ``executor`` are deliberately ``Any``: this helper is
duck-typed on both sides — it feature-detects ``reconcile_pending_audio``
rather than requiring :class:`VoiceAgentAdapter`, and reads the recording off
the executor with ``getattr`` so the lightweight test doubles the recorder
already tolerates work here too. Narrowing either to a concrete class would
describe a contract this function does not actually enforce.
"""
reconcile = getattr(adapter, "reconcile_pending_audio", None)
if reconcile is None:
return
try:
leftover = await reconcile()
except Exception: # noqa: BLE001 — opportunistic cleanup must never fail a turn
logger.warning(
"%s: turn-boundary reconcile raised; continuing.",
type(adapter).__name__,
exc_info=True,
)
return
if leftover is None or not leftover.data:
return
recording = getattr(executor, "_voice_recording", None) if executor else None
segments = getattr(recording, "segments", None) if recording is not None else None
if segments and segments[-1].speaker == "agent":
prior = segments[-1]
prior.audio += leftover.data
prior.end_time = max(prior.end_time, now)
_fire_audio_chunk(executor, leftover)
logger.warning(
"%s: recovered %d bytes of agent audio stranded by an early turn "
"close and attributed them to the preceding agent turn. Raise "
"response_tail_silence if this recurs.",
type(adapter).__name__,
len(leftover.data),
)
else:
logger.warning(
"%s: discarded %d bytes of agent audio at a user-turn boundary — "
"there is no preceding agent segment to attribute them to, and "
"growing an out-of-order segment would corrupt the recording. If "
"your script does not lead with agent(), this is the on-connect "
"greeting arriving after the first user turn was queued; lead with "
"agent() so the greeting is drained as its own turn.",
type(adapter).__name__,
len(leftover.data),
)
def _append_segment(executor, speaker: str, start: float, end: float, chunk: AudioChunk) -> None:
recording = getattr(executor, "_voice_recording", None)
if recording is None:
return
recording.segments.append(
AudioSegment(
speaker=speaker, # type: ignore[arg-type]
start_time=start,
end_time=end,
audio=chunk.data,
transcript=chunk.transcript,
)
)
def _append_event(executor, event: VoiceEvent) -> None:
timeline = getattr(executor, "_voice_timeline", None)
if timeline is None:
return
timeline.append(event)
hook = getattr(executor, "_on_voice_event", None)
if hook is not None:
try:
hook(event)
except Exception:
logger.warning(
"on_voice_event callback raised; continuing scenario.",
exc_info=True,
)
def _fire_audio_chunk(executor, chunk: AudioChunk) -> None:
hook = getattr(executor, "_on_audio_chunk", None)
if hook is None:
return
try:
hook(chunk)
except Exception:
logger.warning(
"on_audio_chunk callback raised; continuing scenario.",
exc_info=True,
)
def _record_latency(executor, latency: float) -> None:
metrics = getattr(executor, "_voice_latency", None)
if metrics is None:
return
metrics.measurements.append(latency)
if metrics.time_to_first_byte is None:
metrics.time_to_first_byte = latencyFunctions
async def reconcile_prior_agent_audio(adapter: Any, executor: Any, now: float) ‑> None-
Sweep up agent audio stranded by an early turn close and give it back to the utterance that produced it (issue #749; TypeScript parity with #748).
The drain ends a turn on
response_tail_silence. A delivery gap longer than that leaves the rest of the agent's utterance in flight, and the next drain shifts it out as the opening audio of the NEXT agent turn — so turn N+1 appears to answer question N. Run at the pre-user-send_audioboundary, where the agent cannot yet have begun its next reply, so anything still arriving is unambiguously the prior turn's tail.Attribution is position-gated, mirroring the TypeScript reconcile:
- Cursor-safe (an AGENT segment is still last — the user segment for this
turn has not been written yet): grow it. Extending its
end_timeand audio cannot overlap a later segment. The transcript is left ALONE: the turn was cut short in AUDIO only — the provider's ownagent_responsetext already covers the whole utterance — so appending the tail makes the two consistent. Clearing it would discard a correct transcript and, for an audio-capable judge (which never runs the STT back-fill), leave the segment with none at all. - Cursor-unsafe (no recording, the opening greeting with no prior agent segment, or a barge-in where a user segment is last): the audio is already off the wire, so the bleed is prevented either way; it is dropped with a warning rather than corrupting the append-only cursor.
Adapters that expose no
reconcile_pending_audioare untouched.adapterandexecutorare deliberatelyAny: this helper is duck-typed on both sides — it feature-detectsreconcile_pending_audiorather than requiring :class:VoiceAgentAdapter, and reads the recording off the executor withgetattrso the lightweight test doubles the recorder already tolerates work here too. Narrowing either to a concrete class would describe a contract this function does not actually enforce.Expand source code
async def reconcile_prior_agent_audio( adapter: Any, executor: Any, now: float ) -> None: """Sweep up agent audio stranded by an early turn close and give it back to the utterance that produced it (issue #749; TypeScript parity with #748). The drain ends a turn on ``response_tail_silence``. A delivery gap longer than that leaves the rest of the agent's utterance in flight, and the next drain shifts it out as the opening audio of the NEXT agent turn — so turn N+1 appears to answer question N. Run at the pre-user-``send_audio`` boundary, where the agent cannot yet have begun its next reply, so anything still arriving is unambiguously the prior turn's tail. Attribution is position-gated, mirroring the TypeScript reconcile: - Cursor-safe (an AGENT segment is still last — the user segment for this turn has not been written yet): grow it. Extending its ``end_time`` and audio cannot overlap a later segment. The transcript is left ALONE: the turn was cut short in AUDIO only — the provider's own ``agent_response`` text already covers the whole utterance — so appending the tail makes the two consistent. Clearing it would discard a correct transcript and, for an audio-capable judge (which never runs the STT back-fill), leave the segment with none at all. - Cursor-unsafe (no recording, the opening greeting with no prior agent segment, or a barge-in where a user segment is last): the audio is already off the wire, so the bleed is prevented either way; it is dropped with a warning rather than corrupting the append-only cursor. Adapters that expose no ``reconcile_pending_audio`` are untouched. ``adapter`` and ``executor`` are deliberately ``Any``: this helper is duck-typed on both sides — it feature-detects ``reconcile_pending_audio`` rather than requiring :class:`VoiceAgentAdapter`, and reads the recording off the executor with ``getattr`` so the lightweight test doubles the recorder already tolerates work here too. Narrowing either to a concrete class would describe a contract this function does not actually enforce. """ reconcile = getattr(adapter, "reconcile_pending_audio", None) if reconcile is None: return try: leftover = await reconcile() except Exception: # noqa: BLE001 — opportunistic cleanup must never fail a turn logger.warning( "%s: turn-boundary reconcile raised; continuing.", type(adapter).__name__, exc_info=True, ) return if leftover is None or not leftover.data: return recording = getattr(executor, "_voice_recording", None) if executor else None segments = getattr(recording, "segments", None) if recording is not None else None if segments and segments[-1].speaker == "agent": prior = segments[-1] prior.audio += leftover.data prior.end_time = max(prior.end_time, now) _fire_audio_chunk(executor, leftover) logger.warning( "%s: recovered %d bytes of agent audio stranded by an early turn " "close and attributed them to the preceding agent turn. Raise " "response_tail_silence if this recurs.", type(adapter).__name__, len(leftover.data), ) else: logger.warning( "%s: discarded %d bytes of agent audio at a user-turn boundary — " "there is no preceding agent segment to attribute them to, and " "growing an out-of-order segment would corrupt the recording. If " "your script does not lead with agent(), this is the on-connect " "greeting arriving after the first user turn was queued; lead with " "agent() so the greeting is drained as its own turn.", type(adapter).__name__, len(leftover.data), ) - Cursor-safe (an AGENT segment is still last — the user segment for this
turn has not been written yet): grow it. Extending its
def write_user_segment(executor, chunk: AudioChunk, start: float, end: float) ‑> None-
Append a finalised user segment + start/stop timeline events.
Single path that both
_AdapterRecorder.record_user(the defaultcall()flow) andScenarioExecutor._record_interrupt_user_segment(the barge-in flow that bypasses the recorder) call into. Previously those two paths each open-coded the same four-step sequence (_fire_audio_chunk+_append_segment+ two_append_events), drifting apart as the timing model evolved.Expand source code
def write_user_segment(executor, chunk: AudioChunk, start: float, end: float) -> None: """Append a finalised user segment + start/stop timeline events. Single path that both ``_AdapterRecorder.record_user`` (the default ``call()`` flow) and ``ScenarioExecutor._record_interrupt_user_segment`` (the barge-in flow that bypasses the recorder) call into. Previously those two paths each open-coded the same four-step sequence (``_fire_audio_chunk`` + ``_append_segment`` + two ``_append_event``s), drifting apart as the timing model evolved. """ if executor is None or not chunk.data: return _fire_audio_chunk(executor, chunk) _append_segment(executor, "user", start, end, chunk) _append_event(executor, VoiceEvent(time=start, type="user_start_speaking")) _append_event(executor, VoiceEvent(time=end, type="user_stop_speaking"))
Classes
class AgentStreamEndedError (*args, **kwargs)-
An adapter's recv_audio raises this when the agent's audio transport has TERMINATED — a background read-loop crash or a clean close by the peer — so no further audio can arrive on this connection.
WHY distinct from asyncio.TimeoutError: a timeout is TRANSIENT (the agent may still be mid-think; audio could still arrive), a stream-ended is TERMINAL (the connection is done). _drain_agent_response treats them differently: on the FIRST chunk it propagates this unchanged (it already names the real cause — this is the #498 diagnostic fix), on TAIL chunks it ends the turn normally (the peer closed after the agent finished speaking). Subclasses (e.g. PipecatRecvError) carry a transport-specific message and chain the underlying cause via cause.
Expand source code
class AgentStreamEndedError(Exception): """An adapter's recv_audio raises this when the agent's audio transport has TERMINATED — a background read-loop crash or a clean close by the peer — so no further audio can arrive on this connection. WHY distinct from asyncio.TimeoutError: a timeout is TRANSIENT (the agent may still be mid-think; audio could still arrive), a stream-ended is TERMINAL (the connection is done). _drain_agent_response treats them differently: on the FIRST chunk it propagates this unchanged (it already names the real cause — this is the #498 diagnostic fix), on TAIL chunks it ends the turn normally (the peer closed after the agent finished speaking). Subclasses (e.g. PipecatRecvError) carry a transport-specific message and chain the underlying cause via __cause__. """Ancestors
- builtins.Exception
- builtins.BaseException
Subclasses
class FirstChunkTimeoutError (*, timeout: float)-
Raised when the agent fails to send its first audio chunk within
response_timeout.WHY this subclass exists: operators could not distinguish a first-chunk hang (agent never spoke — wrong endpoint, VAD never fired, response_timeout too short) from a tail-silence cutoff (agent finished speaking normally). The bare
asyncio.TimeoutErrorthat escaped previously had an emptystr()and no structured attributes, so log aggregators and re-raise chains had no signal. This class embeds the phase marker (_FIRST_CHUNK_PHASE) in its message, a machine-readable.timeoutattribute, and chains the original transport error via__cause__.Expand source code
class FirstChunkTimeoutError(asyncio.TimeoutError): """Raised when the agent fails to send its first audio chunk within ``response_timeout``. WHY this subclass exists: operators could not distinguish a first-chunk hang (agent never spoke — wrong endpoint, VAD never fired, response_timeout too short) from a tail-silence cutoff (agent finished speaking normally). The bare ``asyncio.TimeoutError`` that escaped previously had an empty ``str()`` and no structured attributes, so log aggregators and re-raise chains had no signal. This class embeds the phase marker (``_FIRST_CHUNK_PHASE``) in its message, a machine-readable ``.timeout`` attribute, and chains the original transport error via ``__cause__``. """ def __init__(self, *, timeout: float) -> None: self.timeout = timeout self.phase = _FIRST_CHUNK_PHASE super().__init__( f"agent did not send its first audio chunk within {timeout}s " f"(phase={_FIRST_CHUNK_PHASE})" )Ancestors
- builtins.TimeoutError
- builtins.OSError
- builtins.Exception
- builtins.BaseException
class VoiceAgentAdapter-
Abstract base for voice agents that exchange audio with the agent under test.
Subclasses implement
connect,disconnect,send_audio, andrecv_audio. The defaultcallimplementation threads audio extracted from the last incoming message through the transport and wraps the response back into an assistant message.Attributes
capabilities- Declaration of what the adapter can and cannot do. Each concrete subclass must set this as a class attribute.
response_timeout-
Seconds to wait for agent audio after sending user audio. Defaults to 60 seconds.
60 seconds covers a typical real-world STT → LLM → TTS round-trip including backoff/retry inside each provider, tool calls, and RAG lookups. If you see TimeoutError flakes against a fast LLM-only chain, you can lower this; if your agent does heavy processing (MCP roundtrips, multi-step tool chains), consider raising it.
Override per-adapter at construction time::
adapter = MyVoiceAdapter() adapter.response_timeout = 90.0 # slow tool-call chain
Expand source code
class VoiceAgentAdapter(AgentAdapter): """ Abstract base for voice agents that exchange audio with the agent under test. Subclasses implement ``connect``, ``disconnect``, ``send_audio``, and ``recv_audio``. The default ``call`` implementation threads audio extracted from the last incoming message through the transport and wraps the response back into an assistant message. Attributes: capabilities: Declaration of what the adapter can and cannot do. Each concrete subclass must set this as a class attribute. response_timeout: Seconds to wait for agent audio after sending user audio. Defaults to 60 seconds. 60 seconds covers a typical real-world STT → LLM → TTS round-trip including backoff/retry inside each provider, tool calls, and RAG lookups. If you see TimeoutError flakes against a fast LLM-only chain, you can lower this; if your agent does heavy processing (MCP roundtrips, multi-step tool chains), consider raising it. Override per-adapter at construction time:: adapter = MyVoiceAdapter() adapter.response_timeout = 90.0 # slow tool-call chain """ role: ClassVar[AgentRole] = AgentRole.AGENT capabilities: ClassVar[AdapterCapabilities] = AdapterCapabilities() response_timeout: float = 60.0 # 60s: STT + LLM + TTS budget (see docstring) # Tail silence: once the first agent chunk arrives, keep draining recv_audio # until no chunk shows up within this many seconds — that's how we detect the # agent finished talking. Without this, demos record only the first ~100ms. response_tail_silence: float = 0.6 # Hard cap on a single agent turn's audio. Prevents runaway loops if a # transport never signals end-of-stream. 30s = a long sentence. response_max_duration: float = 30.0 # Live OTel context of the CURRENT ``voice.turn``, published by ``call()`` # for background-receive-loop adapters (Pipecat/Twilio) to parent their # detached-task recv spans under the turn (#774). ``None`` between turns. # Class-level default so it exists even for a subclass that skips # ``super().__init__()`` (mirrors the ``_agent_speaking`` safety-net). _voice_turn_context: Optional[Context] = None #: SET when the AGENT deliberately ended the call (e.g. an ElevenLabs hosted #: agent invoking the ``end_call`` system tool), as opposed to the transport #: dropping. A scripted turn that arrives after this concludes the #: conversation instead of failing the run — the agent behaved as designed. #: Assertions and judges can read it to reason about WHO ended the call. #: #: Class-level default for the same reason as ``_voice_turn_context``: a #: subclass that skips ``super().__init__()`` would otherwise raise #: AttributeError from the ``call()`` gate instead of the intended #: TransportNotConnectedError. Matches the TS field default (``agentHungUp``). agent_hung_up: bool = False def __init__(self) -> None: # Per-instance event used by the interruption path to wait until # the agent is actually speaking before firing an interrupt — so # we don't fire ``clear`` at a silent SUT. Subclasses that # override ``__init__`` must call ``super().__init__()``. self._agent_speaking = asyncio.Event() @property def _agent_speaking_event(self) -> asyncio.Event: """Event set when the agent emits its first chunk of the current turn.""" # Safety net for subclasses that pre-date this base ``__init__`` # contract and didn't call ``super().__init__()``. They get a # one-shot lazy event so the interruption path doesn't crash. # We emit a single warning per subclass — silent fallback masks # bugs, but a warning per call would spam the timing-critical # interruption path. New adapters must call super().__init__(). ev = getattr(self, "_agent_speaking", None) if ev is None: cls = type(self) if not getattr(cls, "_agent_speaking_lazy_warned", False): logger.warning( "%s.__init__() did not call super().__init__(); " "lazily initialising _agent_speaking event. " "Add super().__init__() to silence this warning.", cls.__name__, ) # setattr() form: pyright won't infer this dynamic class attr # otherwise (reportAttributeAccessIssue). Functionally identical # to cls._agent_speaking_lazy_warned = True. setattr(cls, "_agent_speaking_lazy_warned", True) ev = asyncio.Event() self._agent_speaking = ev return ev def is_connected(self) -> bool: """Whether the transport is open and ready to exchange audio. Base default is ``True``: adapters without a persistent socket (or that manage liveness elsewhere) are always considered ready, so the pre-turn guard in :meth:`call` never blocks them. Transports with a real socket override this — e.g. :class:`ElevenLabsAgentAdapter` returns ``self._ws is not None and not self._ws.closed`` (parity with the TS ``isConnected()`` override, ``adapters/elevenlabs.ts:531-534``). """ return True @abstractmethod async def connect(self) -> None: """Open the transport and prepare to exchange audio.""" @abstractmethod async def disconnect(self) -> None: """Close the transport and release resources.""" @abstractmethod async def send_audio(self, chunk: AudioChunk) -> None: """Transmit an AudioChunk to the agent under test.""" @abstractmethod async def recv_audio(self, timeout: float) -> AudioChunk: """Receive the next AudioChunk from the agent.""" async def __aenter__(self): # Default async context manager: subclasses don't need to # reimplement this — they get connect/disconnect sandwiching # for free. Override only if a transport needs extra setup # ordering around connect. await self.connect() return self async def __aexit__(self, *exc_info: Any) -> None: await self.disconnect() async def interrupt(self) -> None: """Send a first-class interrupt signal to the agent under test. Adapters that advertise ``capabilities.interruption=True`` override this to send the transport-native interrupt (e.g., Twilio ``clear``, OpenAI Realtime ``response.cancel``). The agent stops generating audio immediately — much more deterministic than racing VAD against a wall-clock sleep. The default raises ``UnsupportedCapabilityError``. Callers (``scenario.interrupt()``) check ``capabilities.interruption`` and fall back to timing-based barge-in (sending audio while the agent is speaking) when this returns False. """ from .capabilities import UnsupportedCapabilityError raise UnsupportedCapabilityError( type(self).__name__, "interruption", hint=( "This adapter has no native interrupt signal. Use the " "timing-based barge-in pattern instead: " "agent(wait=False) + sleep(N) + user(content), where the " "user audio overlaps with the agent's TTS and the SUT's " "VAD detects it." ), ) @contextlib.contextmanager def _voice_turn_context_scope(self) -> Iterator[None]: """Publish the live ``voice.turn`` OTel context for background-loop adapters. Pipecat/Twilio run their real receive in a background task/callback whose OTel context was frozen at ``connect()`` (a now-closed span). They read :attr:`_voice_turn_context` to parent those detached recv spans under the CURRENT ``voice.turn`` (#774 — the reusable pattern Twilio PR5 inherits). Entered INSIDE the ``voice.turn`` span so ``get_current()`` captures it; cleared on exit so a late background frame BETWEEN turns finds ``None`` and skips its span rather than parenting under a closed turn. """ self._voice_turn_context = otel_context.get_current() try: yield finally: self._voice_turn_context = None async def call(self, input: AgentInput) -> AgentReturnTypes: """ Default implementation: extract audio from the latest user message, send it, drain the agent's full response (multiple recv_audio chunks until tail silence), record once, return as one assistant audio message. Why drain instead of taking one chunk: TTS and realtime APIs stream their response in many small chunks. A single recv_audio() returns the first one only — the recorder would log ~100ms of agent audio per turn and the judge would receive a truncated response. Draining until tail-silence (no new chunk for ``response_tail_silence`` seconds) gives the natural "agent finished talking" signal that works across adapters without each one needing to know its transport's done event. Subclasses may override this for specialised flows but will usually inherit it. """ # Uniform pre-turn connected-state gate (mirror TS # ``adapter.runtime.ts:249-254``): a call() issued before the # executor's connect() — or after a dropped transport — fails once with # a clear error naming the adapter, rather than a transport-specific # null-deref or a silent hang. Checked ONCE, BEFORE send_audio/ # recv_audio. It does NOT suppress ``FirstChunkTimeoutError``: a # connected adapter whose first chunk never arrives still surfaces that # timeout from the drain below. # # We raise ``TransportNotConnectedError`` (a subclass of # ``PendingTransportError``, so the TS-parity ``except # PendingTransportError`` gate still catches it) whose message is # actionable for a real, implemented adapter — "call connect()/reconnect" # — rather than the base "implement your transport" guidance meant for # unshipped stubs. if not self.is_connected(): if self.agent_hung_up: # The AGENT ended the call on purpose (issue #839) — hosted # agents routinely invoke a hangup tool right after their # farewell, which closes the transport. Any scripted turn left # in the script has nobody to talk to, but the agent did # exactly what it was designed to do, so concluding here and # letting the script fall through to the judge is the correct # outcome. Failing the run would punish correct behaviour. # Returning no messages leaves the transcript ending on the # agent's farewell, which is what the judge should assess. logger.info( "%s: agent ended the call; concluding the conversation " "instead of failing the remaining scripted turn(s)", type(self).__name__, ) return [] from .adapters._stub import TransportNotConnectedError raise TransportNotConnectedError(type(self).__name__) # One ``voice.turn`` span per call(), nesting under the executor's # existing ``{cls}.call`` agent span (ambient OTel context — no parent # passed). The transport spans below (send/receive) nest under it. turn_index = getattr( getattr(input, "scenario_state", None), "current_turn", None ) with voice_span( "voice.turn", { "voice.adapter.class": type(self).__name__, "voice.turn.index": turn_index, }, ) as _turn_span, self._voice_turn_context_scope(): _turn_started = time.monotonic() # Clear the speaking-event for this turn — set in _drain on first chunk. self._agent_speaking_event.clear() recorder = _AdapterRecorder(input) incoming = ( extract_audio(input.new_messages[-1]) if input.new_messages else None ) if incoming is not None: # BEFORE the user's audio goes out (and before the user segment # is written, so an agent segment is still last on the cursor): # sweep up any agent audio an early turn close stranded in # flight, so it lands on the turn that produced it instead of # bleeding out as the next turn's opening audio (#749). await reconcile_prior_agent_audio( self, recorder._executor, recorder._offset() ) # The sweep is cleanup of the PREVIOUS turn, so re-stamp the # start: leaving it would bill this turn for time spent draining # the last one and inflate the reported turn latency. _turn_started = time.monotonic() # Wrap send_audio so user.start = "we began transmitting" and # user.end = "we finished transmitting" — both real flow points. recorder.mark_user_start() with voice_span( "voice.audio.send", {"voice.audio.bytes": len(incoming.data)} ): await self.send_audio(incoming) recorder.record_user(incoming) # Drain. Recorder grabs agent.start at first chunk via # mark_agent_start, so agent.start is "first chunk on the wire," # not "now minus merged.duration." merged = await self._drain_agent_response( on_first_chunk=recorder.mark_agent_start ) # Mark agent.end BEFORE the STT round-trip below — the agent stopped # speaking when drain settled, not after transcription returned. recorder.mark_agent_end() _turn_span.set_attribute( "voice.turn.latency_ms", round((time.monotonic() - _turn_started) * 1000), ) if incoming is not None and incoming.data: _turn_span.set_attribute( "voice.turn.user_audio_bytes", len(incoming.data) ) if merged.data: _turn_span.set_attribute( "voice.turn.agent_audio_bytes", len(merged.data) ) merged = await self._ensure_transcript(merged) recorder.record_agent(merged) return create_audio_message(merged, role="assistant") async def _ensure_transcript(self, merged: AudioChunk) -> AudioChunk: """Best-effort runtime STT for adapters whose transport carries no text. The assistant message built from this chunk feeds the conversation history that the user simulator (a text-only LLM) reads. Without a text part the simulator sees the ``[audio message]`` placeholder for every agent turn and replies blind — both sides talk past each other while the judge, which transcribes the recording post-hoc, still renders a perfectly readable transcript. Transcribing here closes that gap; it also fills the recording segment's transcript, so the judge fallback (``transcribe_segments(only_missing=True)``) skips these turns. Adapters that already ship transcripts on their chunks (realtime APIs) return them merged — ``merged.transcript`` is set and this is a no-op. STT failures are logged and the audio-only chunk is returned unchanged, same contract as ``transcribe_segments``. """ if not merged.data or merged.transcript: return merged try: # ``voice.stt.transcribe`` — one span per per-turn STT provider call, # nesting under this turn's ``voice.turn`` span (#776). Python runs # STT per-turn (here); the TS mirror is a per-RUN back-fill batch # (``voice.stt.backfill`` → ``voice.stt.transcribe``). Shared # attributes across BOTH languages: ``voice.stt.scope`` / # ``.speaker`` / ``.audio_bytes`` / ``.transcript_chars``. # ``voice.adapter.class`` is Python-ONLY here: the per-turn STT always # transcribes THIS adapter's own agent output, whereas the TS per-run # back-fill is adapter-agnostic over recorded segments. Disambiguate # cross-language by ``voice.stt.scope`` (turn vs run). with voice_span( "voice.stt.transcribe", { "voice.adapter.class": type(self).__name__, "voice.stt.scope": "turn", "voice.stt.speaker": "agent", "voice.stt.audio_bytes": len(merged.data), }, ) as stt_span: try: text = await transcribe(merged) except Exception as exc: # Sanitize BEFORE the span records it: provider SDK errors # (OpenAI/ElevenLabs) can embed the raw response body — and a # key fragment on a 401 — in the message, which # ``voice_span``'s ``record_exception`` would EXPORT to # telemetry. Log the detail locally at DEBUG (non-exported); # raise a minimal, provider-agnostic error (``from None`` so # the chained original is not formatted into the recorded # traceback) so the span still marks ERROR without leaking. # Mirrors ``ElevenLabsSTTProvider`` in ``voice/stt.py``. logger.debug( "voice: STT provider error detail", exc_info=True ) raise RuntimeError( f"STT provider failed: {type(exc).__name__}" ) from None if text: stt_span.set_attribute("voice.stt.transcript_chars", len(text)) except Exception: logger.warning( "voice: agent-turn STT failed; the user simulator will see " "'[audio message]' instead of this turn's words", ) logger.debug("voice: agent-turn STT failure detail", exc_info=True) return merged if not text: return merged return dataclasses.replace(merged, transcript=text) async def _drain_agent_response( self, on_first_chunk: Optional[Callable[[], None]] = None ) -> AudioChunk: """Loop ``recv_audio`` until tail silence or max duration; merge result. ``on_first_chunk`` is invoked synchronously the moment the first non-empty audio chunk arrives — used by the recorder to capture agent.start at a real flow point rather than back-computing from the merged-chunk duration. """ # ``voice.audio.receive`` — THE user-story span: a receiveAudio timeout # surfaces as ERROR here, so the trace shows WHY a run failed. with voice_span("voice.audio.receive") as _recv_span: _recv_started = time.monotonic() try: first = await self.recv_audio(timeout=self.response_timeout) except asyncio.TimeoutError as err: _recv_span.set_attribute( "voice.audio.terminated_reason", "first_chunk_timeout" ) raise FirstChunkTimeoutError(timeout=self.response_timeout) from err # An AgentStreamEndedError is intentionally NOT caught here: it is not a # TimeoutError, so it propagates past this handler unchanged. That # preserves the real terminal cause (recv-loop crash or clean peer close) # for the #498 diagnostic fix instead of masking it as a first-chunk # timeout. Do NOT add a bare ``except Exception`` here. It exits the # span as ERROR, which is exactly what the trace should show. _recv_span.set_attribute( "voice.audio.first_chunk_latency_ms", round((time.monotonic() - _recv_started) * 1000), ) # First chunk arrived → agent is now speaking. Wakes anyone awaiting # _agent_speaking_event (the interruption path). if first.data and on_first_chunk is not None: on_first_chunk() self._agent_speaking_event.set() chunks: List[AudioChunk] = [first] accumulated = first.duration_seconds # Default reason: the loop condition fell through, i.e. the runaway # backstop (Python-only — TS terminates on a hard ceiling instead). terminated_reason = "max_duration" while accumulated < self.response_max_duration: try: nxt = await self.recv_audio(timeout=self.response_tail_silence) except asyncio.TimeoutError: terminated_reason = "tail_silence" break except AgentStreamEndedError: # The stream ended after the agent already spoke — a normal # end-of-turn (the peer closed once it finished). Return the # audio collected so far instead of surfacing the terminal cause. terminated_reason = "stream_ended" break if not nxt.data: terminated_reason = "terminal_chunk" break chunks.append(nxt) accumulated += nxt.duration_seconds merged = _merge_chunks(chunks) _recv_span.set_attribute("voice.audio.terminated_reason", terminated_reason) _recv_span.set_attribute("voice.audio.chunk_count", len(chunks)) _recv_span.set_attribute( "voice.audio.bytes", len(merged.data) if merged.data else 0 ) return mergedAncestors
- AgentAdapter
- abc.ABC
Subclasses
- ComposableVoiceAgent
- ElevenLabsAgentAdapter
- GeminiLiveAgentAdapter
- LiveKitAgentAdapter
- OpenAIRealtimeAgentAdapter
- PipecatAgentAdapter
- TwilioAgentAdapter
- VapiAgentAdapter
- WebRTCAgentAdapter
- WebSocketAgentAdapter
Class variables
var agent_hung_up : bool-
Class-level default for the same reason as
_voice_turn_context: a subclass that skipssuper().__init__()would otherwise raise AttributeError from thecall()gate instead of the intended TransportNotConnectedError. Matches the TS field default (agentHungUp). var capabilities : ClassVar[AdapterCapabilities]var response_max_duration : floatvar response_tail_silence : floatvar response_timeout : floatvar role : ClassVar[AgentRole]
Methods
async def call(self, input: AgentInput) ‑> str | openai.types.chat.chat_completion_developer_message_param.ChatCompletionDeveloperMessageParam | openai.types.chat.chat_completion_system_message_param.ChatCompletionSystemMessageParam | openai.types.chat.chat_completion_user_message_param.ChatCompletionUserMessageParam | openai.types.chat.chat_completion_assistant_message_param.ChatCompletionAssistantMessageParam | openai.types.chat.chat_completion_tool_message_param.ChatCompletionToolMessageParam | openai.types.chat.chat_completion_function_message_param.ChatCompletionFunctionMessageParam | List[openai.types.chat.chat_completion_developer_message_param.ChatCompletionDeveloperMessageParam | openai.types.chat.chat_completion_system_message_param.ChatCompletionSystemMessageParam | openai.types.chat.chat_completion_user_message_param.ChatCompletionUserMessageParam | openai.types.chat.chat_completion_assistant_message_param.ChatCompletionAssistantMessageParam | openai.types.chat.chat_completion_tool_message_param.ChatCompletionToolMessageParam | openai.types.chat.chat_completion_function_message_param.ChatCompletionFunctionMessageParam] | ScenarioResult-
Default implementation: extract audio from the latest user message, send it, drain the agent's full response (multiple recv_audio chunks until tail silence), record once, return as one assistant audio message.
Why drain instead of taking one chunk: TTS and realtime APIs stream their response in many small chunks. A single recv_audio() returns the first one only — the recorder would log ~100ms of agent audio per turn and the judge would receive a truncated response. Draining until tail-silence (no new chunk for
response_tail_silenceseconds) gives the natural "agent finished talking" signal that works across adapters without each one needing to know its transport's done event.Subclasses may override this for specialised flows but will usually inherit it.
Expand source code
async def call(self, input: AgentInput) -> AgentReturnTypes: """ Default implementation: extract audio from the latest user message, send it, drain the agent's full response (multiple recv_audio chunks until tail silence), record once, return as one assistant audio message. Why drain instead of taking one chunk: TTS and realtime APIs stream their response in many small chunks. A single recv_audio() returns the first one only — the recorder would log ~100ms of agent audio per turn and the judge would receive a truncated response. Draining until tail-silence (no new chunk for ``response_tail_silence`` seconds) gives the natural "agent finished talking" signal that works across adapters without each one needing to know its transport's done event. Subclasses may override this for specialised flows but will usually inherit it. """ # Uniform pre-turn connected-state gate (mirror TS # ``adapter.runtime.ts:249-254``): a call() issued before the # executor's connect() — or after a dropped transport — fails once with # a clear error naming the adapter, rather than a transport-specific # null-deref or a silent hang. Checked ONCE, BEFORE send_audio/ # recv_audio. It does NOT suppress ``FirstChunkTimeoutError``: a # connected adapter whose first chunk never arrives still surfaces that # timeout from the drain below. # # We raise ``TransportNotConnectedError`` (a subclass of # ``PendingTransportError``, so the TS-parity ``except # PendingTransportError`` gate still catches it) whose message is # actionable for a real, implemented adapter — "call connect()/reconnect" # — rather than the base "implement your transport" guidance meant for # unshipped stubs. if not self.is_connected(): if self.agent_hung_up: # The AGENT ended the call on purpose (issue #839) — hosted # agents routinely invoke a hangup tool right after their # farewell, which closes the transport. Any scripted turn left # in the script has nobody to talk to, but the agent did # exactly what it was designed to do, so concluding here and # letting the script fall through to the judge is the correct # outcome. Failing the run would punish correct behaviour. # Returning no messages leaves the transcript ending on the # agent's farewell, which is what the judge should assess. logger.info( "%s: agent ended the call; concluding the conversation " "instead of failing the remaining scripted turn(s)", type(self).__name__, ) return [] from .adapters._stub import TransportNotConnectedError raise TransportNotConnectedError(type(self).__name__) # One ``voice.turn`` span per call(), nesting under the executor's # existing ``{cls}.call`` agent span (ambient OTel context — no parent # passed). The transport spans below (send/receive) nest under it. turn_index = getattr( getattr(input, "scenario_state", None), "current_turn", None ) with voice_span( "voice.turn", { "voice.adapter.class": type(self).__name__, "voice.turn.index": turn_index, }, ) as _turn_span, self._voice_turn_context_scope(): _turn_started = time.monotonic() # Clear the speaking-event for this turn — set in _drain on first chunk. self._agent_speaking_event.clear() recorder = _AdapterRecorder(input) incoming = ( extract_audio(input.new_messages[-1]) if input.new_messages else None ) if incoming is not None: # BEFORE the user's audio goes out (and before the user segment # is written, so an agent segment is still last on the cursor): # sweep up any agent audio an early turn close stranded in # flight, so it lands on the turn that produced it instead of # bleeding out as the next turn's opening audio (#749). await reconcile_prior_agent_audio( self, recorder._executor, recorder._offset() ) # The sweep is cleanup of the PREVIOUS turn, so re-stamp the # start: leaving it would bill this turn for time spent draining # the last one and inflate the reported turn latency. _turn_started = time.monotonic() # Wrap send_audio so user.start = "we began transmitting" and # user.end = "we finished transmitting" — both real flow points. recorder.mark_user_start() with voice_span( "voice.audio.send", {"voice.audio.bytes": len(incoming.data)} ): await self.send_audio(incoming) recorder.record_user(incoming) # Drain. Recorder grabs agent.start at first chunk via # mark_agent_start, so agent.start is "first chunk on the wire," # not "now minus merged.duration." merged = await self._drain_agent_response( on_first_chunk=recorder.mark_agent_start ) # Mark agent.end BEFORE the STT round-trip below — the agent stopped # speaking when drain settled, not after transcription returned. recorder.mark_agent_end() _turn_span.set_attribute( "voice.turn.latency_ms", round((time.monotonic() - _turn_started) * 1000), ) if incoming is not None and incoming.data: _turn_span.set_attribute( "voice.turn.user_audio_bytes", len(incoming.data) ) if merged.data: _turn_span.set_attribute( "voice.turn.agent_audio_bytes", len(merged.data) ) merged = await self._ensure_transcript(merged) recorder.record_agent(merged) return create_audio_message(merged, role="assistant") async def connect(self) ‑> None-
Open the transport and prepare to exchange audio.
Expand source code
@abstractmethod async def connect(self) -> None: """Open the transport and prepare to exchange audio.""" async def disconnect(self) ‑> None-
Close the transport and release resources.
Expand source code
@abstractmethod async def disconnect(self) -> None: """Close the transport and release resources.""" async def interrupt(self) ‑> None-
Send a first-class interrupt signal to the agent under test.
Adapters that advertise
capabilities.interruption=Trueoverride this to send the transport-native interrupt (e.g., Twilioclear, OpenAI Realtimeresponse.cancel). The agent stops generating audio immediately — much more deterministic than racing VAD against a wall-clock sleep.The default raises
UnsupportedCapabilityError. Callers (interrupt()) checkcapabilities.interruptionand fall back to timing-based barge-in (sending audio while the agent is speaking) when this returns False.Expand source code
async def interrupt(self) -> None: """Send a first-class interrupt signal to the agent under test. Adapters that advertise ``capabilities.interruption=True`` override this to send the transport-native interrupt (e.g., Twilio ``clear``, OpenAI Realtime ``response.cancel``). The agent stops generating audio immediately — much more deterministic than racing VAD against a wall-clock sleep. The default raises ``UnsupportedCapabilityError``. Callers (``scenario.interrupt()``) check ``capabilities.interruption`` and fall back to timing-based barge-in (sending audio while the agent is speaking) when this returns False. """ from .capabilities import UnsupportedCapabilityError raise UnsupportedCapabilityError( type(self).__name__, "interruption", hint=( "This adapter has no native interrupt signal. Use the " "timing-based barge-in pattern instead: " "agent(wait=False) + sleep(N) + user(content), where the " "user audio overlaps with the agent's TTS and the SUT's " "VAD detects it." ), ) def is_connected(self) ‑> bool-
Whether the transport is open and ready to exchange audio.
Base default is
True: adapters without a persistent socket (or that manage liveness elsewhere) are always considered ready, so the pre-turn guard in :meth:callnever blocks them. Transports with a real socket override this — e.g. :class:ElevenLabsAgentAdapterreturnsself._ws is not None and not self._ws.closed(parity with the TSisConnected()override,adapters/elevenlabs.ts:531-534).Expand source code
def is_connected(self) -> bool: """Whether the transport is open and ready to exchange audio. Base default is ``True``: adapters without a persistent socket (or that manage liveness elsewhere) are always considered ready, so the pre-turn guard in :meth:`call` never blocks them. Transports with a real socket override this — e.g. :class:`ElevenLabsAgentAdapter` returns ``self._ws is not None and not self._ws.closed`` (parity with the TS ``isConnected()`` override, ``adapters/elevenlabs.ts:531-534``). """ return True async def recv_audio(self, timeout: float) ‑> AudioChunk-
Receive the next AudioChunk from the agent.
Expand source code
@abstractmethod async def recv_audio(self, timeout: float) -> AudioChunk: """Receive the next AudioChunk from the agent.""" async def send_audio(self, chunk: AudioChunk) ‑> None-
Transmit an AudioChunk to the agent under test.
Expand source code
@abstractmethod async def send_audio(self, chunk: AudioChunk) -> None: """Transmit an AudioChunk to the agent under test."""