Platform-specific voice adapters (Phase 2).

Per the proposal (§7.3): per-platform classes over a unified VoiceAgent(transport=...). PipecatAgentAdapter means "test my Pipecat agent"; TwilioAgentAdapter means "test via phone call"; each has platform-specific constructor parameters that don't fit cleanly on a generic class.

"""
Platform-specific voice adapters (Phase 2).

Per the proposal (§7.3): per-platform classes over a unified
``VoiceAgent(transport=...)``. ``PipecatAgentAdapter`` means "test my Pipecat agent";
``TwilioAgentAdapter`` means "test via phone call"; each has platform-specific
constructor parameters that don't fit cleanly on a generic class.
"""

from __future__ import annotations

from ._stub import PendingTransportError
from .composable import ComposableVoiceAgent, ElevenLabsVoiceAgent
from .elevenlabs import ElevenLabsAgentAdapter
from .gemini_live import GeminiLiveAgentAdapter
from .livekit import LiveKitAgentAdapter
from .openai_realtime import OpenAIRealtimeAgentAdapter
from .pipecat import PipecatAgentAdapter, PipecatRecvError
from .twilio import TunnelNotReadyError, TunnelReadiness, TwilioAgentAdapter
from .vapi import VapiAgentAdapter
from .webrtc import WebRTCAgentAdapter
from .websocket import WebSocketAgentAdapter, WebSocketProtocol

__all__ = [
    "ComposableVoiceAgent",
    "ElevenLabsAgentAdapter",
    "ElevenLabsVoiceAgent",
    "GeminiLiveAgentAdapter",
    "LiveKitAgentAdapter",
    "OpenAIRealtimeAgentAdapter",
    "PendingTransportError",
    "PipecatAgentAdapter",
    "PipecatRecvError",
    "TunnelNotReadyError",
    "TunnelReadiness",
    "TwilioAgentAdapter",
    "VapiAgentAdapter",
    "WebRTCAgentAdapter",
    "WebSocketAgentAdapter",
    "WebSocketProtocol",
]

Sub-modules

scenario.voice.adapters.composable

Composable and provider-branded voice agents …

scenario.voice.adapters.elevenlabs

ElevenLabsAgentAdapter: connect to ElevenLabs Conversational AI via their WebSocket …

scenario.voice.adapters.gemini_live

GeminiLiveAgentAdapter: direct-to-model adapter for Gemini Live native-audio …

scenario.voice.adapters.livekit

LiveKitAgentAdapter: join a LiveKit room as a participant and exchange audio …

scenario.voice.adapters.openai_realtime

OpenAIRealtimeAgentAdapter: direct-to-model adapter — the model IS the agent …

scenario.voice.adapters.pipecat

PipecatAgentAdapter: WebSocket client to a user-run Pipecat bot …

scenario.voice.adapters.twilio

TwilioAgentAdapter: bidirectional real phone transport via Twilio Media Streams …

scenario.voice.adapters.vapi

VapiAgentAdapter: call Vapi's REST API to create a call, then connect to the returned websocketCallUrl …

scenario.voice.adapters.webrtc

Generic WebRTC adapter …

scenario.voice.adapters.websocket

Generic WebSocket adapter: bring-your-own protocol …

Classes

class ComposableVoiceAgent (stt: STTProvider, llm: str, tts: str, *, system_prompt: Optional[str] = None)

Locally-executed STT → LLM → TTS voice agent.

stt transcribes incoming user audio, the result is fed to llm (a litellm model string) along with conversation history, and the response is synthesised via the tts voice string using the existing synthesize() router.

Each seam is independently swappable — change any one without touching the other two. Intermediate results are surfaced on instance attributes so the scenario harness can assert on them.

Attributes

last_user_transcript

Transcript of the most-recent user audio turn.

last_llm_response

Text produced by the LLM for the most-recent turn.

Args

stt

STTProvider implementation for the user's audio.

llm

litellm-style model identifier, e.g. COMPOSABLE_VOICE_LLM_MODEL.

tts

TTS voice string in "provider/voice" format, e.g. "openai/nova" or "elevenlabs/rachel".

system_prompt

Optional system prompt seeded at turn zero so the LLM has guidance before the first user message. Defaults to a generic helpful-assistant prompt.

Expand source code

class ComposableVoiceAgent(VoiceAgentAdapter):
    """
    Locally-executed STT → LLM → TTS voice agent.

    ``stt`` transcribes incoming user audio, the result is fed to ``llm``
    (a litellm model string) along with conversation history, and the response
    is synthesised via the ``tts`` voice string using the existing
    ``scenario.voice.synthesize`` router.

    Each seam is independently swappable — change any one without touching the
    other two. Intermediate results are surfaced on instance attributes so the
    scenario harness can assert on them.

    Attributes:
        last_user_transcript: Transcript of the most-recent user audio turn.
        last_llm_response: Text produced by the LLM for the most-recent turn.
    """

    capabilities: ClassVar[AdapterCapabilities] = AdapterCapabilities(
        streaming_transcripts=True,
        native_vad=False,
        dtmf=False,
        input_formats=["pcm16/24000"],
        output_formats=["pcm16/24000"],
    )

    DEFAULT_SYSTEM_PROMPT = (
        "You are a helpful voice assistant. Respond naturally and conversationally "
        "as this is an audio conversation — be concise, friendly, and clear."
    )

    def __init__(
        self,
        stt: STTProvider,
        llm: str,
        tts: str,
        *,
        system_prompt: Optional[str] = None,
    ) -> None:
        """
        Args:
            stt: STTProvider implementation for the user's audio.
            llm: litellm-style model identifier, e.g. ``COMPOSABLE_VOICE_LLM_MODEL``.
            tts: TTS voice string in ``"provider/voice"`` format,
                 e.g. ``"openai/nova"`` or ``"elevenlabs/rachel"``.
            system_prompt: Optional system prompt seeded at turn zero so the
                LLM has guidance before the first user message. Defaults to a
                generic helpful-assistant prompt.
        """
        super().__init__()
        self.stt = stt
        self.llm = llm
        self.tts = tts

        self.last_user_transcript: Optional[str] = None
        self.last_llm_response: Optional[str] = None

        # Seed history with a system prompt so the first recv_audio call (which
        # can happen before any user audio when the agent speaks first) doesn't
        # send an empty messages array to the LLM.
        self._history: List[dict] = [
            {"role": "system", "content": system_prompt or self.DEFAULT_SYSTEM_PROMPT}
        ]
        # Turn-output guard. ``recv_audio`` synthesises ONE chunk per
        # user turn. The default ``call()`` drains by re-calling
        # ``recv_audio`` until tail-silence — on this adapter that would
        # kick a second LLM call, cancelled later by timeout (wasted
        # credits + latency). The guard makes subsequent ``recv_audio``
        # calls in the same turn return an empty chunk, which the drain
        # loop interprets as end-of-stream.
        #
        # Reset boundary: ``send_audio`` (new user audio → new turn).
        # Set boundary: end of ``recv_audio`` (LLM+TTS completed).
        self._turn_output_emitted: bool = False

    def __repr__(self) -> str:
        return f"ComposableVoiceAgent(llm={self.llm!r}, tts={self.tts!r})"

    # ------------------------------------------------------------------ lifecycle

    async def connect(self) -> None:
        """No-op — no external transport to open."""

    async def disconnect(self) -> None:
        """No-op — nothing to tear down."""

    # ------------------------------------------------------------------ I/O

    async def send_audio(self, chunk: AudioChunk) -> None:
        """Transcribe the chunk via STT and store for the next recv_audio call."""
        transcript = await self.stt.transcribe(chunk)
        self.last_user_transcript = transcript
        self._history.append({"role": "user", "content": transcript})
        # New user turn → next recv_audio is allowed to synthesise.
        self._turn_output_emitted = False

    async def recv_audio(self, timeout: float) -> AudioChunk:
        """
        Run the LLM on the current history, synthesise the response via TTS,
        and return the resulting AudioChunk.

        ``timeout`` is honoured for the combined LLM+TTS call via
        ``asyncio.wait_for``. Subsequent calls in the same turn (the
        default ``call()`` drains until tail-silence) return an empty
        chunk so the drain loop exits without billing a second LLM
        round-trip — see ``_turn_output_emitted`` for the guard contract.
        """
        if self._turn_output_emitted:
            return AudioChunk(data=b"")

        import asyncio

        async def _run() -> AudioChunk:
            import litellm  # type: ignore
            from litellm.types.utils import Choices, ModelResponse
            from typing import cast as _cast

            from ..tts import synthesize

            completion = await litellm.acompletion(
                model=self.llm,
                messages=self._history,
            )
            # Non-streaming acompletion returns ModelResponse with Choices;
            # cast satisfies pyright without runtime isinstance overhead.
            completion = _cast(ModelResponse, completion)
            choice = _cast(Choices, completion.choices[0])
            response_text: str = choice.message.content or ""
            self.last_llm_response = response_text
            self._history.append({"role": "assistant", "content": response_text})

            return await synthesize(response_text, self.tts)

        chunk = await asyncio.wait_for(_run(), timeout=timeout)
        self._turn_output_emitted = True
        return chunk

Ancestors

• VoiceAgentAdapter
• AgentAdapter
• abc.ABC

Subclasses

• ElevenLabsVoiceAgent

Class variables

var DEFAULT_SYSTEM_PROMPT

var capabilities : ClassVar[AdapterCapabilities]

Methods

async def connect(self) ‑> None

No-op — no external transport to open.

Expand source code

async def connect(self) -> None:
    """No-op — no external transport to open."""

async def disconnect(self) ‑> None

No-op — nothing to tear down.

Expand source code

async def disconnect(self) -> None:
    """No-op — nothing to tear down."""

async def recv_audio(self, timeout: float) ‑> AudioChunk

Run the LLM on the current history, synthesise the response via TTS, and return the resulting AudioChunk.

timeout is honoured for the combined LLM+TTS call via asyncio.wait_for. Subsequent calls in the same turn (the default call() drains until tail-silence) return an empty chunk so the drain loop exits without billing a second LLM round-trip — see _turn_output_emitted for the guard contract.

Expand source code

async def recv_audio(self, timeout: float) -> AudioChunk:
    """
    Run the LLM on the current history, synthesise the response via TTS,
    and return the resulting AudioChunk.

    ``timeout`` is honoured for the combined LLM+TTS call via
    ``asyncio.wait_for``. Subsequent calls in the same turn (the
    default ``call()`` drains until tail-silence) return an empty
    chunk so the drain loop exits without billing a second LLM
    round-trip — see ``_turn_output_emitted`` for the guard contract.
    """
    if self._turn_output_emitted:
        return AudioChunk(data=b"")

    import asyncio

    async def _run() -> AudioChunk:
        import litellm  # type: ignore
        from litellm.types.utils import Choices, ModelResponse
        from typing import cast as _cast

        from ..tts import synthesize

        completion = await litellm.acompletion(
            model=self.llm,
            messages=self._history,
        )
        # Non-streaming acompletion returns ModelResponse with Choices;
        # cast satisfies pyright without runtime isinstance overhead.
        completion = _cast(ModelResponse, completion)
        choice = _cast(Choices, completion.choices[0])
        response_text: str = choice.message.content or ""
        self.last_llm_response = response_text
        self._history.append({"role": "assistant", "content": response_text})

        return await synthesize(response_text, self.tts)

    chunk = await asyncio.wait_for(_run(), timeout=timeout)
    self._turn_output_emitted = True
    return chunk

async def send_audio(self, chunk: AudioChunk) ‑> None

Transcribe the chunk via STT and store for the next recv_audio call.

Expand source code

async def send_audio(self, chunk: AudioChunk) -> None:
    """Transcribe the chunk via STT and store for the next recv_audio call."""
    transcript = await self.stt.transcribe(chunk)
    self.last_user_transcript = transcript
    self._history.append({"role": "user", "content": transcript})
    # New user turn → next recv_audio is allowed to synthesise.
    self._turn_output_emitted = False

Inherited members

• VoiceAgentAdapter:
  • agent_hung_up
  • call
  • interrupt
  • is_connected

class ElevenLabsAgentAdapter (agent_id: str, api_key: str, *, base_url: Optional[str] = None, mint: Union[ElevenLabsMintEndpoint, bool, None] = None, system_prompt_override: Optional[str] = None, first_message_override: Optional[str] = None, dynamic_variables: Optional[dict[str, Any]] = None, overrides: Optional[dict[str, Any]] = None, turn_commit_mode: TurnCommitMode = 'audio', silence_tail_bytes: int = 16000)

ElevenLabs hosted Conversational AI adapter.

Connects to ElevenLabs' hosted endpoint where the STT→LLM→TTS loop runs on their infrastructure. All audio is PCM16 @ 24kHz mono — no conversion needed at either edge.

Not to be confused with :class:ElevenLabsVoiceAgent (in scenario.voice.adapters.composable), which is the typed composable preset that runs locally with separate STT, LLM, and TTS providers. The two complement each other:

• ElevenLabsAgentAdapter (this class): black-box hosted EL ConvAI; you provide an agent_id provisioned in the EL dashboard and EL runs the whole pipeline server-side.
• :class:ElevenLabsVoiceAgent: composes ElevenLabsSTTProvider + any LLM + ElevenLabs TTS on your side; you control the prompts, model choice, and tool calls.

Intermediate transcripts are tracked on last_user_transcript and last_agent_transcript for scenario observability.

Example::

adapter = ElevenLabsAgentAdapter(agent_id="abc123", api_key="sk-...")
async with adapter:
    # scenario.run() feeds send_audio / recv_audio ...

Expand source code

class ElevenLabsAgentAdapter(VoiceAgentAdapter):
    """
    ElevenLabs **hosted** Conversational AI adapter.

    Connects to ElevenLabs' hosted endpoint where the STT→LLM→TTS loop runs
    on their infrastructure. All audio is PCM16 @ 24kHz mono — no conversion
    needed at either edge.

    Not to be confused with :class:`ElevenLabsVoiceAgent` (in
    ``scenario.voice.adapters.composable``), which is the typed composable
    preset that runs locally with separate STT, LLM, and TTS providers. The
    two complement each other:

    - ``ElevenLabsAgentAdapter`` (this class): black-box hosted EL ConvAI;
      you provide an ``agent_id`` provisioned in the EL dashboard and EL
      runs the whole pipeline server-side.
    - :class:`ElevenLabsVoiceAgent`: composes ``ElevenLabsSTTProvider`` +
      any LLM + ElevenLabs TTS on your side; you control the prompts,
      model choice, and tool calls.

    Intermediate transcripts are tracked on ``last_user_transcript`` and
    ``last_agent_transcript`` for scenario observability.

    Example::

        adapter = ElevenLabsAgentAdapter(agent_id="abc123", api_key="sk-...")
        async with adapter:
            # scenario.run() feeds send_audio / recv_audio ...
    """

    capabilities: ClassVar[AdapterCapabilities] = AdapterCapabilities(
        streaming_transcripts=True,
        native_vad=True,
        dtmf=False,
        input_formats=["pcm16/24000"],
        output_formats=["pcm16/24000"],
    )

    def __init__(
        self,
        agent_id: str,
        api_key: str,
        *,
        base_url: Optional[str] = None,
        mint: Union[ElevenLabsMintEndpoint, bool, None] = None,
        system_prompt_override: Optional[str] = None,
        first_message_override: Optional[str] = None,
        dynamic_variables: Optional[dict[str, Any]] = None,
        overrides: Optional[dict[str, Any]] = None,
        turn_commit_mode: TurnCommitMode = "audio",
        silence_tail_bytes: int = SILENCE_TAIL_BYTES,
    ) -> None:
        super().__init__()
        self.agent_id = agent_id
        self._api_key = api_key
        # Where to mint the session's signed URL. ``False`` skips the mint and
        # dials the vendor's own websocket directly with ``xi-api-key``, which
        # is what an unconfigured adapter has always done. An explicit
        # endpoint, ``base_url=``, or ELEVENLABS_BASE_URL points the mint at a
        # LangWatch AI Gateway, which mirrors the vendor's signed-URL path:
        # the gateway checks the virtual key's budget and session cap, mints
        # the signed URL, and bills the call as one spend record. The
        # websocket that URL names still belongs to ElevenLabs, so the media
        # stream runs client to vendor and nothing about latency or the wire
        # protocol changes. ``api_key`` then carries the virtual key.
        self._mint: Optional[ElevenLabsMintEndpoint]
        if mint is False:
            self._mint = None
        elif isinstance(mint, ElevenLabsMintEndpoint):
            self._mint = mint
        else:
            self._mint = resolve_elevenlabs_mint_endpoint(
                base_url=base_url, api_key=api_key
            )
        #: The gateway's id for the current session, empty unless a gateway
        #: minted it.
        self._broker_session_id: str = ""
        # Per-session overrides applied via conversation_initiation_client_data
        # at the start of every WS connect. Used by demos that need a
        # different prompt shape (e.g. verbose for interrupt demos) without
        # mutating the shared test agent's persistent config.
        #
        # WARNING (issue #838): ``system_prompt_override`` replaces the agent's
        # ENTIRE prompt object server-side, ``tool_ids`` included. A hosted
        # agent that relies on server tools silently loses them and then stalls
        # waiting for tool responses that can never arrive. To personalise a
        # deployed agent without dropping its tools, use ``dynamic_variables``
        # (fills the deployed prompt template) plus a narrow ``overrides``
        # entry, and leave the prompt alone.
        self._system_prompt_override = system_prompt_override
        self._first_message_override = first_message_override
        #: Per-call dynamic variables, forwarded natively as the init
        #: handshake's ``dynamic_variables``. EL personalises a hosted agent per
        #: call from these. Values keep their JSON type — str / int / float /
        #: bool — with NO ``str()`` coercion, matching EL's typed support.
        #: Applied only if the agent declares them (server-side allowlist).
        #: When unset, no ``dynamic_variables`` key is sent at all (not ``{}``).
        self._dynamic_variables = dynamic_variables
        #: Per-call conversation-config overrides, DEEP-merged UNDER the narrow
        #: prompt/first-message knobs above. Use for anything those do not cover
        #: (e.g. ``{"agent": {"language": "es"}, "tts": {"stability": 0.3}}``).
        #: Applied only if the agent allowlists the key server-side.
        self._overrides = overrides
        # How a user turn is committed. Defaults to ``"audio"``: the turn's real
        # PCM goes on the wire and the pump's unbounded closing silence closes
        # the turn, so the agent's own STT and VAD run on the user's voice.
        # ``"text"`` and ``"silence"`` are opt-in. See ``TurnCommitMode``.
        if turn_commit_mode not in ("audio", "text", "silence"):
            raise ValueError(
                f"Unknown turn_commit_mode: {turn_commit_mode!r}. "
                'Expected "audio", "text" or "silence".'
            )
        if not isinstance(silence_tail_bytes, int) or silence_tail_bytes <= 0:
            raise ValueError(
                f"silence_tail_bytes must be a positive integer, got {silence_tail_bytes!r}."
            )
        self._turn_commit_mode: TurnCommitMode = turn_commit_mode
        # Zero-byte silence-tail length, consulted only on the ``"silence"``
        # path (and the ``"text"`` fallback when no transcript is available).
        self._silence_tail_bytes = silence_tail_bytes
        self._ws: Any = None

        # ---- continuous mic pump state ----
        # The pump is the SINGLE owner of idle silence on the wire: a
        # background task ticks every :data:`PUMP_INTERVAL_S` and feeds one
        # 20 ms frame — queued speech while the user is talking, the closing
        # SILENCE_FRAME while the turn is still closing, and NOTHING once the
        # agent has responded (the post-response pause, #705). Mirrors the TS
        # pump (``adapters/elevenlabs.ts:563-668``), adapted to Python's raw-WS
        # send seam (there is no ``inputCallback`` SDK seam here).
        self._pump_task: Optional[asyncio.Task[None]] = None
        self._outbound_frames: Deque[bytes] = deque()
        #: SET when agent audio begins (pauses the idle mic so EL's idle prompt
        #: never trips); CLEARED when a new user turn is enqueued.
        self.awaiting_user_turn: bool = False
        #: Server-side tools the agent started this turn and has not answered yet
        #: (``agent_tool_request`` minus ``agent_tool_response``). While one runs
        #: the agent is mid-turn even if it already spoke; it keeps
        #: :attr:`_agent_tool_turn_open` set past audio that arrives mid-tool.
        #: Cleared with :attr:`awaiting_user_turn`.
        self._agent_tools_in_flight: int = 0
        #: TRUE from the agent's first ``agent_tool_request`` of a turn until its
        #: audio resumes after the last tool answered. Tool completion is not
        #: speech: the agent still has to generate the spoken answer from the
        #: result, which can take longer than ``response_tail_silence``. While
        #: set, every frame re-arms the recv deadline and the idle budget is
        #: ``response_timeout``, so a silent tool or a slow post-tool answer is
        #: not cut off by the sub-second tail probe. Cleared with
        #: :attr:`awaiting_user_turn`.
        self._agent_tool_turn_open: bool = False

        # Transcript observability — updated on each transcript event.
        self.last_user_transcript: Optional[str] = None
        self.last_agent_transcript: Optional[str] = None

        #: How many user turns were committed as REAL audio. Counted once per
        #: non-empty :meth:`enqueue_speech`, not per 20 ms frame, so it counts
        #: turns. Lets a test assert the agent actually heard the user rather
        #: than merely replying (mirror TS ``audioCommitCount``).
        self.audio_commit_count: int = 0

    @property
    def url(self) -> str:
        return CONVAI_URL_TEMPLATE.format(agent_id=self.agent_id)

    @property
    def brokered(self) -> bool:
        """Whether the live session was minted by a gateway, not the vendor.

        Only true after ``connect()``, and only when the mint answered with
        ``X-LangWatch-Session-Id``. Configuration cannot set it, because the
        response is the one thing that cannot be told a lie about what
        answered.
        """
        return self._broker_session_id != ""

    def __repr__(self) -> str:  # redact credentials
        return f"ElevenLabsAgentAdapter(agent_id={self.agent_id!r}, api_key='***')"  # noqa: S105

    # ------------------------------------------------------------------ lifecycle

    async def connect(self) -> None:
        """Open the WebSocket to ElevenLabs' ConvAI endpoint.

        We send ``conversation_initiation_client_data`` on every connect.
        The EL docs neither require nor forbid this (the reference SDK
        sample sends it unconditionally with an empty body); empirically
        we've seen ``first_message`` skipped on bare connects and reliably
        fire when the init message is sent, even with an empty override
        block. If EL's behavior changes, this is the first thing to
        revisit.
        """
        import websockets

        self._broker_session_id = ""

        # A minted signed URL already carries the session's authentication in
        # its query string, so no key travels on the socket. Without a mint
        # endpoint the adapter dials the vendor's own websocket with
        # ``xi-api-key``, exactly as it always has.
        target_url = self.url
        headers = {"xi-api-key": self._api_key}
        if self._mint is not None:
            result = await mint_elevenlabs_signed_url(self._mint, self.agent_id)
            if result.minted:
                target_url = result.credential
                headers = {}
                self._broker_session_id = result.session_id
            else:
                # No mint route: a plain proxy, or a gateway older than this
                # feature. Dial the vendor directly, and say so, because an
                # unbilled session otherwise looks identical to a billed one.
                # A refusal never reaches here: it raises inside the mint.
                warn_direct_dial_fallback(
                    self._mint.base_url,
                    ELEVENLABS_SIGNED_URL_PATH,
                    "the api_key passed to the adapter",
                )

        self._ws = await websockets.connect(
            target_url,
            additional_headers=headers,
        )
        logger.debug("ElevenLabsAgentAdapter: connected (brokered=%s)", self.brokered)

        # Stamp EL-specific attrs onto the active ``voice.adapter.connect`` span
        # (opened by the executor connect loop). Base spans are name-owned; the
        # adapter contributes attributes, never a parallel span name.
        from opentelemetry import trace as _otel_trace
        from .._telemetry import set_span_attributes

        set_span_attributes(
            _otel_trace.get_current_span(),
            {
                "voice.elevenlabs.agent_id": self.agent_id,
                "voice.elevenlabs.brokered": self.brokered,
                "voice.elevenlabs.session_id": self._broker_session_id or None,
            },
        )

        # The NARROW prompt/first-message knobs build an `agent` override that is
        # always sent (an empty `agent` object is a no-op) so the handshake shape
        # is stable; it carries the overrides when set.
        agent_override: dict[str, Any] = {}
        if self._system_prompt_override:
            agent_override["prompt"] = {"prompt": self._system_prompt_override}
        if self._first_message_override:
            agent_override["first_message"] = self._first_message_override

        # DEEP-merge the caller's `overrides` (the base, lower precedence) UNDER
        # the narrow `{"agent": agent_override}` (the layer, higher precedence),
        # so a caller's `agent.language` and our `agent.prompt` BOTH survive.
        conversation_config_override = _deep_merge(
            self._overrides or {}, {"agent": agent_override}
        )

        init: dict[str, Any] = {
            "type": "conversation_initiation_client_data",
            "conversation_config_override": conversation_config_override,
        }
        # Omit the key entirely when unset — EL treats an empty object
        # differently from an absent one.
        if self._dynamic_variables is not None:
            init["dynamic_variables"] = self._dynamic_variables
        await self._ws.send(json.dumps(init))
        logger.debug(
            "ElevenLabsAgentAdapter: sent conversation_initiation_client_data "
            "with config_override_keys=%s dynamic_variables=%s",
            sorted(conversation_config_override.keys()) or "none",
            sorted(self._dynamic_variables.keys())
            if self._dynamic_variables
            else "none",
        )

        # Start the continuous mic pump now that the socket is open, so the
        # interval never outlives the socket (mirror ``onAudioStart`` →
        # ``startPump``, ``adapters/elevenlabs.ts:497,563``).
        self.start_pump()

    def is_connected(self) -> bool:
        """Whether the WS session is open and ready to exchange audio.

        The websockets-lib equivalent of the TS
        ``conversation?.isSessionActive()`` check
        (``adapters/elevenlabs.ts:531-534``): open iff we hold a socket that
        is not closed.

        Liveness is read from the connection ``state`` on modern
        ``websockets`` (>=13, the ``websockets.asyncio.client.ClientConnection``
        that :func:`websockets.connect` returns) — that class exposes a
        ``state`` enum, NOT a ``closed`` attribute. We fall back to ``closed``
        for the legacy ``WebSocketClientProtocol`` and for in-memory test
        doubles that model ``closed``. Note ``is_connected()`` is a
        best-effort *hint* for the pre-turn guard and the pump active-check;
        it is never the sole guard on a send — the pump's send is wrapped in a
        raced-close swallow, and ``recv_audio`` still handles ``ConnectionClosed``
        directly, because a socket can drop between this check and the I/O.
        """
        ws = self._ws
        if ws is None:
            return False
        state = getattr(ws, "state", None)
        if state is not None:
            # Modern websockets: OPEN iff state is OPEN. CONNECTING/CLOSING/
            # CLOSED all read as not-ready for a turn.
            try:
                from websockets.protocol import State

                return state is State.OPEN
            except Exception:
                # State enum unavailable — compare by name as a last resort.
                return getattr(state, "name", "") == "OPEN"
        # Legacy protocol / test double exposing `closed`.
        return not getattr(ws, "closed", False)

    async def disconnect(self) -> None:
        """Close the WebSocket if open.

        The pump is stopped and awaited FIRST (mirror TS ``disconnect`` stop-
        before-close, ``adapters/elevenlabs.ts:539``) so no frame is fed once
        teardown begins — even if disconnect() races a close/error that has
        already nulled the socket.
        """
        await self.stop_pump()
        # Stamp pump counters onto the active ``voice.adapter.disconnect`` span
        # (the pump has no per-tick span — H1). Read before the socket teardown.
        _stats = getattr(self, "_pump_stats", None)
        if _stats is not None:
            from opentelemetry import trace as _otel_trace
            from .._telemetry import set_span_attributes

            set_span_attributes(
                _otel_trace.get_current_span(),
                {
                    "voice.elevenlabs.pump.ticks_total": _stats.get("ticks_total", 0),
                    "voice.elevenlabs.pump.speech_frames_sent": _stats.get(
                        "speech_frames_sent", 0
                    ),
                    "voice.elevenlabs.pump.silence_frames_sent": _stats.get(
                        "silence_frames_sent", 0
                    ),
                    "voice.elevenlabs.pump.unexpected_errors": _stats.get(
                        "unexpected_errors", 0
                    ),
                },
            )
        if self._ws is not None:
            try:
                await self._ws.close()
            except Exception:
                # Best-effort: connection may already be half-closed or
                # in an error state when disconnect() is called. We're
                # tearing down regardless — propagating here would just
                # leak the WS reference.
                pass
            finally:
                self._ws = None
            logger.debug("ElevenLabsAgentAdapter: disconnected")

    # ------------------------------------------------------- continuous mic pump

    def start_pump(self) -> None:
        """Start the continuous mic pump. Idempotent — a double call yields
        exactly one running task (mirror ``startPump``,
        ``adapters/elevenlabs.ts:588-595``)."""
        if self._pump_task is None or self._pump_task.done():
            # Fresh per-pump-lifetime counters (surfaced on voice.adapter.disconnect;
            # the 20 ms pump gets NO per-tick span — its OTel context is frozen at
            # task creation, so a per-tick span would misparent + flood).
            self._pump_stats = {
                "ticks_total": 0,
                "speech_frames_sent": 0,
                "silence_frames_sent": 0,
                "unexpected_errors": 0,
            }
            self._pump_task = asyncio.ensure_future(self._pump_loop())

    async def stop_pump(self) -> None:
        """Stop the pump, drop any unsent frames, and reset the post-response
        pause so a reconnect starts in the closing-silence state.

        Cancels AND awaits the task so no orphan is left pending at loop close
        (mirror ``stopPump``, ``adapters/elevenlabs.ts:601-612``).
        """
        task = self._pump_task
        self._pump_task = None
        if task is not None and not task.done():
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
                # Expected: awaiting a just-cancelled task re-raises the
                # CancelledError here. Intentionally suppressed — we requested
                # the cancel as part of an orderly pump shutdown.
                pass
        self._outbound_frames.clear()
        # Reset the post-response pause so a reconnect starts in the
        # closing-silence state — the next user turn streams its silence again.
        self.awaiting_user_turn = False
        self._agent_tools_in_flight = 0
        self._agent_tool_turn_open = False

    async def _pump_loop(self) -> None:
        """Tick every :data:`PUMP_INTERVAL_S` until cancelled."""
        while True:
            await asyncio.sleep(PUMP_INTERVAL_S)
            await self._pump_tick()

    async def _pump_tick(self) -> None:
        """One pump tick — one of three outcomes, gated on the session being
        active so a frame that races a close cannot reach a dead socket
        (mirror ``pumpTick``, ``adapters/elevenlabs.ts:640-668``):

        * QUEUED SPEECH present → feed it (the user is speaking).
        * else AWAITING the next user turn (:attr:`awaiting_user_turn`) → feed
          NOTHING this tick (the #705 post-response pause, so EL's idle prompt
          never trips in the inter-turn gap).
        * else → feed one :data:`SILENCE_FRAME`: the closing silence after the
          user's speech, which EL's server VAD measures end-of-turn against.
        """
        if not self.is_connected():
            return

        _stats = getattr(self, "_pump_stats", None)
        if _stats is not None:
            _stats["ticks_total"] += 1

        if self._outbound_frames:
            frame = self._outbound_frames.popleft()
            if _stats is not None:
                _stats["speech_frames_sent"] += 1
        elif self.awaiting_user_turn:
            return
        else:
            frame = SILENCE_FRAME
            if _stats is not None:
                _stats["silence_frames_sent"] += 1

        import websockets  # for the ConnectionClosed close-race classes

        try:
            b64 = base64.b64encode(frame).decode()
            await self._ws.send(json.dumps({"user_audio_chunk": b64}))
        except (
            websockets.exceptions.ConnectionClosed,
            ConnectionError,
            OSError,
            RuntimeError,
        ):
            # The EXPECTED close race: the socket closed between the
            # active-check above and this feed. Drop the frame; the
            # disconnect/close path tears the pump down (mirror the raced-close
            # swallow, ``adapters/elevenlabs.ts:661-663``).
            logger.debug("ElevenLabsAgentAdapter: pump tick raced a close; frame dropped")
        except Exception:  # noqa: BLE001 — background task must never propagate
            # An UNEXPECTED failure (serialization/protocol bug, not a close
            # race). Do NOT silently swallow it as a close: surface it at
            # WARNING so the bug is visible, but still don't propagate out of
            # the background task (that would kill the pump loop and leave an
            # unhandled-task exception). The next tick retries.
            if _stats is not None:
                _stats["unexpected_errors"] += 1
            logger.warning(
                "ElevenLabsAgentAdapter: unexpected error feeding pump frame; "
                "dropping frame and continuing",
                exc_info=True,
            )

    def enqueue_speech(self, data: bytes) -> None:
        """Slice a user turn's PCM into fixed 20 ms frames and enqueue them for
        the pump; a non-empty turn also CLEARS :attr:`awaiting_user_turn` (a new
        user turn is starting, so the closing silence must stream again once the
        speech drains). Mirror ``enqueueSpeech``,
        ``adapters/elevenlabs.ts:703-731``.
        """
        if not data:
            # An empty chunk carries no speech: don't count it as a real-audio
            # turn, don't disturb the pause, don't enqueue a meaningless frame.
            return
        # A real user turn is starting → lift the post-response pause.
        self.awaiting_user_turn = False
        self._agent_tools_in_flight = 0
        self._agent_tool_turn_open = False
        # Count the turn ONCE per non-empty call (not once per 20 ms frame) so
        # the counter still counts turns.
        self.audio_commit_count += 1
        for offset in range(0, len(data), PUMP_FRAME_BYTES):
            slice_ = data[offset : offset + PUMP_FRAME_BYTES]
            if len(slice_) < PUMP_FRAME_BYTES:
                # Pad the final partial frame to a full 20 ms with trailing
                # zeros so the pump only ever feeds fixed-size frames.
                slice_ = slice_ + b"\x00" * (PUMP_FRAME_BYTES - len(slice_))
            self._outbound_frames.append(slice_)

    async def reconcile_pending_audio(self) -> Optional[AudioChunk]:
        """Collect agent audio still in flight at a user-turn boundary.

        The drain closes a turn on ``response_tail_silence``. EL delivers audio
        in bursts, so a mid-utterance delivery gap longer than that silence ends
        the turn while the agent is still speaking. Left alone, the remainder is
        read by the NEXT drain and surfaces as the next agent turn's opening
        audio — an answer to the previous question attributed to the current one
        (the split-utterance bleed, issue #749; fixed for TypeScript in #748).

        Called at the pre-user-``send_audio`` boundary and never while a drain is
        in flight. At that instant the agent cannot have begun its next reply —
        the user has not spoken yet — so anything still arriving is unambiguously
        leftover from the PRIOR agent turn. That is what makes attributing it
        backwards safe.

        Bounded by :data:`RECONCILE_BUDGET_S` of wall clock so a turn boundary
        never stalls: this recovers the burst already on the wire, it does not
        wait out a slow agent.

        Duck-typed convention (symmetric with ``last_agent_transcript``): the
        shared runtime feature-detects this method, so adapters without buffered
        audio are untouched.
        """
        if not self.is_connected():
            return None
        collected: list[bytes] = []
        loop = asyncio.get_running_loop()
        deadline = loop.time() + RECONCILE_BUDGET_S
        while loop.time() < deadline:
            try:
                chunk = await asyncio.wait_for(
                    self.recv_audio(timeout=RECONCILE_POLL_S),
                    timeout=max(RECONCILE_POLL_S, deadline - loop.time()),
                )
            except asyncio.TimeoutError:
                # Nothing more immediately available — the prior turn is done.
                break
            except Exception:  # noqa: BLE001 — best-effort sweep
                # The socket went away mid-reconcile, or the transport raised.
                # This is opportunistic cleanup at a turn boundary; surface it
                # but never fail the turn over it.
                logger.debug(
                    "ElevenLabsAgentAdapter: reconcile sweep ended on transport "
                    "error; keeping what was collected",
                    exc_info=True,
                )
                break
            if not chunk.data:
                break
            collected.append(chunk.data)
        if not collected:
            return None
        return AudioChunk(data=b"".join(collected))

    def _on_agent_audio_begin(self) -> None:
        """Agent turn audio has arrived → engage the post-response pause: the
        pump stops streaming idle silence until the next user turn. Idempotent —
        the first agent frame is the real transition; later frames re-assert it
        (mirror ``onAgentAudio`` setting ``awaitingUserTurn``,
        ``adapters/elevenlabs.ts:513-514``).
        """
        self.awaiting_user_turn = True
        # Audio after the last tool answered is the post-tool answer: back to
        # the tail probe.
        if self._agent_tools_in_flight == 0:
            self._agent_tool_turn_open = False

    # ------------------------------------------------------------------ I/O

    async def send_audio(self, chunk: AudioChunk) -> None:
        """Commit a user turn to EL ConvAI.

        EL ConvAI exposes NO audio-flush / end-of-turn client event (verified
        against the official EL Python + JS SDKs — the full client→server union
        is pong | client_tool_result | conversation_initiation_client_data |
        feedback | contextual_update | user_message | user_activity |
        multimodal_message, plus the bare user_audio_chunk; none commit audio).
        Server-side turn detection (ConvAI 2.0 hybrid VAD + DL turn-detector)
        does NOT reliably fire on a scripted, non-mic stream, so the legacy
        "stream audio + silence tail" path stalls on turn 2+ (issue #567).

        Three commit modes (see ``TurnCommitMode``):

        - ``"audio"`` (default): enqueue the turn's real PCM for the pump and
          stop. The pump feeds it as 20 ms ``user_audio_chunk`` frames at
          microphone cadence, then streams *unbounded* closing silence until
          the agent responds — the audio→silence transition EL's server VAD
          measures end-of-turn against. No bounded tail to tune. This is the
          only mode that puts the user's voice on the wire, so it is the only
          mode in which the agent's own STT and VAD are exercised.
        - ``"text"``: send ONLY ``{"type": "user_message", "text": …}``. The
          audio is DISCARDED — EL's STT never runs and a green run proves
          nothing about the agent's transcription or turn-taking. Opt-in only,
          for agents whose turn-taking cannot be driven by scripted audio.
        - ``"silence"``: stream the speech then a fixed ``silence_tail_bytes``
          zero-byte tail. Superseded by ``"audio"``'s unbounded closing
          silence; retained for callers pinned to a bounded tail.

        Silence-tail size rationale (``"silence"`` path): 16000 zero bytes at
        pcm_24000 = ~333ms — empirically the sweet spot for the greeting →
        first-turn exchange. Removing it entirely, or doubling to 24000, both
        reproduced the stall pattern.
        """
        if self._ws is None:
            raise RuntimeError("ElevenLabsAgentAdapter: not connected")

        transcript = chunk.transcript.strip() if chunk.transcript else ""

        # A real user turn is starting → lift the post-response pause so the
        # pump streams the closing silence again after this turn (mirror
        # ``enqueueSpeech`` clearing ``awaitingUserTurn``,
        # ``adapters/elevenlabs.ts:710``). Empty chunks carry no turn.
        if transcript or chunk.data:
            self.awaiting_user_turn = False
            self._agent_tools_in_flight = 0
            self._agent_tool_turn_open = False

        if self._turn_commit_mode == "text" and transcript:
            # Text-only commit: no user_audio_chunk is sent, so EL's STT never
            # sees this turn. See ``TurnCommitMode`` for why this is opt-in.
            await self._send_user_message(transcript)
            return

        # The pump is the SINGLE owner of WS audio writes: rather than writing
        # `chunk.data` DIRECTLY to `self._ws` (which would race the always-
        # running background pump — two concurrent writers producing
        # interleaved/oversized non-20ms `user_audio_chunk` frames), we ENQUEUE
        # the speech as fixed 960-byte pump frames. The pump drains them at the
        # same 20ms cadence as every other frame, so there is exactly one writer
        # and a consistent frame size on the wire. Returns promptly
        # (continuous-mic model) — it does not block until the queue drains.
        self.enqueue_speech(chunk.data)
        if self._turn_commit_mode != "audio":
            # "silence" (and the "text" fallback when a chunk carries no
            # transcript): append the bounded legacy tail. On the "audio" path
            # the pump's unbounded closing silence already provides the
            # end-of-turn transition, so a fixed tail would only cap it.
            self._enqueue_silence_tail()

    async def _send_user_message(self, text: str) -> None:
        """Explicit turn-commit: tell EL the user is done and force an agent
        response without relying on mic-style server VAD (issue #567). Wire
        shape matches the official SDK's ``user_message`` event.

        This is a single control frame (`user_message`), not streamed audio,
        so it does not contend with the pump's `user_audio_chunk` cadence.
        """
        await self._ws.send(json.dumps({"type": "user_message", "text": text}))

    def _enqueue_silence_tail(self) -> None:
        """Queue the legacy closing-silence tail as fixed 960-byte pump frames.

        The legacy path coaxed server VAD with a fixed ``_silence_tail_bytes``
        zero-byte blob written directly to the socket. To keep the pump the
        single WS writer, express that same total silence as
        ``ceil(_silence_tail_bytes / PUMP_FRAME_BYTES)`` all-zero 960-byte pump
        frames on the outbound queue, drained at the 20ms cadence. Rounds UP so
        the emitted silence is never less than the legacy tail.
        """
        num_frames = -(-self._silence_tail_bytes // PUMP_FRAME_BYTES)  # ceil div
        for _ in range(num_frames):
            self._outbound_frames.append(SILENCE_FRAME)

    async def recv_audio(self, timeout: float) -> AudioChunk:
        """
        Receive the next audio chunk from ElevenLabs.

        ``timeout`` bounds **inter-message silence** — the maximum gap between
        any two received frames — NOT the total duration of the call. Until
        the agent has spoken this turn, every received frame (**keep-alive
        pings included**) resets the idle deadline, so this returns when an
        ``audio`` event arrives and raises :class:`asyncio.TimeoutError` only
        after ``timeout`` seconds elapse with **no message of any kind**. Once
        the agent has spoken (and no server tool it started is still running),
        only :data:`AGENT_TURN_ACTIVITY_TYPES` reset it, so the drain's tail
        probe ends on audio silence. Pings are replied to inline;
        transcript events update instance attributes for observability; most
        other event types are swallowed without error.

        Terminal (non-audio) completions return an **empty** ``AudioChunk``
        rather than hanging to the deadline (issue #648): a ``client_tool_call``
        (tool-only turn — this adapter has no ``client_tool_result`` path) and a
        socket close mid-receive both end the drain cleanly, mirroring the
        #646/PR647 reference fix and the Gemini Live / Pipecat idiom.

        Design decision (issue #493 — intentional, not an oversight): because
        a received ping is treated as proof of liveness, a hosted agent that
        keeps pinging but never sends audio (e.g. a wedged tool/RAG call) will
        make this method wait past ``timeout``. A legitimate 30s+
        silent-but-pinging stretch must not abort the turn, which a cumulative
        budget would do — so pings keep re-arming the idle deadline below.
        But EL ConvAI can also ping *indefinitely* on a turn it will never
        answer with audio (e.g. after it ends/transfers its turn), which would
        otherwise wedge the whole multi-turn run forever. To bound that case
        without punishing a merely slow agent, :data:`KEEPALIVE_HARD_CEILING_S`
        sets an absolute wall-clock ceiling, computed ONCE per call and NOT
        reset by pings (issue #829; the backstop explicitly deferred in #493).
        The caller's ``response_max_duration`` is checked *between*
        ``recv_audio`` calls and does **not** bound a single in-progress recv.
        """
        import websockets  # for the ConnectionClosed terminal (issue #648)

        if self._ws is None:
            raise RuntimeError("ElevenLabsAgentAdapter: not connected")

        from opentelemetry import trace as _otel_trace
        from .._telemetry import set_span_attributes

        start = asyncio.get_running_loop().time()
        # The idle budget the deadline was last armed with: ``timeout``, or
        # ``response_timeout`` while an agent tool turn is open (see
        # ``_agent_tool_turn_open``).
        tool_turn_budget = max(timeout, self.response_timeout)
        idle_budget = tool_turn_budget if self._agent_tool_turn_open else timeout
        deadline = start + idle_budget
        # Absolute wall-clock ceiling that keepalive pings do NOT reset (#829 /
        # the deferred #493 backstop). EL ConvAI pings indefinitely on a turn it
        # will never answer with audio (e.g. after it ends or transfers its
        # turn); with only the ping-resettable idle ``deadline`` this loop would
        # wedge forever. The ceiling bounds that pings-but-no-audio case. At
        # least ``timeout`` so it never pre-empts the idle deadline.
        ceiling = max(timeout, KEEPALIVE_HARD_CEILING_S)
        hard_deadline = start + ceiling
        # Whether ANY inbound frame arrived. It is what tells the two rejections
        # apart when both bounds land on the same instant, which is the default
        # case: idle 60s, ceiling max(60, 45) = 60s. Nothing received means the
        # socket was silent throughout, so the idle diagnosis is the true one.
        saw_inbound_frame = False
        # Wait diagnosis: which EL message types reached this recv. Logged when
        # it ends on a deadline, so a slow turn names its cause: a frame type
        # re-arming the wait, or (nothing arrived, yet the deadline fired late) a
        # stalled event loop in the host.
        frames_by_type: dict[str, int] = {}

        def timeout_error() -> asyncio.TimeoutError:
            """The rejection for whichever bound expired. A socket that went
            completely quiet and one that pings steadily without ever speaking
            are different problems, so they get different messages."""
            is_ceiling = saw_inbound_frame and hard_deadline <= deadline
            now = asyncio.get_running_loop().time()
            late_s = max(0.0, now - min(deadline, hard_deadline))
            details = {
                "end": "ceiling" if is_ceiling else "idle",
                "timeout_s": idle_budget,
                "waited_s": round(now - start, 3),
                "late_s": round(late_s, 3),
                "after_agent_audio": self.awaiting_user_turn,
                "frames_by_type": frames_by_type,
            }
            # Same attrs as the JS adapter, onto the active ``voice.audio.receive``
            # span, so the trace shows why the wait ended without the debug log.
            set_span_attributes(
                _otel_trace.get_current_span(),
                {
                    "voice.elevenlabs.receive_wait_end": details["end"],
                    "voice.elevenlabs.receive_wait_ms": round((now - start) * 1000),
                    "voice.elevenlabs.receive_wait_late_ms": round(late_s * 1000),
                    "voice.elevenlabs.receive_wait_frames": json.dumps(frames_by_type),
                },
            )
            if is_ceiling:
                logger.warning(
                    "ElevenLabsAgentAdapter: recv_audio ended on the absolute "
                    "ceiling; frames_by_type shows what kept the wait open: %s",
                    details,
                )
                return asyncio.TimeoutError(_ceiling_timeout_message(timeout, ceiling))
            if late_s > LATE_TIMER_WARN_S:
                logger.warning(
                    "ElevenLabsAgentAdapter: recv_audio deadline fired late; the "
                    "host event loop was stalled: %s",
                    details,
                )
            else:
                logger.debug("ElevenLabsAgentAdapter: recv_audio ended on its idle deadline: %s", details)
            return asyncio.TimeoutError(_idle_timeout_message(idle_budget))

        while True:
            now = asyncio.get_running_loop().time()
            remaining = min(deadline, hard_deadline) - now
            if remaining <= 0:
                raise timeout_error()

            try:
                raw = await asyncio.wait_for(self._ws.recv(), timeout=remaining)
            except asyncio.TimeoutError as err:
                # ``wait_for`` raises a bare TimeoutError with an empty message,
                # which is how both bounds used to reach the caller indistinguishable.
                raise timeout_error() from err
            except websockets.exceptions.ConnectionClosed:
                # Issue #648: the hosted agent finished its turn and the server
                # closed the socket WITHOUT a trailing audio frame (a silent /
                # tool-only turn). Mirror the #646/PR647 reference pattern (and
                # the Gemini Live / Pipecat idiom): return an empty AudioChunk so
                # the base ``_drain_agent_response`` loop exits cleanly, instead
                # of letting ConnectionClosed propagate — the drain only catches
                # asyncio.TimeoutError, so an unhandled close would crash the turn.
                logger.debug(
                    "ElevenLabsAgentAdapter: socket closed during recv; "
                    "ending turn with empty chunk"
                )
                return AudioChunk(data=b"")
            saw_inbound_frame = True
            try:
                event = json.loads(raw) if isinstance(raw, str) else json.loads(raw.decode())
            except Exception:
                event = None
            etype = event.get("type", "") if isinstance(event, dict) else ""

            # Track running server-side tools before the re-arm decision, so a
            # frame that arrives while one runs, or before the agent speaks its
            # post-tool answer, keeps the turn open.
            if etype == "agent_tool_request":
                self._agent_tools_in_flight += 1
                self._agent_tool_turn_open = True
            elif etype == "agent_tool_response":
                self._agent_tools_in_flight = max(0, self._agent_tools_in_flight - 1)

            # A received message (ping included) proves the socket is alive, so
            # re-arm the idle deadline, even for a non-JSON/malformed frame. Once
            # the agent has spoken this turn and no tool turn is open, only turn
            # activity re-arms it (AGENT_TURN_ACTIVITY_TYPES), so the tail probe
            # measures audio silence rather than socket silence. While a tool
            # turn is open every frame re-arms it, on the response budget.
            key = etype or "unknown"
            frames_by_type[key] = frames_by_type.get(key, 0) + 1
            tool_turn = self._agent_tool_turn_open
            agent_quiet_candidate = self.awaiting_user_turn and not tool_turn
            if not agent_quiet_candidate or etype in AGENT_TURN_ACTIVITY_TYPES:
                idle_budget = tool_turn_budget if tool_turn else timeout
                deadline = asyncio.get_running_loop().time() + idle_budget

            if not isinstance(event, dict):
                logger.debug("ElevenLabsAgentAdapter: non-JSON message, skipping")
                continue

            logger.debug("ElevenLabsAgentAdapter: recv event %s", etype)

            if etype == "audio":
                audio_event = event.get("audio_event", {})
                b64 = audio_event.get("audio_base_64", "")
                pcm = base64.b64decode(b64)
                # Ensure even byte count (PCM16 invariant).
                if len(pcm) % 2 == 1:
                    pcm = pcm[:-1]
                # Agent turn audio has arrived → engage the post-response pause
                # so the pump stops streaming idle silence into the inter-turn
                # gap (mirror ``onAgentAudio`` setting ``awaitingUserTurn``).
                self._on_agent_audio_begin()
                return AudioChunk(data=pcm)

            elif etype == "ping":
                # Per EL docs, ping wire shape is
                #   {"type": "ping", "ping_event": {"event_id": <int>, "ping_ms": <int>}}
                # Pong must echo the event_id at the top level. The
                # fallback to top-level event_id covers any older shape.
                ping_event = event.get("ping_event") or {}
                event_id = ping_event.get("event_id")
                if event_id is None:
                    event_id = event.get("event_id")
                if event_id is None:
                    logger.debug("ElevenLabsAgentAdapter: ping with no event_id, skipping pong: %r", event)
                    continue
                pong = json.dumps({"type": "pong", "event_id": event_id})
                await self._ws.send(pong)

            elif etype == "user_transcript":
                self.last_user_transcript = event.get("user_transcription_event", {}).get("user_transcript")

            elif etype == "agent_response":
                self.last_agent_transcript = event.get("agent_response_event", {}).get("agent_response")

            elif etype == "agent_tool_response":
                # The agent invoked one of its own (server-side) tools. When
                # that tool is the ``end_call`` system tool, the agent has
                # deliberately hung up and EL closes the socket right after
                # this frame with a clean 1000 (issue #839).
                #
                # Wire shape (captured live):
                #   {"type": "agent_tool_response",
                #    "agent_tool_response": {"tool_name": "end_call",
                #      "tool_type": "system", "is_error": false,
                #      "is_blocked": false, "is_called": true, ...}}
                #
                # Recording it as a deliberate hangup — rather than letting the
                # ensuing close look like a dropped transport — is what lets a
                # leftover scripted turn conclude gracefully instead of failing
                # a run in which the agent behaved exactly as designed.
                tool = event.get("agent_tool_response", {}) or {}
                if (
                    tool.get("tool_name") in HANGUP_TOOL_NAMES
                    and tool.get("is_called")
                    and not tool.get("is_error")
                    and not tool.get("is_blocked")
                ):
                    logger.info(
                        "ElevenLabsAgentAdapter: agent invoked %r — treating the "
                        "upcoming close as a deliberate hangup",
                        tool.get("tool_name"),
                    )
                    self.agent_hung_up = True

            elif etype == "agent_response_correction":
                # EL signals a corrected agent reply (post server-side
                # barge-in detection). The corrected text replaces the
                # last_agent_transcript so consumers see what the agent
                # ACTUALLY said after our interrupt landed, not the
                # pre-correction draft.
                #
                # Wire shape:
                #   {"type": "agent_response_correction",
                #    "agent_response_correction_event": {
                #      "original_agent_response": "...",
                #      "corrected_agent_response": "..."}}
                correction = event.get("agent_response_correction_event", {}) or {}
                corrected = correction.get("corrected_agent_response")
                if corrected:
                    self.last_agent_transcript = corrected

            elif etype == "conversation_initiation_metadata":
                # EL reports the agent's actual configured audio formats
                # here. Our adapter capabilities advertise pcm16/24000,
                # matching the test agent we provision. If a caller
                # points the adapter at an agent configured differently,
                # this is where the mismatch becomes visible — warn so
                # the codec mismatch is logged rather than silently
                # garbling audio.
                #
                # Wire shape (per docs):
                #   {"type": "conversation_initiation_metadata",
                #    "conversation_initiation_metadata_event": {
                #      "conversation_id": "...",
                #      "agent_output_audio_format": "pcm_24000",
                #      "user_input_audio_format": "pcm_24000"}}
                meta = event.get("conversation_initiation_metadata_event", {}) or {}
                out_fmt = meta.get("agent_output_audio_format")
                in_fmt = meta.get("user_input_audio_format")
                if out_fmt and out_fmt != "pcm_24000":
                    logger.warning(
                        "ElevenLabsAgentAdapter: agent_output_audio_format=%r "
                        "differs from advertised pcm16/24000 capability; "
                        "audio may pitch-shift or fail to decode.",
                        out_fmt,
                    )
                if in_fmt and in_fmt != "pcm_24000":
                    logger.warning(
                        "ElevenLabsAgentAdapter: user_input_audio_format=%r "
                        "differs from advertised pcm16/24000 capability; "
                        "the agent may not understand audio we send.",
                        in_fmt,
                    )

            elif etype == "client_tool_call":
                # Issue #648: EL ConvAI emits ``client_tool_call`` when the agent
                # invokes a CLIENT-side tool. This adapter is a black-box test
                # harness and does NOT send ``client_tool_result`` back, so the
                # hosted agent will never produce spoken audio for this turn — it
                # is a tool-only / non-audio terminal turn. Mirror the #646/PR647
                # reference pattern: return an empty AudioChunk so the drain exits
                # cleanly instead of looping to the ``response_timeout`` deadline
                # and raising. The tool call is observable on the wire; we surface
                # the turn's completion, not its payload.
                logger.debug(
                    "ElevenLabsAgentAdapter: client_tool_call (tool-only turn); "
                    "ending turn with empty chunk"
                )
                return AudioChunk(data=b"")

            elif etype == "interruption":
                pass  # documented non-audio event, no action needed

            else:
                logger.debug("ElevenLabsAgentAdapter: unknown event type %r, skipping", etype)

Ancestors

• VoiceAgentAdapter
• AgentAdapter
• abc.ABC

Class variables

var capabilities : ClassVar[AdapterCapabilities]

Instance variables

var audio_commit_count

How many user turns were committed as REAL audio. Counted once per non-empty :meth:enqueue_speech, not per 20 ms frame, so it counts turns. Lets a test assert the agent actually heard the user rather than merely replying (mirror TS audioCommitCount).

var awaiting_user_turn

SET when agent audio begins (pauses the idle mic so EL's idle prompt never trips); CLEARED when a new user turn is enqueued.

var brokered : bool

Whether the live session was minted by a gateway, not the vendor.

Only true after connect(), and only when the mint answered with X-LangWatch-Session-Id. Configuration cannot set it, because the response is the one thing that cannot be told a lie about what answered.

Expand source code

@property
def brokered(self) -> bool:
    """Whether the live session was minted by a gateway, not the vendor.

    Only true after ``connect()``, and only when the mint answered with
    ``X-LangWatch-Session-Id``. Configuration cannot set it, because the
    response is the one thing that cannot be told a lie about what
    answered.
    """
    return self._broker_session_id != ""

var url : str

Expand source code

@property
def url(self) -> str:
    return CONVAI_URL_TEMPLATE.format(agent_id=self.agent_id)

Methods

async def connect(self) ‑> None

Open the WebSocket to ElevenLabs' ConvAI endpoint.

We send conversation_initiation_client_data on every connect. The EL docs neither require nor forbid this (the reference SDK sample sends it unconditionally with an empty body); empirically we've seen first_message skipped on bare connects and reliably fire when the init message is sent, even with an empty override block. If EL's behavior changes, this is the first thing to revisit.

Expand source code

async def connect(self) -> None:
    """Open the WebSocket to ElevenLabs' ConvAI endpoint.

    We send ``conversation_initiation_client_data`` on every connect.
    The EL docs neither require nor forbid this (the reference SDK
    sample sends it unconditionally with an empty body); empirically
    we've seen ``first_message`` skipped on bare connects and reliably
    fire when the init message is sent, even with an empty override
    block. If EL's behavior changes, this is the first thing to
    revisit.
    """
    import websockets

    self._broker_session_id = ""

    # A minted signed URL already carries the session's authentication in
    # its query string, so no key travels on the socket. Without a mint
    # endpoint the adapter dials the vendor's own websocket with
    # ``xi-api-key``, exactly as it always has.
    target_url = self.url
    headers = {"xi-api-key": self._api_key}
    if self._mint is not None:
        result = await mint_elevenlabs_signed_url(self._mint, self.agent_id)
        if result.minted:
            target_url = result.credential
            headers = {}
            self._broker_session_id = result.session_id
        else:
            # No mint route: a plain proxy, or a gateway older than this
            # feature. Dial the vendor directly, and say so, because an
            # unbilled session otherwise looks identical to a billed one.
            # A refusal never reaches here: it raises inside the mint.
            warn_direct_dial_fallback(
                self._mint.base_url,
                ELEVENLABS_SIGNED_URL_PATH,
                "the api_key passed to the adapter",
            )

    self._ws = await websockets.connect(
        target_url,
        additional_headers=headers,
    )
    logger.debug("ElevenLabsAgentAdapter: connected (brokered=%s)", self.brokered)

    # Stamp EL-specific attrs onto the active ``voice.adapter.connect`` span
    # (opened by the executor connect loop). Base spans are name-owned; the
    # adapter contributes attributes, never a parallel span name.
    from opentelemetry import trace as _otel_trace
    from .._telemetry import set_span_attributes

    set_span_attributes(
        _otel_trace.get_current_span(),
        {
            "voice.elevenlabs.agent_id": self.agent_id,
            "voice.elevenlabs.brokered": self.brokered,
            "voice.elevenlabs.session_id": self._broker_session_id or None,
        },
    )

    # The NARROW prompt/first-message knobs build an `agent` override that is
    # always sent (an empty `agent` object is a no-op) so the handshake shape
    # is stable; it carries the overrides when set.
    agent_override: dict[str, Any] = {}
    if self._system_prompt_override:
        agent_override["prompt"] = {"prompt": self._system_prompt_override}
    if self._first_message_override:
        agent_override["first_message"] = self._first_message_override

    # DEEP-merge the caller's `overrides` (the base, lower precedence) UNDER
    # the narrow `{"agent": agent_override}` (the layer, higher precedence),
    # so a caller's `agent.language` and our `agent.prompt` BOTH survive.
    conversation_config_override = _deep_merge(
        self._overrides or {}, {"agent": agent_override}
    )

    init: dict[str, Any] = {
        "type": "conversation_initiation_client_data",
        "conversation_config_override": conversation_config_override,
    }
    # Omit the key entirely when unset — EL treats an empty object
    # differently from an absent one.
    if self._dynamic_variables is not None:
        init["dynamic_variables"] = self._dynamic_variables
    await self._ws.send(json.dumps(init))
    logger.debug(
        "ElevenLabsAgentAdapter: sent conversation_initiation_client_data "
        "with config_override_keys=%s dynamic_variables=%s",
        sorted(conversation_config_override.keys()) or "none",
        sorted(self._dynamic_variables.keys())
        if self._dynamic_variables
        else "none",
    )

    # Start the continuous mic pump now that the socket is open, so the
    # interval never outlives the socket (mirror ``onAudioStart`` →
    # ``startPump``, ``adapters/elevenlabs.ts:497,563``).
    self.start_pump()

async def disconnect(self) ‑> None

Close the WebSocket if open.

The pump is stopped and awaited FIRST (mirror TS disconnect stop- before-close, adapters/elevenlabs.ts:539) so no frame is fed once teardown begins — even if disconnect() races a close/error that has already nulled the socket.

Expand source code

async def disconnect(self) -> None:
    """Close the WebSocket if open.

    The pump is stopped and awaited FIRST (mirror TS ``disconnect`` stop-
    before-close, ``adapters/elevenlabs.ts:539``) so no frame is fed once
    teardown begins — even if disconnect() races a close/error that has
    already nulled the socket.
    """
    await self.stop_pump()
    # Stamp pump counters onto the active ``voice.adapter.disconnect`` span
    # (the pump has no per-tick span — H1). Read before the socket teardown.
    _stats = getattr(self, "_pump_stats", None)
    if _stats is not None:
        from opentelemetry import trace as _otel_trace
        from .._telemetry import set_span_attributes

        set_span_attributes(
            _otel_trace.get_current_span(),
            {
                "voice.elevenlabs.pump.ticks_total": _stats.get("ticks_total", 0),
                "voice.elevenlabs.pump.speech_frames_sent": _stats.get(
                    "speech_frames_sent", 0
                ),
                "voice.elevenlabs.pump.silence_frames_sent": _stats.get(
                    "silence_frames_sent", 0
                ),
                "voice.elevenlabs.pump.unexpected_errors": _stats.get(
                    "unexpected_errors", 0
                ),
            },
        )
    if self._ws is not None:
        try:
            await self._ws.close()
        except Exception:
            # Best-effort: connection may already be half-closed or
            # in an error state when disconnect() is called. We're
            # tearing down regardless — propagating here would just
            # leak the WS reference.
            pass
        finally:
            self._ws = None
        logger.debug("ElevenLabsAgentAdapter: disconnected")

def enqueue_speech(self, data: bytes) ‑> None

Slice a user turn's PCM into fixed 20 ms frames and enqueue them for the pump; a non-empty turn also CLEARS :attr:awaiting_user_turn (a new user turn is starting, so the closing silence must stream again once the speech drains). Mirror enqueueSpeech, adapters/elevenlabs.ts:703-731.

Expand source code

def enqueue_speech(self, data: bytes) -> None:
    """Slice a user turn's PCM into fixed 20 ms frames and enqueue them for
    the pump; a non-empty turn also CLEARS :attr:`awaiting_user_turn` (a new
    user turn is starting, so the closing silence must stream again once the
    speech drains). Mirror ``enqueueSpeech``,
    ``adapters/elevenlabs.ts:703-731``.
    """
    if not data:
        # An empty chunk carries no speech: don't count it as a real-audio
        # turn, don't disturb the pause, don't enqueue a meaningless frame.
        return
    # A real user turn is starting → lift the post-response pause.
    self.awaiting_user_turn = False
    self._agent_tools_in_flight = 0
    self._agent_tool_turn_open = False
    # Count the turn ONCE per non-empty call (not once per 20 ms frame) so
    # the counter still counts turns.
    self.audio_commit_count += 1
    for offset in range(0, len(data), PUMP_FRAME_BYTES):
        slice_ = data[offset : offset + PUMP_FRAME_BYTES]
        if len(slice_) < PUMP_FRAME_BYTES:
            # Pad the final partial frame to a full 20 ms with trailing
            # zeros so the pump only ever feeds fixed-size frames.
            slice_ = slice_ + b"\x00" * (PUMP_FRAME_BYTES - len(slice_))
        self._outbound_frames.append(slice_)

def is_connected(self) ‑> bool

Whether the WS session is open and ready to exchange audio.

The websockets-lib equivalent of the TS conversation?.isSessionActive() check (adapters/elevenlabs.ts:531-534): open iff we hold a socket that is not closed.

Liveness is read from the connection state on modern websockets (>=13, the websockets.asyncio.client.ClientConnection that :func:websockets.connect returns) — that class exposes a state enum, NOT a closed attribute. We fall back to closed for the legacy WebSocketClientProtocol and for in-memory test doubles that model closed. Note is_connected() is a best-effort hint for the pre-turn guard and the pump active-check; it is never the sole guard on a send — the pump's send is wrapped in a raced-close swallow, and recv_audio still handles ConnectionClosed directly, because a socket can drop between this check and the I/O.

Expand source code

def is_connected(self) -> bool:
    """Whether the WS session is open and ready to exchange audio.

    The websockets-lib equivalent of the TS
    ``conversation?.isSessionActive()`` check
    (``adapters/elevenlabs.ts:531-534``): open iff we hold a socket that
    is not closed.

    Liveness is read from the connection ``state`` on modern
    ``websockets`` (>=13, the ``websockets.asyncio.client.ClientConnection``
    that :func:`websockets.connect` returns) — that class exposes a
    ``state`` enum, NOT a ``closed`` attribute. We fall back to ``closed``
    for the legacy ``WebSocketClientProtocol`` and for in-memory test
    doubles that model ``closed``. Note ``is_connected()`` is a
    best-effort *hint* for the pre-turn guard and the pump active-check;
    it is never the sole guard on a send — the pump's send is wrapped in a
    raced-close swallow, and ``recv_audio`` still handles ``ConnectionClosed``
    directly, because a socket can drop between this check and the I/O.
    """
    ws = self._ws
    if ws is None:
        return False
    state = getattr(ws, "state", None)
    if state is not None:
        # Modern websockets: OPEN iff state is OPEN. CONNECTING/CLOSING/
        # CLOSED all read as not-ready for a turn.
        try:
            from websockets.protocol import State

            return state is State.OPEN
        except Exception:
            # State enum unavailable — compare by name as a last resort.
            return getattr(state, "name", "") == "OPEN"
    # Legacy protocol / test double exposing `closed`.
    return not getattr(ws, "closed", False)

async def reconcile_pending_audio(self) ‑> AudioChunk | None

Collect agent audio still in flight at a user-turn boundary.

The drain closes a turn on response_tail_silence. EL delivers audio in bursts, so a mid-utterance delivery gap longer than that silence ends the turn while the agent is still speaking. Left alone, the remainder is read by the NEXT drain and surfaces as the next agent turn's opening audio — an answer to the previous question attributed to the current one (the split-utterance bleed, issue #749; fixed for TypeScript in #748).

Called at the pre-user-send_audio boundary and never while a drain is in flight. At that instant the agent cannot have begun its next reply — the user has not spoken yet — so anything still arriving is unambiguously leftover from the PRIOR agent turn. That is what makes attributing it backwards safe.

Bounded by :data:RECONCILE_BUDGET_S of wall clock so a turn boundary never stalls: this recovers the burst already on the wire, it does not wait out a slow agent.

Duck-typed convention (symmetric with last_agent_transcript): the shared runtime feature-detects this method, so adapters without buffered audio are untouched.

Expand source code

async def reconcile_pending_audio(self) -> Optional[AudioChunk]:
    """Collect agent audio still in flight at a user-turn boundary.

    The drain closes a turn on ``response_tail_silence``. EL delivers audio
    in bursts, so a mid-utterance delivery gap longer than that silence ends
    the turn while the agent is still speaking. Left alone, the remainder is
    read by the NEXT drain and surfaces as the next agent turn's opening
    audio — an answer to the previous question attributed to the current one
    (the split-utterance bleed, issue #749; fixed for TypeScript in #748).

    Called at the pre-user-``send_audio`` boundary and never while a drain is
    in flight. At that instant the agent cannot have begun its next reply —
    the user has not spoken yet — so anything still arriving is unambiguously
    leftover from the PRIOR agent turn. That is what makes attributing it
    backwards safe.

    Bounded by :data:`RECONCILE_BUDGET_S` of wall clock so a turn boundary
    never stalls: this recovers the burst already on the wire, it does not
    wait out a slow agent.

    Duck-typed convention (symmetric with ``last_agent_transcript``): the
    shared runtime feature-detects this method, so adapters without buffered
    audio are untouched.
    """
    if not self.is_connected():
        return None
    collected: list[bytes] = []
    loop = asyncio.get_running_loop()
    deadline = loop.time() + RECONCILE_BUDGET_S
    while loop.time() < deadline:
        try:
            chunk = await asyncio.wait_for(
                self.recv_audio(timeout=RECONCILE_POLL_S),
                timeout=max(RECONCILE_POLL_S, deadline - loop.time()),
            )
        except asyncio.TimeoutError:
            # Nothing more immediately available — the prior turn is done.
            break
        except Exception:  # noqa: BLE001 — best-effort sweep
            # The socket went away mid-reconcile, or the transport raised.
            # This is opportunistic cleanup at a turn boundary; surface it
            # but never fail the turn over it.
            logger.debug(
                "ElevenLabsAgentAdapter: reconcile sweep ended on transport "
                "error; keeping what was collected",
                exc_info=True,
            )
            break
        if not chunk.data:
            break
        collected.append(chunk.data)
    if not collected:
        return None
    return AudioChunk(data=b"".join(collected))

async def recv_audio(self, timeout: float) ‑> AudioChunk

Receive the next audio chunk from ElevenLabs.

timeout bounds inter-message silence — the maximum gap between any two received frames — NOT the total duration of the call. Until the agent has spoken this turn, every received frame (keep-alive pings included) resets the idle deadline, so this returns when an audio event arrives and raises :class:asyncio.TimeoutError only after timeout seconds elapse with no message of any kind. Once the agent has spoken (and no server tool it started is still running), only :data:AGENT_TURN_ACTIVITY_TYPES reset it, so the drain's tail probe ends on audio silence. Pings are replied to inline; transcript events update instance attributes for observability; most other event types are swallowed without error.

Terminal (non-audio) completions return an empty AudioChunk rather than hanging to the deadline (issue #648): a client_tool_call (tool-only turn — this adapter has no client_tool_result path) and a socket close mid-receive both end the drain cleanly, mirroring the

646/PR647 reference fix and the Gemini Live / Pipecat idiom.

Design decision (issue #493 — intentional, not an oversight): because a received ping is treated as proof of liveness, a hosted agent that keeps pinging but never sends audio (e.g. a wedged tool/RAG call) will make this method wait past timeout. A legitimate 30s+ silent-but-pinging stretch must not abort the turn, which a cumulative budget would do — so pings keep re-arming the idle deadline below. But EL ConvAI can also ping indefinitely on a turn it will never answer with audio (e.g. after it ends/transfers its turn), which would otherwise wedge the whole multi-turn run forever. To bound that case without punishing a merely slow agent, :data:KEEPALIVE_HARD_CEILING_S sets an absolute wall-clock ceiling, computed ONCE per call and NOT reset by pings (issue #829; the backstop explicitly deferred in #493). The caller's response_max_duration is checked between recv_audio calls and does not bound a single in-progress recv.

Expand source code

async def recv_audio(self, timeout: float) -> AudioChunk:
    """
    Receive the next audio chunk from ElevenLabs.

    ``timeout`` bounds **inter-message silence** — the maximum gap between
    any two received frames — NOT the total duration of the call. Until
    the agent has spoken this turn, every received frame (**keep-alive
    pings included**) resets the idle deadline, so this returns when an
    ``audio`` event arrives and raises :class:`asyncio.TimeoutError` only
    after ``timeout`` seconds elapse with **no message of any kind**. Once
    the agent has spoken (and no server tool it started is still running),
    only :data:`AGENT_TURN_ACTIVITY_TYPES` reset it, so the drain's tail
    probe ends on audio silence. Pings are replied to inline;
    transcript events update instance attributes for observability; most
    other event types are swallowed without error.

    Terminal (non-audio) completions return an **empty** ``AudioChunk``
    rather than hanging to the deadline (issue #648): a ``client_tool_call``
    (tool-only turn — this adapter has no ``client_tool_result`` path) and a
    socket close mid-receive both end the drain cleanly, mirroring the
    #646/PR647 reference fix and the Gemini Live / Pipecat idiom.

    Design decision (issue #493 — intentional, not an oversight): because
    a received ping is treated as proof of liveness, a hosted agent that
    keeps pinging but never sends audio (e.g. a wedged tool/RAG call) will
    make this method wait past ``timeout``. A legitimate 30s+
    silent-but-pinging stretch must not abort the turn, which a cumulative
    budget would do — so pings keep re-arming the idle deadline below.
    But EL ConvAI can also ping *indefinitely* on a turn it will never
    answer with audio (e.g. after it ends/transfers its turn), which would
    otherwise wedge the whole multi-turn run forever. To bound that case
    without punishing a merely slow agent, :data:`KEEPALIVE_HARD_CEILING_S`
    sets an absolute wall-clock ceiling, computed ONCE per call and NOT
    reset by pings (issue #829; the backstop explicitly deferred in #493).
    The caller's ``response_max_duration`` is checked *between*
    ``recv_audio`` calls and does **not** bound a single in-progress recv.
    """
    import websockets  # for the ConnectionClosed terminal (issue #648)

    if self._ws is None:
        raise RuntimeError("ElevenLabsAgentAdapter: not connected")

    from opentelemetry import trace as _otel_trace
    from .._telemetry import set_span_attributes

    start = asyncio.get_running_loop().time()
    # The idle budget the deadline was last armed with: ``timeout``, or
    # ``response_timeout`` while an agent tool turn is open (see
    # ``_agent_tool_turn_open``).
    tool_turn_budget = max(timeout, self.response_timeout)
    idle_budget = tool_turn_budget if self._agent_tool_turn_open else timeout
    deadline = start + idle_budget
    # Absolute wall-clock ceiling that keepalive pings do NOT reset (#829 /
    # the deferred #493 backstop). EL ConvAI pings indefinitely on a turn it
    # will never answer with audio (e.g. after it ends or transfers its
    # turn); with only the ping-resettable idle ``deadline`` this loop would
    # wedge forever. The ceiling bounds that pings-but-no-audio case. At
    # least ``timeout`` so it never pre-empts the idle deadline.
    ceiling = max(timeout, KEEPALIVE_HARD_CEILING_S)
    hard_deadline = start + ceiling
    # Whether ANY inbound frame arrived. It is what tells the two rejections
    # apart when both bounds land on the same instant, which is the default
    # case: idle 60s, ceiling max(60, 45) = 60s. Nothing received means the
    # socket was silent throughout, so the idle diagnosis is the true one.
    saw_inbound_frame = False
    # Wait diagnosis: which EL message types reached this recv. Logged when
    # it ends on a deadline, so a slow turn names its cause: a frame type
    # re-arming the wait, or (nothing arrived, yet the deadline fired late) a
    # stalled event loop in the host.
    frames_by_type: dict[str, int] = {}

    def timeout_error() -> asyncio.TimeoutError:
        """The rejection for whichever bound expired. A socket that went
        completely quiet and one that pings steadily without ever speaking
        are different problems, so they get different messages."""
        is_ceiling = saw_inbound_frame and hard_deadline <= deadline
        now = asyncio.get_running_loop().time()
        late_s = max(0.0, now - min(deadline, hard_deadline))
        details = {
            "end": "ceiling" if is_ceiling else "idle",
            "timeout_s": idle_budget,
            "waited_s": round(now - start, 3),
            "late_s": round(late_s, 3),
            "after_agent_audio": self.awaiting_user_turn,
            "frames_by_type": frames_by_type,
        }
        # Same attrs as the JS adapter, onto the active ``voice.audio.receive``
        # span, so the trace shows why the wait ended without the debug log.
        set_span_attributes(
            _otel_trace.get_current_span(),
            {
                "voice.elevenlabs.receive_wait_end": details["end"],
                "voice.elevenlabs.receive_wait_ms": round((now - start) * 1000),
                "voice.elevenlabs.receive_wait_late_ms": round(late_s * 1000),
                "voice.elevenlabs.receive_wait_frames": json.dumps(frames_by_type),
            },
        )
        if is_ceiling:
            logger.warning(
                "ElevenLabsAgentAdapter: recv_audio ended on the absolute "
                "ceiling; frames_by_type shows what kept the wait open: %s",
                details,
            )
            return asyncio.TimeoutError(_ceiling_timeout_message(timeout, ceiling))
        if late_s > LATE_TIMER_WARN_S:
            logger.warning(
                "ElevenLabsAgentAdapter: recv_audio deadline fired late; the "
                "host event loop was stalled: %s",
                details,
            )
        else:
            logger.debug("ElevenLabsAgentAdapter: recv_audio ended on its idle deadline: %s", details)
        return asyncio.TimeoutError(_idle_timeout_message(idle_budget))

    while True:
        now = asyncio.get_running_loop().time()
        remaining = min(deadline, hard_deadline) - now
        if remaining <= 0:
            raise timeout_error()

        try:
            raw = await asyncio.wait_for(self._ws.recv(), timeout=remaining)
        except asyncio.TimeoutError as err:
            # ``wait_for`` raises a bare TimeoutError with an empty message,
            # which is how both bounds used to reach the caller indistinguishable.
            raise timeout_error() from err
        except websockets.exceptions.ConnectionClosed:
            # Issue #648: the hosted agent finished its turn and the server
            # closed the socket WITHOUT a trailing audio frame (a silent /
            # tool-only turn). Mirror the #646/PR647 reference pattern (and
            # the Gemini Live / Pipecat idiom): return an empty AudioChunk so
            # the base ``_drain_agent_response`` loop exits cleanly, instead
            # of letting ConnectionClosed propagate — the drain only catches
            # asyncio.TimeoutError, so an unhandled close would crash the turn.
            logger.debug(
                "ElevenLabsAgentAdapter: socket closed during recv; "
                "ending turn with empty chunk"
            )
            return AudioChunk(data=b"")
        saw_inbound_frame = True
        try:
            event = json.loads(raw) if isinstance(raw, str) else json.loads(raw.decode())
        except Exception:
            event = None
        etype = event.get("type", "") if isinstance(event, dict) else ""

        # Track running server-side tools before the re-arm decision, so a
        # frame that arrives while one runs, or before the agent speaks its
        # post-tool answer, keeps the turn open.
        if etype == "agent_tool_request":
            self._agent_tools_in_flight += 1
            self._agent_tool_turn_open = True
        elif etype == "agent_tool_response":
            self._agent_tools_in_flight = max(0, self._agent_tools_in_flight - 1)

        # A received message (ping included) proves the socket is alive, so
        # re-arm the idle deadline, even for a non-JSON/malformed frame. Once
        # the agent has spoken this turn and no tool turn is open, only turn
        # activity re-arms it (AGENT_TURN_ACTIVITY_TYPES), so the tail probe
        # measures audio silence rather than socket silence. While a tool
        # turn is open every frame re-arms it, on the response budget.
        key = etype or "unknown"
        frames_by_type[key] = frames_by_type.get(key, 0) + 1
        tool_turn = self._agent_tool_turn_open
        agent_quiet_candidate = self.awaiting_user_turn and not tool_turn
        if not agent_quiet_candidate or etype in AGENT_TURN_ACTIVITY_TYPES:
            idle_budget = tool_turn_budget if tool_turn else timeout
            deadline = asyncio.get_running_loop().time() + idle_budget

        if not isinstance(event, dict):
            logger.debug("ElevenLabsAgentAdapter: non-JSON message, skipping")
            continue

        logger.debug("ElevenLabsAgentAdapter: recv event %s", etype)

        if etype == "audio":
            audio_event = event.get("audio_event", {})
            b64 = audio_event.get("audio_base_64", "")
            pcm = base64.b64decode(b64)
            # Ensure even byte count (PCM16 invariant).
            if len(pcm) % 2 == 1:
                pcm = pcm[:-1]
            # Agent turn audio has arrived → engage the post-response pause
            # so the pump stops streaming idle silence into the inter-turn
            # gap (mirror ``onAgentAudio`` setting ``awaitingUserTurn``).
            self._on_agent_audio_begin()
            return AudioChunk(data=pcm)

        elif etype == "ping":
            # Per EL docs, ping wire shape is
            #   {"type": "ping", "ping_event": {"event_id": <int>, "ping_ms": <int>}}
            # Pong must echo the event_id at the top level. The
            # fallback to top-level event_id covers any older shape.
            ping_event = event.get("ping_event") or {}
            event_id = ping_event.get("event_id")
            if event_id is None:
                event_id = event.get("event_id")
            if event_id is None:
                logger.debug("ElevenLabsAgentAdapter: ping with no event_id, skipping pong: %r", event)
                continue
            pong = json.dumps({"type": "pong", "event_id": event_id})
            await self._ws.send(pong)

        elif etype == "user_transcript":
            self.last_user_transcript = event.get("user_transcription_event", {}).get("user_transcript")

        elif etype == "agent_response":
            self.last_agent_transcript = event.get("agent_response_event", {}).get("agent_response")

        elif etype == "agent_tool_response":
            # The agent invoked one of its own (server-side) tools. When
            # that tool is the ``end_call`` system tool, the agent has
            # deliberately hung up and EL closes the socket right after
            # this frame with a clean 1000 (issue #839).
            #
            # Wire shape (captured live):
            #   {"type": "agent_tool_response",
            #    "agent_tool_response": {"tool_name": "end_call",
            #      "tool_type": "system", "is_error": false,
            #      "is_blocked": false, "is_called": true, ...}}
            #
            # Recording it as a deliberate hangup — rather than letting the
            # ensuing close look like a dropped transport — is what lets a
            # leftover scripted turn conclude gracefully instead of failing
            # a run in which the agent behaved exactly as designed.
            tool = event.get("agent_tool_response", {}) or {}
            if (
                tool.get("tool_name") in HANGUP_TOOL_NAMES
                and tool.get("is_called")
                and not tool.get("is_error")
                and not tool.get("is_blocked")
            ):
                logger.info(
                    "ElevenLabsAgentAdapter: agent invoked %r — treating the "
                    "upcoming close as a deliberate hangup",
                    tool.get("tool_name"),
                )
                self.agent_hung_up = True

        elif etype == "agent_response_correction":
            # EL signals a corrected agent reply (post server-side
            # barge-in detection). The corrected text replaces the
            # last_agent_transcript so consumers see what the agent
            # ACTUALLY said after our interrupt landed, not the
            # pre-correction draft.
            #
            # Wire shape:
            #   {"type": "agent_response_correction",
            #    "agent_response_correction_event": {
            #      "original_agent_response": "...",
            #      "corrected_agent_response": "..."}}
            correction = event.get("agent_response_correction_event", {}) or {}
            corrected = correction.get("corrected_agent_response")
            if corrected:
                self.last_agent_transcript = corrected

        elif etype == "conversation_initiation_metadata":
            # EL reports the agent's actual configured audio formats
            # here. Our adapter capabilities advertise pcm16/24000,
            # matching the test agent we provision. If a caller
            # points the adapter at an agent configured differently,
            # this is where the mismatch becomes visible — warn so
            # the codec mismatch is logged rather than silently
            # garbling audio.
            #
            # Wire shape (per docs):
            #   {"type": "conversation_initiation_metadata",
            #    "conversation_initiation_metadata_event": {
            #      "conversation_id": "...",
            #      "agent_output_audio_format": "pcm_24000",
            #      "user_input_audio_format": "pcm_24000"}}
            meta = event.get("conversation_initiation_metadata_event", {}) or {}
            out_fmt = meta.get("agent_output_audio_format")
            in_fmt = meta.get("user_input_audio_format")
            if out_fmt and out_fmt != "pcm_24000":
                logger.warning(
                    "ElevenLabsAgentAdapter: agent_output_audio_format=%r "
                    "differs from advertised pcm16/24000 capability; "
                    "audio may pitch-shift or fail to decode.",
                    out_fmt,
                )
            if in_fmt and in_fmt != "pcm_24000":
                logger.warning(
                    "ElevenLabsAgentAdapter: user_input_audio_format=%r "
                    "differs from advertised pcm16/24000 capability; "
                    "the agent may not understand audio we send.",
                    in_fmt,
                )

        elif etype == "client_tool_call":
            # Issue #648: EL ConvAI emits ``client_tool_call`` when the agent
            # invokes a CLIENT-side tool. This adapter is a black-box test
            # harness and does NOT send ``client_tool_result`` back, so the
            # hosted agent will never produce spoken audio for this turn — it
            # is a tool-only / non-audio terminal turn. Mirror the #646/PR647
            # reference pattern: return an empty AudioChunk so the drain exits
            # cleanly instead of looping to the ``response_timeout`` deadline
            # and raising. The tool call is observable on the wire; we surface
            # the turn's completion, not its payload.
            logger.debug(
                "ElevenLabsAgentAdapter: client_tool_call (tool-only turn); "
                "ending turn with empty chunk"
            )
            return AudioChunk(data=b"")

        elif etype == "interruption":
            pass  # documented non-audio event, no action needed

        else:
            logger.debug("ElevenLabsAgentAdapter: unknown event type %r, skipping", etype)

async def send_audio(self, chunk: AudioChunk) ‑> None

Commit a user turn to EL ConvAI.

EL ConvAI exposes NO audio-flush / end-of-turn client event (verified against the official EL Python + JS SDKs — the full client→server union is pong | client_tool_result | conversation_initiation_client_data | feedback | contextual_update | user_message | user_activity | multimodal_message, plus the bare user_audio_chunk; none commit audio). Server-side turn detection (ConvAI 2.0 hybrid VAD + DL turn-detector) does NOT reliably fire on a scripted, non-mic stream, so the legacy "stream audio + silence tail" path stalls on turn 2+ (issue #567).

Three commit modes (see TurnCommitMode):

• "audio" (default): enqueue the turn's real PCM for the pump and stop. The pump feeds it as 20 ms user_audio_chunk frames at microphone cadence, then streams unbounded closing silence until the agent responds — the audio→silence transition EL's server VAD measures end-of-turn against. No bounded tail to tune. This is the only mode that puts the user's voice on the wire, so it is the only mode in which the agent's own STT and VAD are exercised.
• "text": send ONLY {"type": "user_message", "text": …}. The audio is DISCARDED — EL's STT never runs and a green run proves nothing about the agent's transcription or turn-taking. Opt-in only, for agents whose turn-taking cannot be driven by scripted audio.
• "silence": stream the speech then a fixed silence_tail_bytes zero-byte tail. Superseded by "audio"'s unbounded closing silence; retained for callers pinned to a bounded tail.

Silence-tail size rationale ("silence" path): 16000 zero bytes at pcm_24000 = ~333ms — empirically the sweet spot for the greeting → first-turn exchange. Removing it entirely, or doubling to 24000, both reproduced the stall pattern.

Expand source code

async def send_audio(self, chunk: AudioChunk) -> None:
    """Commit a user turn to EL ConvAI.

    EL ConvAI exposes NO audio-flush / end-of-turn client event (verified
    against the official EL Python + JS SDKs — the full client→server union
    is pong | client_tool_result | conversation_initiation_client_data |
    feedback | contextual_update | user_message | user_activity |
    multimodal_message, plus the bare user_audio_chunk; none commit audio).
    Server-side turn detection (ConvAI 2.0 hybrid VAD + DL turn-detector)
    does NOT reliably fire on a scripted, non-mic stream, so the legacy
    "stream audio + silence tail" path stalls on turn 2+ (issue #567).

    Three commit modes (see ``TurnCommitMode``):

    - ``"audio"`` (default): enqueue the turn's real PCM for the pump and
      stop. The pump feeds it as 20 ms ``user_audio_chunk`` frames at
      microphone cadence, then streams *unbounded* closing silence until
      the agent responds — the audio→silence transition EL's server VAD
      measures end-of-turn against. No bounded tail to tune. This is the
      only mode that puts the user's voice on the wire, so it is the only
      mode in which the agent's own STT and VAD are exercised.
    - ``"text"``: send ONLY ``{"type": "user_message", "text": …}``. The
      audio is DISCARDED — EL's STT never runs and a green run proves
      nothing about the agent's transcription or turn-taking. Opt-in only,
      for agents whose turn-taking cannot be driven by scripted audio.
    - ``"silence"``: stream the speech then a fixed ``silence_tail_bytes``
      zero-byte tail. Superseded by ``"audio"``'s unbounded closing
      silence; retained for callers pinned to a bounded tail.

    Silence-tail size rationale (``"silence"`` path): 16000 zero bytes at
    pcm_24000 = ~333ms — empirically the sweet spot for the greeting →
    first-turn exchange. Removing it entirely, or doubling to 24000, both
    reproduced the stall pattern.
    """
    if self._ws is None:
        raise RuntimeError("ElevenLabsAgentAdapter: not connected")

    transcript = chunk.transcript.strip() if chunk.transcript else ""

    # A real user turn is starting → lift the post-response pause so the
    # pump streams the closing silence again after this turn (mirror
    # ``enqueueSpeech`` clearing ``awaitingUserTurn``,
    # ``adapters/elevenlabs.ts:710``). Empty chunks carry no turn.
    if transcript or chunk.data:
        self.awaiting_user_turn = False
        self._agent_tools_in_flight = 0
        self._agent_tool_turn_open = False

    if self._turn_commit_mode == "text" and transcript:
        # Text-only commit: no user_audio_chunk is sent, so EL's STT never
        # sees this turn. See ``TurnCommitMode`` for why this is opt-in.
        await self._send_user_message(transcript)
        return

    # The pump is the SINGLE owner of WS audio writes: rather than writing
    # `chunk.data` DIRECTLY to `self._ws` (which would race the always-
    # running background pump — two concurrent writers producing
    # interleaved/oversized non-20ms `user_audio_chunk` frames), we ENQUEUE
    # the speech as fixed 960-byte pump frames. The pump drains them at the
    # same 20ms cadence as every other frame, so there is exactly one writer
    # and a consistent frame size on the wire. Returns promptly
    # (continuous-mic model) — it does not block until the queue drains.
    self.enqueue_speech(chunk.data)
    if self._turn_commit_mode != "audio":
        # "silence" (and the "text" fallback when a chunk carries no
        # transcript): append the bounded legacy tail. On the "audio" path
        # the pump's unbounded closing silence already provides the
        # end-of-turn transition, so a fixed tail would only cap it.
        self._enqueue_silence_tail()

def start_pump(self) ‑> None

Start the continuous mic pump. Idempotent — a double call yields exactly one running task (mirror startPump, adapters/elevenlabs.ts:588-595).

Expand source code

def start_pump(self) -> None:
    """Start the continuous mic pump. Idempotent — a double call yields
    exactly one running task (mirror ``startPump``,
    ``adapters/elevenlabs.ts:588-595``)."""
    if self._pump_task is None or self._pump_task.done():
        # Fresh per-pump-lifetime counters (surfaced on voice.adapter.disconnect;
        # the 20 ms pump gets NO per-tick span — its OTel context is frozen at
        # task creation, so a per-tick span would misparent + flood).
        self._pump_stats = {
            "ticks_total": 0,
            "speech_frames_sent": 0,
            "silence_frames_sent": 0,
            "unexpected_errors": 0,
        }
        self._pump_task = asyncio.ensure_future(self._pump_loop())

async def stop_pump(self) ‑> None

Stop the pump, drop any unsent frames, and reset the post-response pause so a reconnect starts in the closing-silence state.

Cancels AND awaits the task so no orphan is left pending at loop close (mirror stopPump, adapters/elevenlabs.ts:601-612).

Expand source code

async def stop_pump(self) -> None:
    """Stop the pump, drop any unsent frames, and reset the post-response
    pause so a reconnect starts in the closing-silence state.

    Cancels AND awaits the task so no orphan is left pending at loop close
    (mirror ``stopPump``, ``adapters/elevenlabs.ts:601-612``).
    """
    task = self._pump_task
    self._pump_task = None
    if task is not None and not task.done():
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            # Expected: awaiting a just-cancelled task re-raises the
            # CancelledError here. Intentionally suppressed — we requested
            # the cancel as part of an orderly pump shutdown.
            pass
    self._outbound_frames.clear()
    # Reset the post-response pause so a reconnect starts in the
    # closing-silence state — the next user turn streams its silence again.
    self.awaiting_user_turn = False
    self._agent_tools_in_flight = 0
    self._agent_tool_turn_open = False

Inherited members

• VoiceAgentAdapter:
  • agent_hung_up
  • call
  • interrupt

class ElevenLabsVoiceAgent (api_key: str, *, llm: str = 'openai/gpt-5.4-mini', voice: Optional[str] = None, stt: Optional[STTProvider] = None, system_prompt: Optional[str] = None)

Composable voice agent with ElevenLabs-opinionated defaults.

Not to be confused with :class:ElevenLabsAgentAdapter (in scenario.voice.adapters.elevenlabs) — that one talks to ElevenLabs' hosted Conversational AI endpoint where EL runs the full STT→LLM→TTS loop. This class is local: you compose ElevenLabsSTTProvider + any LLM + ElevenLabs TTS yourself, keeping full control over prompts, model choice, and tool calls.

Instantiate with just an api_key to get an ElevenLabs STT + LLM (default COMPOSABLE_VOICE_LLM_MODEL) + elevenlabs/rachel TTS stack. Each piece can be overridden independently without changing the others.

Example::

# Defaults — all ElevenLabs STT, GPT-4o-mini, ElevenLabs TTS
agent = ElevenLabsVoiceAgent(api_key="sk-...")

# Override just the LLM
agent = ElevenLabsVoiceAgent(api_key="sk-...", llm="openai/gpt-4o")

# Bring your own STT
agent = ElevenLabsVoiceAgent(api_key="sk-...", stt=MyCustomSTT())

Args

api_key

ElevenLabs API key. Redacted in __repr__.

llm

litellm-style model identifier. Defaults to COMPOSABLE_VOICE_LLM_MODEL.

voice

TTS voice string in "elevenlabs/<voice_id>" format. Defaults to the ELEVENLABS_VOICE_ID environment variable when set, otherwise falls back to "Sarah" ("elevenlabs/EXAVITQu4vr4xnSDxMaL") — premade and accessible on the ElevenLabs free tier as of 2026-05. Other premade voices (e.g. "Rachel" 21m00Tcm4TlvDq8ikWAM) returned 402 paid_plan_required from the EL TTS API; gating differs per voice. Set ELEVENLABS_VOICE_ID to override.

stt

STTProvider override. Defaults to ElevenLabsSTTProvider(api_key=api_key).

system_prompt

Optional system prompt. Defaults to ComposableVoiceAgent.DEFAULT_SYSTEM_PROMPT.

Expand source code

class ElevenLabsVoiceAgent(ComposableVoiceAgent):
    """
    Composable voice agent with ElevenLabs-opinionated defaults.

    Not to be confused with :class:`ElevenLabsAgentAdapter` (in
    ``scenario.voice.adapters.elevenlabs``) — that one talks to ElevenLabs'
    **hosted** Conversational AI endpoint where EL runs the full
    STT→LLM→TTS loop. This class is local: you compose ``ElevenLabsSTTProvider``
    + any LLM + ElevenLabs TTS yourself, keeping full control over prompts,
    model choice, and tool calls.

    Instantiate with just an ``api_key`` to get an ElevenLabs STT +
    LLM (default ``COMPOSABLE_VOICE_LLM_MODEL``) + ``elevenlabs/rachel`` TTS stack. Each piece
    can be overridden independently without changing the others.

    Example::

        # Defaults — all ElevenLabs STT, GPT-4o-mini, ElevenLabs TTS
        agent = ElevenLabsVoiceAgent(api_key="sk-...")

        # Override just the LLM
        agent = ElevenLabsVoiceAgent(api_key="sk-...", llm="openai/gpt-4o")

        # Bring your own STT
        agent = ElevenLabsVoiceAgent(api_key="sk-...", stt=MyCustomSTT())
    """

    def __init__(
        self,
        api_key: str,
        *,
        llm: str = COMPOSABLE_VOICE_LLM_MODEL,
        voice: Optional[str] = None,
        stt: Optional[STTProvider] = None,
        system_prompt: Optional[str] = None,
    ) -> None:
        """
        Args:
            api_key: ElevenLabs API key. Redacted in ``__repr__``.
            llm: litellm-style model identifier. Defaults to
                ``COMPOSABLE_VOICE_LLM_MODEL``.
            voice: TTS voice string in ``"elevenlabs/<voice_id>"`` format.
                Defaults to the ``ELEVENLABS_VOICE_ID`` environment variable
                when set, otherwise falls back to "Sarah"
                (``"elevenlabs/EXAVITQu4vr4xnSDxMaL"``) — premade and
                accessible on the ElevenLabs free tier as of 2026-05.
                Other premade voices (e.g. "Rachel"
                ``21m00Tcm4TlvDq8ikWAM``) returned 402 paid_plan_required
                from the EL TTS API; gating differs per voice.  Set
                ``ELEVENLABS_VOICE_ID`` to override.
            stt: STTProvider override. Defaults to
                ``ElevenLabsSTTProvider(api_key=api_key)``.
            system_prompt: Optional system prompt. Defaults to
                ``ComposableVoiceAgent.DEFAULT_SYSTEM_PROMPT``.
        """
        import os

        if voice is None:
            env_voice_id = os.environ.get("ELEVENLABS_VOICE_ID")
            voice = (
                f"elevenlabs/{env_voice_id}"
                if env_voice_id
                else "elevenlabs/EXAVITQu4vr4xnSDxMaL"  # "Sarah" — free-tier premade
            )
        resolved_stt = stt if stt is not None else ElevenLabsSTTProvider(api_key=api_key)
        super().__init__(stt=resolved_stt, llm=llm, tts=voice, system_prompt=system_prompt)
        self._api_key = api_key
        self.voice = voice

    def __repr__(self) -> str:  # redact credentials
        return (
            f"ElevenLabsVoiceAgent("
            f"api_key='***', llm={self.llm!r}, voice={self.voice!r})"
        )

Ancestors

• ComposableVoiceAgent
• VoiceAgentAdapter
• AgentAdapter
• abc.ABC

Inherited members

• ComposableVoiceAgent:
  • agent_hung_up
  • call
  • connect
  • disconnect
  • interrupt
  • is_connected
  • recv_audio
  • send_audio

class GeminiLiveAgentAdapter (model: str = 'gemini-2.5-flash-native-audio-latest', voice: str = 'Algieba', system_instruction: str = '', api_key: Optional[str] = None)

Gemini Live native-audio adapter.

Connects directly to the Gemini Live API via the official google-genai SDK. STT, LLM, and TTS all run on Google's infrastructure; audio flows bidirectionally as raw PCM16.

Example::

adapter = GeminiLiveAgentAdapter(
    model=GEMINI_LIVE_MODEL,
    system_instruction="You are a helpful assistant.",
)
async with adapter:
    # scenario.run() feeds send_audio / recv_audio ...

Attributes

last_agent_transcript

Most-recent output transcript received from the server (if transcription is available), for observability.

Expand source code

class GeminiLiveAgentAdapter(VoiceAgentAdapter):
    """
    Gemini Live native-audio adapter.

    Connects directly to the Gemini Live API via the official ``google-genai``
    SDK.  STT, LLM, and TTS all run on Google's infrastructure; audio flows
    bidirectionally as raw PCM16.

    Example::

        adapter = GeminiLiveAgentAdapter(
            model=GEMINI_LIVE_MODEL,
            system_instruction="You are a helpful assistant.",
        )
        async with adapter:
            # scenario.run() feeds send_audio / recv_audio ...

    Attributes:
        last_agent_transcript: Most-recent output transcript received from
            the server (if transcription is available), for observability.
    """

    capabilities: ClassVar[AdapterCapabilities] = AdapterCapabilities(
        streaming_transcripts=True,
        native_vad=True,
        dtmf=False,
        # ``interruption=True``: with explicit Activity markers and
        # ``activity_handling=START_OF_ACTIVITY_INTERRUPTS`` (see
        # ``connect``), the next ``activity_start`` we send while the
        # model is replying causes Gemini to cut its in-flight audio.
        # ``interrupt()`` itself just drains stale chunks out of the
        # local queue so the recovery agent turn doesn't replay them.
        interruption=True,
        input_formats=["pcm16/16000"],
        output_formats=["pcm16/24000"],
    )

    def __init__(
        self,
        model: str = GEMINI_LIVE_MODEL,
        voice: str = "Algieba",
        system_instruction: str = "",
        api_key: Optional[str] = None,
    ) -> None:
        super().__init__()
        self.model = model
        self.voice = voice
        self.system_instruction = system_instruction
        # Resolve key: explicit arg > env var.
        self._api_key: str = api_key or os.environ.get("GEMINI_API_KEY", "")

        # Populated when the background session task is live.
        self._session: Optional[Any] = None
        self._session_task: Optional[asyncio.Task[None]] = None
        self._session_ready: Optional[asyncio.Event] = None
        self._shutdown: Optional[asyncio.Event] = None
        self._session_error: Optional[BaseException] = None

        # Cached async iterator on ``session.receive()``. Acquired lazily
        # on the first ``recv_audio`` call so we can iterate the same
        # stream across consecutive agent turns. Without caching, each
        # ``recv_audio`` would call ``session.receive()`` afresh — which
        # the SDK does not support cleanly across turns.
        self._recv_iter: Optional[Any] = None
        # Tracks whether any audio was received on the CURRENT iterator
        # (reset whenever ``_recv_iter`` is recreated). Used by
        # ``recv_audio`` to distinguish a spurious empty-interrupt turn
        # (no audio at all) from a real mid-reply interrupt (audio
        # arrived before the interrupt landed).
        self._iter_had_audio: bool = False
        # Count of spurious interrupt→turn_complete re-entries during the CURRENT
        # turn's drain (reset per turn in ``send_audio``). Stamped onto the
        # inherited ``voice.audio.receive`` span at turn_complete — recv_audio
        # never opens a span of its own (exactly one receive span per turn).
        self._spurious_retry_count: int = 0

        # Observability.
        self.last_agent_transcript: Optional[str] = None

    def __repr__(self) -> str:
        # Never leak the API key.
        masked = "***" if self._api_key else ""
        return (
            f"GeminiLiveAgentAdapter("
            f"model={self.model!r}, "
            f"voice={self.voice!r}, "
            f"api_key={masked!r})"
        )

    # ------------------------------------------------------------------ lifecycle

    async def connect(self) -> None:
        """Open a Gemini Live session.

        Spawns a background task that holds the ``async with`` SDK context open
        for the adapter's lifetime.  Returns once the session handshake is
        complete and audio can flow.
        """
        from google import genai  # type: ignore[attr-defined]  # noqa: PLC0415 — lazy import
        from google.genai import types  # noqa: PLC0415

        self._session_ready = asyncio.Event()
        self._shutdown = asyncio.Event()
        loop = asyncio.get_running_loop()
        session_future: asyncio.Future[Any] = loop.create_future()

        config = types.LiveConnectConfig(
            response_modalities=[types.Modality.AUDIO],
            system_instruction=self.system_instruction or None,
            speech_config=types.SpeechConfig(
                voice_config=types.VoiceConfig(
                    prebuilt_voice_config=types.PrebuiltVoiceConfig(
                        voice_name=self.voice,
                    )
                )
            ),
            # Disable Automatic Activity Detection. AAD requires a clean
            # trailing silence to fire its end-of-speech detector, which is
            # unreliable across the audio shapes scenario produces (TTS'd
            # user-sim audio, scripted clips, layered interruptions). With
            # AAD off we drive turn boundaries explicitly via
            # ``activity_start`` / ``activity_end`` in ``send_audio``;
            # Gemini replies the moment we close the turn instead of
            # waiting on its own VAD heuristic. Activity handling is left
            # at its default (START_OF_ACTIVITY_INTERRUPTS): when we send
            # a new ``activity_start`` while Gemini is mid-reply, the
            # model treats it as a barge-in and cuts its in-flight audio.
            #
            # Subtlety: even after ``generation_complete`` on turn N, the
            # next ``activity_start`` opening turn N+1 is still treated as
            # a barge-in on the just-completed turn. The server emits a
            # spurious ``interrupted → turn_complete`` pair (with no model
            # output) BEFORE actually producing turn N+1's reply. The
            # ``recv_audio`` loop transparently skips that empty pair and
            # re-enters ``session.receive()`` to read the real reply.
            realtime_input_config=types.RealtimeInputConfig(
                automatic_activity_detection=types.AutomaticActivityDetection(
                    disabled=True,
                ),
            ),
            # Enable transcripts so the recv loop can populate
            # last_agent_transcript / chunk.transcript. Without these,
            # audio still flows but consumers (judge, manifest) get
            # no readable text.
            input_audio_transcription=types.AudioTranscriptionConfig(),
            output_audio_transcription=types.AudioTranscriptionConfig(),
        )

        client = genai.Client(api_key=self._api_key)

        async def _session_lifetime() -> None:
            """Hold the SDK context manager open; expose session via future."""
            try:
                async with client.aio.live.connect(
                    model=self.model, config=config
                ) as session:
                    if not session_future.done():
                        session_future.set_result(session)
                    assert self._session_ready is not None
                    self._session_ready.set()
                    # Stay alive until disconnect() fires the shutdown event.
                    assert self._shutdown is not None
                    await self._shutdown.wait()
            except Exception as exc:
                self._session_error = exc
                if not session_future.done():
                    session_future.set_exception(exc)
                assert self._session_ready is not None
                self._session_ready.set()  # unblock connect() even on error

        self._session_task = asyncio.create_task(_session_lifetime())

        # Wait until the session is ready (or errored).
        assert self._session_ready is not None
        await self._session_ready.wait()

        if self._session_error is not None:
            raise self._session_error

        self._session = await session_future
        self._recv_iter = None
        # Stamp Gemini-specific attrs onto the active ``voice.adapter.connect``
        # span (opened by the executor connect loop). Base spans are name-owned;
        # the adapter contributes attributes, never a parallel voice.gemini.* span.
        set_span_attributes(
            _otel_trace.get_current_span(),
            {"voice.gemini.model": self.model, "voice.gemini.voice": self.voice},
        )
        logger.debug("GeminiLiveAgentAdapter: connected model=%s", self.model)

    async def disconnect(self) -> None:
        """Close the Gemini Live session."""
        if self._recv_iter is not None:
            try:
                await self._recv_iter.aclose()  # type: ignore[attr-defined]
            except Exception:
                # Best-effort teardown: the iterator may already be
                # closed or in an invalid state during shutdown. Any
                # exception here is non-actionable since we're tearing
                # down anyway.
                pass
            self._recv_iter = None
        if self._shutdown is not None:
            self._shutdown.set()
        if self._session_task is not None:
            try:
                await asyncio.wait_for(self._session_task, timeout=5.0)
            except (asyncio.TimeoutError, Exception):
                # Timeout: task didn't finish in 5s — proceed with
                # teardown anyway, can't block disconnect indefinitely.
                # Other Exception: task error during shutdown is
                # non-actionable; we're discarding the session.
                pass
        self._session = None
        self._session_task = None
        self._session_ready = None
        self._shutdown = None
        self._session_error = None
        logger.debug("GeminiLiveAgentAdapter: disconnected")

    # ------------------------------------------------------------------ I/O

    async def send_audio(self, chunk: AudioChunk) -> None:
        """Send a canonical 24kHz AudioChunk to Gemini Live as a complete turn.

        Resamples from 24kHz → 16kHz at the wire boundary so the adapter
        speaks Gemini's expected ``audio/pcm;rate=16000`` format while the rest
        of the framework stays at the canonical 24kHz.

        Wraps the audio in explicit ``activity_start`` / ``activity_end``
        markers because we connect with Automatic Activity Detection
        disabled (see ``connect``). Each ``send_audio`` call is therefore a
        complete user turn from Gemini's perspective: it triggers the
        model to reply immediately on ``activity_end`` instead of waiting
        on its own VAD heuristic to detect end-of-speech. This is critical
        for the interrupt path — when the user barges in, we send a fresh
        turn boundary on top of the agent's in-flight reply, which Gemini
        treats as a deterministic interruption signal.
        """
        if self._session is None:
            raise RuntimeError("GeminiLiveAgentAdapter: not connected")
        from google.genai import types  # noqa: PLC0415

        pcm_16k = _resample_pcm16(chunk.data, CANONICAL_RATE, GEMINI_INPUT_RATE)
        if not pcm_16k:
            # Sub-sample chunk resampled to empty: nothing to send. Stamp
            # wire_bytes=0 onto the active ``voice.audio.send`` span (base-owned
            # name) and return WITHOUT emitting activity markers.
            set_span_attributes(
                _otel_trace.get_current_span(),
                {"voice.gemini.audio.wire_bytes": 0},
            )
            return
        # Stamp the resampled wire size onto the active ``voice.audio.send`` span;
        # the base owns the span name, the adapter contributes voice.gemini.* attrs.
        set_span_attributes(
            _otel_trace.get_current_span(),
            {"voice.gemini.audio.wire_bytes": len(pcm_16k)},
        )
        # New user turn → reset transcript and the per-turn receive
        # iterator so the next ``recv_audio`` enters
        # ``session.receive()`` fresh for this turn.
        self._reset_turn_transcript()
        if self._recv_iter is not None:
            try:
                await self._recv_iter.aclose()  # type: ignore[attr-defined]
            except Exception:
                # Best-effort: prior turn's receive iterator may already be
                # closed or in an error state. We're resetting to start a new
                # turn — propagating here would block legitimate new turns.
                pass
            self._recv_iter = None
        await self._session.send_realtime_input(activity_start=types.ActivityStart())
        blob = types.Blob(
            data=pcm_16k,
            mime_type="audio/pcm;rate=16000",
        )
        await self._session.send_realtime_input(audio=blob)
        await self._session.send_realtime_input(activity_end=types.ActivityEnd())

    async def recv_audio(self, timeout: float) -> AudioChunk:
        """Receive the next audio fragment from Gemini Live for the current turn.

        The SDK's ``session.receive()`` async generator yields messages
        for ONE model turn then stops at ``turn_complete``. We cache the
        per-turn iterator on ``self._recv_iter`` and reset it when the
        previous turn ended (StopAsyncIteration), so each user turn
        sent via ``send_audio`` can read its full reply across multiple
        ``recv_audio`` calls without us re-entering ``session.receive()``
        mid-turn (which would skip messages already buffered server-side).

        Returns the next non-empty audio chunk as soon as it arrives
        so the executor's ``_drain_agent_response`` can set
        ``_agent_speaking_event`` early — the interruption path depends
        on this.

        On ``turn_complete`` returns an empty AudioChunk so the drain
        loop's tail-silence path exits.

        Raises ``asyncio.TimeoutError`` if no chunk arrives within
        ``timeout`` seconds.
        """
        if self._session is None:
            raise RuntimeError("GeminiLiveAgentAdapter: not connected")
        if self._recv_iter is None:
            self._recv_iter = self._session.receive().__aiter__()  # type: ignore[union-attr]
            self._iter_had_audio = False
            # Reset the per-turn spurious-retry counter at drain start (alongside
            # _iter_had_audio) so the count stamped at turn_complete reflects only
            # THIS turn — correct even for a no-incoming turn that skips send_audio.
            self._spurious_retry_count = 0

        async def _next_chunk() -> AudioChunk:
            pending_delta = ""
            # Local-to-call: detects the spurious empty-interrupt turn
            # pattern (server emits ``interrupted=True`` then
            # ``turn_complete=True`` with no audio at all when a fresh
            # ``activity_start`` arrives during turn N's post-
            # ``generation_complete`` playback delay). Combined with
            # ``self._iter_had_audio`` (iterator-scope) we can tell the
            # difference between a spurious turn (no audio ever, on this
            # iterator) and a real mid-reply interrupt (audio arrived
            # earlier on this iterator).
            saw_interrupted = False
            while True:
                try:
                    assert self._recv_iter is not None
                    message = await self._recv_iter.__anext__()  # type: ignore[union-attr]
                except StopAsyncIteration:
                    # The previous turn ended (turn_complete already
                    # consumed). Surface end-of-turn to the drain loop
                    # and reset the iterator so the next user turn
                    # can re-enter session.receive() afresh.
                    self._recv_iter = None
                    return AudioChunk(
                        data=b"",
                        transcript=pending_delta or None,
                    )

                if message.go_away is not None:
                    raise RuntimeError(
                        f"GeminiLiveAgentAdapter: server sent go_away: {message.go_away}"
                    )

                sc = message.server_content
                if sc is None:
                    continue

                if getattr(sc, "interrupted", None):
                    saw_interrupted = True

                if sc.output_transcription is not None:
                    transcript_text = getattr(sc.output_transcription, "text", None)
                    if transcript_text:
                        pending_delta += transcript_text
                        existing = self.last_agent_transcript or ""
                        self.last_agent_transcript = existing + transcript_text

                if sc.model_turn is not None and sc.model_turn.parts:
                    audio_bytes = b""
                    for part in sc.model_turn.parts:
                        if part.inline_data is not None and part.inline_data.data:
                            audio_bytes += part.inline_data.data
                    if audio_bytes:
                        if len(audio_bytes) % 2 == 1:
                            audio_bytes = audio_bytes[:-1]
                        if audio_bytes:
                            self._iter_had_audio = True
                            return AudioChunk(
                                data=audio_bytes,
                                transcript=pending_delta or None,
                            )

                if sc.turn_complete:
                    # Spurious empty-interrupt turn? When activity_start
                    # opens turn N+1 after turn N's generation_complete,
                    # the server emits ``interrupted → turn_complete`` with
                    # no audio FIRST, then the real reply in a separate
                    # turn. Detect that pattern (saw interrupted=True, no
                    # audio on THIS iterator, no transcript) and re-enter
                    # ``session.receive()`` to read the actual reply.
                    #
                    # We gate on ``self._iter_had_audio`` (iterator-scope)
                    # rather than this call's audio: a real mid-reply
                    # interrupt earlier in the same turn would have yielded
                    # audio chunks before this point, even if THIS call sees
                    # only the trailing ``interrupted → turn_complete`` pair.
                    if (
                        saw_interrupted
                        and not self._iter_had_audio
                        and not pending_delta
                    ):
                        # Spurious re-entry — count it so the running total lands
                        # on the inherited ``voice.audio.receive`` span at the real
                        # turn_complete below (additive across the N recv_audio
                        # calls in one drain).
                        self._spurious_retry_count += 1
                        self._recv_iter = self._session.receive().__aiter__()  # type: ignore[union-attr]
                        self._iter_had_audio = False
                        saw_interrupted = False
                        continue
                    # Real end-of-turn — stamp Gemini attrs onto the active
                    # ``voice.audio.receive`` span (base-owned name; recv_audio
                    # opens NO span of its own — exactly one receive span per
                    # turn), then yield the empty AudioChunk and reset the
                    # iterator. The next ``recv_audio`` call (for the next user
                    # turn) will re-enter ``session.receive()``.
                    set_span_attributes(
                        _otel_trace.get_current_span(),
                        {
                            "voice.gemini.turn_complete": True,
                            "voice.gemini.spurious_retry_count": self._spurious_retry_count,
                        },
                    )
                    self._recv_iter = None
                    return AudioChunk(
                        data=b"",
                        transcript=pending_delta or None,
                    )

        return await asyncio.wait_for(_next_chunk(), timeout=timeout)

    async def interrupt(self) -> None:
        """Drain leftover chunks from the in-flight agent turn so the
        recovery agent's ``recv_audio`` doesn't pick them up as a fake
        first reply.

        On Gemini Live, when we send a fresh ``activity_start`` (the
        next ``send_audio``) while the model is mid-reply, the server
        cuts its in-flight audio AND emits ``turn_complete`` for that
        cancelled turn. ``session.receive()`` is a one-turn generator,
        so the cancelled turn's tail messages still need to be consumed
        before the next ``session.receive()`` invocation can read the
        recovery turn cleanly. ``interrupt()`` consumes them up to that
        ``turn_complete`` and resets the cached iterator.

        Best-effort: bounded by 2 seconds so a stuck stream doesn't
        block the executor's interrupt sequence.
        """
        if self._session is None or self._recv_iter is None:
            # No active turn to drain. Stamp the no-op outcome onto the active
            # ``voice.adapter.interrupt`` span (opened by the executor interrupt
            # site) and return.
            set_span_attributes(
                _otel_trace.get_current_span(),
                {
                    "voice.gemini.interrupt.outcome": "no_op_not_connected",
                    "voice.gemini.interrupt.drained_chunks": 0,
                },
            )
            return
        drained_chunks = 0
        outcome = "drained_to_turn_complete"
        try:
            async with asyncio.timeout(2.0):
                while True:
                    try:
                        message = await self._recv_iter.__anext__()  # type: ignore[union-attr]
                    except StopAsyncIteration:
                        break
                    drained_chunks += 1
                    sc = getattr(message, "server_content", None)
                    if sc is not None and sc.turn_complete:
                        break
        except asyncio.TimeoutError:
            # Bounded drain: if the server doesn't close out the turn within
            # 2s after we sent the activity_end, give up and proceed. The
            # finally block will still close the iterator.
            outcome = "drain_timeout_2s"
        finally:
            try:
                await self._recv_iter.aclose()  # type: ignore[attr-defined]
            except Exception:
                # Best-effort close; the iterator may already be exhausted
                # or in an error state. Don't mask the original outcome.
                pass
            self._recv_iter = None
        # Stamp the interrupt outcome onto the active ``voice.adapter.interrupt``
        # span (base-owned name; the adapter contributes voice.gemini.* attrs).
        # py DRAINS the cancelled turn with a 2s bound (pull iterator); the TS
        # mirror is a passive abort-sentinel (push callback) — see
        # ``gemini-live.ts`` ``interrupt()`` for the genuine SDK asymmetry.
        set_span_attributes(
            _otel_trace.get_current_span(),
            {
                "voice.gemini.interrupt.outcome": outcome,
                "voice.gemini.interrupt.drained_chunks": drained_chunks,
            },
        )

    def _reset_turn_transcript(self) -> None:
        """Clear the running transcript before each new agent turn.

        Called from ``send_audio`` so each turn starts fresh — otherwise
        the agent's first reply text would be permanently prefixed onto
        all subsequent turns' transcripts.
        """
        self.last_agent_transcript = None

Ancestors

• VoiceAgentAdapter
• AgentAdapter
• abc.ABC

Class variables

var capabilities : ClassVar[AdapterCapabilities]

Methods

async def connect(self) ‑> None

Open a Gemini Live session.

Spawns a background task that holds the async with SDK context open for the adapter's lifetime. Returns once the session handshake is complete and audio can flow.

Expand source code

async def connect(self) -> None:
    """Open a Gemini Live session.

    Spawns a background task that holds the ``async with`` SDK context open
    for the adapter's lifetime.  Returns once the session handshake is
    complete and audio can flow.
    """
    from google import genai  # type: ignore[attr-defined]  # noqa: PLC0415 — lazy import
    from google.genai import types  # noqa: PLC0415

    self._session_ready = asyncio.Event()
    self._shutdown = asyncio.Event()
    loop = asyncio.get_running_loop()
    session_future: asyncio.Future[Any] = loop.create_future()

    config = types.LiveConnectConfig(
        response_modalities=[types.Modality.AUDIO],
        system_instruction=self.system_instruction or None,
        speech_config=types.SpeechConfig(
            voice_config=types.VoiceConfig(
                prebuilt_voice_config=types.PrebuiltVoiceConfig(
                    voice_name=self.voice,
                )
            )
        ),
        # Disable Automatic Activity Detection. AAD requires a clean
        # trailing silence to fire its end-of-speech detector, which is
        # unreliable across the audio shapes scenario produces (TTS'd
        # user-sim audio, scripted clips, layered interruptions). With
        # AAD off we drive turn boundaries explicitly via
        # ``activity_start`` / ``activity_end`` in ``send_audio``;
        # Gemini replies the moment we close the turn instead of
        # waiting on its own VAD heuristic. Activity handling is left
        # at its default (START_OF_ACTIVITY_INTERRUPTS): when we send
        # a new ``activity_start`` while Gemini is mid-reply, the
        # model treats it as a barge-in and cuts its in-flight audio.
        #
        # Subtlety: even after ``generation_complete`` on turn N, the
        # next ``activity_start`` opening turn N+1 is still treated as
        # a barge-in on the just-completed turn. The server emits a
        # spurious ``interrupted → turn_complete`` pair (with no model
        # output) BEFORE actually producing turn N+1's reply. The
        # ``recv_audio`` loop transparently skips that empty pair and
        # re-enters ``session.receive()`` to read the real reply.
        realtime_input_config=types.RealtimeInputConfig(
            automatic_activity_detection=types.AutomaticActivityDetection(
                disabled=True,
            ),
        ),
        # Enable transcripts so the recv loop can populate
        # last_agent_transcript / chunk.transcript. Without these,
        # audio still flows but consumers (judge, manifest) get
        # no readable text.
        input_audio_transcription=types.AudioTranscriptionConfig(),
        output_audio_transcription=types.AudioTranscriptionConfig(),
    )

    client = genai.Client(api_key=self._api_key)

    async def _session_lifetime() -> None:
        """Hold the SDK context manager open; expose session via future."""
        try:
            async with client.aio.live.connect(
                model=self.model, config=config
            ) as session:
                if not session_future.done():
                    session_future.set_result(session)
                assert self._session_ready is not None
                self._session_ready.set()
                # Stay alive until disconnect() fires the shutdown event.
                assert self._shutdown is not None
                await self._shutdown.wait()
        except Exception as exc:
            self._session_error = exc
            if not session_future.done():
                session_future.set_exception(exc)
            assert self._session_ready is not None
            self._session_ready.set()  # unblock connect() even on error

    self._session_task = asyncio.create_task(_session_lifetime())

    # Wait until the session is ready (or errored).
    assert self._session_ready is not None
    await self._session_ready.wait()

    if self._session_error is not None:
        raise self._session_error

    self._session = await session_future
    self._recv_iter = None
    # Stamp Gemini-specific attrs onto the active ``voice.adapter.connect``
    # span (opened by the executor connect loop). Base spans are name-owned;
    # the adapter contributes attributes, never a parallel voice.gemini.* span.
    set_span_attributes(
        _otel_trace.get_current_span(),
        {"voice.gemini.model": self.model, "voice.gemini.voice": self.voice},
    )
    logger.debug("GeminiLiveAgentAdapter: connected model=%s", self.model)

async def disconnect(self) ‑> None

Close the Gemini Live session.

Expand source code

async def disconnect(self) -> None:
    """Close the Gemini Live session."""
    if self._recv_iter is not None:
        try:
            await self._recv_iter.aclose()  # type: ignore[attr-defined]
        except Exception:
            # Best-effort teardown: the iterator may already be
            # closed or in an invalid state during shutdown. Any
            # exception here is non-actionable since we're tearing
            # down anyway.
            pass
        self._recv_iter = None
    if self._shutdown is not None:
        self._shutdown.set()
    if self._session_task is not None:
        try:
            await asyncio.wait_for(self._session_task, timeout=5.0)
        except (asyncio.TimeoutError, Exception):
            # Timeout: task didn't finish in 5s — proceed with
            # teardown anyway, can't block disconnect indefinitely.
            # Other Exception: task error during shutdown is
            # non-actionable; we're discarding the session.
            pass
    self._session = None
    self._session_task = None
    self._session_ready = None
    self._shutdown = None
    self._session_error = None
    logger.debug("GeminiLiveAgentAdapter: disconnected")

async def interrupt(self) ‑> None

Drain leftover chunks from the in-flight agent turn so the recovery agent's recv_audio doesn't pick them up as a fake first reply.

On Gemini Live, when we send a fresh activity_start (the next send_audio) while the model is mid-reply, the server cuts its in-flight audio AND emits turn_complete for that cancelled turn. session.receive() is a one-turn generator, so the cancelled turn's tail messages still need to be consumed before the next session.receive() invocation can read the recovery turn cleanly. interrupt() consumes them up to that turn_complete and resets the cached iterator.

Best-effort: bounded by 2 seconds so a stuck stream doesn't block the executor's interrupt sequence.

Expand source code

async def interrupt(self) -> None:
    """Drain leftover chunks from the in-flight agent turn so the
    recovery agent's ``recv_audio`` doesn't pick them up as a fake
    first reply.

    On Gemini Live, when we send a fresh ``activity_start`` (the
    next ``send_audio``) while the model is mid-reply, the server
    cuts its in-flight audio AND emits ``turn_complete`` for that
    cancelled turn. ``session.receive()`` is a one-turn generator,
    so the cancelled turn's tail messages still need to be consumed
    before the next ``session.receive()`` invocation can read the
    recovery turn cleanly. ``interrupt()`` consumes them up to that
    ``turn_complete`` and resets the cached iterator.

    Best-effort: bounded by 2 seconds so a stuck stream doesn't
    block the executor's interrupt sequence.
    """
    if self._session is None or self._recv_iter is None:
        # No active turn to drain. Stamp the no-op outcome onto the active
        # ``voice.adapter.interrupt`` span (opened by the executor interrupt
        # site) and return.
        set_span_attributes(
            _otel_trace.get_current_span(),
            {
                "voice.gemini.interrupt.outcome": "no_op_not_connected",
                "voice.gemini.interrupt.drained_chunks": 0,
            },
        )
        return
    drained_chunks = 0
    outcome = "drained_to_turn_complete"
    try:
        async with asyncio.timeout(2.0):
            while True:
                try:
                    message = await self._recv_iter.__anext__()  # type: ignore[union-attr]
                except StopAsyncIteration:
                    break
                drained_chunks += 1
                sc = getattr(message, "server_content", None)
                if sc is not None and sc.turn_complete:
                    break
    except asyncio.TimeoutError:
        # Bounded drain: if the server doesn't close out the turn within
        # 2s after we sent the activity_end, give up and proceed. The
        # finally block will still close the iterator.
        outcome = "drain_timeout_2s"
    finally:
        try:
            await self._recv_iter.aclose()  # type: ignore[attr-defined]
        except Exception:
            # Best-effort close; the iterator may already be exhausted
            # or in an error state. Don't mask the original outcome.
            pass
        self._recv_iter = None
    # Stamp the interrupt outcome onto the active ``voice.adapter.interrupt``
    # span (base-owned name; the adapter contributes voice.gemini.* attrs).
    # py DRAINS the cancelled turn with a 2s bound (pull iterator); the TS
    # mirror is a passive abort-sentinel (push callback) — see
    # ``gemini-live.ts`` ``interrupt()`` for the genuine SDK asymmetry.
    set_span_attributes(
        _otel_trace.get_current_span(),
        {
            "voice.gemini.interrupt.outcome": outcome,
            "voice.gemini.interrupt.drained_chunks": drained_chunks,
        },
    )

async def recv_audio(self, timeout: float) ‑> AudioChunk

Receive the next audio fragment from Gemini Live for the current turn.

The SDK's session.receive() async generator yields messages for ONE model turn then stops at turn_complete. We cache the per-turn iterator on self._recv_iter and reset it when the previous turn ended (StopAsyncIteration), so each user turn sent via send_audio can read its full reply across multiple recv_audio calls without us re-entering session.receive() mid-turn (which would skip messages already buffered server-side).

Returns the next non-empty audio chunk as soon as it arrives so the executor's _drain_agent_response can set _agent_speaking_event early — the interruption path depends on this.

On turn_complete returns an empty AudioChunk so the drain loop's tail-silence path exits.

Raises asyncio.TimeoutError if no chunk arrives within timeout seconds.

Expand source code

async def recv_audio(self, timeout: float) -> AudioChunk:
    """Receive the next audio fragment from Gemini Live for the current turn.

    The SDK's ``session.receive()`` async generator yields messages
    for ONE model turn then stops at ``turn_complete``. We cache the
    per-turn iterator on ``self._recv_iter`` and reset it when the
    previous turn ended (StopAsyncIteration), so each user turn
    sent via ``send_audio`` can read its full reply across multiple
    ``recv_audio`` calls without us re-entering ``session.receive()``
    mid-turn (which would skip messages already buffered server-side).

    Returns the next non-empty audio chunk as soon as it arrives
    so the executor's ``_drain_agent_response`` can set
    ``_agent_speaking_event`` early — the interruption path depends
    on this.

    On ``turn_complete`` returns an empty AudioChunk so the drain
    loop's tail-silence path exits.

    Raises ``asyncio.TimeoutError`` if no chunk arrives within
    ``timeout`` seconds.
    """
    if self._session is None:
        raise RuntimeError("GeminiLiveAgentAdapter: not connected")
    if self._recv_iter is None:
        self._recv_iter = self._session.receive().__aiter__()  # type: ignore[union-attr]
        self._iter_had_audio = False
        # Reset the per-turn spurious-retry counter at drain start (alongside
        # _iter_had_audio) so the count stamped at turn_complete reflects only
        # THIS turn — correct even for a no-incoming turn that skips send_audio.
        self._spurious_retry_count = 0

    async def _next_chunk() -> AudioChunk:
        pending_delta = ""
        # Local-to-call: detects the spurious empty-interrupt turn
        # pattern (server emits ``interrupted=True`` then
        # ``turn_complete=True`` with no audio at all when a fresh
        # ``activity_start`` arrives during turn N's post-
        # ``generation_complete`` playback delay). Combined with
        # ``self._iter_had_audio`` (iterator-scope) we can tell the
        # difference between a spurious turn (no audio ever, on this
        # iterator) and a real mid-reply interrupt (audio arrived
        # earlier on this iterator).
        saw_interrupted = False
        while True:
            try:
                assert self._recv_iter is not None
                message = await self._recv_iter.__anext__()  # type: ignore[union-attr]
            except StopAsyncIteration:
                # The previous turn ended (turn_complete already
                # consumed). Surface end-of-turn to the drain loop
                # and reset the iterator so the next user turn
                # can re-enter session.receive() afresh.
                self._recv_iter = None
                return AudioChunk(
                    data=b"",
                    transcript=pending_delta or None,
                )

            if message.go_away is not None:
                raise RuntimeError(
                    f"GeminiLiveAgentAdapter: server sent go_away: {message.go_away}"
                )

            sc = message.server_content
            if sc is None:
                continue

            if getattr(sc, "interrupted", None):
                saw_interrupted = True

            if sc.output_transcription is not None:
                transcript_text = getattr(sc.output_transcription, "text", None)
                if transcript_text:
                    pending_delta += transcript_text
                    existing = self.last_agent_transcript or ""
                    self.last_agent_transcript = existing + transcript_text

            if sc.model_turn is not None and sc.model_turn.parts:
                audio_bytes = b""
                for part in sc.model_turn.parts:
                    if part.inline_data is not None and part.inline_data.data:
                        audio_bytes += part.inline_data.data
                if audio_bytes:
                    if len(audio_bytes) % 2 == 1:
                        audio_bytes = audio_bytes[:-1]
                    if audio_bytes:
                        self._iter_had_audio = True
                        return AudioChunk(
                            data=audio_bytes,
                            transcript=pending_delta or None,
                        )

            if sc.turn_complete:
                # Spurious empty-interrupt turn? When activity_start
                # opens turn N+1 after turn N's generation_complete,
                # the server emits ``interrupted → turn_complete`` with
                # no audio FIRST, then the real reply in a separate
                # turn. Detect that pattern (saw interrupted=True, no
                # audio on THIS iterator, no transcript) and re-enter
                # ``session.receive()`` to read the actual reply.
                #
                # We gate on ``self._iter_had_audio`` (iterator-scope)
                # rather than this call's audio: a real mid-reply
                # interrupt earlier in the same turn would have yielded
                # audio chunks before this point, even if THIS call sees
                # only the trailing ``interrupted → turn_complete`` pair.
                if (
                    saw_interrupted
                    and not self._iter_had_audio
                    and not pending_delta
                ):
                    # Spurious re-entry — count it so the running total lands
                    # on the inherited ``voice.audio.receive`` span at the real
                    # turn_complete below (additive across the N recv_audio
                    # calls in one drain).
                    self._spurious_retry_count += 1
                    self._recv_iter = self._session.receive().__aiter__()  # type: ignore[union-attr]
                    self._iter_had_audio = False
                    saw_interrupted = False
                    continue
                # Real end-of-turn — stamp Gemini attrs onto the active
                # ``voice.audio.receive`` span (base-owned name; recv_audio
                # opens NO span of its own — exactly one receive span per
                # turn), then yield the empty AudioChunk and reset the
                # iterator. The next ``recv_audio`` call (for the next user
                # turn) will re-enter ``session.receive()``.
                set_span_attributes(
                    _otel_trace.get_current_span(),
                    {
                        "voice.gemini.turn_complete": True,
                        "voice.gemini.spurious_retry_count": self._spurious_retry_count,
                    },
                )
                self._recv_iter = None
                return AudioChunk(
                    data=b"",
                    transcript=pending_delta or None,
                )

    return await asyncio.wait_for(_next_chunk(), timeout=timeout)

async def send_audio(self, chunk: AudioChunk) ‑> None

Send a canonical 24kHz AudioChunk to Gemini Live as a complete turn.

Resamples from 24kHz → 16kHz at the wire boundary so the adapter speaks Gemini's expected audio/pcm;rate=16000 format while the rest of the framework stays at the canonical 24kHz.

Wraps the audio in explicit activity_start / activity_end markers because we connect with Automatic Activity Detection disabled (see connect). Each send_audio call is therefore a complete user turn from Gemini's perspective: it triggers the model to reply immediately on activity_end instead of waiting on its own VAD heuristic to detect end-of-speech. This is critical for the interrupt path — when the user barges in, we send a fresh turn boundary on top of the agent's in-flight reply, which Gemini treats as a deterministic interruption signal.

Expand source code

async def send_audio(self, chunk: AudioChunk) -> None:
    """Send a canonical 24kHz AudioChunk to Gemini Live as a complete turn.

    Resamples from 24kHz → 16kHz at the wire boundary so the adapter
    speaks Gemini's expected ``audio/pcm;rate=16000`` format while the rest
    of the framework stays at the canonical 24kHz.

    Wraps the audio in explicit ``activity_start`` / ``activity_end``
    markers because we connect with Automatic Activity Detection
    disabled (see ``connect``). Each ``send_audio`` call is therefore a
    complete user turn from Gemini's perspective: it triggers the
    model to reply immediately on ``activity_end`` instead of waiting
    on its own VAD heuristic to detect end-of-speech. This is critical
    for the interrupt path — when the user barges in, we send a fresh
    turn boundary on top of the agent's in-flight reply, which Gemini
    treats as a deterministic interruption signal.
    """
    if self._session is None:
        raise RuntimeError("GeminiLiveAgentAdapter: not connected")
    from google.genai import types  # noqa: PLC0415

    pcm_16k = _resample_pcm16(chunk.data, CANONICAL_RATE, GEMINI_INPUT_RATE)
    if not pcm_16k:
        # Sub-sample chunk resampled to empty: nothing to send. Stamp
        # wire_bytes=0 onto the active ``voice.audio.send`` span (base-owned
        # name) and return WITHOUT emitting activity markers.
        set_span_attributes(
            _otel_trace.get_current_span(),
            {"voice.gemini.audio.wire_bytes": 0},
        )
        return
    # Stamp the resampled wire size onto the active ``voice.audio.send`` span;
    # the base owns the span name, the adapter contributes voice.gemini.* attrs.
    set_span_attributes(
        _otel_trace.get_current_span(),
        {"voice.gemini.audio.wire_bytes": len(pcm_16k)},
    )
    # New user turn → reset transcript and the per-turn receive
    # iterator so the next ``recv_audio`` enters
    # ``session.receive()`` fresh for this turn.
    self._reset_turn_transcript()
    if self._recv_iter is not None:
        try:
            await self._recv_iter.aclose()  # type: ignore[attr-defined]
        except Exception:
            # Best-effort: prior turn's receive iterator may already be
            # closed or in an error state. We're resetting to start a new
            # turn — propagating here would block legitimate new turns.
            pass
        self._recv_iter = None
    await self._session.send_realtime_input(activity_start=types.ActivityStart())
    blob = types.Blob(
        data=pcm_16k,
        mime_type="audio/pcm;rate=16000",
    )
    await self._session.send_realtime_input(audio=blob)
    await self._session.send_realtime_input(activity_end=types.ActivityEnd())

Inherited members

• VoiceAgentAdapter:
  • agent_hung_up
  • call
  • is_connected

class LiveKitAgentAdapter (url: str, api_key: str, api_secret: str, room: str)

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

Expand source code

class LiveKitAgentAdapter(VoiceAgentAdapter):
    capabilities: ClassVar[AdapterCapabilities] = AdapterCapabilities(
        streaming_transcripts=True,
        native_vad=True,
        dtmf=False,
        input_formats=["pcm16/48000"],
        output_formats=["pcm16/48000"],
    )

    def __init__(self, url: str, api_key: str, api_secret: str, room: str):
        super().__init__()
        self.url = url
        self.api_key = api_key
        self.api_secret = api_secret
        self.room = room
        self._room: Optional[object] = None

    def __repr__(self) -> str:  # redact credentials
        return f"LiveKitAgentAdapter(url={self.url!r}, room={self.room!r}, api_key='***', api_secret='***')"

    async def connect(self) -> None:
        self._room = object()

    async def disconnect(self) -> None:
        self._room = None

    async def send_audio(self, chunk: AudioChunk) -> None:
        if self._room is None:
            raise RuntimeError("LiveKitAgentAdapter: not connected")
        raise PendingTransportError("LiveKitAgentAdapter")

    async def recv_audio(self, timeout: float) -> AudioChunk:
        if self._room is None:
            raise RuntimeError("LiveKitAgentAdapter: not connected")
        raise PendingTransportError("LiveKitAgentAdapter")

Ancestors

• VoiceAgentAdapter
• AgentAdapter
• abc.ABC

Class variables

var capabilities : ClassVar[AdapterCapabilities]

Inherited members

• VoiceAgentAdapter:
  • agent_hung_up
  • call
  • connect
  • disconnect
  • interrupt
  • is_connected
  • recv_audio
  • send_audio

class OpenAIRealtimeAgentAdapter (model: str = 'gpt-realtime-mini', voice: str = 'alloy', instructions: str = '', tools: Optional[List[Any]] = None, *, api_key: Optional[str] = None, role: AgentRole = AgentRole.AGENT, speaks_first: bool = False, mint: Union[RealtimeMintEndpoint, bool, None] = None)

Exercise OpenAI's Realtime API as either the agent under test (role=AGENT, default) or as the voice-enabled user simulator (role=USER, per §7.2 L1164-1171).

When role=USER, scripted user("text") steps route text through the realtime session's text-input channel rather than triggering TTS.

Transcript observability: - last_user_transcript — set from conversation.item.input_audio_transcription.completed - last_agent_transcript — accumulated from response.output_audio_transcript.delta / reset on done

Example::

adapter = OpenAIRealtimeAgentAdapter(
    model=OPENAI_REALTIME_MODEL,
    voice="alloy",
    instructions="You are a helpful assistant.",
)
async with adapter:
    # scenario.run() feeds send_audio / recv_audio ...

Expand source code

class OpenAIRealtimeAgentAdapter(VoiceAgentAdapter):
    """
    Exercise OpenAI's Realtime API as either the agent under test
    (role=AGENT, default) or as the voice-enabled user simulator
    (role=USER, per §7.2 L1164-1171).

    When role=USER, scripted ``user("text")`` steps route text through the
    realtime session's text-input channel rather than triggering TTS.

    Transcript observability:
        - ``last_user_transcript`` — set from
          ``conversation.item.input_audio_transcription.completed``
        - ``last_agent_transcript`` — accumulated from
          ``response.output_audio_transcript.delta`` / reset on done

    Example::

        adapter = OpenAIRealtimeAgentAdapter(
            model=OPENAI_REALTIME_MODEL,
            voice="alloy",
            instructions="You are a helpful assistant.",
        )
        async with adapter:
            # scenario.run() feeds send_audio / recv_audio ...
    """

    capabilities: ClassVar[AdapterCapabilities] = AdapterCapabilities(
        streaming_transcripts=True,
        native_vad=True,
        dtmf=False,
        # OpenAI Realtime exposes ``response.cancel`` as a first-class
        # interrupt event — the model stops generating immediately. Mapped
        # below in ``interrupt()``.
        interruption=True,
        input_formats=["pcm16/24000"],
        output_formats=["pcm16/24000"],
    )

    def __init__(
        self,
        model: str = OPENAI_REALTIME_MODEL,
        voice: str = "alloy",
        instructions: str = "",
        tools: Optional[List[Any]] = None,
        *,
        api_key: Optional[str] = None,
        role: AgentRole = AgentRole.AGENT,
        speaks_first: bool = False,
        mint: Union[RealtimeMintEndpoint, bool, None] = None,
    ):
        super().__init__()
        self.model = model
        self.voice = voice
        self.instructions = instructions
        self.tools = tools or []
        self.role = role  # type: ignore[misc]
        # The fallback key for dialling the vendor directly: explicit param,
        # then OPENAI_REALTIME_API_KEY, then OPENAI_API_KEY.
        # OPENAI_REALTIME_API_KEY exists so a deployment whose OPENAI_API_KEY
        # is a virtual key can still hold a real provider key for this socket.
        #
        # A minted session uses none of these on the socket. It dials with the
        # ephemeral secret the mint returned, so a long-lived key never
        # reaches the socket. See ``mint``.
        self._api_key: str = (
            api_key
            or os.environ.get("OPENAI_REALTIME_API_KEY")
            or os.environ.get("OPENAI_API_KEY", "")
        )
        # Which of the three the key came from. Only used to name it in the
        # direct-dial warning, where naming the wrong one sends the reader to
        # a variable that is not set.
        self._api_key_source: str = (
            "the api_key passed to the adapter"
            if api_key is not None
            else (
                "OPENAI_REALTIME_API_KEY"
                if os.environ.get("OPENAI_REALTIME_API_KEY")
                else "OPENAI_API_KEY"
            )
        )
        # Where to mint the session credential. ``False`` skips the mint and
        # dials the vendor directly; an explicit endpoint overrides
        # OPENAI_BASE_URL and OPENAI_API_KEY; ``None`` resolves from those two.
        #
        # ``POST /v1/realtime/client_secrets`` is OpenAI's own path, and a
        # LangWatch AI Gateway mirrors it, so the same request works against
        # either. Against OpenAI it returns an ephemeral client secret.
        # Against a gateway it also checks the virtual key's budget and
        # open-session cap and opens one spend record, naming the session on
        # the ``X-LangWatch-Session-Id`` response header. Usage is reported
        # back only when that header is present.
        self._mint: Optional[RealtimeMintEndpoint]
        if mint is False:
            self._mint = None
        elif isinstance(mint, RealtimeMintEndpoint):
            self._mint = mint
        else:
            self._mint = resolve_realtime_mint_endpoint()
        #: The gateway's id for the current session, empty unless a gateway
        #: minted it.
        self._broker_session_id: str = ""
        #: What the socket has reported so far this session, summed across
        #: responses and sent when the session ends.
        self._session_usage: Optional[Dict[str, Any]] = None
        self._ws: Any = None

        # Transcript observability — updated on incoming transcript events.
        self.last_user_transcript: Optional[str] = None
        self.last_agent_transcript: Optional[str] = None

        # Accumulation buffer for streaming agent transcript deltas.
        self._agent_transcript_buf: str = ""

        # Bytes appended to input_audio_buffer since last commit. Non-zero
        # means recv_audio should commit + request a response before awaiting.
        self._pending_audio_bytes: int = 0

        # Tracks which legacy (pre-GA) event names have already triggered a
        # one-time warning, so the log isn't spammed on every audio frame.
        self._legacy_events_warned: set[str] = set()

        # --- Gap 1: agent-speaks-first support ---
        # When speaks_first=True, the adapter is configured for an agent-initiated
        # scenario where the agent must speak first without any user audio.
        self._speaks_first: bool = speaks_first

        # True while a response is in flight (set on response.created, cleared on
        # response.done / response.cancelled). Prevents double-firing response.create
        # and allows drain re-entries after a completed response to return empty
        # (clean drain exit).
        self._response_active: bool = False

        # Set to True the first time response.created is received. Guards the
        # drain re-entry short-circuit in recv_audio: the empty-chunk early-return
        # must only fire AFTER at least one response has been active and completed,
        # not on the very first recv_audio call (which would break direct-call
        # tests that don't go through notify_agent_turn).
        self._response_ever_active: bool = False

        # Per-turn signal: set by notify_agent_turn() before each agent step so
        # recv_audio knows this is a genuine agent-initiated turn (not a silent
        # proceed()/resume). Consumed (cleared) when the kick fires.
        self._agent_turn_pending: bool = speaks_first  # first turn armed if speaks_first

        # Issue #657: distinct deferral flag for the user-audio race guard.
        # Set when user audio arrives while a response is in flight; cleared and
        # consumed in the response.done/cancelled handler to fire the deferred
        # response.create.  Kept separate from _agent_turn_pending so each flag
        # retains its single original meaning.
        # Known limitation: a second user-audio commit while one create is already
        # deferred collapses into this single boolean.  Manual-VAD single-turn flow
        # makes that path unreachable in practice; FSM refactor tracked as follow-up.
        self._deferred_response_create: bool = False

        # --- Issue #630: realtime function-call (tool-call) surfacing ---
        # In-progress function calls keyed by call_id. Each value is a dict
        # {"name": str|None, "arguments": str} assembled from the streaming
        # `response.function_call_arguments.delta`/`.done` events and the
        # `response.output_item.added`/`.done` (function_call item) events.
        # recv_audio() returns an AudioChunk, so it cannot carry tool calls in
        # its return value — instead the function-call branches accumulate here
        # and the overridden call() drains them into result.messages.
        self._tool_call_accumulators: dict[str, dict[str, Any]] = {}
        # Finalized OpenAI-chat tool_call dicts for the CURRENT turn, in arrival
        # order. Cleared at the start of each call() so calls never leak across
        # turns (mirrors the transcript-state reset in _drain_agent_response).
        # Shape per entry:
        #   {"id": call_id, "type": "function",
        #    "function": {"name": name, "arguments": <json-str>}}
        self._completed_tool_calls: List[dict[str, Any]] = []

    @property
    def url(self) -> str:
        return REALTIME_URL_TEMPLATE.format(model=self.model)

    @property
    def brokered(self) -> bool:
        """Whether the live session was minted by a gateway, not the vendor.

        Only true after ``connect()``, and only when the mint answered with
        ``X-LangWatch-Session-Id``. Configuration cannot set it, because the
        response is the one thing that cannot be told a lie about what
        answered.
        """
        return self._broker_session_id != ""

    def __repr__(self) -> str:  # redact credentials
        return (
            f"OpenAIRealtimeAgentAdapter("
            f"model={self.model!r}, "
            f"voice={self.voice!r}, "
            f"role={self.role!r}, "
            f"api_key='***')"
        )

    def _warn_if_legacy(self, received: str, ga_name: str) -> None:
        """Emit a one-time WARNING when a pre-GA (beta) event name is seen.

        Only fires once per distinct legacy name per adapter instance, so a
        multi-chunk audio response doesn't flood the log. The GA-named event
        itself never triggers a warning.
        """
        if received != ga_name and received not in self._legacy_events_warned:
            self._legacy_events_warned.add(received)
            logger.warning(
                "OpenAIRealtimeAgentAdapter: received legacy event %r; "
                "GA uses %r — accepting defensively",
                received,
                ga_name,
            )

    # ------------------------------------------------------------- tool calls
    # Issue #630: surface OpenAI Realtime function-call events as OpenAI-chat
    # tool_calls on the assistant message in result.messages. The Realtime wire
    # describes ONE logical call across several events
    # (`response.function_call_arguments.delta` × N → `.done`, and/or
    # `response.output_item.added`/`.done` carrying the function_call item).
    # We accumulate per call_id, then finalize into self._completed_tool_calls,
    # de-duplicating so each call_id yields exactly one tool_call (AC6).

    def _accumulate_tool_call_delta(self, call_id: str, delta: str) -> None:
        """Append a streaming arguments fragment for ``call_id``."""
        acc = self._tool_call_accumulators.setdefault(
            call_id, {"name": None, "arguments": ""}
        )
        acc["arguments"] = (acc["arguments"] or "") + (delta or "")

    def _note_tool_call_name(self, call_id: str, name: Optional[str]) -> None:
        """Record the function name for ``call_id`` (the item/added event)."""
        if not name:
            return
        acc = self._tool_call_accumulators.setdefault(
            call_id, {"name": None, "arguments": ""}
        )
        acc["name"] = name

    def _finalize_tool_call(
        self,
        call_id: Optional[str],
        *,
        name: Optional[str] = None,
        arguments: Optional[str] = None,
    ) -> None:
        """Resolve one logical function call into ``self._completed_tool_calls``.

        Idempotent on ``call_id`` (AC6): the streaming-args path and the
        output-item path both describe the SAME call, so a second finalize for
        an already-completed ``call_id`` MERGES (fills a missing name / upgrades
        to a more complete arguments string) rather than appending a duplicate.

        Degrades safely (AC7): a finalize with NO ``call_id`` is skipped with a
        DEBUG log and emits nothing. Missing arguments become ``"{}"``; a
        malformed (non-JSON) arguments string is passed through verbatim — the
        adapter never parses-and-reraises.
        """
        if not call_id:
            logger.debug(
                "OpenAIRealtimeAgentAdapter: function-call event with no "
                "call_id; skipping (AC7 degraded path)"
            )
            return

        acc = self._tool_call_accumulators.get(call_id, {})
        resolved_name = name or acc.get("name")
        # Pick the most complete arguments source: explicit (item/done) arg if
        # non-empty, else the accumulated streaming deltas, else "{}".
        candidates = [arguments, acc.get("arguments")]
        resolved_args = next(
            (c for c in candidates if c is not None and c != ""), None
        )
        if resolved_args is None:
            resolved_args = "{}"

        # De-dup / merge on call_id (AC6).
        for existing in self._completed_tool_calls:
            if existing["id"] == call_id:
                fn = existing["function"]
                if not fn.get("name") and resolved_name:
                    fn["name"] = resolved_name
                # Upgrade to a longer/more-complete arguments string if the new
                # source carries more (item arguments arriving after deltas).
                if resolved_args not in ("", "{}") and len(resolved_args) > len(
                    fn.get("arguments") or ""
                ):
                    fn["arguments"] = resolved_args
                return

        self._completed_tool_calls.append(
            {
                "id": call_id,
                "type": "function",
                "function": {
                    "name": resolved_name or "",
                    "arguments": resolved_args,
                },
            }
        )

    def notify_agent_turn(self) -> None:
        """Signal that an agent turn is about to be dispatched.

        Called by the executor before each agent step so recv_audio can fire
        a bare response.create for agent-initiated turns (where no user audio
        has been committed). This per-turn signal handles both turn 1 (opening)
        and subsequent agent turns in multi-turn scripts like [agent(), user(),
        agent()].

        Only meaningful when role=AGENT. Safe to call on every agent step —
        recv_audio consumes and clears the flag.
        """
        if self.role == AgentRole.AGENT:
            self._agent_turn_pending = True

    # ------------------------------------------------------------------ lifecycle

    async def connect(self) -> None:
        """Open the Realtime WebSocket and send the initial session.update."""
        import websockets

        # Issue #662 / reconnect hygiene (JS parity — openai-realtime.ts connect()):
        # a reused adapter instance must start each connection with a clean slate.
        # Clearing _response_active matters most: a disconnect mid-response would
        # otherwise leave it True, and the next turn's user-audio commit would take
        # the defer branch forever (no response.done ever arrives to fire it), draining
        # recv_audio to timeout.
        self._response_active = False
        self._deferred_response_create = False

        self._broker_session_id = ""
        self._session_usage = None

        # The socket's bearer is the ephemeral secret the mint returned: a
        # credential for exactly this call, so no long-lived key reaches the
        # socket. Whether OpenAI or a gateway answers the mint is a property
        # of OPENAI_BASE_URL, not of this adapter.
        socket_key = self._api_key
        if self._mint is not None:
            result = await mint_openai_realtime_session(self._mint, self.model)
            if result.minted:
                socket_key = result.credential
                self._broker_session_id = result.session_id
                # Start the session's total at zero rather than at nothing.
                # A session that never completes a response captures no usage,
                # and a total of None means disconnect() reports nothing at
                # all, which leaves the record open and holding one of the
                # key's slots until the gateway's grace expires. Zero is also
                # the truth for a session that did no work.
                if self._broker_session_id:
                    self._session_usage = zero_realtime_usage()
            else:
                # No mint route: a plain proxy, or a gateway older than this
                # feature. Dial the vendor directly, and say so, because an
                # unbilled session otherwise looks identical to a billed one.
                # A refusal never reaches here: it raises inside the mint.
                warn_direct_dial_fallback(
                    self._mint.base_url, REALTIME_MINT_PATH, self._api_key_source
                )
        if socket_key == self._api_key and not self._api_key:
            raise RuntimeError(
                "OpenAIRealtimeAgentAdapter: no API key. Set OPENAI_API_KEY or "
                "pass `api_key=` to the constructor."
            )

        try:
            self._ws = await websockets.connect(
                self.url,
                additional_headers={
                    "Authorization": f"Bearer {socket_key}",
                },
            )
        except BaseException:
            # The mint booked a session and the socket never opened, so
            # nothing will ever report against it. Closing it at zero is the
            # truth: an ephemeral secret that opened no socket used nothing.
            # Leaving it open would hold one of the key's session slots until
            # the gateway's own window expires, and book a call that never
            # happened.
            await self._report_usage()
            raise
        logger.debug("OpenAIRealtimeAgentAdapter: connected to %s", self.url)

        # Configure session: audio formats, voice, instructions, tools.
        # Disable server-side VAD (session.audio.input.turn_detection=None) so
        # we control turn boundaries explicitly via input_audio_buffer.commit +
        # response.create after each send_audio.
        session_config: dict[str, Any] = {
            "type": "realtime",
            "audio": {
                "input": {
                    "format": {"type": "audio/pcm", "rate": 24000},
                    "turn_detection": None,
                    "transcription": {"model": OPENAI_STT_MODEL},
                },
                "output": {
                    "format": {"type": "audio/pcm", "rate": 24000},
                    "voice": self.voice,
                },
            },
        }
        if self.instructions:
            session_config["instructions"] = self.instructions
        if self.tools:
            session_config["tools"] = self.tools

        await self._ws.send(
            json.dumps({"type": "session.update", "session": session_config})
        )
        logger.debug("OpenAIRealtimeAgentAdapter: session.update sent")

        # Stamp realtime-specific attrs onto the active ``voice.adapter.connect``
        # span (opened by the executor connect loop). Base spans are name-owned;
        # the adapter contributes attributes, never a parallel span name (mirrors
        # ElevenLabs stamping ``voice.elevenlabs.agent_id``).
        from opentelemetry import trace as _otel_trace
        from .._telemetry import set_span_attributes

        set_span_attributes(
            _otel_trace.get_current_span(),
            {
                "voice.realtime.model": self.model,
                "voice.realtime.voice": self.voice,
                "voice.realtime.session_type": "realtime",
                "voice.realtime.tool_count": len(self.tools),
                "voice.realtime.brokered": self.brokered,
                "voice.realtime.session_id": self._broker_session_id or None,
            },
        )

    def _capture_usage(self, event: dict[str, Any]) -> None:
        """Add the usage OpenAI reports on ``response.done`` to the session.

        Read from every inbound event rather than from the audio branch,
        because a drain that ends on tail silence never reaches the terminal
        event and a session whose usage was never captured bills as
        cost-unknown.

        Summed rather than replaced. OpenAI reports usage per response, so a
        session that kept only the last ``response.done`` would bill for its
        last turn and nothing else.
        """
        if not self._broker_session_id:
            return
        if event.get("type") != "response.done":
            return
        response = event.get("response")
        if not isinstance(response, dict):
            return
        usage = response.get("usage")
        if isinstance(usage, dict):
            self._session_usage = accumulate_realtime_usage(
                self._session_usage, usage
            )

    async def _report_usage(self) -> None:
        """Close the session's spend record with what the socket measured.

        Also the failure path for a mint that succeeded and a socket that
        never opened, because the total starts at zero: the report is the same
        request either way, so there is one method rather than two that
        disagree.

        Clearing the captured usage first is what makes a second disconnect a
        no-op, so a session cannot be reported twice.
        """
        if (
            self._mint is None
            or not self._broker_session_id
            or not self._session_usage
        ):
            return
        usage = self._session_usage
        session_id = self._broker_session_id
        self._session_usage = None
        error = await report_openai_realtime_usage(self._mint, session_id, usage)
        if error is not None:
            logger.debug(
                "OpenAIRealtimeAgentAdapter: usage report failed: %r", error
            )

    async def disconnect(self) -> None:
        """Close the WebSocket if open, then close the session's spend record."""
        await self._report_usage()
        if self._ws is not None:
            try:
                await self._ws.close()
            except Exception:
                # Best-effort: connection may already be half-closed or in an
                # error state when disconnect() is called. We're tearing down
                # regardless — propagating here would just leak the WS reference.
                pass
            finally:
                self._ws = None
            # Issue #662 / reconnect hygiene (JS parity): clear response-lifecycle
            # guard state on teardown so a reused adapter instance does not carry a
            # stale active/deferred flag into its next connection.
            self._response_active = False
            self._deferred_response_create = False
            logger.debug("OpenAIRealtimeAgentAdapter: disconnected")

    # ------------------------------------------------------------------ I/O

    async def send_audio(self, chunk: AudioChunk) -> None:
        """
        Append a PCM16 audio chunk to the model's input audio buffer.

        Only emits ``input_audio_buffer.append`` — the commit + response are
        deferred to the next ``recv_audio`` call. The scenario executor may
        call ``send_audio`` many times for a single user turn (TTS streams
        audio as chunks); committing per-chunk would confuse the server with
        sub-second turn boundaries. By deferring commit to recv_audio, we
        get one server turn per user turn.
        """
        if self._ws is None:
            raise RuntimeError("OpenAIRealtimeAgentAdapter: not connected")
        b64 = base64.b64encode(chunk.data).decode()
        await self._ws.send(
            json.dumps({"type": "input_audio_buffer.append", "audio": b64})
        )
        self._pending_audio_bytes += len(chunk.data)

    async def interrupt(self) -> None:
        """Send ``response.cancel`` — the OpenAI Realtime API's first-class
        interrupt. The model stops generating audio and text immediately.
        No timing race against VAD: deterministic stop, then the next user
        turn flows normally through ``send_audio`` + ``recv_audio``.
        """
        if self._ws is None:
            raise RuntimeError("OpenAIRealtimeAgentAdapter: not connected")
        await self._ws.send(json.dumps({"type": "response.cancel"}))
        logger.debug("OpenAIRealtimeAgentAdapter: sent response.cancel (interrupt)")

    async def recv_audio(self, timeout: float) -> AudioChunk:
        """
        Commit any pending audio, request a response, and return the first
        audio chunk the model produces.

        If ``send_audio`` was called since the last ``recv_audio``, this
        method commits the buffer and emits ``response.create`` before
        awaiting the reply. Subsequent recv calls without new send calls
        just await the next audio delta (for multi-chunk responses).

        Loops over incoming events until a ``response.output_audio.delta``
        event arrives (GA name), then returns decoded PCM16. The legacy
        ``response.audio.delta`` name is accepted defensively with a one-time
        warning. Transcript events update the instance's
        ``last_user_transcript`` / ``last_agent_transcript`` attributes.
        An ``error`` event raises a ``RuntimeError``. All other housekeeping
        events are ignored and the loop continues.

        Raises:
            asyncio.TimeoutError: if no audio arrives within ``timeout``.
            RuntimeError: if the server sends an error event.
        """
        if self._ws is None:
            raise RuntimeError("OpenAIRealtimeAgentAdapter: not connected")

        # R2 markers/attrs land on the active ``voice.audio.receive`` span (opened
        # by the base drain that invokes recv_audio) via the ambient OTel context —
        # this method runs INSIDE that span. ``get_current_span().add_event`` is a
        # safe no-op on a non-recording span, so no extra guard is needed.
        from opentelemetry import trace as _otel_trace
        from .._telemetry import set_span_attributes

        # If send_audio was called since last recv, commit and request response.
        if self._pending_audio_bytes > 0:
            await self._ws.send(json.dumps({"type": "input_audio_buffer.commit"}))
            self._pending_audio_bytes = 0
            self._agent_turn_pending = False  # user spoke → per-turn signal consumed (always)
            if self._response_active:
                # Race guard (#657): server rejects/ignores a duplicate
                # response.create while a response is in flight, which causes the
                # recv loop to drain to timeout instead of completing.  Defer via
                # _deferred_response_create (distinct from _agent_turn_pending) so
                # the response.done handler fires response.create once the in-flight
                # response clears.
                self._deferred_response_create = True
            else:
                await self._ws.send(json.dumps({"type": "response.create"}))

        # Gap 1: agent-speaks-first / multi-turn agent initiation.
        # When the executor signals an agent turn via notify_agent_turn() and
        # no user audio has been committed and no response is already in flight,
        # send a bare content-free response.create so the model speaks.
        # Opening words come from the session instructions — no text is injected.
        # Self-limiting on user-first scripts: if user audio was committed above,
        # _agent_turn_pending was already cleared.
        # Also serves as the clean drain exit: if a response already completed
        # (_response_active is False after response.done cleared it) and
        # _agent_turn_pending is False, a drain re-entry returns an empty chunk.
        elif self._agent_turn_pending and not self._response_active:
            await self._ws.send(json.dumps({"type": "response.create"}))
            self._agent_turn_pending = False
            self._response_active = True

        elif not self._agent_turn_pending and not self._response_active and self._response_ever_active:
            # No pending audio, no agent-turn signal, no response in flight,
            # AND at least one response has already completed this session:
            # this is a drain re-entry after a completed response. Return empty
            # chunk so _drain_agent_response's tail-silence loop exits cleanly.
            # Guard on _response_ever_active so that a fresh recv_audio call
            # (before any response.created fires) does NOT short-circuit —
            # that would break direct recv_audio callers (e.g. unit tests that
            # call recv_audio without going through notify_agent_turn).
            return AudioChunk(data=b"")

        deadline = asyncio.get_running_loop().time() + timeout
        while True:
            remaining = deadline - asyncio.get_running_loop().time()
            if remaining <= 0:
                raise asyncio.TimeoutError(
                    "OpenAIRealtimeAgentAdapter: recv_audio timed out"
                )

            raw = await asyncio.wait_for(self._ws.recv(), timeout=remaining)
            try:
                event = json.loads(raw) if isinstance(raw, str) else json.loads(raw.decode())
            except Exception:
                logger.debug(
                    "OpenAIRealtimeAgentAdapter: non-JSON message, skipping"
                )
                continue

            etype = event.get("type", "")
            # Every inbound event passes here, which is why the usage capture
            # lives here and not in the terminal branch below.
            self._capture_usage(event)

            if etype in ("response.output_audio.delta", "response.audio.delta"):
                # Accept both the GA event name and its retired beta alias —
                # live gpt-realtime* models have been observed still emitting
                # the beta names. These legacy arms should be removed once the
                # GA names are confirmed stable at a live endpoint (issue #602).
                self._warn_if_legacy(etype, "response.output_audio.delta")
                b64 = event.get("delta", "")
                pcm = base64.b64decode(b64)
                # Enforce PCM16 invariant: even byte count.
                if len(pcm) % 2 == 1:
                    pcm = pcm[:-1]
                return AudioChunk(data=pcm)

            elif etype == "response.created":
                # Response is now in flight — mark it so subsequent recv_audio
                # drain re-entries don't fire a spurious second response.create.
                self._response_active = True
                self._response_ever_active = True
                # R2 marker: record the response start on the active receive span.
                _otel_trace.get_current_span().add_event(
                    "voice.realtime.response.created"
                )

            elif etype in (
                "response.output_audio_transcript.delta",
                "response.audio_transcript.delta",
            ):
                # Accumulate streaming agent transcript.
                self._warn_if_legacy(etype, "response.output_audio_transcript.delta")
                self._agent_transcript_buf += event.get("delta", "")

            elif etype in (
                "response.output_audio_transcript.done",
                "response.audio_transcript.done",
            ):
                # Finalise; the `transcript` field may have the full text.
                self._warn_if_legacy(etype, "response.output_audio_transcript.done")
                transcript = event.get("transcript", "")
                if transcript:
                    self.last_agent_transcript = transcript
                elif self._agent_transcript_buf:
                    self.last_agent_transcript = self._agent_transcript_buf
                self._agent_transcript_buf = ""

            elif etype in ("response.done", "response.cancelled"):
                # Response finished or was cancelled — mark it so the next
                # drain re-entry returns an empty chunk (clean exit).
                self._response_active = False
                # R2 markers: record the response terminal as a span event and
                # stamp the FINAL tool-call count onto the active
                # ``voice.audio.receive`` span. Done BEFORE the deferred re-fire and
                # the tool-only empty-chunk return below, so the count reflects THIS
                # response — set even when 0, for every response.done/.cancelled
                # actually observed. A drain that ends on tail-silence before a
                # terminal event is processed won't stamp it (pre-existing drain
                # timing, not new here).
                _otel_trace.get_current_span().add_event(f"voice.realtime.{etype}")
                set_span_attributes(
                    _otel_trace.get_current_span(),
                    {"voice.realtime.tool_call_count": len(self._completed_tool_calls)},
                )
                # Issue #657: if user audio was committed while a response was
                # in flight, _deferred_response_create was set as a deferral flag.
                # Now that _response_active is cleared, fire the deferred create.
                if self._deferred_response_create:
                    await self._ws.send(json.dumps({"type": "response.create"}))
                    self._deferred_response_create = False
                    self._response_active = True
                # Issue #646: a tool-only turn (function call, NO audio delta) would
                # otherwise loop here to the deadline and raise — the accumulated tool
                # call is parsed but never returned. When the response is done and at
                # least one tool call has been finalized this turn, return an empty
                # chunk so the drain exits cleanly and call() surfaces the tool_calls
                # message. A genuinely empty turn (done + EMPTY accumulator) must still
                # fall through to the timeout — the non-empty accumulator is the
                # discriminator, NOT response.done alone.
                # (Distinct from the drain-re-entry empty-chunk path above at the
                # _response_ever_active guard, which handles a COMPLETED prior response.)
                if self._completed_tool_calls:
                    return AudioChunk(data=b"")

            elif etype == "conversation.item.input_audio_transcription.completed":
                # User-side transcript from Whisper.
                self.last_user_transcript = event.get("transcript", "")

            elif etype == "response.function_call_arguments.delta":
                # Issue #630: streaming arguments fragment for a function call.
                # Accumulate per call_id; the call is finalized on `.done` or
                # the function_call output-item event.
                self._accumulate_tool_call_delta(
                    event.get("call_id", ""), event.get("delta", "")
                )

            elif etype == "response.function_call_arguments.done":
                # Issue #630: streaming-args path complete. `name` is typically
                # NOT on this event (it arrives via the output_item), so resolve
                # it from the accumulator. A missing call_id degrades safely.
                self._finalize_tool_call(
                    event.get("call_id"),
                    name=event.get("name"),
                    arguments=event.get("arguments"),
                )

            elif etype in ("response.output_item.added", "response.output_item.done"):
                # Issue #630: the output-item form of a function call carries the
                # authoritative `name` + `call_id` (and, on `.done`, the full
                # `arguments`). For non-function items (e.g. an audio message
                # item) this is benign housekeeping — fall through silently.
                item = event.get("item") or {}
                if isinstance(item, dict) and item.get("type") == "function_call":
                    call_id = item.get("call_id")
                    name = item.get("name")
                    if etype == "response.output_item.added":
                        # Shell arrives before args stream — record the name so a
                        # later delta/done can attach to it. Do not finalize yet.
                        self._note_tool_call_name(call_id or "", name)
                    else:
                        # `.done`: authoritative full call. Finalize (idempotent
                        # on call_id — merges with any streaming-args entry, AC6).
                        self._finalize_tool_call(
                            call_id, name=name, arguments=item.get("arguments")
                        )
                else:
                    logger.debug(
                        "OpenAIRealtimeAgentAdapter: ignoring non-function "
                        "output_item event %r",
                        etype,
                    )

            elif etype == "error":
                error_detail = event.get("error", {})
                msg = error_detail.get("message", str(error_detail))
                raise RuntimeError(
                    f"OpenAIRealtimeAgentAdapter: server error — {msg}"
                )

            else:
                # Housekeeping events — session.created, session.updated, etc. —
                # are benign. Log at DEBUG and keep the loop running.
                logger.debug(
                    "OpenAIRealtimeAgentAdapter: ignoring event type %r", etype
                )

    async def call(self, input: AgentInput) -> AgentReturnTypes:
        """Surface realtime tool calls alongside the spoken audio turn (#630).

        The base ``call()`` returns a single assistant audio message and does
        all the recording bookkeeping. We keep that intact and, when the agent
        called any tools this turn, append ONE extra assistant message carrying
        every tool call as OpenAI-chat ``tool_calls`` — the shape
        ``state.has_tool_call`` / ``state.last_tool_call`` consume.

        Returns:
            - the single audio message (dict) when no tools were called — byte
              identical to the base behaviour (AC8 regression), OR
            - ``[audio_message, tool_call_message]`` when ≥1 tool was called
              (AC1/AC9/AC10). ``convert_agent_return_types_to_openai_messages``
              passes a list of dicts through verbatim into result.messages.

        The completed-calls list is reset HERE (turn start) so tool calls never
        leak across turns; the function-call events for THIS turn are consumed
        inside the ``super().call()`` drain and finalized onto the list.
        """
        # Reset per-turn tool-call state so a prior turn's calls don't bleed
        # through (mirrors the transcript reset in _drain_agent_response).
        self._deferred_response_create = False
        self._completed_tool_calls = []
        self._tool_call_accumulators = {}

        audio_message = await super().call(input)

        if not self._completed_tool_calls:
            return audio_message

        # One assistant message carrying ALL of this turn's tool calls — the
        # conventional OpenAI shape (one assistant turn, many tool_calls).
        tool_call_message: ChatCompletionMessageParam = cast(
            ChatCompletionMessageParam,
            {
                "role": "assistant",
                "content": None,
                "tool_calls": list(self._completed_tool_calls),
            },
        )
        return [cast(ChatCompletionMessageParam, audio_message), tool_call_message]

    async def _drain_agent_response(
        self, on_first_chunk=None
    ) -> "AudioChunk":
        """Override to surface the spoken transcript after draining.

        The base class drains recv_audio chunks until tail silence. After
        draining, ``response.output_audio_transcript.done`` will have already
        fired (it arrives before ``response.done``), so ``self.last_agent_transcript``
        is populated. We rebuild the merged chunk with ``transcript=`` set so
        ``create_audio_message`` attaches a text part to the assistant message.

        This puts the transcript in ``result.messages`` (AC2/AC5) without
        modifying messages.py, tts.py, or composable.py (AC6).

        The transcript text is an ordinary ``{"type":"text","text":...}`` part —
        no extra keys. Echo-safety is handled in ``_strip_audio_content``
        (user_simulator_agent.py) by detecting the structural pattern: an
        assistant message that carries both an ``input_audio`` part AND a ``text``
        part is a voiced agent turn, so the text is reframed as third-person
        context before ``reverse_roles`` runs (AC4/AC11).
        """
        from ..audio_chunk import AudioChunk as _AudioChunk

        # Clear transcript state at the start of each turn so a stale value
        # from a prior turn doesn't bleed through if this turn's event never
        # fires (AC8: degraded case — absent transcript → no text part).
        self.last_agent_transcript = None
        self._agent_transcript_buf = ""

        merged = await super()._drain_agent_response(on_first_chunk=on_first_chunk)

        transcript = self.last_agent_transcript
        if transcript:
            # Rebuild with transcript attached so create_audio_message adds
            # the text part. The original merged.data is unchanged.
            return _AudioChunk(data=merged.data, transcript=transcript)
        # No transcript (AC8): return the merge unchanged (audio-only).
        return merged

    async def send_text(self, text: str) -> None:
        """
        Inject scripted text into the realtime session as a user message.

        Used when this adapter is the user simulator (role=USER): scripted
        ``user("text")`` steps route through here instead of spawning TTS.
        The model synthesises the text into spoken audio with natural prosody,
        which is then delivered via ``recv_audio``.

        NOTE: per §7.2, OpenAI Realtime cannot populate assistant audio
        messages retroactively; the downstream transcript reflects what the
        model actually emitted, not what was scripted.

        Raises:
            RuntimeError: if called before ``connect()``.
        """
        if self._ws is None:
            raise RuntimeError("OpenAIRealtimeAgentAdapter: not connected")

        # Create a user conversation item with the scripted text.
        await self._ws.send(
            json.dumps(
                {
                    "type": "conversation.item.create",
                    "item": {
                        "type": "message",
                        "role": "user",
                        "content": [{"type": "input_text", "text": text}],
                    },
                }
            )
        )
        # Prompt the model to generate audio output.
        # Guard: if a response is already in flight, defer the create rather than
        # firing unconditionally — mirrors the recv_audio user-audio branch guard.
        if self._response_active:
            self._deferred_response_create = True
            return
        await self._ws.send(json.dumps({"type": "response.create"}))
        logger.debug(
            "OpenAIRealtimeAgentAdapter: send_text injected text (%d chars)", len(text)
        )

Ancestors

• VoiceAgentAdapter
• AgentAdapter
• abc.ABC

Class variables

var capabilities : ClassVar[AdapterCapabilities]

Instance variables

var brokered : bool

Whether the live session was minted by a gateway, not the vendor.

Only true after connect(), and only when the mint answered with X-LangWatch-Session-Id. Configuration cannot set it, because the response is the one thing that cannot be told a lie about what answered.

Expand source code

@property
def brokered(self) -> bool:
    """Whether the live session was minted by a gateway, not the vendor.

    Only true after ``connect()``, and only when the mint answered with
    ``X-LangWatch-Session-Id``. Configuration cannot set it, because the
    response is the one thing that cannot be told a lie about what
    answered.
    """
    return self._broker_session_id != ""

var url : str

Expand source code

@property
def url(self) -> str:
    return REALTIME_URL_TEMPLATE.format(model=self.model)

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

Surface realtime tool calls alongside the spoken audio turn (#630).

The base call() returns a single assistant audio message and does all the recording bookkeeping. We keep that intact and, when the agent called any tools this turn, append ONE extra assistant message carrying every tool call as OpenAI-chat tool_calls — the shape state.has_tool_call / state.last_tool_call consume.

Returns

• the single audio message (dict) when no tools were called — byte identical to the base behaviour (AC8 regression), OR
• [audio_message, tool_call_message] when ≥1 tool was called (AC1/AC9/AC10). convert_agent_return_types_to_openai_messages passes a list of dicts through verbatim into result.messages. The completed-calls list is reset HERE (turn start) so tool calls never leak across turns; the function-call events for THIS turn are consumed inside the super().call() drain and finalized onto the list.

Expand source code

async def call(self, input: AgentInput) -> AgentReturnTypes:
    """Surface realtime tool calls alongside the spoken audio turn (#630).

    The base ``call()`` returns a single assistant audio message and does
    all the recording bookkeeping. We keep that intact and, when the agent
    called any tools this turn, append ONE extra assistant message carrying
    every tool call as OpenAI-chat ``tool_calls`` — the shape
    ``state.has_tool_call`` / ``state.last_tool_call`` consume.

    Returns:
        - the single audio message (dict) when no tools were called — byte
          identical to the base behaviour (AC8 regression), OR
        - ``[audio_message, tool_call_message]`` when ≥1 tool was called
          (AC1/AC9/AC10). ``convert_agent_return_types_to_openai_messages``
          passes a list of dicts through verbatim into result.messages.

    The completed-calls list is reset HERE (turn start) so tool calls never
    leak across turns; the function-call events for THIS turn are consumed
    inside the ``super().call()`` drain and finalized onto the list.
    """
    # Reset per-turn tool-call state so a prior turn's calls don't bleed
    # through (mirrors the transcript reset in _drain_agent_response).
    self._deferred_response_create = False
    self._completed_tool_calls = []
    self._tool_call_accumulators = {}

    audio_message = await super().call(input)

    if not self._completed_tool_calls:
        return audio_message

    # One assistant message carrying ALL of this turn's tool calls — the
    # conventional OpenAI shape (one assistant turn, many tool_calls).
    tool_call_message: ChatCompletionMessageParam = cast(
        ChatCompletionMessageParam,
        {
            "role": "assistant",
            "content": None,
            "tool_calls": list(self._completed_tool_calls),
        },
    )
    return [cast(ChatCompletionMessageParam, audio_message), tool_call_message]

async def connect(self) ‑> None

Open the Realtime WebSocket and send the initial session.update.

Expand source code

async def connect(self) -> None:
    """Open the Realtime WebSocket and send the initial session.update."""
    import websockets

    # Issue #662 / reconnect hygiene (JS parity — openai-realtime.ts connect()):
    # a reused adapter instance must start each connection with a clean slate.
    # Clearing _response_active matters most: a disconnect mid-response would
    # otherwise leave it True, and the next turn's user-audio commit would take
    # the defer branch forever (no response.done ever arrives to fire it), draining
    # recv_audio to timeout.
    self._response_active = False
    self._deferred_response_create = False

    self._broker_session_id = ""
    self._session_usage = None

    # The socket's bearer is the ephemeral secret the mint returned: a
    # credential for exactly this call, so no long-lived key reaches the
    # socket. Whether OpenAI or a gateway answers the mint is a property
    # of OPENAI_BASE_URL, not of this adapter.
    socket_key = self._api_key
    if self._mint is not None:
        result = await mint_openai_realtime_session(self._mint, self.model)
        if result.minted:
            socket_key = result.credential
            self._broker_session_id = result.session_id
            # Start the session's total at zero rather than at nothing.
            # A session that never completes a response captures no usage,
            # and a total of None means disconnect() reports nothing at
            # all, which leaves the record open and holding one of the
            # key's slots until the gateway's grace expires. Zero is also
            # the truth for a session that did no work.
            if self._broker_session_id:
                self._session_usage = zero_realtime_usage()
        else:
            # No mint route: a plain proxy, or a gateway older than this
            # feature. Dial the vendor directly, and say so, because an
            # unbilled session otherwise looks identical to a billed one.
            # A refusal never reaches here: it raises inside the mint.
            warn_direct_dial_fallback(
                self._mint.base_url, REALTIME_MINT_PATH, self._api_key_source
            )
    if socket_key == self._api_key and not self._api_key:
        raise RuntimeError(
            "OpenAIRealtimeAgentAdapter: no API key. Set OPENAI_API_KEY or "
            "pass `api_key=` to the constructor."
        )

    try:
        self._ws = await websockets.connect(
            self.url,
            additional_headers={
                "Authorization": f"Bearer {socket_key}",
            },
        )
    except BaseException:
        # The mint booked a session and the socket never opened, so
        # nothing will ever report against it. Closing it at zero is the
        # truth: an ephemeral secret that opened no socket used nothing.
        # Leaving it open would hold one of the key's session slots until
        # the gateway's own window expires, and book a call that never
        # happened.
        await self._report_usage()
        raise
    logger.debug("OpenAIRealtimeAgentAdapter: connected to %s", self.url)

    # Configure session: audio formats, voice, instructions, tools.
    # Disable server-side VAD (session.audio.input.turn_detection=None) so
    # we control turn boundaries explicitly via input_audio_buffer.commit +
    # response.create after each send_audio.
    session_config: dict[str, Any] = {
        "type": "realtime",
        "audio": {
            "input": {
                "format": {"type": "audio/pcm", "rate": 24000},
                "turn_detection": None,
                "transcription": {"model": OPENAI_STT_MODEL},
            },
            "output": {
                "format": {"type": "audio/pcm", "rate": 24000},
                "voice": self.voice,
            },
        },
    }
    if self.instructions:
        session_config["instructions"] = self.instructions
    if self.tools:
        session_config["tools"] = self.tools

    await self._ws.send(
        json.dumps({"type": "session.update", "session": session_config})
    )
    logger.debug("OpenAIRealtimeAgentAdapter: session.update sent")

    # Stamp realtime-specific attrs onto the active ``voice.adapter.connect``
    # span (opened by the executor connect loop). Base spans are name-owned;
    # the adapter contributes attributes, never a parallel span name (mirrors
    # ElevenLabs stamping ``voice.elevenlabs.agent_id``).
    from opentelemetry import trace as _otel_trace
    from .._telemetry import set_span_attributes

    set_span_attributes(
        _otel_trace.get_current_span(),
        {
            "voice.realtime.model": self.model,
            "voice.realtime.voice": self.voice,
            "voice.realtime.session_type": "realtime",
            "voice.realtime.tool_count": len(self.tools),
            "voice.realtime.brokered": self.brokered,
            "voice.realtime.session_id": self._broker_session_id or None,
        },
    )

async def disconnect(self) ‑> None

Close the WebSocket if open, then close the session's spend record.

Expand source code

async def disconnect(self) -> None:
    """Close the WebSocket if open, then close the session's spend record."""
    await self._report_usage()
    if self._ws is not None:
        try:
            await self._ws.close()
        except Exception:
            # Best-effort: connection may already be half-closed or in an
            # error state when disconnect() is called. We're tearing down
            # regardless — propagating here would just leak the WS reference.
            pass
        finally:
            self._ws = None
        # Issue #662 / reconnect hygiene (JS parity): clear response-lifecycle
        # guard state on teardown so a reused adapter instance does not carry a
        # stale active/deferred flag into its next connection.
        self._response_active = False
        self._deferred_response_create = False
        logger.debug("OpenAIRealtimeAgentAdapter: disconnected")

async def interrupt(self) ‑> None

Send response.cancel — the OpenAI Realtime API's first-class interrupt. The model stops generating audio and text immediately. No timing race against VAD: deterministic stop, then the next user turn flows normally through send_audio + recv_audio.

Expand source code

async def interrupt(self) -> None:
    """Send ``response.cancel`` — the OpenAI Realtime API's first-class
    interrupt. The model stops generating audio and text immediately.
    No timing race against VAD: deterministic stop, then the next user
    turn flows normally through ``send_audio`` + ``recv_audio``.
    """
    if self._ws is None:
        raise RuntimeError("OpenAIRealtimeAgentAdapter: not connected")
    await self._ws.send(json.dumps({"type": "response.cancel"}))
    logger.debug("OpenAIRealtimeAgentAdapter: sent response.cancel (interrupt)")

def notify_agent_turn(self) ‑> None

Signal that an agent turn is about to be dispatched.

Called by the executor before each agent step so recv_audio can fire a bare response.create for agent-initiated turns (where no user audio has been committed). This per-turn signal handles both turn 1 (opening) and subsequent agent turns in multi-turn scripts like [agent(), user(), agent()].

Only meaningful when role=AGENT. Safe to call on every agent step — recv_audio consumes and clears the flag.

Expand source code

def notify_agent_turn(self) -> None:
    """Signal that an agent turn is about to be dispatched.

    Called by the executor before each agent step so recv_audio can fire
    a bare response.create for agent-initiated turns (where no user audio
    has been committed). This per-turn signal handles both turn 1 (opening)
    and subsequent agent turns in multi-turn scripts like [agent(), user(),
    agent()].

    Only meaningful when role=AGENT. Safe to call on every agent step —
    recv_audio consumes and clears the flag.
    """
    if self.role == AgentRole.AGENT:
        self._agent_turn_pending = True

async def recv_audio(self, timeout: float) ‑> AudioChunk

Commit any pending audio, request a response, and return the first audio chunk the model produces.

If send_audio was called since the last recv_audio, this method commits the buffer and emits response.create before awaiting the reply. Subsequent recv calls without new send calls just await the next audio delta (for multi-chunk responses).

Loops over incoming events until a response.output_audio.delta event arrives (GA name), then returns decoded PCM16. The legacy response.audio.delta name is accepted defensively with a one-time warning. Transcript events update the instance's last_user_transcript / last_agent_transcript attributes. An error event raises a RuntimeError. All other housekeeping events are ignored and the loop continues.

Raises

asyncio.TimeoutError

if no audio arrives within timeout.

RuntimeError

if the server sends an error event.

Expand source code

async def recv_audio(self, timeout: float) -> AudioChunk:
    """
    Commit any pending audio, request a response, and return the first
    audio chunk the model produces.

    If ``send_audio`` was called since the last ``recv_audio``, this
    method commits the buffer and emits ``response.create`` before
    awaiting the reply. Subsequent recv calls without new send calls
    just await the next audio delta (for multi-chunk responses).

    Loops over incoming events until a ``response.output_audio.delta``
    event arrives (GA name), then returns decoded PCM16. The legacy
    ``response.audio.delta`` name is accepted defensively with a one-time
    warning. Transcript events update the instance's
    ``last_user_transcript`` / ``last_agent_transcript`` attributes.
    An ``error`` event raises a ``RuntimeError``. All other housekeeping
    events are ignored and the loop continues.

    Raises:
        asyncio.TimeoutError: if no audio arrives within ``timeout``.
        RuntimeError: if the server sends an error event.
    """
    if self._ws is None:
        raise RuntimeError("OpenAIRealtimeAgentAdapter: not connected")

    # R2 markers/attrs land on the active ``voice.audio.receive`` span (opened
    # by the base drain that invokes recv_audio) via the ambient OTel context —
    # this method runs INSIDE that span. ``get_current_span().add_event`` is a
    # safe no-op on a non-recording span, so no extra guard is needed.
    from opentelemetry import trace as _otel_trace
    from .._telemetry import set_span_attributes

    # If send_audio was called since last recv, commit and request response.
    if self._pending_audio_bytes > 0:
        await self._ws.send(json.dumps({"type": "input_audio_buffer.commit"}))
        self._pending_audio_bytes = 0
        self._agent_turn_pending = False  # user spoke → per-turn signal consumed (always)
        if self._response_active:
            # Race guard (#657): server rejects/ignores a duplicate
            # response.create while a response is in flight, which causes the
            # recv loop to drain to timeout instead of completing.  Defer via
            # _deferred_response_create (distinct from _agent_turn_pending) so
            # the response.done handler fires response.create once the in-flight
            # response clears.
            self._deferred_response_create = True
        else:
            await self._ws.send(json.dumps({"type": "response.create"}))

    # Gap 1: agent-speaks-first / multi-turn agent initiation.
    # When the executor signals an agent turn via notify_agent_turn() and
    # no user audio has been committed and no response is already in flight,
    # send a bare content-free response.create so the model speaks.
    # Opening words come from the session instructions — no text is injected.
    # Self-limiting on user-first scripts: if user audio was committed above,
    # _agent_turn_pending was already cleared.
    # Also serves as the clean drain exit: if a response already completed
    # (_response_active is False after response.done cleared it) and
    # _agent_turn_pending is False, a drain re-entry returns an empty chunk.
    elif self._agent_turn_pending and not self._response_active:
        await self._ws.send(json.dumps({"type": "response.create"}))
        self._agent_turn_pending = False
        self._response_active = True

    elif not self._agent_turn_pending and not self._response_active and self._response_ever_active:
        # No pending audio, no agent-turn signal, no response in flight,
        # AND at least one response has already completed this session:
        # this is a drain re-entry after a completed response. Return empty
        # chunk so _drain_agent_response's tail-silence loop exits cleanly.
        # Guard on _response_ever_active so that a fresh recv_audio call
        # (before any response.created fires) does NOT short-circuit —
        # that would break direct recv_audio callers (e.g. unit tests that
        # call recv_audio without going through notify_agent_turn).
        return AudioChunk(data=b"")

    deadline = asyncio.get_running_loop().time() + timeout
    while True:
        remaining = deadline - asyncio.get_running_loop().time()
        if remaining <= 0:
            raise asyncio.TimeoutError(
                "OpenAIRealtimeAgentAdapter: recv_audio timed out"
            )

        raw = await asyncio.wait_for(self._ws.recv(), timeout=remaining)
        try:
            event = json.loads(raw) if isinstance(raw, str) else json.loads(raw.decode())
        except Exception:
            logger.debug(
                "OpenAIRealtimeAgentAdapter: non-JSON message, skipping"
            )
            continue

        etype = event.get("type", "")
        # Every inbound event passes here, which is why the usage capture
        # lives here and not in the terminal branch below.
        self._capture_usage(event)

        if etype in ("response.output_audio.delta", "response.audio.delta"):
            # Accept both the GA event name and its retired beta alias —
            # live gpt-realtime* models have been observed still emitting
            # the beta names. These legacy arms should be removed once the
            # GA names are confirmed stable at a live endpoint (issue #602).
            self._warn_if_legacy(etype, "response.output_audio.delta")
            b64 = event.get("delta", "")
            pcm = base64.b64decode(b64)
            # Enforce PCM16 invariant: even byte count.
            if len(pcm) % 2 == 1:
                pcm = pcm[:-1]
            return AudioChunk(data=pcm)

        elif etype == "response.created":
            # Response is now in flight — mark it so subsequent recv_audio
            # drain re-entries don't fire a spurious second response.create.
            self._response_active = True
            self._response_ever_active = True
            # R2 marker: record the response start on the active receive span.
            _otel_trace.get_current_span().add_event(
                "voice.realtime.response.created"
            )

        elif etype in (
            "response.output_audio_transcript.delta",
            "response.audio_transcript.delta",
        ):
            # Accumulate streaming agent transcript.
            self._warn_if_legacy(etype, "response.output_audio_transcript.delta")
            self._agent_transcript_buf += event.get("delta", "")

        elif etype in (
            "response.output_audio_transcript.done",
            "response.audio_transcript.done",
        ):
            # Finalise; the `transcript` field may have the full text.
            self._warn_if_legacy(etype, "response.output_audio_transcript.done")
            transcript = event.get("transcript", "")
            if transcript:
                self.last_agent_transcript = transcript
            elif self._agent_transcript_buf:
                self.last_agent_transcript = self._agent_transcript_buf
            self._agent_transcript_buf = ""

        elif etype in ("response.done", "response.cancelled"):
            # Response finished or was cancelled — mark it so the next
            # drain re-entry returns an empty chunk (clean exit).
            self._response_active = False
            # R2 markers: record the response terminal as a span event and
            # stamp the FINAL tool-call count onto the active
            # ``voice.audio.receive`` span. Done BEFORE the deferred re-fire and
            # the tool-only empty-chunk return below, so the count reflects THIS
            # response — set even when 0, for every response.done/.cancelled
            # actually observed. A drain that ends on tail-silence before a
            # terminal event is processed won't stamp it (pre-existing drain
            # timing, not new here).
            _otel_trace.get_current_span().add_event(f"voice.realtime.{etype}")
            set_span_attributes(
                _otel_trace.get_current_span(),
                {"voice.realtime.tool_call_count": len(self._completed_tool_calls)},
            )
            # Issue #657: if user audio was committed while a response was
            # in flight, _deferred_response_create was set as a deferral flag.
            # Now that _response_active is cleared, fire the deferred create.
            if self._deferred_response_create:
                await self._ws.send(json.dumps({"type": "response.create"}))
                self._deferred_response_create = False
                self._response_active = True
            # Issue #646: a tool-only turn (function call, NO audio delta) would
            # otherwise loop here to the deadline and raise — the accumulated tool
            # call is parsed but never returned. When the response is done and at
            # least one tool call has been finalized this turn, return an empty
            # chunk so the drain exits cleanly and call() surfaces the tool_calls
            # message. A genuinely empty turn (done + EMPTY accumulator) must still
            # fall through to the timeout — the non-empty accumulator is the
            # discriminator, NOT response.done alone.
            # (Distinct from the drain-re-entry empty-chunk path above at the
            # _response_ever_active guard, which handles a COMPLETED prior response.)
            if self._completed_tool_calls:
                return AudioChunk(data=b"")

        elif etype == "conversation.item.input_audio_transcription.completed":
            # User-side transcript from Whisper.
            self.last_user_transcript = event.get("transcript", "")

        elif etype == "response.function_call_arguments.delta":
            # Issue #630: streaming arguments fragment for a function call.
            # Accumulate per call_id; the call is finalized on `.done` or
            # the function_call output-item event.
            self._accumulate_tool_call_delta(
                event.get("call_id", ""), event.get("delta", "")
            )

        elif etype == "response.function_call_arguments.done":
            # Issue #630: streaming-args path complete. `name` is typically
            # NOT on this event (it arrives via the output_item), so resolve
            # it from the accumulator. A missing call_id degrades safely.
            self._finalize_tool_call(
                event.get("call_id"),
                name=event.get("name"),
                arguments=event.get("arguments"),
            )

        elif etype in ("response.output_item.added", "response.output_item.done"):
            # Issue #630: the output-item form of a function call carries the
            # authoritative `name` + `call_id` (and, on `.done`, the full
            # `arguments`). For non-function items (e.g. an audio message
            # item) this is benign housekeeping — fall through silently.
            item = event.get("item") or {}
            if isinstance(item, dict) and item.get("type") == "function_call":
                call_id = item.get("call_id")
                name = item.get("name")
                if etype == "response.output_item.added":
                    # Shell arrives before args stream — record the name so a
                    # later delta/done can attach to it. Do not finalize yet.
                    self._note_tool_call_name(call_id or "", name)
                else:
                    # `.done`: authoritative full call. Finalize (idempotent
                    # on call_id — merges with any streaming-args entry, AC6).
                    self._finalize_tool_call(
                        call_id, name=name, arguments=item.get("arguments")
                    )
            else:
                logger.debug(
                    "OpenAIRealtimeAgentAdapter: ignoring non-function "
                    "output_item event %r",
                    etype,
                )

        elif etype == "error":
            error_detail = event.get("error", {})
            msg = error_detail.get("message", str(error_detail))
            raise RuntimeError(
                f"OpenAIRealtimeAgentAdapter: server error — {msg}"
            )

        else:
            # Housekeeping events — session.created, session.updated, etc. —
            # are benign. Log at DEBUG and keep the loop running.
            logger.debug(
                "OpenAIRealtimeAgentAdapter: ignoring event type %r", etype
            )

async def send_audio(self, chunk: AudioChunk) ‑> None

Append a PCM16 audio chunk to the model's input audio buffer.

Only emits input_audio_buffer.append — the commit + response are deferred to the next recv_audio call. The scenario executor may call send_audio many times for a single user turn (TTS streams audio as chunks); committing per-chunk would confuse the server with sub-second turn boundaries. By deferring commit to recv_audio, we get one server turn per user turn.

Expand source code

async def send_audio(self, chunk: AudioChunk) -> None:
    """
    Append a PCM16 audio chunk to the model's input audio buffer.

    Only emits ``input_audio_buffer.append`` — the commit + response are
    deferred to the next ``recv_audio`` call. The scenario executor may
    call ``send_audio`` many times for a single user turn (TTS streams
    audio as chunks); committing per-chunk would confuse the server with
    sub-second turn boundaries. By deferring commit to recv_audio, we
    get one server turn per user turn.
    """
    if self._ws is None:
        raise RuntimeError("OpenAIRealtimeAgentAdapter: not connected")
    b64 = base64.b64encode(chunk.data).decode()
    await self._ws.send(
        json.dumps({"type": "input_audio_buffer.append", "audio": b64})
    )
    self._pending_audio_bytes += len(chunk.data)

async def send_text(self, text: str) ‑> None

Inject scripted text into the realtime session as a user message.

Used when this adapter is the user simulator (role=USER): scripted user("text") steps route through here instead of spawning TTS. The model synthesises the text into spoken audio with natural prosody, which is then delivered via recv_audio.

NOTE: per §7.2, OpenAI Realtime cannot populate assistant audio messages retroactively; the downstream transcript reflects what the model actually emitted, not what was scripted.

Raises

RuntimeError

if called before connect().

Expand source code

async def send_text(self, text: str) -> None:
    """
    Inject scripted text into the realtime session as a user message.

    Used when this adapter is the user simulator (role=USER): scripted
    ``user("text")`` steps route through here instead of spawning TTS.
    The model synthesises the text into spoken audio with natural prosody,
    which is then delivered via ``recv_audio``.

    NOTE: per §7.2, OpenAI Realtime cannot populate assistant audio
    messages retroactively; the downstream transcript reflects what the
    model actually emitted, not what was scripted.

    Raises:
        RuntimeError: if called before ``connect()``.
    """
    if self._ws is None:
        raise RuntimeError("OpenAIRealtimeAgentAdapter: not connected")

    # Create a user conversation item with the scripted text.
    await self._ws.send(
        json.dumps(
            {
                "type": "conversation.item.create",
                "item": {
                    "type": "message",
                    "role": "user",
                    "content": [{"type": "input_text", "text": text}],
                },
            }
        )
    )
    # Prompt the model to generate audio output.
    # Guard: if a response is already in flight, defer the create rather than
    # firing unconditionally — mirrors the recv_audio user-audio branch guard.
    if self._response_active:
        self._deferred_response_create = True
        return
    await self._ws.send(json.dumps({"type": "response.create"}))
    logger.debug(
        "OpenAIRealtimeAgentAdapter: send_text injected text (%d chars)", len(text)
    )

Inherited members

• VoiceAgentAdapter:
  • agent_hung_up
  • is_connected

class PendingTransportError (adapter_name: str)

Raised by stub adapters when their transport code has not landed yet.

Expand source code

class PendingTransportError(NotImplementedError):
    """Raised by stub adapters when their transport code has not landed yet."""

    def __init__(self, adapter_name: str) -> None:
        super().__init__(
            f"{adapter_name}: transport implementation is not yet wired up. "
            "Options: (1) run this scenario as an @integration test against a "
            f"live endpoint, (2) subclass {adapter_name} and implement "
            "send_audio/recv_audio — and re-audit the inherited "
            "`capabilities` ClassVar so the matrix matches what your subclass "
            "can actually do. Claiming streaming_transcripts=True in a "
            "subclass without a real transcript stream will silently break "
            "after_words interruption."
        )
        self.adapter_name = adapter_name

Ancestors

• builtins.NotImplementedError
• builtins.RuntimeError
• builtins.Exception
• builtins.BaseException

Subclasses

• scenario.voice.adapters._stub.TransportNotConnectedError

class PipecatAgentAdapter (url: Optional[str] = None, *, signaling_url: Optional[str] = None, transport: "Literal['scenario.voice.adapters.websocket', 'scenario.voice.adapters.webrtc']" = 'websocket', audio_format: str = 'mulaw', sample_rate: int = 8000, stream_sid: Optional[str] = None, call_sid: Optional[str] = None)

Test a running Pipecat bot via its exposed WebSocket endpoint.

Transport is selected by the transport argument: - "websocket" (default): Twilio Media Streams protocol over WS. Scenario sends a synthetic start event, then media frames. Pipecat's TwilioFrameSerializer on the bot side handles the wire format. - "webrtc": SmallWebRTC-style negotiation. Raises PendingTransportError; tracked as a follow-up.

Expand source code

class PipecatAgentAdapter(VoiceAgentAdapter):
    """
    Test a running Pipecat bot via its exposed WebSocket endpoint.

    Transport is selected by the ``transport`` argument:
        - ``"websocket"`` (default): Twilio Media Streams protocol over WS.
          Scenario sends a synthetic ``start`` event, then ``media`` frames.
          Pipecat's ``TwilioFrameSerializer`` on the bot side handles the
          wire format.
        - ``"webrtc"``: SmallWebRTC-style negotiation. Raises
          ``PendingTransportError``; tracked as a follow-up.
    """

    capabilities: ClassVar[AdapterCapabilities] = AdapterCapabilities(
        # The Twilio Media Streams wire protocol carries audio frames only,
        # never transcript events. Advertising streaming_transcripts made
        # interrupt(after_words=N) poll a ``streaming_transcript`` attribute
        # this adapter never populates, hanging the script step forever.
        # Turn-level transcripts come from the base call()'s runtime STT.
        streaming_transcripts=False,
        native_vad=True,
        dtmf=False,
        # Pipecat over the Twilio WS transport speaks the Twilio Media Streams
        # protocol; the ``clear`` event drops all buffered outbound audio on
        # the bot side. That's first-class interrupt — no VAD timing race.
        interruption=True,
        input_formats=["pcm16/24000", "mulaw/8000", "opus"],
        output_formats=["pcm16/24000", "mulaw/8000", "opus"],
    )

    def __init__(
        self,
        url: Optional[str] = None,
        *,
        signaling_url: Optional[str] = None,
        transport: Literal["websocket", "webrtc"] = "websocket",
        audio_format: str = "mulaw",
        sample_rate: int = 8000,
        stream_sid: Optional[str] = None,
        call_sid: Optional[str] = None,
    ) -> None:
        super().__init__()
        if transport == "websocket" and url is None:
            raise ValueError("PipecatAgentAdapter(transport='websocket') requires url=")
        if transport == "webrtc" and signaling_url is None:
            raise ValueError("PipecatAgentAdapter(transport='webrtc') requires signaling_url=")

        self.url = url
        self.signaling_url = signaling_url
        self.transport = transport
        self.audio_format = audio_format
        self.sample_rate = sample_rate
        # Synthetic SIDs pipecat's TwilioFrameSerializer needs in the `start`
        # event. If caller doesn't supply them, we fabricate UUIDs. Pipecat
        # uses them for logging and the auto-hangup REST call; both are no-ops
        # when we're not actually going through Twilio.
        self.stream_sid = stream_sid
        self.call_sid = call_sid

        self._ws: Any = None
        self._recv_task: Optional[asyncio.Task] = None
        # Carries AudioChunks plus the _RECV_LOOP_DONE sentinel (hence Any), so a
        # waiting recv_audio learns the loop ended instead of blocking forever.
        self._inbound_queue: Optional[asyncio.Queue[Any]] = None
        # Set by _recv_loop when it crashes; recv_audio reads it to name the root
        # cause (chained via __cause__) on the PipecatRecvError it raises (#498).
        self._recv_loop_exc: Optional[BaseException] = None
        # Set True when _recv_loop terminates (crash or clean close). Lets
        # recv_audio fail fast on a drained queue without re-reading it (#498).
        self._recv_loop_done: bool = False
        # The turn context we last emitted a background ``voice.audio.receive``
        # span for — so ``_deliver`` spans only the FIRST wire delivery of each
        # turn (one span/turn, no per-100ms-chunk flood). Reset implicitly: the
        # next turn publishes a distinct ``_voice_turn_context`` object (#774).
        self._bg_span_turn_context: Optional[Context] = None
        # Serialises concurrent send_audio() calls — without it two paced
        # senders would interleave 20-ms mulaw frames on the wire and the
        # bot would receive corrupted audio. Used for the interruption case
        # where the executor calls send_audio() while a previous turn's
        # send is still in flight.
        self._send_lock: Optional[asyncio.Lock] = None

    @property
    def transport_format(self) -> str:
        return f"{self.audio_format}/{self.sample_rate}"

    # ------------------------------------------------------------------ lifecycle

    async def connect(self) -> None:
        if self.transport == "webrtc":
            from ._stub import PendingTransportError

            raise PendingTransportError(
                "PipecatAgentAdapter(transport='webrtc')"
            )

        # Lazy import so `import scenario` doesn't require websockets at the
        # top of the module-load path (it's already a hard dep, but being
        # consistent with the Twilio adapter style).
        import websockets

        assert self.url is not None  # validated in __init__
        self._ws = await websockets.connect(
            self.url, ping_interval=None, ping_timeout=None
        )
        self._inbound_queue = asyncio.Queue()
        self._recv_loop_exc = None  # reset per fresh connection
        self._recv_loop_done = False
        self._send_lock = asyncio.Lock()

        # Send the synthetic `start` event that pipecat's TwilioFrameSerializer
        # requires to learn the stream/call SIDs and start deserializing
        # media frames.
        if self.stream_sid is None:
            self.stream_sid = f"MZ{uuid.uuid4().hex}"
        if self.call_sid is None:
            self.call_sid = f"CA{uuid.uuid4().hex}"

        await self._ws.send(json.dumps({"event": "connected", "protocol": "Call", "version": "1.0.0"}))
        await self._ws.send(
            json.dumps(
                {
                    "event": "start",
                    "streamSid": self.stream_sid,
                    "start": {
                        "streamSid": self.stream_sid,
                        "callSid": self.call_sid,
                        "mediaFormat": {
                            "encoding": "audio/x-mulaw",
                            "sampleRate": 8000,
                            "channels": 1,
                        },
                    },
                }
            )
        )

        # Stamp Pipecat transport attrs onto the active ``voice.adapter.connect``
        # span (opened by the executor connect loop). Base spans are name-owned;
        # the adapter contributes attributes, never a parallel span name — mirror
        # ElevenLabs' ``voice.elevenlabs.agent_id`` seam (``adapters/elevenlabs.py``).
        from opentelemetry import trace as _otel_trace
        from .._telemetry import set_span_attributes

        set_span_attributes(
            _otel_trace.get_current_span(),
            {
                "voice.pipecat.transport": self.transport,
                "voice.pipecat.transport_format": self.transport_format,
            },
        )

        self._recv_task = asyncio.create_task(self._recv_loop())
        logger.debug("PipecatAgentAdapter: connected to %s (stream=%s)", self.url, self.stream_sid)

    async def disconnect(self) -> None:
        ws = self._ws
        if ws is None:
            return

        # Send `stop` event so the bot can clean up its pipeline gracefully.
        try:
            if self.stream_sid:
                await ws.send(json.dumps({"event": "stop", "streamSid": self.stream_sid}))
        except Exception:
            logger.debug("PipecatAgentAdapter: failed to send stop frame", exc_info=True)

        if self._recv_task is not None:
            self._recv_task.cancel()
            try:
                await self._recv_task
            except asyncio.CancelledError:
                # Expected: we just cancelled it.
                pass
            except Exception:
                # Unexpected teardown error — already logging enough context
                # elsewhere; disconnect() is best-effort.
                logger.debug("PipecatAgentAdapter: recv_task raised during cancel", exc_info=True)
            self._recv_task = None

        try:
            await ws.close()
        except Exception:
            # WS may already be closed by the peer; disconnect() is best-effort.
            logger.debug("PipecatAgentAdapter: ws.close() raised", exc_info=True)

        self._ws = None
        self._inbound_queue = None
        self.stream_sid = None
        self.call_sid = None

    # ------------------------------------------------------------------ I/O

    async def send_audio(self, chunk: AudioChunk) -> None:
        # Pace at real-time (TWILIO_FRAME_MS/1000s per 20-ms frame). Matches what
        # a real caller produces over a PSTN line — the SUT sees normal speech
        # rhythm, not a synthetic dump.
        #
        # After the last frame we send a Twilio ``mark`` named "utterance_end".
        # Real-time pacing means TTS-induced inter-phrase pauses survive on the
        # wire, and a stateless inactivity-timer on the receiver can't
        # distinguish "speaker paused after a comma" from "speaker finished
        # their turn." The mark is an explicit, non-ambiguous end-of-turn
        # signal: cooperating SUTs flush on the mark; legacy SUTs fall back to
        # VAD timing.
        self._assert_connected()
        assert self._ws is not None and self.stream_sid is not None and self._send_lock is not None
        mulaw = pcm16_24k_to_mulaw8k(chunk.data)
        frame_secs = TWILIO_FRAME_MS / 1000
        async with self._send_lock:
            for frame in iter_mulaw_frames(mulaw):
                if not frame:
                    continue
                await self._ws.send(build_media_frame(self.stream_sid, frame))
                await asyncio.sleep(frame_secs)
            await self._ws.send(build_mark_frame(self.stream_sid, "utterance_end"))

    async def recv_audio(self, timeout: float) -> AudioChunk:
        self._assert_connected()
        assert self._inbound_queue is not None
        # The loop already terminated and its queue is drained → fail fast with
        # the terminal cause instead of blocking for `timeout` on a queue nothing
        # will fill. The flag + drained-queue check is the terminal state for the
        # rest of this connection (no await, no re-pushed sentinel to leak).
        if self._recv_loop_done and self._inbound_queue.empty():
            raise self._recv_loop_ended_error() from self._recv_loop_exc
        item = await asyncio.wait_for(self._inbound_queue.get(), timeout=timeout)
        if item is _RECV_LOOP_DONE:
            raise self._recv_loop_ended_error() from self._recv_loop_exc
        return item

    def _recv_loop_ended_error(self) -> PipecatRecvError:
        # Chaining is done at the raise site (``raise ... from self._recv_loop_exc``)
        # so ``__suppress_context__`` is set correctly and the clean-close branch
        # (exc is None → ``from None``) gets a true empty cause.
        exc = self._recv_loop_exc
        if exc is not None:
            return PipecatRecvError(
                "pipecat recv loop crashed; no further audio will arrive: "
                f"{type(exc).__name__}: {exc}"
            )
        return PipecatRecvError(
            "pipecat bot closed the WebSocket; no further audio will arrive — the "
            "bot hung up or its pipeline stopped without responding"
        )

    async def interrupt(self) -> None:
        """Send a Twilio ``clear`` frame — the bot drops all buffered outbound
        audio immediately. Cooperating Pipecat bots (and any code wired to
        the Media Streams protocol) treat ``clear`` as "stop talking now."
        Use this in preference to timing-based barge-in when the SUT
        supports it: it's deterministic, doesn't depend on VAD detection
        windows, and matches the same protocol used in production.
        """
        self._assert_connected()
        assert self._ws is not None and self.stream_sid is not None
        await self._ws.send(build_clear_frame(self.stream_sid))
        logger.debug("PipecatAgentAdapter: sent clear frame (interrupt)")

    # ------------------------------------------------------------------ background

    async def _recv_loop(self) -> None:
        """Read frames from pipecat, decode µ-law → PCM16 24k, enqueue."""
        assert self._ws is not None and self._inbound_queue is not None
        queue = self._inbound_queue
        buffered_mulaw = bytearray()
        BATCH_MS = 100

        try:
            async for raw in self._ws:
                if isinstance(raw, bytes):
                    # pipecat sometimes emits binary frames for audio; treat
                    # as raw µ-law payload if we see one.
                    buffered_mulaw.extend(raw)
                    if len(buffered_mulaw) >= (BATCH_MS * 8):
                        self._deliver(bytes(buffered_mulaw))
                        buffered_mulaw.clear()
                    continue

                frame = parse_media_stream_frame(raw)
                if frame is None:
                    continue
                if frame.event == "media" and frame.payload_mulaw:
                    buffered_mulaw.extend(frame.payload_mulaw)
                    if len(buffered_mulaw) >= (BATCH_MS * 8):
                        self._deliver(bytes(buffered_mulaw))
                        buffered_mulaw.clear()
                elif frame.event == "stop":
                    if buffered_mulaw:
                        self._deliver(bytes(buffered_mulaw))
                        buffered_mulaw.clear()
                    return
        except asyncio.CancelledError:
            raise
        except Exception as exc:
            # #498: do NOT swallow. A crash here (decode error, transport reset,
            # bot pipeline failure) used to only log + fall through, leaving the
            # inbound queue silent so recv_audio blocked the full response_timeout
            # with no attributable cause. Record it; recv_audio raises
            # PipecatRecvError naming this as the root cause (chained via __cause__).
            self._recv_loop_exc = exc
            logger.warning("PipecatAgentAdapter: recv loop crashed", exc_info=True)
        finally:
            # Mark terminal, then wake any pending recv_audio: no more audio will
            # arrive on this connection. The flag lets later recv_audio calls fail
            # fast on the drained queue; the sentinel unblocks a getter currently
            # awaiting an empty queue. Together they turn an indefinite wait into
            # an immediate, attributable PipecatRecvError. (No await between the
            # two lines, so a waiter can't observe a half-set terminal state.)
            self._recv_loop_done = True
            queue.put_nowait(_RECV_LOOP_DONE)

    def _deliver(self, mulaw: bytes) -> None:
        """Decode a coalesced µ-law batch to PCM16/24k and enqueue it.

        On the FIRST wire delivery of a turn — the base ``call()`` published its
        OTel context via :attr:`VoiceAgentAdapter._voice_turn_context` — wrap the
        decode+enqueue in a ``voice.audio.receive`` span parented to THAT turn
        (#774), so the background ``_recv_loop`` task's receive work is attributed
        to the turn instead of the task's frozen connect-time context (a
        now-closed span).

        This is a producer-side **delivery marker** (it carries the delivered
        byte count); the consumer-side wait and the timeout/ERROR semantics (P3)
        live on the base ``voice.audio.receive`` span that wraps ``recv_audio``.
        The span is disambiguated from that base span by
        ``voice.pipecat.recv.source=background_loop``.

        Emitted at most ONCE per turn (only the first delivery under a given turn
        context) — matching the base's one-receive-span-per-turn granularity and
        the epic's no-per-tick-flood rule (the EL pump H1), so a multi-second turn
        of 100 ms chunks is not exploded into dozens of sibling spans. Between
        turns the context is ``None`` and we enqueue WITHOUT a span, so no
        detached/closed-parent span can leak from the background task. The body is
        synchronous (``put_nowait`` on the unbounded queue), so a ``disconnect()``
        cancel can never land mid-span.

        Coverage limits (by design — this is a marker, not exhaustive accounting):
        - ``voice.audio.bytes`` is the DELIVERED coalesced-chunk size, which may
          fold in a sub-100 ms µ-law tail carried over from the prior turn (the
          recv-loop coalesces to 800-byte batches across the turn boundary — a
          pre-existing property the base drain chunks share; not corrected here
          because flushing a partial batch at the boundary would drop that audio).
        - A turn whose agent audio was ALREADY buffered before ``call()`` published
          the turn context (an opening greeting delivered during ``connect``) is
          drained from the queue without a fresh wire delivery, so it gets no
          background marker — the base ``voice.audio.receive`` span still covers
          its consumption. The marker records in-turn wire deliveries, not every
          turn that consumes audio (the deterministic turn-liveness gate).
        """
        assert self._inbound_queue is not None
        parent = self._voice_turn_context
        if parent is None or parent is self._bg_span_turn_context:
            self._inbound_queue.put_nowait(AudioChunk(data=mulaw8k_to_pcm16_24k(mulaw)))
            return
        self._bg_span_turn_context = parent
        with voice_span(
            "voice.audio.receive",
            {
                "voice.adapter.class": type(self).__name__,
                "voice.pipecat.recv.source": "background_loop",
            },
            parent=parent,
        ) as span:
            pcm = mulaw8k_to_pcm16_24k(mulaw)
            span.set_attribute("voice.audio.bytes", len(pcm))
            self._inbound_queue.put_nowait(AudioChunk(data=pcm))

    # ------------------------------------------------------------------ assertions

    def _assert_connected(self) -> None:
        if self._ws is None:
            raise RuntimeError(
                "PipecatAgentAdapter: not connected. Did you forget to call connect()?"
            )

Ancestors

• VoiceAgentAdapter
• AgentAdapter
• abc.ABC

Class variables

var capabilities : ClassVar[AdapterCapabilities]

Instance variables

var transport_format : str

Expand source code

@property
def transport_format(self) -> str:
    return f"{self.audio_format}/{self.sample_rate}"

Methods

async def interrupt(self) ‑> None

Send a Twilio clear frame — the bot drops all buffered outbound audio immediately. Cooperating Pipecat bots (and any code wired to the Media Streams protocol) treat clear as "stop talking now." Use this in preference to timing-based barge-in when the SUT supports it: it's deterministic, doesn't depend on VAD detection windows, and matches the same protocol used in production.

Expand source code

async def interrupt(self) -> None:
    """Send a Twilio ``clear`` frame — the bot drops all buffered outbound
    audio immediately. Cooperating Pipecat bots (and any code wired to
    the Media Streams protocol) treat ``clear`` as "stop talking now."
    Use this in preference to timing-based barge-in when the SUT
    supports it: it's deterministic, doesn't depend on VAD detection
    windows, and matches the same protocol used in production.
    """
    self._assert_connected()
    assert self._ws is not None and self.stream_sid is not None
    await self._ws.send(build_clear_frame(self.stream_sid))
    logger.debug("PipecatAgentAdapter: sent clear frame (interrupt)")

Inherited members

• VoiceAgentAdapter:
  • agent_hung_up
  • call
  • connect
  • disconnect
  • is_connected
  • recv_audio
  • send_audio

class PipecatRecvError (*args, **kwargs)

recv_audio could get no audio because the background _recv_loop ended.

Names the real reason — a crash in the read loop (decode/transport error, chained via cause) or a clean WebSocket close by the bot — so the #498 2nd-turn hang surfaces an attributable error instead of a blind response_timeout with an empty body.

Expand source code

class PipecatRecvError(AgentStreamEndedError):
    """recv_audio could get no audio because the background _recv_loop ended.

    Names the real reason — a crash in the read loop (decode/transport error,
    chained via __cause__) or a clean WebSocket close by the bot — so the #498
    2nd-turn hang surfaces an attributable error instead of a blind
    response_timeout with an empty body.
    """

Ancestors

• AgentStreamEndedError
• builtins.Exception
• builtins.BaseException

class TunnelNotReadyError (*args, **kwargs)

Raised when the public base URL is not yet reachable from the edge.

A-leg origination hands Twilio our public URL and Twilio opens the media WebSocket against it within seconds. If the tunnel edge is not live yet the call connects to nothing: the caller pays for a dead PSTN call that ends in a confusing stream-connect timeout. Failing fast under a named error before origination is the cheaper failure.

Expand source code

class TunnelNotReadyError(RuntimeError):
    """Raised when the public base URL is not yet reachable from the edge.

    A-leg origination hands Twilio our public URL and Twilio opens the media
    WebSocket against it within seconds. If the tunnel edge is not live yet the
    call connects to nothing: the caller pays for a dead PSTN call that ends in
    a confusing stream-connect timeout. Failing fast under a named error before
    origination is the cheaper failure.
    """

Ancestors

• builtins.RuntimeError
• builtins.Exception
• builtins.BaseException

class TunnelReadiness (*args, **kwargs)

Anything that can tell us our public URL is reachable from the edge.

Structurally satisfied by CloudflareTunnel — the adapter never imports the tunnel (testing depends on adapters, not the other way round), it just calls the method the tunnel already has.

Expand source code

@runtime_checkable
class TunnelReadiness(Protocol):
    """Anything that can tell us our public URL is reachable from the edge.

    Structurally satisfied by ``scenario.voice.testing.CloudflareTunnel`` — the
    adapter never imports the tunnel (``testing`` depends on ``adapters``, not
    the other way round), it just calls the method the tunnel already has.
    """

    async def wait_until_edge_reachable(self) -> None:
        """Return once the public edge answers; raise if it will not."""

Ancestors

• typing.Protocol
• typing.Generic

Methods

async def wait_until_edge_reachable(self) ‑> None

Return once the public edge answers; raise if it will not.

Expand source code

async def wait_until_edge_reachable(self) -> None:
    """Return once the public edge answers; raise if it will not."""

class TwilioAgentAdapter (*, account_sid: str, auth_token: str, phone_number: str, public_base_url: Optional[str] = None, allowed_callers: Optional[list[str]] = None, allowed_callees: Optional[list[str]] = None, tunnel_readiness: Optional[TunnelReadiness] = None, on_dtmf: Optional[Callable[[str], None]] = None, http_port: int = 8765, role: AgentRole = AgentRole.AGENT, validate_signature: bool = True)

Bidirectional Twilio Media Streams adapter.

Same class, same state, either direction:

adapter = TwilioAgentAdapter(
    account_sid=..., auth_token=...,
    phone_number="+14155551234",
)
async with adapter:                   # connect() / disconnect()
    await adapter.place_call(to="+14155557777")  # OR wait_for_call()
    # ... scenario.run(...) feeds send_audio / recv_audio ...

The adapter is the only adapter with dtmf=True. DTMF events received from the callee surface via the on_dtmf callback set at construction time. To send DTMF, use send_dtmf().

interrupt(after_words=N) raises UnsupportedCapabilityError on this adapter — Media Streams delivers raw audio without incremental transcripts. Use interrupt(after=seconds) instead.

Expand source code

class TwilioAgentAdapter(VoiceAgentAdapter):
    """
    Bidirectional Twilio Media Streams adapter.

    Same class, same state, either direction:

        adapter = TwilioAgentAdapter(
            account_sid=..., auth_token=...,
            phone_number="+14155551234",
        )
        async with adapter:                   # connect() / disconnect()
            await adapter.place_call(to="+14155557777")  # OR wait_for_call()
            # ... scenario.run(...) feeds send_audio / recv_audio ...

    The adapter is the *only* adapter with ``dtmf=True``. DTMF events
    received from the callee surface via the ``on_dtmf`` callback set at
    construction time. To send DTMF, use ``send_dtmf()``.

    ``interrupt(after_words=N)`` raises ``UnsupportedCapabilityError`` on this
    adapter — Media Streams delivers raw audio without incremental
    transcripts. Use ``interrupt(after=seconds)`` instead.
    """

    capabilities: ClassVar[AdapterCapabilities] = AdapterCapabilities(
        streaming_transcripts=False,
        native_vad=False,
        dtmf=True,
        # Twilio Media Streams ``clear`` event drops all buffered outbound
        # audio. Used by ``adapter.interrupt()`` already wired below.
        interruption=True,
        input_formats=["mulaw/8000"],
        output_formats=["mulaw/8000"],
    )

    # ------------------------------------------------------------------ init

    def __init__(
        self,
        *,
        account_sid: str,
        auth_token: str,
        phone_number: str,
        public_base_url: Optional[str] = None,
        allowed_callers: Optional[list[str]] = None,
        allowed_callees: Optional[list[str]] = None,
        tunnel_readiness: Optional[TunnelReadiness] = None,
        on_dtmf: Optional[Callable[[str], None]] = None,
        http_port: int = 8765,
        role: AgentRole = AgentRole.AGENT,
        validate_signature: bool = True,
    ) -> None:
        super().__init__()
        validate_e164(phone_number)

        self.account_sid = account_sid
        self.auth_token = auth_token
        self.phone_number = phone_number
        self.public_base_url = public_base_url
        self.allowed_callers = set(allowed_callers) if allowed_callers else None
        # Destination allowlist for a-leg mode (#762 guardrail (c)). ``None``
        # denies every a-leg destination: a-leg dials numbers this account does
        # NOT own, so an unguarded ``to`` is an unbounded dialer and the
        # capability has to be opted into per destination. Entries are
        # normalised (and so validated) here rather than at dial time, so a
        # typo fails at setup instead of when a PSTN call is about to be
        # billed. b-leg is deliberately not gated on this — it can only reach
        # numbers on this account, which is its own guardrail.
        self.allowed_callees = (
            {normalize_e164(n) for n in allowed_callees} if allowed_callees else None
        )
        # Edge-readiness probe consulted before a-leg origination (#762
        # guardrail (c)). ``CloudflareTunnel`` satisfies it structurally;
        # ``TwilioHarness`` wires its own tunnel in. ``None`` means the caller
        # owns a stable public URL and has nothing to wait for.
        self.tunnel_readiness = tunnel_readiness
        self.on_dtmf = on_dtmf
        self.http_port = http_port
        self.role = role  # type: ignore[misc]
        # When True (default), the inbound /twilio/voice route requires a
        # valid X-Twilio-Signature header before accepting the webhook
        # body. The cloudflared tunnel URL is ephemeral and not
        # guessable, but anyone who learns it could otherwise POST fake
        # Twilio webhook events into the harness. Tests that don't have
        # real Twilio credentials disable this with
        # ``validate_signature=False``; production callers should leave
        # it on.
        self.validate_signature = validate_signature
        if not validate_signature:
            logger.warning(
                "TwilioAgentAdapter: validate_signature=False — inbound "
                "webhooks accept any payload without signature checks. "
                "Use only in tests; do not deploy to production."
            )

        # Populated during connect(); None when disconnected.
        self._rest: Optional[TwilioRESTHelper] = None
        self._phone_number_sid: Optional[str] = None
        self._prior_voice_url: Optional[str] = None
        # Set by place_call() when it rewrites the callee's voice_url so
        # B-leg's webhook lands on our harness. Restored in disconnect.
        self._callee_phone_number_sid: Optional[str] = None
        self._prior_callee_voice_url: Optional[str] = None
        # Set by the first of wait_for_call()/place_call(); subsequent calls to
        # the other method raise. "idle" after connect() before either fires.
        self._mode: TwilioAdapterMode = "idle"

        # Server / media stream state.
        self._server_task: Optional[asyncio.Task] = None
        self._server_shutdown: Optional[asyncio.Event] = None
        self._call_sid: Optional[str] = None
        self._stream_sid: Optional[str] = None
        self._stream_connected: Optional[asyncio.Event] = None
        self._stream_ws: Any = None  # starlette WebSocket
        # Call generation (#762 P1). Bumped by connect()/place_call()/
        # wait_for_call() every time we (re)dial. The media loop captures it at
        # the moment a socket is ADOPTED and re-checks it on every later side
        # effect (enqueue, signal-connected, mark-ended, cancel-watchdog, being
        # ``_stream_ws``): a socket adopted under an older generation — one that
        # connected and even adopted BEFORE this dial armed the nonce — is stale
        # the instant the counter moves, and is closed on its next frame instead
        # of injecting audio into, or ending, the current call.
        self._call_generation: int = 0
        # A-leg WS auth (#762 guardrail (a)). Set by place_call in "a-leg" mode
        # only; its non-None-ness is what ARMS media-stream nonce enforcement.
        # In b-leg mode the signed POST /twilio/voice precedes the socket, so
        # the socket inherits that trust and this stays None — enforcement is
        # keyed on the mode we originated in, never on whether the inbound frame
        # happens to carry a nonce (which an attacker could simply omit).
        self._stream_nonce: Optional[str] = None
        # Adapter-side max-call-duration watchdog (#762 guardrail (b)). Armed by
        # place_call in "a-leg" mode; the belt to Twilio's own ``TimeLimit``
        # suspenders, which is the half that survives this process dying.
        self._max_duration_task: Optional[asyncio.Task] = None
        self._inbound_queue: Optional[asyncio.Queue[AudioChunk]] = None
        # Set True by the media-stream loop's terminal path (stop / socket close
        # / throw) the moment it enqueues the end-of-call sentinel. Once the call
        # has ended, recv_audio must keep draining the queue (and hand back the
        # sentinel) WITHOUT re-asserting transport liveness — the real
        # production wrapper nulls _stream_ws/_stream_sid synchronously right
        # after the loop returns, so the drain's second recv_audio would
        # otherwise hit _assert_stream_live and crash (#695). Reset on
        # connect(), disconnect(), and at media-stream-loop entry (per-call
        # scope — a second session on the same connected adapter must not
        # inherit the previous session's flag).
        self._stream_ended: bool = False

        # Call-lifetime counters stamped onto ``voice.adapter.disconnect`` from
        # inside disconnect() (#775 Tier 2b — mirrors ElevenLabs' pump-counter
        # seam). Accumulated by the media loop (`_twilio_server.py`) and the
        # `/twilio/voice` webhook handler; reset on connect()/disconnect() like
        # `_stream_ended` above.
        self._frames_received: int = 0
        self._dtmf_received: int = 0
        self._stream_ended_reason: TwilioStreamEndedReason = "none"
        self._webhook_invocations: int = 0
        self._webhook_rejected: int = 0


    # ------------------------------------------------------------------ repr

    def __repr__(self) -> str:  # redact credentials
        return (
            f"TwilioAgentAdapter("
            f"phone_number={self.phone_number!r}, "
            f"account_sid='***', auth_token='***', "
            f"public_base_url={self.public_base_url!r})"
        )

    # ------------------------------------------------------------------ lifecycle

    async def connect(self) -> None:
        """Resolve number SID and start the FastAPI webhook + WS server.

        Does NOT modify the Twilio account's ``voice_url``. That side-effect
        only happens when ``wait_for_call()`` is invoked — callers (who will
        use ``place_call()``) never overwrite their number's inbound webhook,
        which makes caller-mode adapters safe to run against a shared pool of
        Twilio numbers without clobbering anyone's prod webhook.

        Idempotent: calling connect() on an already-connected adapter is a
        no-op. This lets the scenario executor's auto-connect step
        (``_voice_connect_all``) coexist with explicit harness-driven
        connects (``TwilioHarness`` ``__aenter__``) that already brought
        the adapter up before scenario.run() was called.
        """
        if self._rest is not None:
            return

        if self.public_base_url is None:
            raise RuntimeError(
                "TwilioAgentAdapter: public_base_url is required. Wrap the "
                "adapter in scenario.voice.testing.TwilioHarness, or supply "
                "a stable public HTTPS URL that routes to this machine."
            )

        self._rest = TwilioRESTHelper(self.account_sid, self.auth_token)
        self._phone_number_sid = self._rest.resolve_phone_number_sid(self.phone_number)

        # Stamp Twilio-specific attrs onto the active ``voice.adapter.connect``
        # span (opened by the executor connect loop). Base spans are name-owned;
        # the adapter contributes attributes, never a parallel span name — mirror
        # ElevenLabs' ``voice.elevenlabs.agent_id`` seam (``adapters/elevenlabs.py``).
        # NOTE: no ``voice.twilio.direction`` here — connect() is direction-
        # agnostic (``_mode`` stays "idle" until place_call()/wait_for_call(),
        # after this span has already closed); direction is stamped on the NEW
        # ``voice.adapter.dial`` span instead (see place_call/wait_for_call).
        set_span_attributes(
            _otel_trace.get_current_span(),
            {
                "voice.twilio.phone_number_sid": self._phone_number_sid,
                "voice.twilio.validate_signature": self.validate_signature,
                "voice.twilio.webhook_port": self.http_port,
            },
        )

        self._stream_connected = asyncio.Event()
        self._inbound_queue = asyncio.Queue()
        self._server_shutdown = asyncio.Event()
        self._stream_ended = False
        self._frames_received = 0
        self._dtmf_received = 0
        self._stream_ended_reason = "none"
        self._webhook_invocations = 0
        self._webhook_rejected = 0
        self._stream_nonce = None
        self._cancel_max_duration_timer()
        # Fresh connection lifecycle — bump the generation so any socket left
        # over from a previous connect can never pass the media loop's identity
        # check against this one.
        self._call_generation += 1
        self._mode = "idle"

        # Webhook server is its own unit — see _twilio_server.py. The
        # adapter only orchestrates lifecycle; the routes, signature
        # validation, and WS framing live in TwilioWebhookServer.
        from ._twilio_server import TwilioWebhookServer
        self._webhook_server: Optional[TwilioWebhookServer] = TwilioWebhookServer(self)
        self._server_task = asyncio.create_task(self._run_server())
        # Give uvicorn a beat to bind the port before Twilio hits it.
        await asyncio.sleep(0.2)

    async def disconnect(self) -> None:
        """Restore prior voice_url (answer mode only), tear down server.

        Best-effort on errors. In caller mode we never touched the Twilio
        number's voice_url, so there's nothing to restore.
        """
        if self._rest is None:
            return

        # 1. Restore webhook first so Twilio doesn't keep hitting a dead URL.
        # ``rest_restore_failed`` tracks whether either restore below raised —
        # previously fully swallowed via ``suppress(Exception)``; now surfaced
        # onto the disconnect span (#775 Tier 2b) so a Twilio REST outage during
        # teardown is no longer invisible. The swallow behavior itself (a failed
        # restore must never block disconnect()) is unchanged.
        # Disarm the duration watchdog first: from here on the call is ours to
        # end, and a timer that fired mid-teardown would hang up a SID this
        # adapter may already have replaced with a later call's.
        self._cancel_max_duration_timer()

        rest_restore_failed = False
        if self._mode == "answer" and self._phone_number_sid is not None:
            try:
                prior = self._prior_voice_url or ""
                self._rest.write_voice_url(self._phone_number_sid, prior)
                logger.debug(
                    "TwilioAgentAdapter: restored voice_url=%r on %s",
                    prior,
                    self._phone_number_sid,
                )
            except Exception:
                rest_restore_failed = True
        # place_call() rewrites the CALLEE's voice_url to attach Media
        # Streams to B-leg. Restore that too.
        if self._mode == "call" and self._callee_phone_number_sid is not None:
            try:
                prior_b = self._prior_callee_voice_url or ""
                self._rest.write_voice_url(self._callee_phone_number_sid, prior_b)
                logger.debug(
                    "TwilioAgentAdapter: restored callee voice_url=%r on %s",
                    prior_b,
                    self._callee_phone_number_sid,
                )
            except Exception:
                rest_restore_failed = True

        # 2. Signal server to shut down, then wait for the task. This is what
        # ACTUALLY closes a still-live media stream (uvicorn's graceful
        # shutdown tears down the open WS), which is what lets the media
        # loop's ``finally``/except-handlers (``_twilio_server.py``) make
        # their FINAL writes to ``_frames_received`` / ``_stream_ended_reason``
        # / etc. The counter stamp below MUST run after this await, not
        # before — stamping first (the #775 review-caught bug) reads the
        # pre-teardown snapshot and reports ``stream_ended_reason=="none"`` /
        # undercounted frames for a call that was still live when disconnect()
        # was invoked, even though the call did in fact end (via this very
        # shutdown) moments later.
        if self._server_shutdown is not None:
            self._server_shutdown.set()
        if self._server_task is not None:
            with suppress(Exception):
                await asyncio.wait_for(self._server_task, timeout=3.0)

        # Stamp call-lifetime counters onto the active ``voice.adapter.disconnect``
        # span (opened by the executor disconnect loop) — mirrors ElevenLabs'
        # pump-counter stamp (``adapters/elevenlabs.py``). Sampled AFTER the
        # shutdown-await above (not before) so a still-live call's FINAL
        # counts/reason are captured, not a mid-call snapshot. The disconnect
        # span stays the current span across the await (contextvars survive
        # awaits within the same task), so ``get_current_span()`` here still
        # targets it.
        set_span_attributes(
            _otel_trace.get_current_span(),
            {
                "voice.twilio.frames_received": self._frames_received,
                "voice.twilio.dtmf_received": self._dtmf_received,
                "voice.twilio.stream_ended_reason": self._stream_ended_reason,
                "voice.twilio.webhook_invocations": self._webhook_invocations,
                "voice.twilio.webhook_rejected": self._webhook_rejected,
                "voice.twilio.rest_restore_failed": rest_restore_failed,
            },
        )

        # 3. Reset state.
        self._rest = None
        self._phone_number_sid = None
        self._prior_voice_url = None
        self._callee_phone_number_sid = None
        self._prior_callee_voice_url = None
        self._mode = "idle"
        self._server_task = None
        self._server_shutdown = None
        self._call_sid = None
        self._stream_sid = None
        self._stream_connected = None
        self._stream_ws = None
        self._stream_nonce = None
        self._cancel_max_duration_timer()
        if self._inbound_queue is not None:
            # A recv_audio() task may already be blocked in get() on this queue.
            # Wake that in-flight consumer before dropping our reference; setting
            # _inbound_queue to None alone leaves it parked until its receive
            # timeout expires (#757). An empty chunk is the adapter's existing
            # end-of-stream signal and lets the shared drain finish cleanly.
            self._inbound_queue.put_nowait(AudioChunk(data=b""))
        self._inbound_queue = None
        self._stream_ended = False
        self._frames_received = 0
        self._dtmf_received = 0
        self._stream_ended_reason = "none"
        self._webhook_invocations = 0
        self._webhook_rejected = 0

    # ------------------------------------------------------------------ direction

    def _reset_connection_state_for_dial(self) -> None:
        """Invalidate any socket adopted under a prior call generation (#762 P1).

        Bumping ``_call_generation`` is what evicts a socket that connected — and
        even ADOPTED — before this dial armed the nonce: the media loop captured
        the old generation at adoption and closes itself on its next frame once
        the counter moves. We also reset the per-dial connection state the loop
        writes, so this dial starts from a clean slate and, critically, cannot
        resolve on a stale socket's ``_stream_connected`` signal:

        - clear ``_stream_connected`` — otherwise a socket that adopted un-gated
          before arming has already fired it and ``place_call`` would return
          before any socket authenticates;
        - drop ``_stream_ws``/``_stream_sid`` — the stale socket must not remain
          our outbound transport (``send_audio`` writes there);
        - drain the inbound queue and zero ``_frames_received`` — the stale
          socket's media must not surface as this call's audio;
        - clear the terminal flag/reason — the stale session's ``stop`` must not
          read as this call having ended.

        Called by both ``place_call`` and ``wait_for_call``. Inbound/b-leg
        behaviour is unchanged: no socket has adopted yet when the genuine one
        arrives, so it adopts under the current generation and connects exactly
        as before.
        """
        self._call_generation += 1
        self._stream_ws = None
        self._stream_sid = None
        if self._stream_connected is not None:
            self._stream_connected.clear()
        if self._inbound_queue is not None:
            while not self._inbound_queue.empty():
                self._inbound_queue.get_nowait()
        self._stream_ended = False
        self._stream_ended_reason = "none"
        self._frames_received = 0

    async def place_call(
        self,
        to: str,
        *,
        timeout: float = 120.0,
        attach_stream_to_self: Optional[bool] = None,
        attach_stream: Optional[Literal["a-leg", "b-leg"]] = None,
        max_call_duration_seconds: Optional[int] = None,
    ) -> None:
        """
        Originate an outbound call from this adapter's Twilio number to ``to``.

        Twilio's REST ``Calls.create`` runs TwiML on TWO legs of the
        resulting call:

        - **A-leg** (the originator, ``from_=self.phone_number``): runs the
          inline TwiML passed via ``twiml=`` to ``Calls.create``. We use
          ``<Pause length=120>`` so the originator just holds the bridge
          open while the demo runs.
        - **B-leg** (the callee, ``to=``): when Twilio dials B and B picks
          up, B's number's ``voice_url`` fires — that's where the bridge's
          Media Streams attach. This is identical to the inbound demo's
          flow: B's voice_url returns ``<Connect><Stream>``, the WS opens,
          audio flows.

        So ``place_call`` only makes sense when ``to`` is another Twilio
        number on this account whose ``voice_url`` is set to OUR harness
        webhook. To make that wiring automatic, ``place_call`` temporarily
        rewrites B's ``voice_url`` for the duration of the call and
        restores it on ``disconnect``. The harness on this adapter's own
        number does NOT need to be answer-mode — the Stream attaches to
        B's leg, NOT this adapter's leg, but B's webhook is hosted on this
        adapter's local server, so the WS still lands here.

        The bidirectional audio model is unchanged: ``send_audio`` writes
        frames over the WS (B hears them and bridges to A), ``recv_audio``
        reads inbound frames off the WS (whatever the bridge mixes from
        both legs).

        Limitation of the default (**b-leg**) mode: ``to`` MUST be a phone
        number on this same Twilio account. Calling an external PSTN endpoint
        (a real cell phone) is instead served by **a-leg** mode.

        Stream-attach mode is selected by ``attach_stream``:

        - ``"b-leg"`` — today's default: rewrite the callee's ``voice_url``,
          hold the A-leg open with ``<Say>…<Pause>``, restore on disconnect.
          Only works for numbers this account owns.
        - ``"a-leg"`` — originate with inline ``<Connect><Stream>`` so the
          Media Stream rides OUR own leg. Touches nothing on the callee (no
          ``resolve_phone_number_sid``/``read_voice_url``/``write_voice_url``),
          so ``to`` can be any external number. Because that reaches numbers
          this account does not own, ``to`` must appear in the adapter's
          ``allowed_callees`` (deny-by-default) and the public base URL must be
          reachable from the edge — both checked before origination.

        The legacy ``attach_stream_to_self`` bool is still honoured but
        superseded by ``attach_stream``: ``True``/unset → ``"b-leg"``,
        ``False`` → originator-only (a third mode ``attach_stream`` cannot
        express). Passing both with disagreeing intent raises ``ValueError``.

        Default timeout 120s covers cloudflared cold-start latency.

        ``timeout`` and ``max_call_duration_seconds`` are two different clocks
        and are configured independently:

        - ``timeout`` bounds how long we WAIT FOR THE MEDIA STREAM TO CONNECT
          after origination. It says nothing about how long the call may run.
        - ``max_call_duration_seconds`` bounds HOW LONG THE CALL MAY LIVE
          (a-leg only, default ``DEFAULT_MAX_CALL_DURATION_SECONDS``, hard cap
          ``MAX_CALL_DURATION_CAP_SECONDS``; a larger request raises). Under
          a-leg's ``<Connect>`` the call lives exactly as long as the WebSocket,
          so without a ceiling a hung executor keeps a billing PSTN call open
          forever. Two mechanisms enforce it: Twilio's own ``TimeLimit`` on
          ``Calls.create`` (the load-bearing half — it still fires if this
          process hangs or is killed) and an adapter-side wall-clock timer that
          hangs the call up via REST and closes the socket.

        Raises:
            RuntimeError: If called after ``wait_for_call()`` (modes are
                exclusive per adapter instance), or if ``to`` is not a
                Twilio number on this account.
            ValueError: If ``to`` is not in E.164 format, or — in a-leg mode —
                is not on ``allowed_callees``.
            TunnelNotReadyError: If a-leg mode's readiness probe says the
                public base URL is not reachable from the edge yet.
            asyncio.TimeoutError: If the media stream doesn't open within
                ``timeout`` seconds.
        """
        self._assert_connected()
        self._enter_mode("call")
        validate_e164(to)

        # Resolve the effective stream-attach mode BEFORE any REST call so a
        # conflicting-parameter caller error surfaces before we dial. `attach_stream`
        # (typed) supersedes the legacy `attach_stream_to_self` bool; the bool's
        # third mode (originator-only) is unreachable via `attach_stream` by design.
        mode = _resolve_stream_mode(attach_stream, attach_stream_to_self)
        # Only a-leg loses <Pause>'s implicit ceiling, so only a-leg carries a
        # duration cap. Naming one in another mode is rejected rather than
        # ignored: silently dropping it would leave the caller believing the
        # call is bounded when it is not.
        if mode == "a-leg":
            # Guardrail (c): destinations are deny-by-default in a-leg mode.
            # Checked here, with the other local caller-fixable failures, so a
            # misconfigured allowlist costs nothing and dials nothing.
            self._assert_callee_allowed(to)
            max_call_duration = resolve_max_call_duration(max_call_duration_seconds)
        elif max_call_duration_seconds is not None:
            raise ValueError(
                "place_call: max_call_duration_seconds is only supported with "
                'attach_stream="a-leg"; b-leg and originator-only modes hold the '
                "originator leg with <Pause> and are bounded by it."
            )
        else:
            max_call_duration = None

        # Guardrail (c), second half: a-leg hands Twilio our public URL and
        # Twilio opens the media socket against it seconds later. Probe the edge
        # AFTER the free local checks above and BEFORE origination, so an
        # unreachable tunnel is a named error instead of a billed call that
        # dies in a stream-connect timeout.
        if mode == "a-leg":
            await self._assert_tunnel_ready()

        assert self.public_base_url is not None
        assert self._rest is not None
        assert self._stream_connected is not None

        # NEW ``voice.adapter.dial`` span (#775 Tier 2a): self-instrumented,
        # since no executor span is active by the time place_call()/
        # wait_for_call() run (the ``voice.adapter.connect`` span already
        # closed). Wraps the REST dial + the ``_stream_connected`` wait — the
        # "I placed the call but no media ever streamed" failure surface.
        with voice_span(
            "voice.adapter.dial",
            {
                "voice.adapter.class": type(self).__name__,
                "voice.twilio.direction": "outbound",
                "voice.twilio.to": _redact_e164(to),
                "voice.twilio.from": _redact_e164(self.phone_number),
            },
        ) as _dial:
            # #762 P1: bump the call generation and reset per-dial connection
            # state BEFORE originating, so a socket that adopted un-gated before
            # this dial is evicted and cannot resolve the wait below.
            self._reset_connection_state_for_dial()
            if mode == "b-leg":
                # Resolve B-leg's number SID and snapshot+rewrite its voice_url so
                # B's leg attaches its Media Stream to our harness webhook. We own
                # this number (same Twilio account); disconnect() will restore.
                self._callee_phone_number_sid = self._rest.resolve_phone_number_sid(to)
                self._prior_callee_voice_url = self._rest.read_voice_url(
                    self._callee_phone_number_sid
                )
                webhook_url = self.public_base_url.rstrip("/") + "/twilio/voice"
                self._rest.write_voice_url(self._callee_phone_number_sid, webhook_url)
                logger.info(
                    "TwilioAgentAdapter: rewrote callee %s voice_url to %s",
                    _redact_e164(to),
                    webhook_url,
                )

            # a-leg: the Media Stream rides OUR own leg via inline
            # <Connect><Stream>, so `to` can be any external number and we
            # touch nothing on the callee. Same TwiML shape the inbound webhook
            # returns (`_twilio_server.py`), now on origination.
            #
            # Guardrail (a): the socket is the only inbound signal on this path,
            # so it must authenticate itself. Mint a per-call CSPRNG nonce and
            # ship it as a <Parameter> child; Twilio echoes it back in the
            # `start` frame's customParameters, and the media loop
            # (`_twilio_server.py`) closes any socket that cannot present it.
            #
            # Assigned UNCONDITIONALLY (mirrors the JS twin): the field also
            # ARMS nonce enforcement and the a-leg send_dtmf refusal, so a b-leg
            # dial that left an earlier a-leg call's nonce standing would gate a
            # socket whose `start` frame carries no <Parameter> at all — the
            # b-leg call would never connect.
            nonce = mint_stream_nonce() if mode == "a-leg" else None
            self._stream_nonce = nonce
            if nonce is not None:
                origination_twiml = _build_connect_stream_twiml(
                    stream_ws_url(self.public_base_url),
                    stream_parameters={"nonce": nonce},
                )
            else:
                # A-leg (originator-side) TwiML for b-leg / originator-only modes:
                # play a short deterministic <Say> line, then hold the bridge
                # open. Twilio runs this on the originator side while B's webhook
                # attaches the Media Stream.
                #
                # The <Say> gives the recording a known-good utterance to
                # transcribe. A bare <Pause> alone produces 120s of line silence
                # that Whisper has been observed to hallucinate as non-English
                # text (issue #465 in this PR). The Say is a one-time anchor at
                # call setup; the Media Stream carries the real bidirectional
                # conversation that follows.
                origination_twiml = (
                    '<?xml version="1.0" encoding="UTF-8"?>'
                    "<Response>"
                    f'<Say voice="Polly.Joanna">{PLACE_CALL_A_LEG_SAY_TEXT}</Say>'
                    '<Pause length="120"/>'
                    "</Response>"
                )
            self._call_sid = self._rest.place_call(
                to=to,
                from_=self.phone_number,
                twiml=origination_twiml,
                time_limit=max_call_duration,
            )
            logger.info(
                "TwilioAgentAdapter: placed call %s from %s to %s",
                self._call_sid,
                _redact_e164(self.phone_number),
                _redact_e164(to),
            )
            set_span_attributes(_dial, {"voice.twilio.call_sid": self._call_sid})

            if max_call_duration is not None:
                self._arm_max_duration_timer(max_call_duration, self._call_sid, to)

            if mode != "originator-only":
                # Wait for the media stream to reach us. In b-leg mode it arrives
                # via the callee's rewritten voice_url; in a-leg mode it rides our
                # own <Connect><Stream> leg. In originator-only mode there's no
                # stream coming to us; the callee has its own harness which owns it.
                _dial_wait_started = time.monotonic()
                try:
                    await asyncio.wait_for(self._stream_connected.wait(), timeout=timeout)
                except asyncio.TimeoutError:
                    # The media stream never connected — the marquee "I placed
                    # the call but no media ever streamed" failure. Tag the
                    # outcome BEFORE re-raising; the raised TimeoutError still
                    # marks the span ERROR (voice_span's own exception handling).
                    _dial.set_attribute(
                        "voice.twilio.dial_outcome", "stream_connect_timeout"
                    )
                    if mode == "a-leg" and self._call_sid is not None:
                        # The call is already originated and, under a-leg's
                        # <Connect>, lives to the duration cap even though
                        # nothing will ever stream on it. Hang it up before
                        # handing the caller their TimeoutError — otherwise the
                        # ordinary connect-failure path bills for the cap.
                        await self._abandon_originated_call(self._call_sid)
                    raise
                set_span_attributes(
                    _dial,
                    {
                        "voice.twilio.stream_connect_latency_ms": round(
                            (time.monotonic() - _dial_wait_started) * 1000
                        ),
                    },
                )

    async def wait_for_call(self, timeout: float = 120.0) -> None:
        """
        Block until someone dials in and the media stream is live.

        In **default mode** (no conference_room): the number's ``voice_url``
        is overwritten to point at our webhook so inbound calls reach us.
        Caller (``place_call``) elsewhere will dial this number.

        In **conference mode**: there's no inbound call to wait for — the
        two-Twilio-number demo can't naturally have one side receive an
        inbound call when both legs need conference TwiML. Instead, we
        ORIGINATE an outbound call FROM this adapter's number TO itself
        with inline TwiML that opens the capture stream and dials into
        the shared conference room. This is the symmetric counterpart to
        ``place_call`` in conference mode — both adapters end up as
        participants in the same room, exchanging audio via the bridge.

        Default timeout 120s covers cloudflared cold-start, conference-room
        formation, and Twilio's webhook ramp.

        Raises:
            RuntimeError: If called after ``place_call()``.
            asyncio.TimeoutError: If nobody dials in within ``timeout``.
        """
        self._assert_connected()
        self._enter_mode("answer")

        assert self.public_base_url is not None
        assert self._rest is not None
        assert self._phone_number_sid is not None
        assert self._stream_connected is not None

        # NEW ``voice.adapter.dial`` span (#775 Tier 2a) — see place_call()'s
        # comment for the rationale (self-instrumented, no executor span active
        # here).
        with voice_span(
            "voice.adapter.dial",
            {
                "voice.adapter.class": type(self).__name__,
                "voice.twilio.direction": "inbound",
                "voice.twilio.to": _redact_e164(self.phone_number),
            },
        ) as _dial:
            # #762 P1: bump the call generation and reset per-dial connection
            # state, symmetric with place_call. For inbound this only guards a
            # reconnect/back-to-back dial; a fresh inbound socket still connects
            # exactly as before.
            self._reset_connection_state_for_dial()
            # Snapshot the prior webhook so we can restore it on disconnect, then
            # point the number at our server. Only answer mode does this.
            self._prior_voice_url = self._rest.read_voice_url(self._phone_number_sid)
            webhook_url = self.public_base_url.rstrip("/") + "/twilio/voice"
            self._rest.write_voice_url(self._phone_number_sid, webhook_url)
            logger.info("TwilioAgentAdapter: webhook set to %s", webhook_url)

            _dial_wait_started = time.monotonic()
            try:
                await asyncio.wait_for(self._stream_connected.wait(), timeout=timeout)
            except asyncio.TimeoutError:
                # Nobody dialed in — tag the outcome BEFORE re-raising; the
                # raised TimeoutError still marks the span ERROR (voice_span's
                # own exception handling).
                _dial.set_attribute(
                    "voice.twilio.dial_outcome", "stream_connect_timeout"
                )
                raise
            set_span_attributes(
                _dial,
                {
                    "voice.twilio.call_sid": self._call_sid,
                    "voice.twilio.stream_connect_latency_ms": round(
                        (time.monotonic() - _dial_wait_started) * 1000
                    ),
                },
            )

    def _assert_callee_allowed(self, to: str) -> None:
        """Refuse an a-leg destination that was not explicitly allowlisted.

        Default-deny: no ``allowed_callees`` means no a-leg call at all. The
        message names the field to set, because "denied" without "here is how
        to allow it" just sends the caller reading source.

        Comparison is an exact set lookup on the normalised number, so a
        near-miss — a prefix, a suffix, an extra space — is a refusal.
        """
        if self.allowed_callees is None:
            raise ValueError(
                'place_call: attach_stream="a-leg" requires allowed_callees. '
                "a-leg dials numbers this Twilio account does not own, so "
                "destinations are deny-by-default: pass "
                "TwilioAgentAdapter(allowed_callees=[...]) listing every "
                f"number this adapter may dial, including {_redact_e164(to)}."
            )
        if normalize_e164(to) not in self.allowed_callees:
            raise ValueError(
                f"place_call: destination {_redact_e164(to)} is not in "
                "allowed_callees. Add it to "
                "TwilioAgentAdapter(allowed_callees=[...]) to permit this "
                "destination."
            )

    def _assert_dtmf_supported(self) -> None:
        """Refuse DTMF on a call whose media stream rides our own leg (#762 AC10).

        ``send_dtmf`` works by ``calls(sid).update(twiml=...)``, which REPLACES
        the TwiML the call is executing. Under b-leg that TwiML is a ``<Say>`` +
        ``<Pause>`` hold on our leg, so replacing it costs nothing — the Media
        Stream lives on the callee's leg. Under a-leg the TwiML being replaced
        IS the ``<Connect><Stream>`` verb carrying the scenario, so the same
        REST call would redirect the call away from the socket and end the media
        session mid-run.

        Armed off ``_stream_nonce`` — set only by an a-leg ``place_call``, the
        same signal the media loop uses to arm nonce enforcement — rather than
        re-derived from the caller's arguments here.
        """
        if self._stream_nonce is not None:
            raise RuntimeError(A_LEG_SEND_DTMF_UNSUPPORTED)

    async def _assert_tunnel_ready(self) -> None:
        """Refuse to originate until the public URL is reachable from the edge.

        Delegates to whatever readiness probe was supplied — ``CloudflareTunnel``
        already knows how to answer this. No probe means the caller owns a
        stable public URL and there is nothing to wait for.
        """
        if self.tunnel_readiness is None:
            return
        try:
            await self.tunnel_readiness.wait_until_edge_reachable()
        except TunnelNotReadyError:
            raise
        except Exception as exc:
            raise TunnelNotReadyError(
                f"place_call: public_base_url {self.public_base_url!r} is not "
                "reachable from the edge yet, so Twilio's media stream would "
                "connect to nothing. Not originating."
            ) from exc

    def _enter_mode(self, mode: TwilioAdapterMode) -> None:
        """Transition idle → mode, or raise if already in a different mode.

        Modes are exclusive per connected session: an adapter can place a call
        or answer a call, not both. Disconnect + reconnect to reuse the
        instance in the other direction.
        """
        if self._mode == mode:
            return  # idempotent re-entry (e.g. retrying place_call after timeout)
        if self._mode != "idle":
            raise RuntimeError(
                f"TwilioAgentAdapter: already in {self._mode!r} mode; cannot "
                f"switch to {mode!r}. Disconnect and reconnect to reuse this "
                f"adapter in the other direction."
            )
        self._mode = mode

    # ------------------------------------------------------- max call duration

    def _arm_max_duration_timer(self, seconds: int, call_sid: str, to: str) -> None:
        """Start the wall-clock watchdog that ends ``call_sid`` at ``seconds``.

        Any previously-armed timer is cancelled first, so a second
        ``place_call`` on the same adapter can never leave an older call's timer
        alive to hang up the newer call's SID.
        """
        self._cancel_max_duration_timer()
        # Capture the generation this watchdog is armed under (#762 P2). After
        # the hang-up REST round-trip the handler re-checks it before closing the
        # socket, so an expiry whose REST response resolves AFTER a newer call has
        # taken over cannot close that newer call's authenticated socket.
        generation = self._call_generation
        self._max_duration_task = asyncio.create_task(
            self._run_max_duration_timer(seconds, call_sid, to, generation)
        )

    async def _abandon_originated_call(self, call_sid: str) -> None:
        """Disarm the watchdog and hang up a call nothing will ever stream on.

        Best-effort on the REST hangup: the caller is already raising, and
        Twilio's own ``TimeLimit`` remains the backstop if this leg of the
        teardown fails.
        """
        self._cancel_max_duration_timer()
        rest = self._rest
        if rest is not None:
            # Blocking REST call off-thread, as send_dtmf does.
            with suppress(Exception):
                await asyncio.to_thread(rest.end_call, call_sid)

    def _cancel_max_duration_timer(self) -> None:
        """Disarm the watchdog. Idempotent; safe when none was ever armed."""
        task = self._max_duration_task
        self._max_duration_task = None
        if task is not None and not task.done():
            task.cancel()

    async def _await_max_duration(self, seconds: int) -> None:
        """Sleep until the duration cap elapses.

        Test seam (same role as ``_build_app``/``_media_stream_loop``): tests
        replace this on the instance to drive expiry on controlled time instead
        of sleeping for real.
        """
        await asyncio.sleep(seconds)

    async def _run_max_duration_timer(
        self, seconds: int, call_sid: str, to: str, generation: int
    ) -> None:
        """Hang up the call and close the socket once the cap elapses.

        The belt to Twilio's ``TimeLimit`` suspenders: it fires even if the
        Twilio-side limit was misconfigured or silently dropped. Cancelled (and
        therefore silent) on ``disconnect()``, at stream end, and on re-arm.

        ``generation`` is the call generation this watchdog was armed under
        (#762 P2). Ending the ORIGINATED ``call_sid`` via REST is always correct
        — it targets the expired call by SID. But the OTHER two side effects
        target whatever call is live NOW, so both are gated on still owning the
        generation:

        - The ended-reason stamp (before the REST hang-up): ``"max_duration"``
          wins ended-reason ties permanently, so a stale watchdog — one that
          fires after a newer call (a b-leg / inbound / no-cap dial that bumps
          the generation WITHOUT cancelling this task) has taken over — would
          otherwise mislabel that newer live call for the rest of its life.
        - The socket close (after the REST hang-up ``await``): a newer
          ``place_call`` can adopt a fresh, authenticated socket while our
          hang-up response is still pending, and closing ``_stream_ws`` would
          tear it down.
        """
        await self._await_max_duration(seconds)
        logger.warning(
            "TwilioAgentAdapter: max call duration of %ss reached on call to "
            "%s — ending call %s",
            seconds,
            _redact_e164(to),
            call_sid,
        )
        # Only stamp the ended-reason while we still own the call generation; a
        # stale watchdog must not relabel a newer live call.
        if generation == self._call_generation:
            self._set_stream_ended_reason("max_duration")
        rest = self._rest
        if rest is not None:
            # Blocking REST call off-thread, as send_dtmf does. Targets our own
            # SID, so it is correct even once a newer call has taken over.
            with suppress(Exception):
                await asyncio.to_thread(rest.end_call, call_sid)
        # A newer call took over while the hang-up was in flight — its socket is
        # not ours to close.
        if generation != self._call_generation:
            return
        ws = self._stream_ws
        if ws is not None:
            with suppress(Exception):
                await ws.close()

    def _set_stream_ended_reason(self, reason: TwilioStreamEndedReason) -> None:
        """Record how the media session ended, cap verdict winning ties.

        The watchdog closes the socket itself, so the media loop's own "close"
        verdict lands moments later and would otherwise mask WHY the call
        ended. connect()/disconnect() reset the field directly and so still
        clear it for the next session.
        """
        if self._stream_ended_reason == "max_duration":
            return
        self._stream_ended_reason = reason

    # ------------------------------------------------------------------ I/O

    async def send_audio(self, chunk: AudioChunk) -> None:
        # Pace at real-time (one frame per TWILIO_FRAME_MS). Without pacing the
        # whole utterance arrives in milliseconds, which trips bots' VAD into
        # a clipped-utterance reading.
        self._assert_stream_live()

        ws = self._stream_ws
        stream_sid = self._stream_sid
        assert ws is not None and stream_sid is not None

        mulaw = pcm16_24k_to_mulaw8k(chunk.data)
        frame_secs = TWILIO_FRAME_MS / 1000
        for frame in iter_mulaw_frames(mulaw):
            if not frame:
                continue
            await ws.send_text(build_media_frame(stream_sid, frame))
            await asyncio.sleep(frame_secs)

    async def recv_audio(self, timeout: float) -> AudioChunk:
        # Once the media-stream loop has ended (stop / socket close / throw),
        # the production wrapper (run_stream_session in _twilio_server) nulls
        # _stream_ws/_stream_sid synchronously right after the loop returns. The
        # drain loop (_drain_agent_response) always makes a *second* recv_audio
        # call after the first chunk — by then liveness has flipped. So the
        # liveness assert only guards the genuinely-live-and-idle case; buffered
        # audio and the end-of-call drain bypass it (#695).
        #
        # Two sentinel sources cooperate here, and BOTH are load-bearing: the
        # loop's ``finally`` ENQUEUES one empty chunk — that is what wakes a
        # consumer already blocked in ``queue.get()`` at the moment the call
        # ends (flipping ``_stream_ended`` alone wakes nobody) — and this method
        # SYNTHESIZES further empty chunks for every call after the queue has
        # drained (the tail-silence probe, or any caller that keeps polling).
        # Delete either and a hang comes back.
        if self._inbound_queue is None:
            # disconnect() tore the adapter down mid-drain — nothing left to
            # read. In the current teardown order _assert_connected() raises
            # first (_rest is nulled before _inbound_queue, synchronously), so
            # the explicit raise below is a reordering guard, not a live path —
            # it keeps this method's error honest if disconnect() is ever
            # resequenced.
            self._assert_connected()
            raise RuntimeError(
                "TwilioAgentAdapter: inbound queue is gone; adapter disconnected."
            )

        if not self._inbound_queue.empty():
            # Buffered audio or the loop-enqueued terminal sentinel — drain it,
            # live or not.
            return await asyncio.wait_for(self._inbound_queue.get(), timeout=timeout)
        if self._stream_ended:
            # Call ended and fully drained: synthesize another empty sentinel.
            return AudioChunk(data=b"")
        # Live call, nothing buffered yet: assert liveness, then wait.
        self._assert_stream_live()
        return await asyncio.wait_for(self._inbound_queue.get(), timeout=timeout)

    async def send_dtmf(self, tones: str) -> None:
        """Send DTMF digits on the live call (uses Twilio REST ``<Play digits>``).

        Refused in a-leg mode (#762 AC10) — see ``_assert_dtmf_supported``.
        """
        self._assert_dtmf_supported()
        if self._rest is None or self._call_sid is None:
            raise RuntimeError("TwilioAgentAdapter: no active call; send_dtmf requires an in-progress call")
        # Run blocking REST call off-thread so we don't stall the event loop.
        await asyncio.to_thread(self._rest.send_dtmf_on_call, self._call_sid, tones)

    async def interrupt(self) -> None:
        """Drop any buffered outbound audio on Twilio's side (``clear`` event)."""
        self._assert_stream_live()
        ws = self._stream_ws
        stream_sid = self._stream_sid
        assert ws is not None and stream_sid is not None
        await ws.send_text(build_clear_frame(stream_sid))

    # ------------------------------------------------------------------ server

    async def _run_server(self) -> None:
        """Thin lifecycle wrapper; the real work lives in TwilioWebhookServer."""
        assert self._webhook_server is not None
        await self._webhook_server.run()

    def _build_app(self) -> Any:
        """Test seam: build the FastAPI app for in-process exercise.

        Production code never calls this — the server's ``run()`` builds
        the app itself when uvicorn starts. Existing unit tests use
        ``TestClient(_build_app())`` to exercise the routes without
        binding a port; the delegation keeps that test surface stable.
        """
        assert self._webhook_server is not None
        return self._webhook_server.build_app()

    async def _media_stream_loop(self, ws: Any) -> None:
        """Test seam: kick off the Media Streams WS loop directly.

        The two-adapter-bridge test in
        ``tests/voice/test_twilio_two_adapter_bridge.py`` uses this to
        drive a loopback WS without going through the FastAPI route
        wrapper. Production code reaches the loop via the ``/twilio/stream``
        WebSocket handler defined in ``_twilio_server.build_app``.
        """
        assert self._webhook_server is not None
        await self._webhook_server.media_stream_loop(ws)

    # ------------------------------------------------------------------ assertions

    def _assert_connected(self) -> None:
        if self._rest is None:
            raise RuntimeError("TwilioAgentAdapter: not connected; call connect() or use `async with`.")

    def _assert_stream_live(self) -> None:
        self._assert_connected()
        if self._stream_ws is None or self._stream_sid is None:
            raise RuntimeError(
                "TwilioAgentAdapter: no live media stream. Call place_call() or "
                "wait_for_call() first."
            )

Ancestors

• VoiceAgentAdapter
• AgentAdapter
• abc.ABC

Class variables

var capabilities : ClassVar[AdapterCapabilities]

Methods

async def connect(self) ‑> None

Resolve number SID and start the FastAPI webhook + WS server.

Does NOT modify the Twilio account's voice_url. That side-effect only happens when wait_for_call() is invoked — callers (who will use place_call()) never overwrite their number's inbound webhook, which makes caller-mode adapters safe to run against a shared pool of Twilio numbers without clobbering anyone's prod webhook.

Idempotent: calling connect() on an already-connected adapter is a no-op. This lets the scenario executor's auto-connect step (_voice_connect_all) coexist with explicit harness-driven connects (TwilioHarness __aenter__) that already brought the adapter up before scenario.run() was called.

Expand source code

async def connect(self) -> None:
    """Resolve number SID and start the FastAPI webhook + WS server.

    Does NOT modify the Twilio account's ``voice_url``. That side-effect
    only happens when ``wait_for_call()`` is invoked — callers (who will
    use ``place_call()``) never overwrite their number's inbound webhook,
    which makes caller-mode adapters safe to run against a shared pool of
    Twilio numbers without clobbering anyone's prod webhook.

    Idempotent: calling connect() on an already-connected adapter is a
    no-op. This lets the scenario executor's auto-connect step
    (``_voice_connect_all``) coexist with explicit harness-driven
    connects (``TwilioHarness`` ``__aenter__``) that already brought
    the adapter up before scenario.run() was called.
    """
    if self._rest is not None:
        return

    if self.public_base_url is None:
        raise RuntimeError(
            "TwilioAgentAdapter: public_base_url is required. Wrap the "
            "adapter in scenario.voice.testing.TwilioHarness, or supply "
            "a stable public HTTPS URL that routes to this machine."
        )

    self._rest = TwilioRESTHelper(self.account_sid, self.auth_token)
    self._phone_number_sid = self._rest.resolve_phone_number_sid(self.phone_number)

    # Stamp Twilio-specific attrs onto the active ``voice.adapter.connect``
    # span (opened by the executor connect loop). Base spans are name-owned;
    # the adapter contributes attributes, never a parallel span name — mirror
    # ElevenLabs' ``voice.elevenlabs.agent_id`` seam (``adapters/elevenlabs.py``).
    # NOTE: no ``voice.twilio.direction`` here — connect() is direction-
    # agnostic (``_mode`` stays "idle" until place_call()/wait_for_call(),
    # after this span has already closed); direction is stamped on the NEW
    # ``voice.adapter.dial`` span instead (see place_call/wait_for_call).
    set_span_attributes(
        _otel_trace.get_current_span(),
        {
            "voice.twilio.phone_number_sid": self._phone_number_sid,
            "voice.twilio.validate_signature": self.validate_signature,
            "voice.twilio.webhook_port": self.http_port,
        },
    )

    self._stream_connected = asyncio.Event()
    self._inbound_queue = asyncio.Queue()
    self._server_shutdown = asyncio.Event()
    self._stream_ended = False
    self._frames_received = 0
    self._dtmf_received = 0
    self._stream_ended_reason = "none"
    self._webhook_invocations = 0
    self._webhook_rejected = 0
    self._stream_nonce = None
    self._cancel_max_duration_timer()
    # Fresh connection lifecycle — bump the generation so any socket left
    # over from a previous connect can never pass the media loop's identity
    # check against this one.
    self._call_generation += 1
    self._mode = "idle"

    # Webhook server is its own unit — see _twilio_server.py. The
    # adapter only orchestrates lifecycle; the routes, signature
    # validation, and WS framing live in TwilioWebhookServer.
    from ._twilio_server import TwilioWebhookServer
    self._webhook_server: Optional[TwilioWebhookServer] = TwilioWebhookServer(self)
    self._server_task = asyncio.create_task(self._run_server())
    # Give uvicorn a beat to bind the port before Twilio hits it.
    await asyncio.sleep(0.2)

async def disconnect(self) ‑> None

Restore prior voice_url (answer mode only), tear down server.

Best-effort on errors. In caller mode we never touched the Twilio number's voice_url, so there's nothing to restore.

Expand source code

async def disconnect(self) -> None:
    """Restore prior voice_url (answer mode only), tear down server.

    Best-effort on errors. In caller mode we never touched the Twilio
    number's voice_url, so there's nothing to restore.
    """
    if self._rest is None:
        return

    # 1. Restore webhook first so Twilio doesn't keep hitting a dead URL.
    # ``rest_restore_failed`` tracks whether either restore below raised —
    # previously fully swallowed via ``suppress(Exception)``; now surfaced
    # onto the disconnect span (#775 Tier 2b) so a Twilio REST outage during
    # teardown is no longer invisible. The swallow behavior itself (a failed
    # restore must never block disconnect()) is unchanged.
    # Disarm the duration watchdog first: from here on the call is ours to
    # end, and a timer that fired mid-teardown would hang up a SID this
    # adapter may already have replaced with a later call's.
    self._cancel_max_duration_timer()

    rest_restore_failed = False
    if self._mode == "answer" and self._phone_number_sid is not None:
        try:
            prior = self._prior_voice_url or ""
            self._rest.write_voice_url(self._phone_number_sid, prior)
            logger.debug(
                "TwilioAgentAdapter: restored voice_url=%r on %s",
                prior,
                self._phone_number_sid,
            )
        except Exception:
            rest_restore_failed = True
    # place_call() rewrites the CALLEE's voice_url to attach Media
    # Streams to B-leg. Restore that too.
    if self._mode == "call" and self._callee_phone_number_sid is not None:
        try:
            prior_b = self._prior_callee_voice_url or ""
            self._rest.write_voice_url(self._callee_phone_number_sid, prior_b)
            logger.debug(
                "TwilioAgentAdapter: restored callee voice_url=%r on %s",
                prior_b,
                self._callee_phone_number_sid,
            )
        except Exception:
            rest_restore_failed = True

    # 2. Signal server to shut down, then wait for the task. This is what
    # ACTUALLY closes a still-live media stream (uvicorn's graceful
    # shutdown tears down the open WS), which is what lets the media
    # loop's ``finally``/except-handlers (``_twilio_server.py``) make
    # their FINAL writes to ``_frames_received`` / ``_stream_ended_reason``
    # / etc. The counter stamp below MUST run after this await, not
    # before — stamping first (the #775 review-caught bug) reads the
    # pre-teardown snapshot and reports ``stream_ended_reason=="none"`` /
    # undercounted frames for a call that was still live when disconnect()
    # was invoked, even though the call did in fact end (via this very
    # shutdown) moments later.
    if self._server_shutdown is not None:
        self._server_shutdown.set()
    if self._server_task is not None:
        with suppress(Exception):
            await asyncio.wait_for(self._server_task, timeout=3.0)

    # Stamp call-lifetime counters onto the active ``voice.adapter.disconnect``
    # span (opened by the executor disconnect loop) — mirrors ElevenLabs'
    # pump-counter stamp (``adapters/elevenlabs.py``). Sampled AFTER the
    # shutdown-await above (not before) so a still-live call's FINAL
    # counts/reason are captured, not a mid-call snapshot. The disconnect
    # span stays the current span across the await (contextvars survive
    # awaits within the same task), so ``get_current_span()`` here still
    # targets it.
    set_span_attributes(
        _otel_trace.get_current_span(),
        {
            "voice.twilio.frames_received": self._frames_received,
            "voice.twilio.dtmf_received": self._dtmf_received,
            "voice.twilio.stream_ended_reason": self._stream_ended_reason,
            "voice.twilio.webhook_invocations": self._webhook_invocations,
            "voice.twilio.webhook_rejected": self._webhook_rejected,
            "voice.twilio.rest_restore_failed": rest_restore_failed,
        },
    )

    # 3. Reset state.
    self._rest = None
    self._phone_number_sid = None
    self._prior_voice_url = None
    self._callee_phone_number_sid = None
    self._prior_callee_voice_url = None
    self._mode = "idle"
    self._server_task = None
    self._server_shutdown = None
    self._call_sid = None
    self._stream_sid = None
    self._stream_connected = None
    self._stream_ws = None
    self._stream_nonce = None
    self._cancel_max_duration_timer()
    if self._inbound_queue is not None:
        # A recv_audio() task may already be blocked in get() on this queue.
        # Wake that in-flight consumer before dropping our reference; setting
        # _inbound_queue to None alone leaves it parked until its receive
        # timeout expires (#757). An empty chunk is the adapter's existing
        # end-of-stream signal and lets the shared drain finish cleanly.
        self._inbound_queue.put_nowait(AudioChunk(data=b""))
    self._inbound_queue = None
    self._stream_ended = False
    self._frames_received = 0
    self._dtmf_received = 0
    self._stream_ended_reason = "none"
    self._webhook_invocations = 0
    self._webhook_rejected = 0

async def interrupt(self) ‑> None

Drop any buffered outbound audio on Twilio's side (clear event).

Expand source code

async def interrupt(self) -> None:
    """Drop any buffered outbound audio on Twilio's side (``clear`` event)."""
    self._assert_stream_live()
    ws = self._stream_ws
    stream_sid = self._stream_sid
    assert ws is not None and stream_sid is not None
    await ws.send_text(build_clear_frame(stream_sid))

async def place_call(self, to: str, *, timeout: float = 120.0, attach_stream_to_self: Optional[bool] = None, attach_stream: "Optional[Literal['a-leg', 'b-leg']]" = None, max_call_duration_seconds: Optional[int] = None) ‑> None

Originate an outbound call from this adapter's Twilio number to to.

Twilio's REST Calls.create runs TwiML on TWO legs of the resulting call:

• A-leg (the originator, from_=self.phone_number): runs the inline TwiML passed via twiml= to Calls.create. We use <Pause length=120> so the originator just holds the bridge open while the demo runs.
• B-leg (the callee, to=): when Twilio dials B and B picks up, B's number's voice_url fires — that's where the bridge's Media Streams attach. This is identical to the inbound demo's flow: B's voice_url returns <Connect><Stream>, the WS opens, audio flows.

So place_call only makes sense when to is another Twilio number on this account whose voice_url is set to OUR harness webhook. To make that wiring automatic, place_call temporarily rewrites B's voice_url for the duration of the call and restores it on disconnect. The harness on this adapter's own number does NOT need to be answer-mode — the Stream attaches to B's leg, NOT this adapter's leg, but B's webhook is hosted on this adapter's local server, so the WS still lands here.

The bidirectional audio model is unchanged: send_audio writes frames over the WS (B hears them and bridges to A), recv_audio reads inbound frames off the WS (whatever the bridge mixes from both legs).

Limitation of the default (b-leg) mode: to MUST be a phone number on this same Twilio account. Calling an external PSTN endpoint (a real cell phone) is instead served by a-leg mode.

Stream-attach mode is selected by attach_stream:

• "b-leg" — today's default: rewrite the callee's voice_url, hold the A-leg open with <Say>…<Pause>, restore on disconnect. Only works for numbers this account owns.
• "a-leg" — originate with inline <Connect><Stream> so the Media Stream rides OUR own leg. Touches nothing on the callee (no resolve_phone_number_sid/read_voice_url/write_voice_url), so to can be any external number. Because that reaches numbers this account does not own, to must appear in the adapter's allowed_callees (deny-by-default) and the public base URL must be reachable from the edge — both checked before origination.

The legacy attach_stream_to_self bool is still honoured but superseded by attach_stream: True/unset → "b-leg", False → originator-only (a third mode attach_stream cannot express). Passing both with disagreeing intent raises ValueError.

Default timeout 120s covers cloudflared cold-start latency.

timeout and max_call_duration_seconds are two different clocks and are configured independently:

• timeout bounds how long we WAIT FOR THE MEDIA STREAM TO CONNECT after origination. It says nothing about how long the call may run.
• max_call_duration_seconds bounds HOW LONG THE CALL MAY LIVE (a-leg only, default DEFAULT_MAX_CALL_DURATION_SECONDS, hard cap MAX_CALL_DURATION_CAP_SECONDS; a larger request raises). Under a-leg's <Connect> the call lives exactly as long as the WebSocket, so without a ceiling a hung executor keeps a billing PSTN call open forever. Two mechanisms enforce it: Twilio's own TimeLimit on Calls.create (the load-bearing half — it still fires if this process hangs or is killed) and an adapter-side wall-clock timer that hangs the call up via REST and closes the socket.

Raises

RuntimeError

If called after wait_for_call() (modes are exclusive per adapter instance), or if to is not a Twilio number on this account.

ValueError

If to is not in E.164 format, or — in a-leg mode — is not on allowed_callees.

TunnelNotReadyError

If a-leg mode's readiness probe says the public base URL is not reachable from the edge yet.

asyncio.TimeoutError

If the media stream doesn't open within timeout seconds.

Expand source code

async def place_call(
    self,
    to: str,
    *,
    timeout: float = 120.0,
    attach_stream_to_self: Optional[bool] = None,
    attach_stream: Optional[Literal["a-leg", "b-leg"]] = None,
    max_call_duration_seconds: Optional[int] = None,
) -> None:
    """
    Originate an outbound call from this adapter's Twilio number to ``to``.

    Twilio's REST ``Calls.create`` runs TwiML on TWO legs of the
    resulting call:

    - **A-leg** (the originator, ``from_=self.phone_number``): runs the
      inline TwiML passed via ``twiml=`` to ``Calls.create``. We use
      ``<Pause length=120>`` so the originator just holds the bridge
      open while the demo runs.
    - **B-leg** (the callee, ``to=``): when Twilio dials B and B picks
      up, B's number's ``voice_url`` fires — that's where the bridge's
      Media Streams attach. This is identical to the inbound demo's
      flow: B's voice_url returns ``<Connect><Stream>``, the WS opens,
      audio flows.

    So ``place_call`` only makes sense when ``to`` is another Twilio
    number on this account whose ``voice_url`` is set to OUR harness
    webhook. To make that wiring automatic, ``place_call`` temporarily
    rewrites B's ``voice_url`` for the duration of the call and
    restores it on ``disconnect``. The harness on this adapter's own
    number does NOT need to be answer-mode — the Stream attaches to
    B's leg, NOT this adapter's leg, but B's webhook is hosted on this
    adapter's local server, so the WS still lands here.

    The bidirectional audio model is unchanged: ``send_audio`` writes
    frames over the WS (B hears them and bridges to A), ``recv_audio``
    reads inbound frames off the WS (whatever the bridge mixes from
    both legs).

    Limitation of the default (**b-leg**) mode: ``to`` MUST be a phone
    number on this same Twilio account. Calling an external PSTN endpoint
    (a real cell phone) is instead served by **a-leg** mode.

    Stream-attach mode is selected by ``attach_stream``:

    - ``"b-leg"`` — today's default: rewrite the callee's ``voice_url``,
      hold the A-leg open with ``<Say>…<Pause>``, restore on disconnect.
      Only works for numbers this account owns.
    - ``"a-leg"`` — originate with inline ``<Connect><Stream>`` so the
      Media Stream rides OUR own leg. Touches nothing on the callee (no
      ``resolve_phone_number_sid``/``read_voice_url``/``write_voice_url``),
      so ``to`` can be any external number. Because that reaches numbers
      this account does not own, ``to`` must appear in the adapter's
      ``allowed_callees`` (deny-by-default) and the public base URL must be
      reachable from the edge — both checked before origination.

    The legacy ``attach_stream_to_self`` bool is still honoured but
    superseded by ``attach_stream``: ``True``/unset → ``"b-leg"``,
    ``False`` → originator-only (a third mode ``attach_stream`` cannot
    express). Passing both with disagreeing intent raises ``ValueError``.

    Default timeout 120s covers cloudflared cold-start latency.

    ``timeout`` and ``max_call_duration_seconds`` are two different clocks
    and are configured independently:

    - ``timeout`` bounds how long we WAIT FOR THE MEDIA STREAM TO CONNECT
      after origination. It says nothing about how long the call may run.
    - ``max_call_duration_seconds`` bounds HOW LONG THE CALL MAY LIVE
      (a-leg only, default ``DEFAULT_MAX_CALL_DURATION_SECONDS``, hard cap
      ``MAX_CALL_DURATION_CAP_SECONDS``; a larger request raises). Under
      a-leg's ``<Connect>`` the call lives exactly as long as the WebSocket,
      so without a ceiling a hung executor keeps a billing PSTN call open
      forever. Two mechanisms enforce it: Twilio's own ``TimeLimit`` on
      ``Calls.create`` (the load-bearing half — it still fires if this
      process hangs or is killed) and an adapter-side wall-clock timer that
      hangs the call up via REST and closes the socket.

    Raises:
        RuntimeError: If called after ``wait_for_call()`` (modes are
            exclusive per adapter instance), or if ``to`` is not a
            Twilio number on this account.
        ValueError: If ``to`` is not in E.164 format, or — in a-leg mode —
            is not on ``allowed_callees``.
        TunnelNotReadyError: If a-leg mode's readiness probe says the
            public base URL is not reachable from the edge yet.
        asyncio.TimeoutError: If the media stream doesn't open within
            ``timeout`` seconds.
    """
    self._assert_connected()
    self._enter_mode("call")
    validate_e164(to)

    # Resolve the effective stream-attach mode BEFORE any REST call so a
    # conflicting-parameter caller error surfaces before we dial. `attach_stream`
    # (typed) supersedes the legacy `attach_stream_to_self` bool; the bool's
    # third mode (originator-only) is unreachable via `attach_stream` by design.
    mode = _resolve_stream_mode(attach_stream, attach_stream_to_self)
    # Only a-leg loses <Pause>'s implicit ceiling, so only a-leg carries a
    # duration cap. Naming one in another mode is rejected rather than
    # ignored: silently dropping it would leave the caller believing the
    # call is bounded when it is not.
    if mode == "a-leg":
        # Guardrail (c): destinations are deny-by-default in a-leg mode.
        # Checked here, with the other local caller-fixable failures, so a
        # misconfigured allowlist costs nothing and dials nothing.
        self._assert_callee_allowed(to)
        max_call_duration = resolve_max_call_duration(max_call_duration_seconds)
    elif max_call_duration_seconds is not None:
        raise ValueError(
            "place_call: max_call_duration_seconds is only supported with "
            'attach_stream="a-leg"; b-leg and originator-only modes hold the '
            "originator leg with <Pause> and are bounded by it."
        )
    else:
        max_call_duration = None

    # Guardrail (c), second half: a-leg hands Twilio our public URL and
    # Twilio opens the media socket against it seconds later. Probe the edge
    # AFTER the free local checks above and BEFORE origination, so an
    # unreachable tunnel is a named error instead of a billed call that
    # dies in a stream-connect timeout.
    if mode == "a-leg":
        await self._assert_tunnel_ready()

    assert self.public_base_url is not None
    assert self._rest is not None
    assert self._stream_connected is not None

    # NEW ``voice.adapter.dial`` span (#775 Tier 2a): self-instrumented,
    # since no executor span is active by the time place_call()/
    # wait_for_call() run (the ``voice.adapter.connect`` span already
    # closed). Wraps the REST dial + the ``_stream_connected`` wait — the
    # "I placed the call but no media ever streamed" failure surface.
    with voice_span(
        "voice.adapter.dial",
        {
            "voice.adapter.class": type(self).__name__,
            "voice.twilio.direction": "outbound",
            "voice.twilio.to": _redact_e164(to),
            "voice.twilio.from": _redact_e164(self.phone_number),
        },
    ) as _dial:
        # #762 P1: bump the call generation and reset per-dial connection
        # state BEFORE originating, so a socket that adopted un-gated before
        # this dial is evicted and cannot resolve the wait below.
        self._reset_connection_state_for_dial()
        if mode == "b-leg":
            # Resolve B-leg's number SID and snapshot+rewrite its voice_url so
            # B's leg attaches its Media Stream to our harness webhook. We own
            # this number (same Twilio account); disconnect() will restore.
            self._callee_phone_number_sid = self._rest.resolve_phone_number_sid(to)
            self._prior_callee_voice_url = self._rest.read_voice_url(
                self._callee_phone_number_sid
            )
            webhook_url = self.public_base_url.rstrip("/") + "/twilio/voice"
            self._rest.write_voice_url(self._callee_phone_number_sid, webhook_url)
            logger.info(
                "TwilioAgentAdapter: rewrote callee %s voice_url to %s",
                _redact_e164(to),
                webhook_url,
            )

        # a-leg: the Media Stream rides OUR own leg via inline
        # <Connect><Stream>, so `to` can be any external number and we
        # touch nothing on the callee. Same TwiML shape the inbound webhook
        # returns (`_twilio_server.py`), now on origination.
        #
        # Guardrail (a): the socket is the only inbound signal on this path,
        # so it must authenticate itself. Mint a per-call CSPRNG nonce and
        # ship it as a <Parameter> child; Twilio echoes it back in the
        # `start` frame's customParameters, and the media loop
        # (`_twilio_server.py`) closes any socket that cannot present it.
        #
        # Assigned UNCONDITIONALLY (mirrors the JS twin): the field also
        # ARMS nonce enforcement and the a-leg send_dtmf refusal, so a b-leg
        # dial that left an earlier a-leg call's nonce standing would gate a
        # socket whose `start` frame carries no <Parameter> at all — the
        # b-leg call would never connect.
        nonce = mint_stream_nonce() if mode == "a-leg" else None
        self._stream_nonce = nonce
        if nonce is not None:
            origination_twiml = _build_connect_stream_twiml(
                stream_ws_url(self.public_base_url),
                stream_parameters={"nonce": nonce},
            )
        else:
            # A-leg (originator-side) TwiML for b-leg / originator-only modes:
            # play a short deterministic <Say> line, then hold the bridge
            # open. Twilio runs this on the originator side while B's webhook
            # attaches the Media Stream.
            #
            # The <Say> gives the recording a known-good utterance to
            # transcribe. A bare <Pause> alone produces 120s of line silence
            # that Whisper has been observed to hallucinate as non-English
            # text (issue #465 in this PR). The Say is a one-time anchor at
            # call setup; the Media Stream carries the real bidirectional
            # conversation that follows.
            origination_twiml = (
                '<?xml version="1.0" encoding="UTF-8"?>'
                "<Response>"
                f'<Say voice="Polly.Joanna">{PLACE_CALL_A_LEG_SAY_TEXT}</Say>'
                '<Pause length="120"/>'
                "</Response>"
            )
        self._call_sid = self._rest.place_call(
            to=to,
            from_=self.phone_number,
            twiml=origination_twiml,
            time_limit=max_call_duration,
        )
        logger.info(
            "TwilioAgentAdapter: placed call %s from %s to %s",
            self._call_sid,
            _redact_e164(self.phone_number),
            _redact_e164(to),
        )
        set_span_attributes(_dial, {"voice.twilio.call_sid": self._call_sid})

        if max_call_duration is not None:
            self._arm_max_duration_timer(max_call_duration, self._call_sid, to)

        if mode != "originator-only":
            # Wait for the media stream to reach us. In b-leg mode it arrives
            # via the callee's rewritten voice_url; in a-leg mode it rides our
            # own <Connect><Stream> leg. In originator-only mode there's no
            # stream coming to us; the callee has its own harness which owns it.
            _dial_wait_started = time.monotonic()
            try:
                await asyncio.wait_for(self._stream_connected.wait(), timeout=timeout)
            except asyncio.TimeoutError:
                # The media stream never connected — the marquee "I placed
                # the call but no media ever streamed" failure. Tag the
                # outcome BEFORE re-raising; the raised TimeoutError still
                # marks the span ERROR (voice_span's own exception handling).
                _dial.set_attribute(
                    "voice.twilio.dial_outcome", "stream_connect_timeout"
                )
                if mode == "a-leg" and self._call_sid is not None:
                    # The call is already originated and, under a-leg's
                    # <Connect>, lives to the duration cap even though
                    # nothing will ever stream on it. Hang it up before
                    # handing the caller their TimeoutError — otherwise the
                    # ordinary connect-failure path bills for the cap.
                    await self._abandon_originated_call(self._call_sid)
                raise
            set_span_attributes(
                _dial,
                {
                    "voice.twilio.stream_connect_latency_ms": round(
                        (time.monotonic() - _dial_wait_started) * 1000
                    ),
                },
            )

async def send_dtmf(self, tones: str) ‑> None

Send DTMF digits on the live call (uses Twilio REST <Play digits>).

Refused in a-leg mode (#762 AC10) — see _assert_dtmf_supported.

Expand source code

async def send_dtmf(self, tones: str) -> None:
    """Send DTMF digits on the live call (uses Twilio REST ``<Play digits>``).

    Refused in a-leg mode (#762 AC10) — see ``_assert_dtmf_supported``.
    """
    self._assert_dtmf_supported()
    if self._rest is None or self._call_sid is None:
        raise RuntimeError("TwilioAgentAdapter: no active call; send_dtmf requires an in-progress call")
    # Run blocking REST call off-thread so we don't stall the event loop.
    await asyncio.to_thread(self._rest.send_dtmf_on_call, self._call_sid, tones)

async def wait_for_call(self, timeout: float = 120.0) ‑> None

Block until someone dials in and the media stream is live.

In default mode (no conference_room): the number's voice_url is overwritten to point at our webhook so inbound calls reach us. Caller (place_call) elsewhere will dial this number.

In conference mode: there's no inbound call to wait for — the two-Twilio-number demo can't naturally have one side receive an inbound call when both legs need conference TwiML. Instead, we ORIGINATE an outbound call FROM this adapter's number TO itself with inline TwiML that opens the capture stream and dials into the shared conference room. This is the symmetric counterpart to place_call in conference mode — both adapters end up as participants in the same room, exchanging audio via the bridge.

Default timeout 120s covers cloudflared cold-start, conference-room formation, and Twilio's webhook ramp.

Raises

RuntimeError

If called after place_call().

asyncio.TimeoutError

If nobody dials in within timeout.

Expand source code

async def wait_for_call(self, timeout: float = 120.0) -> None:
    """
    Block until someone dials in and the media stream is live.

    In **default mode** (no conference_room): the number's ``voice_url``
    is overwritten to point at our webhook so inbound calls reach us.
    Caller (``place_call``) elsewhere will dial this number.

    In **conference mode**: there's no inbound call to wait for — the
    two-Twilio-number demo can't naturally have one side receive an
    inbound call when both legs need conference TwiML. Instead, we
    ORIGINATE an outbound call FROM this adapter's number TO itself
    with inline TwiML that opens the capture stream and dials into
    the shared conference room. This is the symmetric counterpart to
    ``place_call`` in conference mode — both adapters end up as
    participants in the same room, exchanging audio via the bridge.

    Default timeout 120s covers cloudflared cold-start, conference-room
    formation, and Twilio's webhook ramp.

    Raises:
        RuntimeError: If called after ``place_call()``.
        asyncio.TimeoutError: If nobody dials in within ``timeout``.
    """
    self._assert_connected()
    self._enter_mode("answer")

    assert self.public_base_url is not None
    assert self._rest is not None
    assert self._phone_number_sid is not None
    assert self._stream_connected is not None

    # NEW ``voice.adapter.dial`` span (#775 Tier 2a) — see place_call()'s
    # comment for the rationale (self-instrumented, no executor span active
    # here).
    with voice_span(
        "voice.adapter.dial",
        {
            "voice.adapter.class": type(self).__name__,
            "voice.twilio.direction": "inbound",
            "voice.twilio.to": _redact_e164(self.phone_number),
        },
    ) as _dial:
        # #762 P1: bump the call generation and reset per-dial connection
        # state, symmetric with place_call. For inbound this only guards a
        # reconnect/back-to-back dial; a fresh inbound socket still connects
        # exactly as before.
        self._reset_connection_state_for_dial()
        # Snapshot the prior webhook so we can restore it on disconnect, then
        # point the number at our server. Only answer mode does this.
        self._prior_voice_url = self._rest.read_voice_url(self._phone_number_sid)
        webhook_url = self.public_base_url.rstrip("/") + "/twilio/voice"
        self._rest.write_voice_url(self._phone_number_sid, webhook_url)
        logger.info("TwilioAgentAdapter: webhook set to %s", webhook_url)

        _dial_wait_started = time.monotonic()
        try:
            await asyncio.wait_for(self._stream_connected.wait(), timeout=timeout)
        except asyncio.TimeoutError:
            # Nobody dialed in — tag the outcome BEFORE re-raising; the
            # raised TimeoutError still marks the span ERROR (voice_span's
            # own exception handling).
            _dial.set_attribute(
                "voice.twilio.dial_outcome", "stream_connect_timeout"
            )
            raise
        set_span_attributes(
            _dial,
            {
                "voice.twilio.call_sid": self._call_sid,
                "voice.twilio.stream_connect_latency_ms": round(
                    (time.monotonic() - _dial_wait_started) * 1000
                ),
            },
        )

Inherited members

• VoiceAgentAdapter:
  • agent_hung_up
  • call
  • is_connected
  • recv_audio
  • send_audio

class VapiAgentAdapter (assistant_id: str, api_key: str)

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

Expand source code

class VapiAgentAdapter(VoiceAgentAdapter):
    capabilities: ClassVar[AdapterCapabilities] = AdapterCapabilities(
        streaming_transcripts=True,
        native_vad=True,
        dtmf=False,
        input_formats=["pcm16/16000"],
        output_formats=["pcm16/16000"],
    )

    def __init__(self, assistant_id: str, api_key: str):
        super().__init__()
        self.assistant_id = assistant_id
        self.api_key = api_key
        self.websocket_call_url: Optional[str] = None
        self._ws: Optional[object] = None

    async def connect(self) -> None:
        # Integration: POST to Vapi REST API to get websocketCallUrl, then
        # open websocket.
        self.websocket_call_url = f"wss://vapi.ai/ws/{self.assistant_id}"
        self._ws = object()

    async def disconnect(self) -> None:
        self._ws = None

    async def send_audio(self, chunk: AudioChunk) -> None:
        if self._ws is None:
            raise RuntimeError("VapiAgentAdapter: not connected")
        raise PendingTransportError("VapiAgentAdapter")

    async def recv_audio(self, timeout: float) -> AudioChunk:
        if self._ws is None:
            raise RuntimeError("VapiAgentAdapter: not connected")
        raise PendingTransportError("VapiAgentAdapter")

    def __repr__(self) -> str:  # redact credentials
        return f"VapiAgentAdapter(assistant_id={self.assistant_id!r}, api_key='***')"

Ancestors

• VoiceAgentAdapter
• AgentAdapter
• abc.ABC

Class variables

var capabilities : ClassVar[AdapterCapabilities]

Inherited members

• VoiceAgentAdapter:
  • agent_hung_up
  • call
  • connect
  • disconnect
  • interrupt
  • is_connected
  • recv_audio
  • send_audio

class WebRTCAgentAdapter (signaling_url: str)

Generic WebRTC adapter that negotiates via an HTTP signaling URL.

Expand source code

class WebRTCAgentAdapter(VoiceAgentAdapter):
    """Generic WebRTC adapter that negotiates via an HTTP signaling URL."""

    capabilities: ClassVar[AdapterCapabilities] = AdapterCapabilities(
        streaming_transcripts=False,
        native_vad=False,
        dtmf=False,
        input_formats=["pcm16/24000"],
        output_formats=["pcm16/24000"],
    )

    def __init__(self, signaling_url: str):
        super().__init__()
        self.signaling_url = signaling_url
        self._pc: Optional[Any] = None
        self._inbound_audio: "asyncio.Queue[AudioChunk]" = asyncio.Queue()

    async def connect(self) -> None:
        # Deferred: actual SDP exchange requires a reachable signaling server.
        # Tested at @integration level with a loopback aiortc peer.
        self._pc = object()  # sentinel — mark "connected"

    async def disconnect(self) -> None:
        self._pc = None

    async def send_audio(self, chunk: AudioChunk) -> None:
        if self._pc is None:
            raise RuntimeError(f"{type(self).__name__}: not connected")
        raise PendingTransportError(type(self).__name__)

    async def recv_audio(self, timeout: float) -> AudioChunk:
        if self._pc is None:
            raise RuntimeError(f"{type(self).__name__}: not connected")
        # If a subclass populated _inbound_audio, return from it.
        if not self._inbound_audio.empty():
            return await asyncio.wait_for(self._inbound_audio.get(), timeout=timeout)
        raise PendingTransportError(type(self).__name__)

Ancestors

• VoiceAgentAdapter
• AgentAdapter
• abc.ABC

Class variables

var capabilities : ClassVar[AdapterCapabilities]

Inherited members

• VoiceAgentAdapter:
  • agent_hung_up
  • call
  • connect
  • disconnect
  • interrupt
  • is_connected
  • recv_audio
  • send_audio

class WebSocketAgentAdapter (url: str, protocol: WebSocketProtocol)

Connects to an arbitrary WebSocket endpoint using a user-supplied protocol.

The protocol's encode_audio is called before sending; decode_response is called on each inbound frame until an AudioChunk is produced.

Expand source code

class WebSocketAgentAdapter(VoiceAgentAdapter):
    """
    Connects to an arbitrary WebSocket endpoint using a user-supplied protocol.

    The protocol's ``encode_audio`` is called before sending; ``decode_response``
    is called on each inbound frame until an AudioChunk is produced.
    """

    capabilities: ClassVar[AdapterCapabilities] = AdapterCapabilities(
        streaming_transcripts=False,
        native_vad=False,
        dtmf=False,
        input_formats=["pcm16/24000"],
        output_formats=["pcm16/24000"],
    )

    def __init__(self, url: str, protocol: WebSocketProtocol):
        super().__init__()
        self.url = url
        self.protocol = protocol
        self._ws: Optional[Any] = None

    async def connect(self) -> None:
        import websockets  # hard dep

        self._ws = await websockets.connect(self.url)

    async def disconnect(self) -> None:
        if self._ws is not None:
            await self._ws.close()
            self._ws = None

    async def send_audio(self, chunk: AudioChunk) -> None:
        if self._ws is None:
            raise RuntimeError(f"{type(self).__name__}: not connected")
        payload = self.protocol.encode_audio(chunk.data)
        await self._ws.send(payload)

    async def recv_audio(self, timeout: float) -> AudioChunk:
        """Loop inbound frames until the protocol decodes an audio chunk.

        A clean server close (end of stream) with no final audio frame is a
        terminal, not an error: ``recv_audio`` returns an empty ``AudioChunk``
        so the base ``_drain_agent_response`` loop exits cleanly (issue #648),
        mirroring the #646/PR647 reference pattern and the Gemini Live / Pipecat
        idiom. ``asyncio.TimeoutError`` is still raised on inter-message silence.
        """
        import websockets  # for the ConnectionClosed terminal (issue #648)

        if self._ws is None:
            raise RuntimeError(f"{type(self).__name__}: not connected")
        loop = asyncio.get_running_loop()
        deadline = loop.time() + timeout
        while True:
            remaining = max(0.0, deadline - loop.time())
            try:
                message = await asyncio.wait_for(self._ws.recv(), timeout=remaining)
            except websockets.exceptions.ConnectionClosed:
                # End of stream: the server closed without a trailing audio
                # frame. Surface a clean terminal rather than letting
                # ConnectionClosed propagate — the drain only catches
                # asyncio.TimeoutError, so an unhandled close crashes the turn.
                return AudioChunk(data=b"")
            chunk = self.protocol.decode_response(message)
            if chunk is not None:
                return chunk

Ancestors

• VoiceAgentAdapter
• AgentAdapter
• abc.ABC

Class variables

var capabilities : ClassVar[AdapterCapabilities]

Methods

async def recv_audio(self, timeout: float) ‑> AudioChunk

Loop inbound frames until the protocol decodes an audio chunk.

A clean server close (end of stream) with no final audio frame is a terminal, not an error: recv_audio returns an empty AudioChunk so the base _drain_agent_response loop exits cleanly (issue #648), mirroring the #646/PR647 reference pattern and the Gemini Live / Pipecat idiom. asyncio.TimeoutError is still raised on inter-message silence.

Expand source code

async def recv_audio(self, timeout: float) -> AudioChunk:
    """Loop inbound frames until the protocol decodes an audio chunk.

    A clean server close (end of stream) with no final audio frame is a
    terminal, not an error: ``recv_audio`` returns an empty ``AudioChunk``
    so the base ``_drain_agent_response`` loop exits cleanly (issue #648),
    mirroring the #646/PR647 reference pattern and the Gemini Live / Pipecat
    idiom. ``asyncio.TimeoutError`` is still raised on inter-message silence.
    """
    import websockets  # for the ConnectionClosed terminal (issue #648)

    if self._ws is None:
        raise RuntimeError(f"{type(self).__name__}: not connected")
    loop = asyncio.get_running_loop()
    deadline = loop.time() + timeout
    while True:
        remaining = max(0.0, deadline - loop.time())
        try:
            message = await asyncio.wait_for(self._ws.recv(), timeout=remaining)
        except websockets.exceptions.ConnectionClosed:
            # End of stream: the server closed without a trailing audio
            # frame. Surface a clean terminal rather than letting
            # ConnectionClosed propagate — the drain only catches
            # asyncio.TimeoutError, so an unhandled close crashes the turn.
            return AudioChunk(data=b"")
        chunk = self.protocol.decode_response(message)
        if chunk is not None:
            return chunk

Inherited members

• VoiceAgentAdapter:
  • agent_hung_up
  • call
  • connect
  • disconnect
  • interrupt
  • is_connected
  • send_audio

class WebSocketProtocol

Encoder/decoder pair for a custom WebSocket audio protocol.

Expand source code

class WebSocketProtocol(ABC):
    """Encoder/decoder pair for a custom WebSocket audio protocol."""

    @abstractmethod
    def encode_audio(self, audio: bytes) -> Any:
        """Convert PCM16 bytes into the wire representation the server expects."""

    @abstractmethod
    def decode_response(self, message: Any) -> Optional[AudioChunk]:
        """Parse a server message into an AudioChunk, or None if it's not audio."""

Ancestors

• abc.ABC

Methods

def decode_response(self, message: Any) ‑> AudioChunk | None

Parse a server message into an AudioChunk, or None if it's not audio.

Expand source code

@abstractmethod
def decode_response(self, message: Any) -> Optional[AudioChunk]:
    """Parse a server message into an AudioChunk, or None if it's not audio."""

def encode_audio(self, audio: bytes) ‑> Any

Convert PCM16 bytes into the wire representation the server expects.

Expand source code

@abstractmethod
def encode_audio(self, audio: bytes) -> Any:
    """Convert PCM16 bytes into the wire representation the server expects."""