diff --git a/src/eva/backend/__init__.py b/src/eva/backend/__init__.py index b9513915..aeb94b61 100644 --- a/src/eva/backend/__init__.py +++ b/src/eva/backend/__init__.py @@ -1,13 +1,19 @@ -"""Provider-agnostic ``Backend`` abstraction (design-only, Step 1 of the refactor). +"""Provider-agnostic ``Backend`` abstraction (see ``docs/refactor-step1.md``). -This package defines the contracts described in ``docs/refactor-step1.md``: -pure API/session objects (``Backend``) that know nothing about role -(assistant vs. user), plus a factory to construct them. Nothing in this -package is wired into the existing ``eva.assistant`` / ``eva.user_simulator`` -code yet -- these are new, additive, currently-unused types. +This package defines pure API/session objects (``Backend``) that know nothing +about role (assistant vs. user), plus a ``BackendFactory`` to construct them. +The worker builds a backend per conversation and drives it through an +``AssistantRole`` / ``UserRole`` for every provider the factory supports. """ -from eva.backend.base import Backend, BackendEvent, BackendEventType, ToolCallRequest, ToolCallResult +from eva.backend.base import ( + Backend, + BackendEvent, + BackendEventType, + BackendSession, + ToolCallRequest, + ToolCallResult, +) from eva.backend.capabilities import BackendCapabilities from eva.backend.factory import BackendFactory @@ -17,6 +23,7 @@ "BackendEvent", "BackendEventType", "BackendFactory", + "BackendSession", "ToolCallRequest", "ToolCallResult", ] diff --git a/src/eva/backend/base.py b/src/eva/backend/base.py index 63bbd278..7d8a8aff 100644 --- a/src/eva/backend/base.py +++ b/src/eva/backend/base.py @@ -1,8 +1,9 @@ """Abstract ``Backend`` contract: pure API/session exchange, no role knowledge. -DESIGN ONLY (Step 1 of the refactor, see docs/refactor-step1.md). This module -defines shapes, not behavior -- every method body is a stub. Nothing in -``eva.assistant`` or ``eva.user_simulator`` depends on this yet. +See docs/refactor-step1.md. This is the live ``Backend`` contract -- +implemented by ``eva.backend.openai_realtime`` and driven by ``AssistantRole`` +/ ``UserRole``. The worker builds one per conversation via ``BackendFactory`` +for every provider the factory supports. A ``Backend`` wraps exactly one provider integration (OpenAI Realtime, Gemini Live, ElevenLabs Agents, a cascade STT->LLM->TTS pipeline, ...) and exposes a @@ -37,11 +38,16 @@ class BackendEventType(StrEnum): """Kinds of events a ``Backend`` can surface via ``receive()``. - Not every ``Backend`` implementation will emit every event type -- a thin, + Every event is normalized: typed fields (``audio`` / ``transcript`` / + ``tool_call_request`` / ``error``) plus normalized ``metadata`` scalars. + Backends never surface raw provider event objects -- all provider-specific + parsing happens inside the backend, so a ``Role`` consuming these stays + fully provider-agnostic. + + Not every ``Backend`` implementation emits every event type -- a thin, end-to-end backend (e.g. ElevenLabs Agents) may only ever emit ``AUDIO_OUTPUT``, ``TRANSCRIPT``, ``TURN_END``, and ``ERROR``, because it - has no separable tool-calling seam of its own that the caller can observe - (tool calls, if any, happen inside the provider and are not surfaced). + has no separable tool-calling seam of its own that the caller can observe. Consumers must treat unhandled event types as ignorable, not as errors. """ @@ -50,9 +56,12 @@ class BackendEventType(StrEnum): simulated user's speech, depending on which role's Backend this is).""" TRANSCRIPT = "transcript" - """A (possibly partial) transcript of something spoken -- either the - backend's own output or, for backends that provide it, the other party's - input as heard by this backend's ASR.""" + """A finalized transcript of something spoken. ``transcript`` holds the + text; ``metadata`` carries normalized descriptors: ``stream`` is + ``"input"`` (what the backend heard from the inbound party) or ``"output"`` + (what the backend's own model said), and for input transcripts + ``metadata["failed"] = True`` marks a transcription failure (empty text). + These are normalized scalars, not raw provider payloads.""" TOOL_CALL_REQUEST = "tool_call_request" """The backend's model wants to invoke a tool. Only emitted by backends @@ -63,11 +72,49 @@ class BackendEventType(StrEnum): "tool execution stays role-side").""" TURN_END = "turn_end" - """The backend's model has finished its current turn (end-of-utterance / - end-of-response signal).""" + """The backend's model finished a response turn. ``transcript`` holds the + backend's best final text for the turn (the backend does any + provider-specific text selection internally). ``metadata`` carries + normalized scalars: ``cancelled`` (the turn was cancelled/interrupted before + completing), ``interrupted`` (the inbound party barged in over a partial + turn -- ``transcript`` is then the partial), ``has_function_calls`` (the + turn produced tool calls), and ``usage`` (``{"prompt_tokens", "completion_tokens"}`` + or ``None``). A consumer decides from these whether/how to record the turn; + no raw provider payload is exposed.""" + + INPUT_SPEECH_STARTED = "input_speech_started" + """The backend's VAD detected that the *inbound* party (whoever is talking + *to* this backend's model) started speaking. Role-agnostic: for an + ``AssistantRole`` backend the inbound party is the caller, for a + ``UserRole`` backend it is the assistant. Emitted only by backends whose + provider surfaces input-side voice-activity boundaries (native S2S realtime + APIs). Added by the OpenAI Realtime migration (docs/refactor-step1.md): the + assistant side needs these for user-turn timestamping, audio-track + alignment, and interrupted-response flushing, and the later turn-taking / + mediator work is built directly on input speech boundaries -- so they are + first-class events rather than ``metadata`` extras. Consumers that don't + care may ignore them like any other event type.""" + + INPUT_SPEECH_STOPPED = "input_speech_stopped" + """The backend's VAD detected that the inbound party stopped speaking. The + end-of-speech counterpart to ``INPUT_SPEECH_STARTED`` (see its docstring).""" + + OUTPUT_TURN_STARTED = "output_turn_started" + """The backend's model began a response turn. The normalized counterpart to + ``TURN_END`` at the start of a turn. Needed by a manually-sequencing + consumer (e.g. a ``UserRole`` gating replies) to know a response is now in + flight; consumers that don't care ignore it.""" + + OUTPUT_AUDIO_DONE = "output_audio_done" + """The backend's model finished emitting output audio for the current turn + (the audio stream is drained, distinct from ``TURN_END`` which also covers + the text/tool bookkeeping). Lets a consumer that paces or gates on playout + flush trailing output; ignorable otherwise.""" ERROR = "error" - """A provider-level error occurred (connection drop, API error, etc.).""" + """A provider-level error occurred (connection drop, API error, etc.). + ``error`` holds the message; ``metadata["code"]`` carries a normalized + error code when the provider supplies one.""" @dataclass @@ -121,33 +168,44 @@ class BackendEvent: tool_call_request: ToolCallRequest | None = None error: str | None = None metadata: dict[str, Any] = field(default_factory=dict) - """Provider-specific extras (e.g. raw event name, timestamps) that don't - warrant a first-class field. Consumers should not rely on specific keys - being present across providers. - - Convention (not enforced by this contract): a backend that proactively - re-engages after a dropped user turn (the turn-end fallback; see - ``AssistantRole``'s ``turn_end_fallback_seconds`` and the shipped - ``eva.assistant.pipeline.fallback``) tags the ``AUDIO_OUTPUT``/ - ``TRANSCRIPT`` event it emits for that turn so callers can distinguish a - fallback nudge from an ordinary model turn (e.g. for audit logging and so - downstream metrics can zero it). The shipped feature records the transcript - marker with ``message_type="turn_fallback"``; a backend surfacing the same - turn here should carry an equivalent flag in ``metadata`` (e.g. - ``metadata["turn_fallback"] = True``). This is *not* a new event type -- a - nudge is just an ordinary turn from the backend's model, triggered by the - backend noticing that a user turn was never detected within the fallback - window rather than by new input; it flows through the same ``receive()`` - surface as anything else.""" + """Normalized descriptors for this event -- plain scalars/dicts, never a raw + provider event object. Which keys are present depends on ``event_type`` and + is documented on each ``BackendEventType`` member (e.g. ``stream`` / + ``failed`` for ``TRANSCRIPT``; ``cancelled`` / ``interrupted`` / + ``has_function_calls`` / ``usage`` for ``TURN_END``; ``code`` for + ``ERROR``). A ``Role`` consumes these normalized fields only, so it stays + provider-agnostic: all provider-specific event parsing happens inside the + backend before emission.""" + + +class BackendSession: + """Opaque handle to one live provider session, returned by ``Backend.open``. + + The ``Backend`` itself is stateless beyond its construction config (model, + key, endpoint): *all* per-exchange state -- the live connection, any + provider-side accumulators -- lives on the session handle, not on the + backend. The caller (a ``Role``, or later a mediator) holds this handle and + passes it back into ``send`` / ``receive`` / ``close``. Concrete backends + subclass this with whatever they need to carry; consumers treat it as + opaque and never introspect it. + + Keeping session state off the backend is deliberate (see + docs/refactor-step1.md discussion): one ``Backend`` instance can then serve + many independent sessions/conversations concurrently, and no exchange data + is smuggled into the backend object. + """ class Backend(ABC): - """Pure API/session exchange with one provider. No role knowledge. + """Stateless adapter to one provider's API. No role knowledge, no session state. - Lifecycle: ``open()`` establishes the session, ``send()`` pushes audio / - text / tool results to the provider, ``receive()`` yields events back, - and ``close()`` tears the session down. A ``Role`` (see - ``eva.role.base``) owns one ``Backend`` instance and drives it. + Lifecycle: ``open()`` establishes a session and returns a + ``BackendSession`` handle, ``send()`` pushes audio / text / tool results to + the provider on a given session, ``receive()`` yields events back for a + session, and ``close()`` tears a session down. The backend holds only its + construction config; the caller holds the session handle. A ``Role`` (see + ``eva.role.base``) is given a ``Backend`` instance (constructed by the + worker via a ``BackendFactory`` the worker owns) and drives it. Implementations are expected to fall along a spectrum: @@ -180,39 +238,51 @@ def capabilities(self) -> BackendCapabilities: """ ... + @property + def input_sample_rate(self) -> int: + """Sample rate (Hz) of PCM the backend expects via ``send(audio=...)``. + + Lets a role convert/record counterparty audio without knowing the + provider. Defaults to 24 kHz (the common realtime rate); backends on a + different rate override. Describes the session's audio format, not turn + state. + """ + return 24000 + + @property + def output_sample_rate(self) -> int: + """Sample rate (Hz) of PCM carried in ``AUDIO_OUTPUT`` events. + + See ``input_sample_rate``; defaults to 24 kHz, overridden per backend. + """ + return 24000 + @abstractmethod - async def open(self, *, system_prompt: str, tools: list[dict[str, Any]] | None, config: dict[str, Any]) -> None: - """Establish the provider session. + async def open(self, *, system_prompt: str, tools: list[dict[str, Any]] | None) -> BackendSession: + """Establish a provider session and return its opaque handle. + + The role supplies only the two things that are genuinely its own -- the + prompt and the tool catalog. All provider-specific session shaping + (model, voice, sample rate, turn-detection, audio formats, ...) is the + backend's own construction config, injected by the worker via the + ``BackendFactory``; the role neither builds nor sees it. That is what + keeps a single generic ``Role`` usable with any backend. Args: - system_prompt: Fully-built system prompt for this session, as - assembled by the owning ``Role`` (``Role.build_prompt()``). - A thin end-to-end backend still receives this even if it - maps it onto a different provider concept (e.g. ElevenLabs - agent overrides). - tools: Tool schemas to expose to the provider's model, in - whatever wire format the concrete backend needs to translate - from the agent's tool definitions. ``None`` or ``[]`` for - backends/roles that don't expose tool calling (e.g. a - ``UserRole`` that only needs an ``end_call`` tool would still - pass that single tool here; a backend with no tool-calling - seam at all may simply ignore this argument). - config: Provider-specific configuration blob (model name, voice, - sample rate, turn-detection parameters, etc.). Deliberately - untyped here -- each concrete ``Backend`` defines and - validates its own config shape; the abstract contract does - not prescribe one, since a native S2S config and a cascade - config share little structure. An ``AssistantRole`` backend - configured for the turn-end fallback (see - ``AssistantRole.turn_end_fallback_seconds``) reads its - threshold from this blob (e.g. a - ``config["turn_end_fallback_seconds"]`` key) the same way -- - the fallback needs no dedicated typed parameter or new - ``Backend`` method, since the resulting nudge is just an - ordinary outbound turn (see ``BackendEvent.metadata``). - - Must be safe to call exactly once per ``Backend`` instance. Must not - block on the other party being ready to exchange data -- readiness to + system_prompt: Fully-built system prompt/instructions for this + session (the role's ``build_prompt()`` output). A thin + end-to-end backend still receives this even if it maps it onto a + different provider concept (e.g. ElevenLabs agent overrides). + tools: Provider-agnostic tool specs -- a list of + ``{"name", "description", "parameters"}`` dicts -- which the + backend translates into its provider's tool-schema wire format. + ``None`` or ``[]`` for roles that expose no tools; a backend + with no tool-calling seam may ignore this argument. + + Each call returns a fresh, independent ``BackendSession``; because the + backend carries no session state, a single ``Backend`` instance may be + opened many times (e.g. one session per conversation). Must not block + on the other party being ready to exchange data -- readiness to *accept* traffic is enough (mirrors today's ``AbstractAssistantServer.start()`` contract: non-blocking, returns once ready). @@ -222,14 +292,16 @@ async def open(self, *, system_prompt: str, tools: list[dict[str, Any]] | None, @abstractmethod async def send( self, + session: BackendSession, *, audio: bytes | None = None, text: str | None = None, tool_result: ToolCallResult | None = None, ) -> None: - """Push data to the provider. Exactly one of the keyword args is set. + """Push data to the provider on ``session``. Exactly one kwarg is set. Args: + session: The handle returned by ``open()`` for this exchange. audio: Raw input audio chunk (format/sample-rate is whatever this backend's ``open(config=...)`` declared; format conversion is the caller's responsibility via the shared audio utilities, @@ -252,8 +324,8 @@ async def send( ... @abstractmethod - def receive(self) -> AsyncIterator[BackendEvent]: - """Yield events from the provider as they arrive. + def receive(self, session: BackendSession) -> AsyncIterator[BackendEvent]: + """Yield events from the provider on ``session`` as they arrive. The single, symmetric inbound stream for both "network-server-like" and "client-like" backends. Must be an async generator (or return an @@ -265,12 +337,26 @@ def receive(self) -> AsyncIterator[BackendEvent]: """ ... + async def trigger_response(self, session: BackendSession) -> None: + """Ask the provider to generate a response now, with no new input. + + Only meaningful for backends whose turn detection is configured *not* + to auto-create responses, so the caller sequences replies itself (e.g. + a ``UserRole`` that gates when the simulated caller speaks). Backends + with no such control -- thin end-to-end providers, or any backend where + responses are always driven by input/tool-results -- leave this as the + default ``NotImplementedError``; consult ``capabilities`` / provider + docs before calling. Not abstract, so those backends need not implement + it. + """ + raise NotImplementedError(f"{type(self).__name__} does not support trigger_response()") + @abstractmethod - async def close(self) -> None: - """Tear down the provider session. + async def close(self, session: BackendSession) -> None: + """Tear down the given provider ``session``. - Must be safe to call even if ``open()`` was never called or the - session already ended on its own (idempotent). Concrete backends are + Must be safe to call even if the session already ended on its own + (idempotent). Concrete backends are responsible for their own provider-specific teardown (closing websockets, cancelling tasks, flushing buffers); this method does not itself define audio/output persistence -- that remains a ``Role`` diff --git a/src/eva/backend/factory.py b/src/eva/backend/factory.py index d0fa61b2..ed386b43 100644 --- a/src/eva/backend/factory.py +++ b/src/eva/backend/factory.py @@ -1,55 +1,59 @@ -"""Factory interface for constructing ``Backend`` instances by name. - -DESIGN ONLY (Step 1 of the refactor). Mirrors the shape of today's -``eva.user_simulator.factory.create_user_simulator`` (lazy per-provider -imports keyed off config type) but is provider-and-role-agnostic: the same -factory is meant to be usable to build a backend for either an -``AssistantRole`` or a ``UserRole``, since a ``Backend`` has no role -knowledge (that's the whole point of the split -- see docs/refactor-step1.md, -"lets any backend act as either role"). +"""Factory that constructs ``Backend`` instances by provider name. + +Mirrors the shape of ``eva.user_simulator.factory.create_user_simulator`` (lazy +per-provider imports keyed off a provider name) but is +provider-and-role-agnostic: the same factory builds a backend for either an +``AssistantRole`` or a ``UserRole``, since a ``Backend`` has no role knowledge +(that's the whole point of the split -- see docs/refactor-step1.md, "lets any +backend act as either role"). + +A single concrete class -- there is no abstract base, since there is only ever +one factory. ``create`` returns ``None`` for a provider that has not been +migrated onto the ``Backend`` contract; the worker uses that as the signal to +fall back to the legacy server/simulator for that provider. """ from __future__ import annotations -from abc import ABC, abstractmethod from typing import Any from eva.backend.base import Backend -class BackendFactory(ABC): +class BackendFactory: """Constructs a ``Backend`` for a named provider from a config blob. - A concrete implementation is expected to hold (or look up) a registry - mapping provider name -> ``Backend`` subclass, analogous to today's - ``create_user_simulator`` / assistant-server construction in - ``orchestrator/runner.py``, and to import each provider module lazily so - that unused providers' SDKs need not be installed/imported. + Providers are added as their backends are migrated onto the ``Backend`` + contract: add a lazy-import branch in ``create``. Each provider's SDK is + imported only when that provider is selected, so unused providers need not + be importable. """ - @abstractmethod - def create(self, name: str, config: dict[str, Any]) -> Backend: - """Construct and return a not-yet-opened ``Backend``. + def create(self, name: str, config: dict[str, Any]) -> Backend | None: + """Construct a not-yet-opened ``Backend``, or ``None`` if unsupported. Args: name: Provider identifier (e.g. ``"openai_realtime"``, - ``"gemini_live"``, ``"elevenlabs"``, ``"cascade"``). The set - of valid names is defined by the concrete factory's registry, - not by this interface. + ``"gemini_live"``, ``"elevenlabs"``, ``"cascade"``). config: Provider-specific configuration understood by that - backend's ``open()`` (see ``Backend.open``). This factory - does not validate the shape of ``config`` beyond dispatching - on ``name`` -- each ``Backend`` subclass is responsible for - validating its own config. + backend. This factory does not validate the shape of + ``config`` beyond dispatching on ``name`` -- each ``Backend`` + subclass validates its own config and assembles its own + provider session (the caller hand-builds no provider JSON). Returns: - A constructed ``Backend`` instance. The returned backend has not - had ``open()`` called on it yet -- construction and session - establishment are separate steps so a ``Role`` can construct its - backend early (e.g. at record setup) and open the session later - (e.g. once the other party is ready). - - Raises: - ValueError: if ``name`` does not match a known provider. + A constructed ``Backend`` for a migrated provider, not yet + ``open()``ed (construction and session establishment are separate + steps, so a ``Role`` can build its backend early and open the + session later). ``None`` if ``name`` is not a migrated provider -- + the worker then falls back to the legacy server/simulator. """ - ... + if name == "openai_realtime": + from eva.backend.openai_realtime import OpenAIRealtimeBackend + + return OpenAIRealtimeBackend(config=config) + if name == "grok_voice": + from eva.backend.grok_voice import GrokVoiceBackend + + return GrokVoiceBackend(config=config) + return None diff --git a/src/eva/backend/grok_voice.py b/src/eva/backend/grok_voice.py new file mode 100644 index 00000000..19396bb7 --- /dev/null +++ b/src/eva/backend/grok_voice.py @@ -0,0 +1,97 @@ +"""Grok Voice ``Backend``: xAI's voice realtime API (OpenAI Realtime-compatible). + +xAI's voice realtime API is event-compatible with OpenAI's Realtime API +(https://docs.x.ai/developers/model-capabilities/audio/voice-agent), so this +backend subclasses ``OpenAIRealtimeBackend`` and overrides only what differs -- +mirroring how ``eva.assistant.grok_voice_server.GrokVoiceAssistantServer`` +subclasses the OpenAI Realtime server: + +- endpoint: point the client at ``https://api.x.ai/v1`` (default ``base_url``); +- default voice: xAI's built-in voices (``eve``/``ara``/``rex``/``sal``/``leo``); +- api key: falls back to ``XAI_API_KEY`` (not ``OPENAI_API_KEY``, which would be + the wrong key for x.ai) when not supplied in config; +- input transcription: xAI fires + ``conversation.item.input_audio_transcription.completed`` multiple times per + turn with progressively longer text, so instead of surfacing each as a final + ``TRANSCRIPT`` (which the OpenAI backend does), buffer it and emit one final + ``TRANSCRIPT`` when the turn settles (next speech start / ``response.done``). + +Everything else -- session assembly, audio, tool round-trip, interruption, +usage, the rest of the event normalization -- is inherited unchanged. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, ClassVar + +from eva.backend.base import BackendEvent, BackendEventType +from eva.backend.openai_realtime import OpenAIRealtimeBackend, OpenAIRealtimeSession +from eva.utils.logging import get_logger + +logger = get_logger(__name__) + +XAI_REALTIME_BASE_URL = "https://api.x.ai/v1" +DEFAULT_VOICE = "eve" + + +@dataclass +class GrokVoiceSession(OpenAIRealtimeSession): + """OpenAI Realtime session state plus xAI's buffered input transcript. + + ``pending_input_transcript`` accumulates the latest (progressively longer) + ``input_audio_transcription.completed`` text for the current turn; it is + flushed as a single final ``TRANSCRIPT`` when the turn settles. + """ + + pending_input_transcript: str = "" + + +class GrokVoiceBackend(OpenAIRealtimeBackend): + """xAI Grok voice realtime behind the role-agnostic ``Backend`` contract.""" + + _SESSION_CLS: ClassVar[type[OpenAIRealtimeSession]] = GrokVoiceSession + _API_KEY_ENV: ClassVar[str] = "XAI_API_KEY" + + def __init__(self, *, config: dict[str, Any]) -> None: + # xAI defaults; any explicit config value wins. api_key falls back to + # XAI_API_KEY in the parent (via _API_KEY_ENV). + merged = {"base_url": XAI_REALTIME_BASE_URL, "voice": DEFAULT_VOICE, **config} + super().__init__(config=merged) + + @staticmethod + def _map_event(session: OpenAIRealtimeSession, event: Any) -> list[BackendEvent]: + """Normalize one xAI event, buffering incremental input transcriptions. + + Defers to ``OpenAIRealtimeBackend._map_event`` for everything except the + progressive ``input_audio_transcription.completed`` stream, which is + buffered and flushed as one final input ``TRANSCRIPT`` when the turn + settles (mirrors ``GrokVoiceAssistantServer``'s deferred transcript). + """ + event_type = getattr(event, "type", "") + + if event_type == "conversation.item.input_audio_transcription.completed": + transcript = (getattr(event, "transcript", "") or "").strip() + if transcript and isinstance(session, GrokVoiceSession): + session.pending_input_transcript = transcript # buffer, don't emit yet + return [] + + # Flush the buffered transcript just before the turn boundary is handled. + if event_type in ("input_audio_buffer.speech_started", "response.done"): + return GrokVoiceBackend._flush_pending_input(session) + OpenAIRealtimeBackend._map_event(session, event) + + return OpenAIRealtimeBackend._map_event(session, event) + + @staticmethod + def _flush_pending_input(session: OpenAIRealtimeSession) -> list[BackendEvent]: + if not isinstance(session, GrokVoiceSession) or not session.pending_input_transcript: + return [] + text = session.pending_input_transcript + session.pending_input_transcript = "" + return [ + BackendEvent( + event_type=BackendEventType.TRANSCRIPT, + transcript=text, + metadata={"stream": "input", "final": True}, + ) + ] diff --git a/src/eva/backend/openai_realtime.py b/src/eva/backend/openai_realtime.py new file mode 100644 index 00000000..8c1237fe --- /dev/null +++ b/src/eva/backend/openai_realtime.py @@ -0,0 +1,492 @@ +"""OpenAI Realtime ``Backend``: normalizing adapter for one provider session. + +Wraps a single OpenAI Realtime API session behind the role-agnostic ``Backend`` +contract (see ``eva.backend.base``). It knows nothing about whether it drives +an ``AssistantRole`` or a ``UserRole``; both use the *same* backend and differ +only in the ``session_config`` the worker constructs it with and in how they +interpret the clean events it emits. + +Responsibilities (all provider-specific work lives here, so roles stay +generic): +- session lifecycle: connect / ``session.update`` / stream audio in / events + out / tool results / close; +- session-config assembly (voice, VAD, formats, transcription) from the + worker-supplied ``session_config`` -- roles never build provider config; +- tool-schema translation: generic ``{name, description, parameters}`` specs + -> OpenAI Realtime ``session.tools`` shape; +- **full event normalization**: every provider event is parsed here and + surfaced as a clean ``BackendEvent`` (typed fields + normalized ``metadata`` + scalars). Roles never see a raw OpenAI event. Provider-specific bookkeeping + (output-transcript accumulation, final-text selection, interruption + detection, token-usage extraction) happens here, with per-turn state carried + on the ``OpenAIRealtimeSession`` handle (the backend object stays stateless). + +Not covered here (role concerns): the transport to the counterparty (the +Twilio WS server / audio-bridge client), audio format conversion for that +transport, recording, prompt building, and tool execution. +""" + +from __future__ import annotations + +import base64 +import json +import os +from collections.abc import AsyncIterator +from dataclasses import dataclass, field +from typing import Any, ClassVar + +from openai import AsyncOpenAI + +from eva.backend.base import ( + Backend, + BackendEvent, + BackendEventType, + BackendSession, + ToolCallRequest, + ToolCallResult, +) +from eva.backend.capabilities import BackendCapabilities +from eva.utils.logging import get_logger + +logger = get_logger(__name__) + +DEFAULT_SAMPLE_RATE = 24000 +PCMU_SAMPLE_RATE = 8000 + + +@dataclass +class OpenAIRealtimeSession(BackendSession): + """Live state for one OpenAI Realtime session. + + Carries the SDK client, the entered realtime-connection context manager + + connection, and the per-turn accumulators the normalizer needs (output + transcript parts, whether a response is in flight, whether it produced tool + calls). All of this is session state, deliberately off the backend object. + """ + + client: AsyncOpenAI + conn_cm: Any + conn: Any + responding: bool = False + output_transcript_parts: list[str] = field(default_factory=list) + output_transcript_done: str = "" + has_function_calls: bool = False + + +class OpenAIRealtimeBackend(Backend): + """One OpenAI Realtime session behind the role-agnostic ``Backend`` contract. + + Construction is cheap and network-free (client + connection are created in + ``open()``), matching ``BackendFactory.create``'s "not-yet-opened" contract. + + Takes a single flat ``config`` of "config things" and assembles the + OpenAI ``session.update`` structure itself -- the caller (worker via the + factory) never hand-builds the provider JSON. Recognized keys: + + - ``model`` (required). ``api_key`` (optional): falls back to the + ``OPENAI_API_KEY`` env var if not provided. ``base_url`` (optional). + - ``accent``: if set, rejected -- this backend can't honor accents (they + are realized via ElevenLabs agent IDs). Fails loud, mirroring the old + ``OpenAIRealtimeUserSimulator`` guard, now backend-side. + - ``voice`` (default ``"marin"``), ``output_sample_rate`` (default 24000). + - ``input_format``: ``"pcm"`` (default) or ``"pcmu"`` (telephony/caller). + - ``vad_settings``: turn-detection tunables (``type`` / ``threshold`` / + ``prefix_padding_ms`` / ``silence_duration_ms``), defaults applied. Named + to match EVA's ``s2s_params["vad_settings"]`` so an assistant can pass its + provider params straight through. + - ``manual_turn_taking``: when True, the model does not auto-create + responses -- adds ``create_response``/``interrupt_response`` false and an + idle timeout (a caller gates replies itself via ``trigger_response``). + - ``transcription_model`` (default ``"whisper-1"``), + ``transcription_language`` (optional). + - ``reasoning_effort`` (optional), ``parallel_tool_calls`` (optional). + """ + + _CAPABILITIES = BackendCapabilities( + emits_continuous_audio=True, + supports_streaming_interruption=True, + owns_playout_clock=False, + ) + + # Session handle class ``open()`` instantiates. Subclasses for API-compatible + # providers (e.g. Grok Voice) override this to carry extra per-turn state. + _SESSION_CLS: ClassVar[type[OpenAIRealtimeSession]] = OpenAIRealtimeSession + + # Env var the api_key falls back to when not supplied in config. Subclasses + # for other OpenAI-compatible providers override it (e.g. Grok -> XAI_API_KEY). + _API_KEY_ENV: ClassVar[str] = "OPENAI_API_KEY" + + # Extra turn-detection fields when the caller gates responses manually. + _MANUAL_TURN_DETECTION = {"create_response": False, "interrupt_response": False, "idle_timeout_ms": 15_000} + + def __init__(self, *, config: dict[str, Any]) -> None: + api_key = config.get("api_key") or os.environ.get(self._API_KEY_ENV) + if not api_key: + raise ValueError(f"{type(self).__name__} requires an api_key (config['api_key'] or {self._API_KEY_ENV})") + if config.get("accent") is not None: + raise ValueError("OpenAI Realtime backend does not support accent variants") + self._model: str = config.get("model") or "" + if not self._model: + raise ValueError(f"{type(self).__name__} requires a 'model' (config['model'])") + self._api_key = api_key + self._base_url = config.get("base_url") + self._input_format: str = config.get("input_format", "pcm") + self._output_sample_rate = int(config.get("output_sample_rate", DEFAULT_SAMPLE_RATE)) + self._session_config = self._assemble_session_config(config) + + @property + def capabilities(self) -> BackendCapabilities: + return self._CAPABILITIES + + @property + def output_sample_rate(self) -> int: + """Sample rate (Hz) of ``AUDIO_OUTPUT`` payloads.""" + return self._output_sample_rate + + @property + def input_sample_rate(self) -> int: + """Sample rate (Hz) the session expects for ``send(audio=...)`` input.""" + return PCMU_SAMPLE_RATE if self._input_format == "pcmu" else self._output_sample_rate + + def _assemble_session_config(self, config: dict[str, Any]) -> dict[str, Any]: + """Build the OpenAI session-shaping block (minus type/instructions/tools) from flat config.""" + input_fmt: dict[str, Any] = ( + {"type": "audio/pcmu"} + if self._input_format == "pcmu" + else {"type": "audio/pcm", "rate": self._output_sample_rate} + ) + vad = config.get("vad_settings") or {} + turn_detection = { + "type": vad.get("type", "server_vad"), + "threshold": vad.get("threshold", 0.5), + "prefix_padding_ms": vad.get("prefix_padding_ms", 300), + "silence_duration_ms": vad.get("silence_duration_ms", 200), + } + if config.get("manual_turn_taking"): + turn_detection.update(self._MANUAL_TURN_DETECTION) + + transcription: dict[str, Any] = {"model": config.get("transcription_model", "whisper-1")} + if config.get("transcription_language"): + transcription["language"] = config["transcription_language"] + + session_config: dict[str, Any] = { + "output_modalities": ["audio"], + "audio": { + "output": { + "voice": config.get("voice", "marin"), + "format": {"type": "audio/pcm", "rate": self._output_sample_rate}, + }, + "input": {"format": input_fmt, "turn_detection": turn_detection, "transcription": transcription}, + }, + } + if config.get("reasoning_effort"): + session_config["reasoning"] = {"effort": config["reasoning_effort"]} + if config.get("parallel_tool_calls") is not None: + session_config["parallel_tool_calls"] = config["parallel_tool_calls"] + return session_config + + # ── Session lifecycle ───────────────────────────────────────────── + + async def open(self, *, system_prompt: str, tools: list[dict[str, Any]] | None) -> OpenAIRealtimeSession: + """Connect and configure a new OpenAI Realtime session; return its handle.""" + client_kwargs: dict[str, Any] = {"api_key": self._api_key} + if self._base_url is not None: + client_kwargs["base_url"] = self._base_url + client = AsyncOpenAI(**client_kwargs) + + conn_cm = client.realtime.connect(model=self._model) + conn = await conn_cm.__aenter__() + + session_update = self._build_session_update(system_prompt, tools) + await conn.session.update(session=session_update) # type: ignore[arg-type] + logger.info(f"OpenAI Realtime session opened (model={self._model})") + return self._SESSION_CLS(client=client, conn_cm=conn_cm, conn=conn) + + def _build_session_update(self, system_prompt: str, tools: list[dict[str, Any]] | None) -> dict[str, Any]: + """Finalize the ``session.update`` payload for ``open()``. + + Takes the session-shaping block assembled at construction and stamps the + per-open fields the backend owns (``type`` / ``instructions`` / + ``tools``), translating generic tool specs to the provider shape. + """ + session_update: dict[str, Any] = dict(self._session_config) + session_update["type"] = "realtime" + session_update["instructions"] = system_prompt + session_update["tools"] = self._format_tools(tools) + return session_update + + @staticmethod + def _format_tools(tools: list[dict[str, Any]] | None) -> list[dict[str, Any]]: + """Translate generic ``{name, description, parameters}`` specs to OpenAI schema.""" + return [ + { + "type": "function", + "name": tool["name"], + "description": tool["description"], + "parameters": tool["parameters"], + } + for tool in (tools or []) + ] + + async def send( + self, + session: BackendSession, + *, + audio: bytes | None = None, + text: str | None = None, + tool_result: ToolCallResult | None = None, + ) -> None: + """Push audio / a text turn / a tool result to ``session`` (exactly one).""" + provided = [x is not None for x in (audio, text, tool_result)] + if sum(provided) != 1: + raise ValueError("send() requires exactly one of audio, text, tool_result") + conn = self._conn(session) + + if audio is not None: + await conn.input_audio_buffer.append(audio=base64.b64encode(audio).decode("ascii")) + return + + if text is not None: + await conn.conversation.item.create( + item={"type": "message", "role": "user", "content": [{"type": "input_text", "text": text}]} + ) + await conn.response.create() + return + + assert tool_result is not None # exactly-one check above + await conn.conversation.item.create( + item={ + "type": "function_call_output", + "call_id": tool_result.call_id, + "output": json.dumps(tool_result.result, ensure_ascii=False), + } + ) + await conn.response.create() + + async def trigger_response(self, session: BackendSession) -> None: + """Manually request a model response (caller-gated turn-taking).""" + await self._conn(session).response.create() + + async def receive(self, session: BackendSession) -> AsyncIterator[BackendEvent]: + """Yield normalized events from ``session``'s connection until it ends.""" + s = self._session(session) + async for event in s.conn: + for be in self._map_event(s, event): + yield be + + @staticmethod + def _session(session: BackendSession) -> OpenAIRealtimeSession: + if not isinstance(session, OpenAIRealtimeSession): + raise TypeError(f"expected OpenAIRealtimeSession, got {type(session).__name__}") + return session + + @classmethod + def _conn(cls, session: BackendSession) -> Any: + return cls._session(session).conn + + @staticmethod + def _map_event(session: OpenAIRealtimeSession, event: Any) -> list[BackendEvent]: + """Normalize one provider event into zero or more clean ``BackendEvent``s. + + Stateful (accumulates output transcript / turn flags on ``session``) so + it can surface a fully-selected final transcript and detect + interruption without the role touching raw provider data. Pure of I/O, + so unit-testable with a fake session + event. + """ + event_type = getattr(event, "type", "") + out: list[BackendEvent] = [] + + match event_type: + case "response.created": + session.responding = True + session.output_transcript_parts = [] + session.output_transcript_done = "" + session.has_function_calls = False + out.append(BackendEvent(event_type=BackendEventType.OUTPUT_TURN_STARTED)) + + case "response.output_audio.delta": + delta_b64 = getattr(event, "delta", "") or "" + if delta_b64: + out.append( + BackendEvent(event_type=BackendEventType.AUDIO_OUTPUT, audio=base64.b64decode(delta_b64)) + ) + + case "response.output_audio_transcript.delta": + session.output_transcript_parts.append(getattr(event, "delta", "") or "") + + case "response.output_audio_transcript.done": + done = (getattr(event, "transcript", "") or "").strip() + session.output_transcript_done = done + text = done or "".join(session.output_transcript_parts).strip() + out.append( + BackendEvent( + event_type=BackendEventType.TRANSCRIPT, + transcript=text, + metadata={"stream": "output", "final": True}, + ) + ) + + case "conversation.item.input_audio_transcription.completed": + transcript = (getattr(event, "transcript", "") or "").strip() + if transcript: + out.append( + BackendEvent( + event_type=BackendEventType.TRANSCRIPT, + transcript=transcript, + metadata={"stream": "input", "final": True}, + ) + ) + + case "conversation.item.input_audio_transcription.failed": + out.append( + BackendEvent( + event_type=BackendEventType.TRANSCRIPT, + transcript="", + metadata={"stream": "input", "failed": True}, + ) + ) + + case "input_audio_buffer.speech_started": + # If the inbound party barges in over a response that has already + # produced text, flush that partial as an interrupted turn before + # signaling the speech start (mirrors the old flush-then-new-turn order). + if session.responding and session.output_transcript_parts: + partial = "".join(session.output_transcript_parts) + out.append( + BackendEvent( + event_type=BackendEventType.TURN_END, + transcript=partial, + metadata={ + "interrupted": True, + "cancelled": False, + "has_function_calls": session.has_function_calls, + "usage": None, + }, + ) + ) + session.responding = False + session.output_transcript_parts = [] + session.output_transcript_done = "" + out.append(BackendEvent(event_type=BackendEventType.INPUT_SPEECH_STARTED)) + + case "input_audio_buffer.speech_stopped": + out.append(BackendEvent(event_type=BackendEventType.INPUT_SPEECH_STOPPED)) + + case "response.function_call_arguments.done": + session.has_function_calls = True + arguments_str = getattr(event, "arguments", "{}") or "{}" + try: + arguments = json.loads(arguments_str) + except json.JSONDecodeError: + arguments = {} + out.append( + BackendEvent( + event_type=BackendEventType.TOOL_CALL_REQUEST, + tool_call_request=ToolCallRequest( + call_id=getattr(event, "call_id", "") or "", + name=getattr(event, "name", "") or "", + arguments=arguments, + ), + ) + ) + + case "response.output_audio.done": + out.append(BackendEvent(event_type=BackendEventType.OUTPUT_AUDIO_DONE)) + + case "response.done": + response = getattr(event, "response", None) + cancelled = bool(response and getattr(response, "status", None) == "cancelled") + final_text = ( + session.output_transcript_done + or "".join(session.output_transcript_parts).strip() + or OpenAIRealtimeBackend._extract_response_text(event) + ) + has_fc = OpenAIRealtimeBackend._response_has_function_calls(event) or session.has_function_calls + out.append( + BackendEvent( + event_type=BackendEventType.TURN_END, + transcript=final_text, + metadata={ + "cancelled": cancelled, + "interrupted": False, + "has_function_calls": has_fc, + "usage": OpenAIRealtimeBackend._extract_usage(response), + }, + ) + ) + session.responding = False + session.output_transcript_parts = [] + session.output_transcript_done = "" + session.has_function_calls = False + + case "error": + error_data = getattr(event, "error", None) + code = getattr(error_data, "code", None) if error_data is not None else None + out.append( + BackendEvent( + event_type=BackendEventType.ERROR, + error=str(error_data) if error_data is not None else "unknown error", + metadata={"code": code}, + ) + ) + + case _: + # session.created/updated, interim transcription deltas, etc.: no + # cross-role meaning -> drop. + pass + + return out + + @staticmethod + def _extract_usage(response: Any) -> dict[str, int] | None: + if not response: + return None + usage = getattr(response, "usage", None) + if not usage: + return None + return { + "prompt_tokens": getattr(usage, "input_tokens", 0) or 0, + "completion_tokens": getattr(usage, "output_tokens", 0) or 0, + } + + @staticmethod + def _response_has_function_calls(event: Any) -> bool: + response = getattr(event, "response", None) + if not response: + return False + output_items = getattr(response, "output", None) or [] + return any(getattr(item, "type", "") == "function_call" for item in output_items) + + @staticmethod + def _extract_response_text(event: Any) -> str: + response = getattr(event, "response", None) + if not response: + return "" + output_items = getattr(response, "output", None) or [] + text_parts: list[str] = [] + for item in output_items: + for part in getattr(item, "content", None) or []: + if getattr(part, "type", "") in ("audio", "text"): + transcript = getattr(part, "transcript", None) or getattr(part, "text", None) or "" + if transcript: + text_parts.append(transcript) + return "".join(text_parts).strip() + + async def close(self, session: BackendSession) -> None: + """Tear down ``session``. Idempotent: safe to call more than once.""" + s = self._session(session) + if s.conn_cm is not None: + try: + await s.conn_cm.__aexit__(None, None, None) + except Exception as e: + logger.debug(f"Error closing OpenAI Realtime connection: {e}") + finally: + s.conn_cm = None + s.conn = None + if s.client is not None: + try: + await s.client.close() + except Exception as e: + logger.debug(f"Error closing OpenAI client: {e}") + finally: + s.client = None # type: ignore[assignment] diff --git a/src/eva/models/config.py b/src/eva/models/config.py index 315bf4eb..e378eaf6 100644 --- a/src/eva/models/config.py +++ b/src/eva/models/config.py @@ -463,17 +463,26 @@ def _warn_extra_fields(cls, data: Any) -> Any: return data -class OpenAIRealtimeSimulatorConfig(BaseModel): - """OpenAI Realtime-specific settings for the user simulator.""" +class S2SSimulatorConfig(BaseModel): + """Native speech-to-speech settings for the user simulator. + + Shared by all OpenAI-Realtime-compatible S2S providers (``openai_realtime``, + ``grok_voice``). Defaults target OpenAI Realtime; other providers should + override ``model`` and the voices (e.g. Grok's ``eve``/``ara``). The backend + resolves the API key from the environment per provider (OPENAI_API_KEY / + XAI_API_KEY). + """ - provider: Literal["openai_realtime"] = "openai_realtime" - model: str = Field("gpt-realtime-1.5", description="OpenAI Realtime model.") + provider: Literal["openai_realtime", "grok_voice"] = "openai_realtime" + model: str = Field( + "gpt-realtime-1.5", description="Native S2S model (OpenAI Realtime default; override per provider)." + ) female_voice: str = Field("marin", description="Voice used for female caller personas.") male_voice: str = Field("cedar", description="Voice used for male caller personas.") UserSimulatorConfig = Annotated[ - ElevenLabsSimulatorConfig | OpenAIRealtimeSimulatorConfig, + ElevenLabsSimulatorConfig | S2SSimulatorConfig, Field(discriminator="provider"), ] @@ -727,13 +736,13 @@ def _check_companion_services(self) -> "RunConfig": config is unused and conflicting env vars are harmless. """ if ( - isinstance(self.user_simulator, OpenAIRealtimeSimulatorConfig) + isinstance(self.user_simulator, S2SSimulatorConfig) and self.perturbation is not None and self.perturbation.accent is not None ): raise ValueError( "Accent perturbations require the ElevenLabs user simulator; " - "OpenAI Realtime supports behavior, noise, and connection perturbations." + "native S2S simulators support behavior, noise, and connection perturbations." ) if self.max_rerun_attempts == 0 or self.aggregate_only: @@ -766,8 +775,12 @@ def _check_companion_services(self) -> "RunConfig": errors.extend( self._validate_service_params("AUDIO_LLM", self.model.audio_llm, self.model.audio_llm_params) ) - case PipelineType.S2S: - errors.extend(self._validate_service_params("S2S", self.model.s2s, self.model.s2s_params)) + # S2S is intentionally not validated here: an S2S run uses a native backend, + # and that backend's construction (via the BackendFactory, exercised in + # orchestrator.preflight) is the single source of truth for its required + # fields/keys. Keeping it out of config avoids duplicating backend knowledge + # (the trade-off: a missing S2S key surfaces as a PreflightError, not a + # pydantic ValidationError). if errors: raise ValidationError.from_exception_data(title=type(self).__name__, line_errors=errors) @@ -833,14 +846,12 @@ def _check_language_personas(self) -> "RunConfig": return self - @model_validator(mode="after") - def _check_openai_realtime_simulator(self) -> "RunConfig": - """When openai_realtime user simulator is selected, OPENAI_API_KEY must be present.""" - if not isinstance(self.user_simulator, OpenAIRealtimeSimulatorConfig): - return self - if not os.environ.get("OPENAI_API_KEY"): - raise ValueError("EVA_USER_SIMULATOR__PROVIDER=openai_realtime requires OPENAI_API_KEY to be set.") - return self + # NOTE: no backend-specific credential validation here. A native S2S user + # simulator's credentials are validated by constructing its backend via the + # BackendFactory (see orchestrator.worker): the backend raises its own precise + # reason (e.g. missing api_key naming OPENAI_API_KEY / XAI_API_KEY). Config-side + # validation stays limited to non-backend concerns (perturbation compatibility, + # companion services, etc.); provider validity is enforced by the Literal types. @model_validator(mode="before") @classmethod diff --git a/src/eva/orchestrator/preflight.py b/src/eva/orchestrator/preflight.py index aedcb903..2becaf1f 100644 --- a/src/eva/orchestrator/preflight.py +++ b/src/eva/orchestrator/preflight.py @@ -12,8 +12,11 @@ - Probes are conservative: a component is reported as failed only on a definitive error (an exception or an ``ErrorFrame``). Ambiguity — no output, an unexpected-but-benign frame — is treated as a pass, so preflight never blocks an otherwise-valid run. -- S2S live probing is not yet supported (each framework needs its own connect path); - S2S relies on the config-level credential validation in ``RunConfig``. +- Native-S2S backends (OpenAI Realtime, Grok Voice) are validated cheaply by + *constructing* them via the ``BackendFactory`` (``_preflight_backends``): the backend's + own ``__init__`` checks that required fields/keys are present, with no API call. This + is the home for that validation (config-side validation was removed). Live S2S probing + (an actual session) is still not supported. """ import asyncio @@ -39,11 +42,15 @@ create_tts_service, ) from eva.assistant.services.llm import LiteLLMClient +from eva.backend.factory import BackendFactory from eva.models.config import PipelineType, RunConfig, get_model_alias_from_params from eva.utils.logging import get_logger logger = get_logger(__name__) +# Stateless; shared across preflight calls (see orchestrator.worker for the same pattern). +_BACKEND_FACTORY = BackendFactory() + # 0.3s of 16 kHz mono 16-bit silence — enough to make a streaming STT service open its # connection (and reveal an auth failure) without depending on actual speech content. _SILENCE_SAMPLE_RATE = 16000 @@ -173,19 +180,63 @@ async def _run_preflight(config: RunConfig) -> list[ProbeResult]: probes.append(("AUDIO_LLM", get_model_alias_from_params(model.audio_llm_params), _probe_audio_llm(config))) probes.append(("TTS", get_model_alias_from_params(model.tts_params), _probe_tts(config))) case PipelineType.S2S: - logger.info("Pre-flight: S2S live probe not yet supported") + # Backend construction is validated separately (_preflight_backends); a live + # S2S session probe is not yet supported. + logger.info("Pre-flight: S2S live probe not yet supported (construction validated)") return [] logger.info(f"Pre-flight: checking {len(probes)} model(s) before the run starts...") return await asyncio.gather(*(_guard(model_type, alias, probe, timeout) for model_type, alias, probe in probes)) +def _check_backend_construction(label: str, name: str, backend_args: dict[str, Any]) -> str | None: + """Validate a native-S2S provider by constructing its backend (no network). + + The backend's own ``__init__`` checks that required fields/keys resolve, so a + successful construction means the run's construction path works. Returns an error + string for a *misconfigured factory backend* (missing key, absent model), or + ``None`` when it's fine — including when the factory doesn't back this provider + (``create()`` -> ``None``): that's a legacy/non-factory provider (e.g. ElevenLabs + Conversational AI, or a cascade pipeline) validated elsewhere, not here. + """ + try: + _BACKEND_FACTORY.create(name, backend_args) + except Exception as e: + return f"{label} {name!r}: {str(e).strip()[:200] or type(e).__name__}" + return None + + +def _preflight_backends(config: RunConfig) -> None: + """Cheap, network-free validation that configured native-S2S backends construct. + + Only native-S2S providers run through the ``BackendFactory``: an S2S assistant + framework, and the user-simulator provider. Cascade/audio-LLM assistants (pipecat + services) and non-factory user providers are left to the live probes / legacy paths. + """ + errors: list[str] = [] + if config.model.pipeline_type == PipelineType.S2S: + assistant_args = { + **(config.model.s2s_params or {}), + "parallel_tool_calls": config.model.parallel_tool_calls, + } + if err := _check_backend_construction("assistant framework", config.framework, assistant_args): + errors.append(err) + sim = config.user_simulator + if err := _check_backend_construction("user simulator", sim.provider, sim.model_dump()): + errors.append(err) + if errors: + raise PreflightError("backend configuration invalid:\n" + "\n".join(errors)) + + async def run_preflight(config: RunConfig) -> None: - """Probe models and raise ``PreflightError`` if any required component fails. + """Validate backends, then probe models; raise ``PreflightError`` on any failure. - No-op when ``config.preflight`` is set or there are no probes to run - (e.g. S2S). Call this immediately before launching simulations. + The cheap, network-free backend-construction check (``_preflight_backends``) ALWAYS + runs -- ``--no-preflight`` only skips the live model probes (the ones that make real + API calls), never the config validation. Call this immediately before launching + simulations. """ + _preflight_backends(config) # always: cheap, no network if not config.preflight: return results = await _run_preflight(config) diff --git a/src/eva/orchestrator/worker.py b/src/eva/orchestrator/worker.py index ef5519c4..94a1dd24 100644 --- a/src/eva/orchestrator/worker.py +++ b/src/eva/orchestrator/worker.py @@ -8,10 +8,14 @@ from typing import Any from eva.assistant.base_server import AbstractAssistantServer +from eva.backend.factory import BackendFactory from eva.models.agents import AgentConfig from eva.models.config import RunConfig from eva.models.record import EvaluationRecord from eva.models.results import ConversationResult, ErrorDetails, LatencyStats +from eva.role.assistant import AssistantRole +from eva.role.user import UserRole +from eva.user_simulator.base import AbstractUserSimulator from eva.user_simulator.factory import create_user_simulator from eva.utils.culture import resolve_scenario_db, resolve_user_config, resolve_user_goal from eva.utils.error_handler import create_error_details @@ -22,6 +26,12 @@ USER_SIMULATOR_SHUTDOWN_GRACE_SECONDS = 20 +# The backend factory is stateless (pure dispatch + lazy per-provider imports), so +# a single shared instance serves every worker. Providers it can build run on the +# Role/Backend path; for the rest create() returns None and we fall back to the +# legacy server/simulator (see _start_assistant / _start_user_simulator). +_BACKEND_FACTORY = BackendFactory() + def _get_server_class(framework: str) -> type[AbstractAssistantServer]: """Return the server class for the given framework name. @@ -116,9 +126,10 @@ def __init__( self.port = port self.output_id = output_id - # Will be set during run - self._assistant_server = None - self._user_simulator = None + # Set during run: a Role (for factory-supported providers) or the legacy + # server/simulator (for the rest). + self._assistant_server: AbstractAssistantServer | AssistantRole | None = None + self._user_simulator: AbstractUserSimulator | UserRole | None = None self._conversation_stats: dict[str, Any] = {} self._log_file_handler = None self.deferred_audio_task: asyncio.Task | None = None @@ -305,26 +316,48 @@ async def run(self) -> ConversationResult: async def _start_assistant(self) -> None: """Start the assistant server using the configured framework.""" - server_cls = _get_server_class(self.config.framework) resolved_db_path = self._materialize_resolved_scenario_db() - # The turn-end fallback applies to the Pipecat pipelines (cascade + audio-LLM), whose - # turn detection can drop a user turn. S2S servers handle turn-taking natively and - # don't accept this kwarg. - server_kwargs: dict[str, Any] = {} - if self.config.framework == "pipecat": - server_kwargs["turn_end_fallback_time"] = self.config.turn_end_fallback_time - self._assistant_server = server_cls( - current_date_time=self.record.current_date_time, - pipeline_config=self.config.model, - agent=self.agent, - agent_config_path=self.agent_config_path, - scenario_db_path=str(resolved_db_path), - output_dir=self.output_dir, - port=self.port, - conversation_id=self.record.id, - language=self.config.language, - **server_kwargs, - ) + + s2s = self.config.model.s2s_params or {} + backend_args = {**s2s, "parallel_tool_calls": self.config.model.parallel_tool_calls} + if backend := _BACKEND_FACTORY.create(self.config.framework, backend_args): + # A generic AssistantRole over a factory-built backend. The backend is + # selected by the configured framework name; its args are the S2S + # provider params passed through (the backend reads what it needs and + # assembles its own session). Providers not yet on the factory fall + # through to the legacy server below. + self._assistant_server = AssistantRole( + backend=backend, + current_date_time=self.record.current_date_time, + pipeline_config=self.config.model, + agent=self.agent, + agent_config_path=self.agent_config_path, + scenario_db_path=str(resolved_db_path), + output_dir=self.output_dir, + port=self.port, + conversation_id=self.record.id, + language=self.config.language, + ) + else: + server_cls = _get_server_class(self.config.framework) + # The turn-end fallback applies to the Pipecat pipelines (cascade + audio-LLM), whose + # turn detection can drop a user turn. S2S servers handle turn-taking natively and + # don't accept this kwarg. + server_kwargs: dict[str, Any] = {} + if self.config.framework == "pipecat": + server_kwargs["turn_end_fallback_time"] = self.config.turn_end_fallback_time + self._assistant_server = server_cls( + current_date_time=self.record.current_date_time, + pipeline_config=self.config.model, + agent=self.agent, + agent_config_path=self.agent_config_path, + scenario_db_path=str(resolved_db_path), + output_dir=self.output_dir, + port=self.port, + conversation_id=self.record.id, + language=self.config.language, + **server_kwargs, + ) await self._assistant_server.start() @@ -366,18 +399,58 @@ async def _start_user_simulator(self) -> None: language, self.record.romanized_culture_overrides, ) - self._user_simulator = create_user_simulator( - self.config.user_simulator, - current_date_time=self.record.current_date_time, - persona_config=resolved_persona, - goal=resolved_goal, - server_url=f"ws://localhost:{self.port}/ws", - output_dir=self.output_dir, - agent_id=self.agent.id, - timeout=self._conversation_guard_timeout_seconds(), - perturbation_config=self.config.perturbation, - language=language, - ) + # Dispatch generically on provider, not on config type: dump the config + # wholesale into a caller-config blob and ask the factory to build a backend + # for sim.provider. If it can (a migrated provider), drive it with a UserRole; + # otherwise create() returns None and we fall through to the legacy simulator. + # This keeps the worker config-type-agnostic -- new providers just register a + # backend in the factory. No api_key here: the backend resolves it from the + # environment per provider. + # + # The blob is read only by backends, which pick the keys they recognize and + # ignore the rest (voice fields are present but unused by, e.g., ElevenLabs, + # which isn't factory-backed anyway). Reading voices from the dumped dict -- + # not attribute access -- is what avoids coupling to a concrete config type. + sim = self.config.user_simulator + caller_config = sim.model_dump() + gender = {1: "F", 2: "M"}.get(resolved_persona.get("user_persona_id")) + voice = caller_config.get("male_voice") if gender == "M" else caller_config.get("female_voice") + backend_args = { + **caller_config, + **UserRole.CALLER_BACKEND_DEFAULTS, + "transcription_language": language, + "accent": self.config.perturbation.accent if self.config.perturbation else None, + } + if voice is not None: + backend_args["voice"] = voice + + if backend := _BACKEND_FACTORY.create(sim.provider, backend_args): + self._user_simulator = UserRole( + backend=backend, + current_date_time=self.record.current_date_time, + persona_config=resolved_persona, + goal=resolved_goal, + server_url=f"ws://localhost:{self.port}/ws", + output_dir=self.output_dir, + agent_id=self.agent.id, + provider=sim.provider, + timeout=self._conversation_guard_timeout_seconds(), + perturbation_config=self.config.perturbation, + language=language, + ) + else: + self._user_simulator = create_user_simulator( + self.config.user_simulator, + current_date_time=self.record.current_date_time, + persona_config=resolved_persona, + goal=resolved_goal, + server_url=f"ws://localhost:{self.port}/ws", + output_dir=self.output_dir, + agent_id=self.agent.id, + timeout=self._conversation_guard_timeout_seconds(), + perturbation_config=self.config.perturbation, + language=language, + ) # Let the simulator tell the assistant the moment the call ends, rather than leaving it # to infer that from transport disconnect — which lands after the simulator's STT grace @@ -399,7 +472,11 @@ async def _run_conversation(self) -> str: if self._user_simulator is None: raise RuntimeError("User simulator not initialized") - ended_reason = await self._user_simulator.run_conversation() + # UserRole drives via run(); the legacy simulator via run_conversation(). + if isinstance(self._user_simulator, UserRole): + ended_reason = await self._user_simulator.run() + else: + ended_reason = await self._user_simulator.run_conversation() return ended_reason diff --git a/src/eva/role/__init__.py b/src/eva/role/__init__.py index c005c0a7..be05d045 100644 --- a/src/eva/role/__init__.py +++ b/src/eva/role/__init__.py @@ -1,13 +1,10 @@ -"""Provider-agnostic ``Role`` abstraction (design-only, Step 1 of the refactor). +"""Provider-agnostic ``Role`` abstraction (see ``docs/refactor-step1.md``). -A ``Role`` owns everything that today is duplicated across the assistant and +A ``Role`` owns everything that was duplicated across the assistant and user-simulator stacks per-provider: prompt construction, tool ownership, and goal/persona/agent-config data. Each ``Role`` holds exactly one ``eva.backend.Backend`` instance, created at runtime via a ``eva.backend.BackendFactory``. - -Nothing in this package is wired into the existing ``eva.assistant`` / -``eva.user_simulator`` code yet. """ from eva.role.assistant import AssistantRole diff --git a/src/eva/role/assistant.py b/src/eva/role/assistant.py index 897dba14..9dc42dbb 100644 --- a/src/eva/role/assistant.py +++ b/src/eva/role/assistant.py @@ -1,132 +1,615 @@ -"""``AssistantRole`` contract: the business-side answering role. - -DESIGN ONLY (Step 1 of the refactor, see docs/refactor-step1.md). Method -bodies are stubs; nothing here is wired into the existing code path yet. - -Plug-in point (where this will eventually replace existing code): - Today the assistant side is a concrete ``AbstractAssistantServer`` - subclass selected by ``eva.orchestrator.worker._get_server_class(framework)`` - (worker.py) and constructed + started inside - ``ConversationWorker._start_assistant()`` (worker.py), which calls - ``server_cls(...).start()``. Its outputs are flushed by - ``ConversationWorker._cleanup()`` via ``server.stop()`` (which internally - calls ``save_outputs()``), and ``get_conversation_stats()`` / - ``get_final_scenario_db()`` are read back in ``ConversationWorker.run()``. - - In a later phase, ``_start_assistant()`` becomes the construction site for - an ``AssistantRole`` (framework string -> ``backend_name`` passed to the - ``BackendFactory``), and the worker drives ``role.run()`` / - ``role.save_outputs()`` / ``role.get_final_scenario_db()`` instead of the - server's own lifecycle methods. The provider-specific server subclasses - collapse into ``Backend`` implementations behind the factory; the - role-agnostic orchestration in ``ConversationWorker`` stays put. This - module is deliberately separate from ``eva.role.user`` so that migration - can land assistant-side first without touching the user-side diff. +"""``AssistantRole``: the business-side answering role (one generic class). + +A single concrete, provider-agnostic role. It holds a ``Backend`` and works +with *any* backend -- swapping the backend swaps the provider. All provider +specifics (session, audio format, event parsing) live in the backend; this +class owns only the role-common concerns shared by every assistant regardless +of provider: + +- the assistant system prompt (built from agent config); +- the agent tool catalog + ``ToolExecutor`` (tool execution stays role-side); +- the ``AuditLog`` and output artifacts (audit_log.json / transcript.jsonl / + scenario DBs / audio WAVs); +- the counterparty transport: a Twilio-framed WebSocket **server** the user + simulator connects to, plus real-time output pacing and audio-track + recording/alignment (all provider-agnostic -- every assistant exposes this + same Twilio WS). + +It consumes only the normalized ``BackendEvent`` stream, so it never touches a +raw provider event. + +Plug-in point: mirrors ``AbstractAssistantServer``'s surface (``start`` / +``stop`` / ``get_conversation_stats`` / ``get_final_scenario_db`` / +``notify_conversation_ending``) so the worker swap is 1:1 -- construct +``AssistantRole(backend=factory.create(...), ...)`` instead of +``server_cls(...)`` and keep every downstream call. The worker takes this path +for every provider the ``BackendFactory`` supports. """ from __future__ import annotations -from abc import abstractmethod +import asyncio +import json +import time +from dataclasses import dataclass +from pathlib import Path from typing import Any -from eva.backend.factory import BackendFactory +import uvicorn +from fastapi import FastAPI, WebSocket, WebSocketDisconnect + +from eva.assistant.agentic.audit_log import AuditLog +from eva.assistant.pipeline.observers import FrameworkLogWriter, MetricsLogWriter +from eva.assistant.tools.tool_executor import ToolExecutor, execute_and_log_tool +from eva.backend.base import Backend, BackendEvent, BackendEventType, ToolCallRequest, ToolCallResult +from eva.models.agents import AgentConfig +from eva.models.config import ModelConfig from eva.role.base import Role +from eva.utils.audio_utils import ( + create_twilio_media_message, + mulaw_8k_to_pcm16_24k, + parse_twilio_media_message, + pcm16_24k_to_mulaw_8k, + pcm16_mix, + save_audio_track, + sync_buffer_to_position, +) +from eva.utils.culture import get_initial_message +from eva.utils.logging import get_logger +from eva.utils.prompt_manager import PromptManager + +logger = get_logger(__name__) + +# Twilio counterparty-transport constants (provider-agnostic). +MULAW_CHUNK_SIZE = 160 # bytes per chunk (20ms at 8kHz mulaw) +MULAW_CHUNK_DURATION_S = 0.02 +# Don't pad the user track to align with the assistant when real user audio +# arrived within this window (the speaking-state flag can go stale under jitter; +# padding then injects a mid-utterance chop). Guard only ever *skips* a pad. +USER_ACTIVE_GUARD_S = 0.3 + + +def _wall_ms() -> str: + """Current wall-clock time as epoch-milliseconds string.""" + return str(int(round(time.time() * 1000))) + + +@dataclass +class _UserTurnRecord: + """State for a single user speech turn (timestamps + transcript flush flag).""" + + speech_started_wall_ms: str = "" + speech_stopped_wall_ms: str = "" + transcript: str = "" + flushed: bool = False + + +@dataclass +class _AssistantTurnState: + """Per-response state the role tracks for recording/metrics/logging.""" + + first_audio_wall_ms: str | None = None + audio_was_streamed: bool = False + responding: bool = False class AssistantRole(Role): - """Role that answers on behalf of the business (today's "assistant server"). - - Carries agent configuration and tool catalog; owns a ``ToolExecutor`` - (constructed by subclasses, not by this contract) to fulfill - ``handle_tool_call_request``. - - Turn-end fallback (self-nudge): the assistant's backstop for a *dropped - user turn*. When VAD / turn detection silently fails to fire for a real - user utterance, the call would otherwise hang until the provider's - inactivity timeout ends it. After the assistant stops speaking, if no user - turn is detected within ``turn_end_fallback_seconds``, the assistant - proactively re-engages with a nudge (acknowledge-and-answer if partial - user speech/audio was captured, otherwise ask the caller to repeat). This - is the seam already shipped as the pipeline-side ``TurnEndFallbackTimer`` - (see ``eva.assistant.pipeline.fallback`` and ``EVA_TURN_END_FALLBACK_TIME``); - it works for both cascade and audio-LLM pipelines. - - Two policies the backend owns, carried over from the shipped feature: - - Give up after a small number of *consecutive* nudges without a real user - turn resetting the count (``MAX_CONSECUTIVE_FALLBACK_NUDGES``), then let - the provider's inactivity backstop end the call. - - Never nudge once the call is ending (a nudge during teardown produces a - phantom assistant turn after the conversation is logically closed). - - Unlike the tool-call/idle-detection seams elsewhere in this contract, the - fallback needs no new ``Role`` method and no new ``Backend`` event type: - the nudge is just an ordinary outbound turn that this role's backend - produces on its own after the timeout, using the same - ``system_prompt``/instructions already established at ``open()`` time (see - ``Backend.open``'s ``config`` docstring). It is surfaced through the normal - ``receive()`` stream and tagged so downstream metrics can identify and zero - it (the shipped feature records the transcript marker with - ``message_type="turn_fallback"``; see ``BackendEvent.metadata``). Whether - the *other* side (a ``UserRole``) needs to do anything special upon - receiving it, versus just treating it as an ordinary assistant turn through - its existing ``run()`` loop, is left open -- see docs/refactor-step1.md - discussion; nothing here requires ``UserRole`` changes to handle it today. - """ + """Generic assistant role. Drives any ``Backend``; owns the Twilio WS transport.""" def __init__( self, *, - backend_factory: BackendFactory, - backend_name: str, - backend_config: dict[str, Any], + backend: Backend, + current_date_time: str, + pipeline_config: ModelConfig, + agent: AgentConfig, agent_config_path: str, scenario_db_path: str, - current_date_time: str, + output_dir: Path, + port: int, + conversation_id: str, + language: str = "en", turn_end_fallback_seconds: float | None = None, ) -> None: - """Initialize the assistant role. - - Args: - backend_factory: Factory used to construct the backend. - backend_name: Key passed to the factory to select a backend. - backend_config: Provider-specific configuration for the backend. - agent_config_path: Path to the agent YAML (role, instructions, - tool schemas) -- mirrors ``AbstractAssistantServer.agent`` / - ``agent_config_path``. - scenario_db_path: Path to the per-record scenario database JSON - consumed by tool execution -- mirrors - ``AbstractAssistantServer.scenario_db_path``. - current_date_time: Current date/time string threaded into both - prompt construction and tool execution (mirrors existing - ``current_date_time`` plumbing throughout the assistant - stack). - turn_end_fallback_seconds: How long after the assistant stops - speaking to wait for a user turn before firing a turn-end - fallback nudge, or ``None`` to disable the fallback entirely - (preserving the old behavior of waiting for the provider's - inactivity timeout). Mirrors the shipped - ``EVA_TURN_END_FALLBACK_TIME`` knob. This is an - ``AssistantRole``-level tuning value, not a - ``BackendCapabilities`` flag (capabilities describe what a - backend *can* do, statically). Wiring it into the constructed - ``self.backend``'s own config (via ``backend_config`` / - ``Backend.open(config=...)``) is left to the concrete - subclass's constructor, same as elsewhere in this contract -- - a ``Role`` does not otherwise reach into backend config after - construction. A backend with no notion of idle timing (e.g. a - thin end-to-end backend that relies on its own provider - backstop) may simply ignore this value. - """ - super().__init__(backend_factory=backend_factory, backend_name=backend_name, backend_config=backend_config) + super().__init__(backend=backend) + self.current_date_time = current_date_time + self.pipeline_config = pipeline_config + self.agent = agent self.agent_config_path = agent_config_path self.scenario_db_path = scenario_db_path - self.current_date_time = current_date_time + self.output_dir = Path(output_dir) + self.port = port + self.conversation_id = conversation_id + self.language = language self.turn_end_fallback_seconds = turn_end_fallback_seconds + self.initial_message = get_initial_message(language) + + # Core components. + self.audit_log = AuditLog() # type: ignore[no-untyped-call] + self.tool_handler = ToolExecutor( + tool_config_path=agent_config_path, + scenario_db_path=scenario_db_path, + tool_module_path=self.agent.tool_module_path, + current_date_time=current_date_time, + ) + + # Recording buffers. Sample rate comes from the backend (provider format). + self._audio_buffer = bytearray() + self.user_audio_buffer = bytearray() + self.assistant_audio_buffer = bytearray() + self._audio_sample_rate = backend.output_sample_rate + + self._fw_log: FrameworkLogWriter | None = None + self._metrics_log: MetricsLogWriter | None = None + + # Server state. + self._app: FastAPI | None = None + self._server: uvicorn.Server | None = None + self._server_task: asyncio.Task[Any] | None = None + self._running = False + + # Prompt + generic tool specs (built once); model name for metrics labels. + self._system_prompt = self.build_prompt() + self._tool_specs = self._build_tool_specs() + self._model = (self.pipeline_config.s2s_params or {}).get("model", "") + + # Per-session/turn state. + self._user_turn: _UserTurnRecord | None = None + self._assistant_turn = _AssistantTurnState() + self._stream_sid = "" + self._user_speaking = False + self._bot_speaking = False + self._audio_interface_speech_start_ts: str | None = None + self._last_user_audio_mono = 0.0 + + # ── Prompt / tool specs (role-owned, provider-agnostic) ─────────── + + def build_prompt(self) -> str: + """Build the assistant system prompt from the agent config.""" + prompt_manager = PromptManager() + prompt = prompt_manager.get_prompt( + "realtime_agent.system_prompt", + agent_personality=self.agent.description, + agent_instructions=self.agent.instructions, + datetime=self.current_date_time, + ) + if self.pipeline_config.pre_tool_speech == "auto": + prompt += "\n\n" + prompt_manager.get_prompt("agent.pre_tool_speech") + return prompt + + def _build_tool_specs(self) -> list[dict[str, Any]]: + """Provider-agnostic tool specs from the agent tools (backend formats to its schema).""" + specs: list[dict[str, Any]] = [] + for tool in self.agent.tools or []: + specs.append( + { + "name": tool.function_name, + "description": f"{tool.name}: {tool.description}", + "parameters": { + "type": "object", + "properties": tool.get_parameter_properties(), + "required": tool.get_required_param_names(), + }, + } + ) + return specs + + # ── Role seams ───────────────────────────────────────────────────── + + async def handle_tool_call_request(self, request: ToolCallRequest) -> ToolCallResult: + """Execute a tool call the backend surfaced and record it in the audit log.""" + result = await execute_and_log_tool(self.tool_handler, self.audit_log, request.name, request.arguments) + return ToolCallResult(call_id=request.call_id, result=result) + + def record_audio(self, source: str, audio_data: bytes) -> None: + """Append PCM16 to the named channel buffer (alignment handled by callers).""" + if source == "user": + self.user_audio_buffer.extend(audio_data) + elif source == "assistant": + self.assistant_audio_buffer.extend(audio_data) + + def notify_conversation_ending(self, reason: str | None = None) -> None: + """No-op: native-S2S turn-taking needs no early-end signal (see AbstractAssistantServer).""" + return None + + def get_conversation_stats(self) -> dict[str, Any]: + return self.audit_log.get_stats() + + def get_initial_scenario_db(self) -> dict[str, Any]: + return self.tool_handler.original_db - @abstractmethod def get_final_scenario_db(self) -> dict[str, Any]: - """Return the (possibly mutated) scenario database state, for metrics. + return self.tool_handler.db + + # ── Server lifecycle ────────────────────────────────────────────── + + async def start(self) -> None: + """Start the FastAPI WebSocket server (non-blocking).""" + if self._running: + logger.warning("Assistant role already running") + return + + self.output_dir.mkdir(parents=True, exist_ok=True) + self._fw_log = FrameworkLogWriter(self.output_dir) + self._metrics_log = MetricsLogWriter(self.output_dir) + + self._app = FastAPI() + + @self._app.websocket("/ws") + async def websocket_endpoint(websocket: WebSocket) -> None: + await websocket.accept() + await self._handle_session(websocket) + + @self._app.websocket("/") + async def websocket_root(websocket: WebSocket) -> None: + await websocket.accept() + await self._handle_session(websocket) + + config = uvicorn.Config(self._app, host="0.0.0.0", port=self.port, log_level="warning", lifespan="off") + self._server = uvicorn.Server(config) + self._running = True + self._server_task = asyncio.create_task(self._server.serve()) + + while not self._server.started: + await asyncio.sleep(0.01) + + logger.info(f"Assistant role started on ws://localhost:{self.port}") + + async def _shutdown(self) -> None: + if not self._running: + return + self._running = False + if self._server: + self._server.should_exit = True + if self._server_task: + try: + await asyncio.wait_for(self._server_task, timeout=5.0) + except TimeoutError: + self._server_task.cancel() + try: + await self._server_task + except asyncio.CancelledError: + pass + except (asyncio.CancelledError, KeyboardInterrupt): + pass + self._server = None + self._server_task = None + logger.info(f"Assistant role stopped on port {self.port}") + + async def stop(self) -> asyncio.Task[None] | None: + """Shut down, extract audio, save outputs (mirrors AbstractAssistantServer.stop).""" + await self._shutdown() + self._ensure_mixed_audio() + + mixed_audio = bytes(self._audio_buffer) + user_audio = bytes(self.user_audio_buffer) + assistant_audio = bytes(self.assistant_audio_buffer) + sample_rate = self._audio_sample_rate + self._audio_buffer.clear() + self.user_audio_buffer.clear() + self.assistant_audio_buffer.clear() + + self.save_outputs() + + if mixed_audio or user_audio or assistant_audio: + return asyncio.create_task( + asyncio.to_thread(self._save_audio_deferred, mixed_audio, user_audio, assistant_audio, sample_rate) + ) + return None + + # ── Output persistence ──────────────────────────────────────────── + + def save_outputs(self) -> None: + self.audit_log.save(self.output_dir / "audit_log.json") + self.audit_log.save_transcript_jsonl(self.output_dir / "transcript.jsonl") + self._save_scenario_dbs() + logger.info(f"Outputs saved to {self.output_dir}") + + def _ensure_mixed_audio(self) -> None: + if self._audio_buffer: + return + if self.user_audio_buffer and self.assistant_audio_buffer: + diff_bytes = abs(len(self.user_audio_buffer) - len(self.assistant_audio_buffer)) + diff_ms = diff_bytes / (2 * self._audio_sample_rate) * 1000 + if diff_ms > 500: + logger.warning( + f"Audio buffer length mismatch: user={len(self.user_audio_buffer)} " + f"assistant={len(self.assistant_audio_buffer)} diff={diff_ms:.0f}ms — mixed recording may be skewed" + ) + self._audio_buffer = bytearray(pcm16_mix(bytes(self.user_audio_buffer), bytes(self.assistant_audio_buffer))) + elif self.user_audio_buffer: + self._audio_buffer = bytearray(self.user_audio_buffer) + elif self.assistant_audio_buffer: + self._audio_buffer = bytearray(self.assistant_audio_buffer) + + def _save_audio_deferred( + self, mixed_audio: bytes, user_audio: bytes, assistant_audio: bytes, sample_rate: int + ) -> None: + save_audio_track(mixed_audio, self.output_dir / "audio_mixed.wav", sample_rate) + save_audio_track(user_audio, self.output_dir / "audio_user.wav", sample_rate) + save_audio_track(assistant_audio, self.output_dir / "audio_assistant.wav", sample_rate) + if mixed_audio or user_audio or assistant_audio: + logger.info(f"Saved audio files to {self.output_dir} ({len(mixed_audio)} bytes mixed)") + + def _save_scenario_dbs(self) -> None: + try: + with open(self.output_dir / "initial_scenario_db.json", "w") as f: + json.dump(self.get_initial_scenario_db(), f, indent=2, sort_keys=True, default=str, ensure_ascii=False) + with open(self.output_dir / "final_scenario_db.json", "w") as f: + json.dump(self.get_final_scenario_db(), f, indent=2, sort_keys=True, default=str, ensure_ascii=False) + logger.info(f"Saved scenario database states to {self.output_dir}") + except Exception as e: + logger.error(f"Error saving scenario database states: {e}", exc_info=True) + raise + + # ── Session handling (Twilio WS <-> backend) ────────────────────── + + async def _handle_session(self, websocket: WebSocket) -> None: + logger.info("Client connected to assistant role") + self._user_turn = None + self._assistant_turn = _AssistantTurnState() + self._stream_sid = self.conversation_id + self._user_speaking = False + self._bot_speaking = False + + session = None + try: + session = await self.backend.open(system_prompt=self._system_prompt, tools=self._tool_specs) + # Trigger the initial greeting. + await self.backend.send(session, text=f"Say: '{self.initial_message}'") + + audio_output_queue: asyncio.Queue[bytes] = asyncio.Queue() + forward_task = asyncio.create_task(self._forward_user_audio(websocket, session)) + receive_task = asyncio.create_task(self._process_backend_events(session, audio_output_queue)) + pacer_task = asyncio.create_task(self._pace_audio_output(websocket, audio_output_queue)) + + done, pending = await asyncio.wait( + [forward_task, receive_task, pacer_task], return_when=asyncio.FIRST_COMPLETED + ) + for task in pending: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + for task in done: + if task.exception(): + logger.error(f"Session task failed: {task.exception()}") + except Exception as e: + logger.error(f"Assistant session error: {e}", exc_info=True) + finally: + if session is not None: + await self.backend.close(session) + logger.info("Client disconnected from assistant role") + + async def _pace_audio_output(self, websocket: WebSocket, audio_output_queue: asyncio.Queue[bytes]) -> None: + """Drain the output queue and forward chunks to Twilio at real-time rate.""" + next_send_time = time.monotonic() + try: + while True: + try: + chunk = await asyncio.wait_for(audio_output_queue.get(), timeout=1.0) + except TimeoutError: + continue + try: + await websocket.send_text(create_twilio_media_message(self._stream_sid, chunk)) + except Exception as e: + logger.error(f"Error sending audio to Twilio WS: {e}") + return + now = time.monotonic() + if next_send_time <= now: + next_send_time = now + next_send_time += MULAW_CHUNK_DURATION_S + sleep_duration = next_send_time - time.monotonic() + if sleep_duration > 0: + await asyncio.sleep(sleep_duration) + except asyncio.CancelledError: + pass + + async def _forward_user_audio(self, websocket: WebSocket, session: Any) -> None: + """Read Twilio media frames and forward audio to the backend.""" + try: + while True: + raw = await websocket.receive_text() + data = json.loads(raw) + event_type = data.get("event") + + if event_type == "start": + self._stream_sid = data.get("start", {}).get("streamSid", self.conversation_id) + continue + if event_type == "stop": + break + if event_type == "user_speech_start": + self._audio_interface_speech_start_ts = data.get("timestamp_ms") + continue + if event_type != "media": + continue + + mulaw_bytes = parse_twilio_media_message(raw) + if mulaw_bytes is None: + continue + + # Twilio 8kHz mulaw -> backend PCM (24kHz converters; the only backend + # today is 24k — a different-rate backend would need rate-generic utils). + pcm = mulaw_8k_to_pcm16_24k(mulaw_bytes) + if not self._bot_speaking: + sync_buffer_to_position(self.assistant_audio_buffer, len(self.user_audio_buffer)) + self.record_audio("user", pcm) + self._last_user_audio_mono = time.monotonic() + + await self.backend.send(session, audio=pcm) + except WebSocketDisconnect: + logger.debug("Twilio WebSocket disconnected") + except asyncio.CancelledError: + pass + except Exception as e: + logger.error(f"Error forwarding user audio: {e}", exc_info=True) + + async def _process_backend_events(self, session: Any, audio_output_queue: asyncio.Queue[bytes]) -> None: + """Consume normalized backend events and produce audit/transcript/audio + tool results.""" + try: + async for event in self.backend.receive(session): + try: + await self._handle_backend_event(event, session, audio_output_queue) + except Exception as e: + logger.error(f"Error handling event {event.event_type}: {e}", exc_info=True) + except asyncio.CancelledError: + pass + except Exception as e: + logger.error(f"Error in backend event loop: {e}", exc_info=True) + + async def _handle_backend_event( + self, event: BackendEvent, session: Any, audio_output_queue: asyncio.Queue[bytes] + ) -> None: + match event.event_type: + case BackendEventType.INPUT_SPEECH_STARTED: + self._on_speech_started() + case BackendEventType.INPUT_SPEECH_STOPPED: + self._on_speech_stopped() + case BackendEventType.TRANSCRIPT: + self._on_transcript(event) + case BackendEventType.AUDIO_OUTPUT: + await self._on_audio_output(event.audio or b"", audio_output_queue) + case BackendEventType.TURN_END: + self._on_turn_end(event) + case BackendEventType.TOOL_CALL_REQUEST: + await self._on_tool_call(event, session) + case BackendEventType.ERROR: + logger.error(f"Backend error: {event.error}") + # OUTPUT_TURN_STARTED / OUTPUT_AUDIO_DONE: not needed by the assistant. + + # ── Event handlers (consume only normalized fields) ─────────────── + + def _on_speech_started(self) -> None: + self._user_speaking = True + # Start a new user turn only if the previous one was flushed (preserves the + # original timestamp when VAD fires multiple speech_started per utterance). + if not self._user_turn or self._user_turn.flushed: + start_ts = self._audio_interface_speech_start_ts or _wall_ms() + self._user_turn = _UserTurnRecord(speech_started_wall_ms=start_ts) + if self._fw_log: + self._fw_log.turn_start(timestamp_ms=int(start_ts)) + self._audio_interface_speech_start_ts = None + + def _on_speech_stopped(self) -> None: + self._user_speaking = False + wall = _wall_ms() + if self._user_turn: + self._user_turn.speech_stopped_wall_ms = wall + else: + self._user_turn = _UserTurnRecord(speech_stopped_wall_ms=wall) + + def _on_transcript(self, event: BackendEvent) -> None: + # Assistant records only the inbound (user) transcript; the outbound + # transcript arrives finalized on TURN_END. + if event.metadata.get("stream") != "input": + return + if event.metadata.get("failed"): + if self._user_turn and not self._user_turn.flushed: + ts = self._user_turn.speech_started_wall_ms or None + self.audit_log.append_user_input("[user speech - transcription unavailable]", timestamp_ms=ts) + self._user_turn.flushed = True + return + transcript = (event.transcript or "").strip() + if not transcript: + return + ts = None + if self._user_turn: + ts = self._user_turn.speech_started_wall_ms or None + self._user_turn.transcript = transcript + self._user_turn.flushed = True + self.audit_log.append_user_input(transcript, timestamp_ms=ts) + + async def _on_audio_output(self, pcm16_bytes: bytes, audio_output_queue: asyncio.Queue[bytes]) -> None: + if not pcm16_bytes: + return + if self._assistant_turn.first_audio_wall_ms is None: + self._assistant_turn.first_audio_wall_ms = _wall_ms() + self._assistant_turn.responding = True + self._bot_speaking = True + # Model response latency: user speech end -> first audio chunk. + if self._user_turn and self._user_turn.speech_stopped_wall_ms and self._metrics_log: + latency_ms = int(self._assistant_turn.first_audio_wall_ms) - int(self._user_turn.speech_stopped_wall_ms) + if 0 < latency_ms < 30_000: + self._metrics_log.write_latency("model_response", latency_ms / 1000, self._model) + + # Skip the user-track pad while the user track is actively receiving audio. + user_recently_active = (time.monotonic() - self._last_user_audio_mono) <= USER_ACTIVE_GUARD_S + if not self._user_speaking and not user_recently_active: + sync_buffer_to_position(self.user_audio_buffer, len(self.assistant_audio_buffer)) + self.record_audio("assistant", pcm16_bytes) + self._assistant_turn.audio_was_streamed = True + + try: + mulaw_bytes = pcm16_24k_to_mulaw_8k(pcm16_bytes) + offset = 0 + while offset < len(mulaw_bytes): + await audio_output_queue.put(mulaw_bytes[offset : offset + MULAW_CHUNK_SIZE]) + offset += MULAW_CHUNK_SIZE + except Exception as e: + logger.error(f"Error converting audio for output queue: {e}") + + def _on_turn_end(self, event: BackendEvent) -> None: + meta = event.metadata + usage = meta.get("usage") + if usage and self._metrics_log: + self._metrics_log.write_token_usage( + processor="openai_realtime", + model=self._model, + prompt_tokens=usage.get("prompt_tokens", 0), + completion_tokens=usage.get("completion_tokens", 0), + ) + + content = (event.transcript or "").strip() + + if meta.get("interrupted"): + if content: + text = content + " [interrupted]" + self.audit_log.append_assistant_output(text, timestamp_ms=self._assistant_turn.first_audio_wall_ms) + if self._fw_log: + self._fw_log.s2s_transcript(text) + self._fw_log.turn_end(was_interrupted=True) + self._reset_assistant_turn() + return + + if meta.get("cancelled"): + self._reset_assistant_turn() + return + + has_fc = bool(meta.get("has_function_calls")) + audio_was_streamed = self._assistant_turn.audio_was_streamed + + # Skip rules (unchanged from the s2s server): tool-call-only, mixed-no-audio, + # audio-without-transcript, and empty turns are not logged as assistant output. + if (not content and has_fc) or (content and not audio_was_streamed and has_fc) or not content: + self._reset_assistant_turn() + return + + timestamp = self._assistant_turn.first_audio_wall_ms or _wall_ms() + self.audit_log.append_assistant_output(content, timestamp_ms=timestamp) + if self._fw_log: + self._fw_log.llm_response(content) + self._fw_log.turn_end(was_interrupted=False) + self._reset_assistant_turn() + + def _reset_assistant_turn(self) -> None: + if self._assistant_turn.first_audio_wall_ms is not None: + self._bot_speaking = False + self._assistant_turn = _AssistantTurnState() - Mirrors ``AbstractAssistantServer.get_final_scenario_db()``. - """ - ... + async def _on_tool_call(self, event: BackendEvent, session: Any) -> None: + request = event.tool_call_request + assert request is not None + logger.info(f"Tool call: {request.name}({json.dumps(request.arguments, ensure_ascii=False)})") + result = await self.handle_tool_call_request(request) + if self._fw_log: + self._fw_log.write( + "tool_call", + { + "frame": "tool_call", + "tool_name": request.name, + "arguments": request.arguments, + "result": result.result, + }, + ) + await self.backend.send(session, tool_result=result) diff --git a/src/eva/role/base.py b/src/eva/role/base.py index a75b4fa1..1b71e28a 100644 --- a/src/eva/role/base.py +++ b/src/eva/role/base.py @@ -1,8 +1,8 @@ """Abstract ``Role`` base contract: prompt/tools/goal ownership over a ``Backend``. -DESIGN ONLY (Step 1 of the refactor, see docs/refactor-step1.md). Every -method body here is a stub -- this module defines shapes, not behavior, and -is not imported by any existing code path. +See docs/refactor-step1.md. Abstract base shared by the concrete +``AssistantRole`` / ``UserRole``, which the worker constructs for any provider +the ``BackendFactory`` supports. This module holds only the shared ``Role`` base. The two concrete roles live in sibling modules -- ``AssistantRole`` in ``eva.role.assistant`` and @@ -12,40 +12,40 @@ separate files keeps each phase's diff scoped to one role. Design choice -- one ``Role`` base with ``AssistantRole``/``UserRole`` -subclasses, rather than two unrelated ABCs: - Both roles share an identical *control loop* shape: construct a backend - via ``BackendFactory``, ``build_prompt()`` before opening it, drive - ``backend.receive()`` and dispatch tool-call requests to - ``handle_tool_call_request()``, and record recorded audio/transcript for - output. What differs between them is only the *data* they carry (agent - config + tool catalog for the assistant; goal + persona + starting - utterance for the user) and how they decide the conversation is over. - That's a difference in constructor args and a couple of abstract methods, - not in control flow -- so one shared base with two thin subclasses avoids - duplicating the event loop, while still keeping tool-ownership and - prompt-building role-specific via abstract methods. If the two roles' - control loops diverge significantly in a later phase, splitting them - apart is a mechanical extraction of ``Role`` into two ABCs -- nothing - here should make that harder. +subclasses: + The two roles share the *seams* that don't depend on transport direction: + ``build_prompt()`` (instructions handed to the backend at open-time), + ``handle_tool_call_request()`` (role-side tool execution), and + ``record_audio()`` (accumulating audio for output). Those live here. + + Their *lifecycle* does differ today, and deliberately so: the assistant is + a WebSocket **server** the user connects to (passive; ``start()`` / + ``stop()``), while the user is the **driver** that dials in and runs the + conversation to completion (``run()``). That asymmetry is a consequence of + there being no mediator yet (docs/refactor-step1.md keeps ``Backend`` + direction-agnostic precisely so a later mediator can absorb the transport + and re-symmetrize the two roles). Rather than force an ill-fitting uniform + ``run()`` onto the server-shaped assistant, the lifecycle entry points live + on the subclasses (``AssistantRole`` / ``UserRole``); the scaffold + anticipated this ("if the two roles' control loops diverge, splitting is a + mechanical extraction"). When the mediator lands, both sides can converge + on a single driven loop. """ from __future__ import annotations from abc import ABC, abstractmethod -from pathlib import Path -from typing import Any from eva.backend.base import Backend, ToolCallRequest, ToolCallResult -from eva.backend.factory import BackendFactory class Role(ABC): - """Owns prompt, tools/goal, and a runtime-created ``Backend``. + """Owns prompt, tools/goal, and drives a worker-injected ``Backend``. - A ``Role`` is the thing that used to be split across - ``AbstractAssistantServer`` (assistant side) and ``AbstractUserSimulator`` - (user side): everything that is *not* pure provider API exchange lives - here instead of in ``Backend``. In particular: + A ``Role`` consolidates what the legacy ``AbstractAssistantServer`` + (assistant side) and ``AbstractUserSimulator`` (user side) each duplicated: + everything that is *not* pure provider API exchange lives here instead of + in ``Backend``. In particular: - Tool execution stays role-side (per docs/refactor-step1.md): a ``Role`` is responsible for turning a ``ToolCallRequest`` surfaced by its @@ -65,20 +65,22 @@ class declares the seam (``record_audio`` / ``save_outputs``) but does not implement the shared helper itself; that helper is later work. """ - def __init__(self, *, backend_factory: BackendFactory, backend_name: str, backend_config: dict[str, Any]) -> None: - """Construct the role's backend (but do not open its session yet). + def __init__(self, *, backend: Backend) -> None: + """Take the (not-yet-opened) backend the role will drive. + + The role does **not** construct its own backend and knows nothing about + the ``BackendFactory``. The worker owns the factory, calls + ``factory.create(name, config)``, and injects the resulting ``Backend`` + here. This keeps backend selection/configuration a worker concern and + lets the same backend be wired to either role. Args: - backend_factory: Factory used to construct ``self.backend``. - backend_name: Provider name passed through to - ``BackendFactory.create``. - backend_config: Provider-specific config passed through to - ``BackendFactory.create`` (not to be confused with the - ``config`` argument of ``Backend.open``, which is also - provider-specific but may be augmented by the role at - open-time, e.g. with a resolved sample rate). + backend: A constructed, not-yet-opened ``Backend`` (see + ``BackendFactory.create``). The role opens a session on it in + ``run()`` and holds the returned ``BackendSession`` handle; + per-exchange state lives on that handle, not on the backend. """ - self.backend: Backend = backend_factory.create(backend_name, backend_config) + self.backend = backend @abstractmethod def build_prompt(self) -> str: @@ -104,30 +106,6 @@ class docstring). Implementations should log the call/result (e.g. """ ... - @abstractmethod - async def run(self) -> str: - """Drive the conversation for this role until it reaches a terminal state. - - Expected shape (left to subclasses to implement, not prescribed in - detail here since the exact loop depends on the backend's - capabilities -- see ``BackendCapabilities``): - 1. ``await self.backend.open(system_prompt=self.build_prompt(), ...)`` - 2. Iterate ``self.backend.receive()``, dispatching - ``TOOL_CALL_REQUEST`` events to ``handle_tool_call_request`` and - feeding the ``ToolCallResult`` back via - ``self.backend.send(tool_result=...)``. - 3. Record audio/transcript events as they arrive (see - ``record_audio``). - 4. On a terminal event (hangup, timeout, transfer, error), call - ``await self.backend.close()`` and return an end-reason string. - - Returns: - A short end-reason string (e.g. ``"goodbye"``, ``"transfer"``, - ``"timeout"``, ``"error"``) -- mirrors the return contract of - today's ``AbstractUserSimulator.run_conversation()``. - """ - ... - @abstractmethod def record_audio(self, source: str, audio_data: bytes) -> None: """Accumulate a chunk of audio for later persistence. @@ -142,18 +120,3 @@ def record_audio(self, source: str, audio_data: bytes) -> None: audio_data: Raw PCM16 bytes at this role's recording sample rate. """ ... - - @abstractmethod - async def save_outputs(self, output_dir: Path) -> None: - """Persist this role's output artifacts to ``output_dir``. - - For ``AssistantRole`` this covers ``audit_log.json``, - ``transcript.jsonl``, scenario DB snapshots (mirrors - ``AbstractAssistantServer.save_outputs``). For ``UserRole`` this - covers ``user_simulator_events.jsonl`` (mirrors the event logger in - ``AbstractUserSimulator``). Audio WAV files are expected to be - written by the shared audio-recording helper referenced in - ``record_audio``, not necessarily by this method -- exact division of - labor is left to the later implementation phase. - """ - ... diff --git a/src/eva/role/user.py b/src/eva/role/user.py index 415d8b7a..d205929c 100644 --- a/src/eva/role/user.py +++ b/src/eva/role/user.py @@ -1,83 +1,491 @@ -"""``UserRole`` contract: the simulated-caller role. - -DESIGN ONLY (Step 1 of the refactor, see docs/refactor-step1.md). Method -bodies are stubs; nothing here is wired into the existing code path yet. - -Plug-in point (where this will eventually replace existing code): - Today the user side is a concrete ``AbstractUserSimulator`` subclass - selected by ``eva.user_simulator.factory.create_user_simulator(config, ...)`` - and constructed inside ``ConversationWorker._start_user_simulator()`` - (worker.py), which passes it ``server_url=f"ws://localhost:{port}/ws"`` - to reach the assistant server. The conversation is driven by - ``ConversationWorker._run_conversation()`` calling - ``user_simulator.run_conversation()``, whose returned end-reason string - becomes the conversation result. - - In a later phase, ``_start_user_simulator()`` becomes the construction - site for a ``UserRole`` (simulator config -> ``backend_name`` + - ``backend_config`` for the ``BackendFactory``), and the worker drives - ``role.run()`` (returning the same end-reason string via - ``get_end_reason()``) instead of ``run_conversation()``. Note the - ``server_url`` handoff is a *transport* detail that today's user side owns - directly; per docs/refactor-step1.md the ``Backend`` contract is kept - direction-agnostic precisely so this WS-connect concern can move into a - ``Backend`` implementation (or, later, a mediator) without the role - caring. This module is deliberately separate from ``eva.role.assistant`` - so the user-side migration can land as its own scoped diff. +"""``UserRole``: the simulated-caller role (one generic class). + +A single concrete, provider-agnostic role that holds a ``Backend`` and works +with any backend. Provider specifics (session, audio format, event parsing) +live in the backend; this class owns only the role-common concerns shared by +every user simulator regardless of provider: + +- the user-simulator system prompt (persona + goal / decision tree); +- the single caller-side ``end_call`` tool; +- the event logger (``user_simulator_events.jsonl``) and clean-user-audio WAV; +- the counterparty transport: the audio-bridge **client** that dials the + assistant, plus a generic "respond after the assistant's turn settles" + sequencing policy that asks the backend to speak via ``trigger_response()`` + (a no-op for self-driving backends). + +It consumes only the normalized ``BackendEvent`` stream, never a raw provider +event. Interpreting the *inbound* transcript as the assistant and the *output* +transcript as the caller is the one role-specific reading of the role-agnostic +events. + +Plug-in point: ``run()`` mirrors ``AbstractUserSimulator.run_conversation()`` +(drives to completion, returns the end-reason) and exposes the same +``on_conversation_ending`` hook, so the worker swap is 1:1. The worker takes +this path for every provider the ``BackendFactory`` supports. """ from __future__ import annotations -from abc import abstractmethod +import asyncio +from collections.abc import Callable +from contextlib import suppress +from pathlib import Path from typing import Any -from eva.backend.factory import BackendFactory +from pipecat.transcriptions.language import Language +from websockets.exceptions import ConnectionClosedOK + +try: + import audioop +except ImportError: + import audioop_lts as audioop # type: ignore[import-not-found,no-redef] + +from eva.backend.base import Backend, BackendEvent, BackendEventType, ToolCallRequest, ToolCallResult +from eva.models.config import LANGUAGE_DISPLAY_NAMES, PerturbationConfig from eva.role.base import Role +from eva.user_simulator.audio_bridge import BotToBotAudioBridge +from eva.user_simulator.base import load_behavior_prompts +from eva.user_simulator.event_logger import UserSimulatorEventLogger +from eva.user_simulator.perturbation import AudioPerturbator +from eva.utils.audio_utils import save_audio_track +from eva.utils.culture import add_user_language_directive +from eva.utils.logging import current_record_id, get_logger +from eva.utils.prompt_manager import PromptManager + +logger = get_logger(__name__) + +BRIDGE_SAMPLE_RATE = 16000 +CALLER_RESPONSE_SETTLE_SECONDS = 2.0 +CALLER_RESPONSE_POLL_SECONDS = 0.05 +CALLER_PLAYBACK_DRAIN_SECONDS = 15.0 +END_CALL_DESCRIPTION = """Use this to end the phone call and hang up. + +Call this function when it is time to end the call and one of the following is true: +1. The agent has confirmed your request is resolved, all steps are completed, and you have said goodbye. +2. The agent has initiated a transfer to a live agent. +3. The agent has been unable to make progress for at least 5 consecutive turns. +4. The agent says goodbye or indicates the conversation is over. +5. The agent indicates that the remainder of your request cannot be fulfilled. +6. The assistant reports an unrecoverable processing error. + +Never call this tool in the same turn that you provide the agent with data, an identifier, +an approval to proceed, a transfer request, or any other information. Say a brief goodbye first.""" class UserRole(Role): - """Role that simulates the human caller (today's "user simulator"). + """Generic simulated-caller role. Drives any ``Backend``; owns the bridge-client transport.""" - Carries goal/persona instead of agent config/tools -- its tool surface, - if any, is limited to caller-side affordances like ``end_call`` (see - ``END_CALL_DESCRIPTION`` in today's ``eva.user_simulator.base``), not a - business tool catalog. - """ + # Generic phone-caller session conventions (provider-agnostic): telephony + # input, manual turn-taking (respond after the assistant settles), caller VAD + # tuning, and no parallel tool calls. Unpacked by the worker into the caller + # backend's args, so the caller session shape lives with the caller role. + CALLER_BACKEND_DEFAULTS: dict[str, Any] = { + "input_format": "pcmu", + "manual_turn_taking": True, + "vad_settings": {"threshold": 0.5, "prefix_padding_ms": 300, "silence_duration_ms": 500}, + "parallel_tool_calls": False, + } def __init__( self, *, - backend_factory: BackendFactory, - backend_name: str, - backend_config: dict[str, Any], - goal: dict[str, Any], - persona_config: dict[str, Any], + backend: Backend, current_date_time: str, + persona_config: dict[str, Any], + goal: dict[str, Any], + server_url: str, + output_dir: Path, + agent_id: str, + provider: str = "unknown", + timeout: int = 600, + perturbation_config: PerturbationConfig | None = None, + language: str = "en", ) -> None: - """Initialize the user role. - - Args: - backend_factory: Factory used to construct the backend. - backend_name: Key passed to the factory to select a backend. - backend_config: Provider-specific configuration for the backend. - goal: User goal / decision-tree data -- mirrors - ``AbstractUserSimulator.goal``. - persona_config: Persona/voice/behavior configuration -- mirrors - ``AbstractUserSimulator.persona_config``. - current_date_time: Threaded into prompt construction, mirroring - existing plumbing. - """ - super().__init__(backend_factory=backend_factory, backend_name=backend_name, backend_config=backend_config) - self.goal = goal + super().__init__(backend=backend) self.persona_config = persona_config + self.goal = goal self.current_date_time = current_date_time + self.server_url = server_url + self.output_dir = Path(output_dir) + self.agent_id = agent_id + self.provider = provider + self.timeout = timeout + self._language = language + self._perturbation_config = perturbation_config + self._perturbator = ( + AudioPerturbator(perturbation_config) + if perturbation_config is not None + and (perturbation_config.background_noise is not None or perturbation_config.connection_degradation) + else None + ) + + self._audio_interface: BotToBotAudioBridge | None = None + self._end_reason = "unknown" + self._conversation_done = asyncio.Event() + self._ending_signaled = False + # Set by the worker to the assistant role's notify_conversation_ending. + self.on_conversation_ending: Callable[[str | None], None] | None = None + self.event_logger = UserSimulatorEventLogger(self.output_dir / "user_simulator_events.jsonl", provider=provider) + self._user_clean_audio = bytearray() + self._record_id = current_record_id.get() + + # Caller response-sequencing state. + self._assistant_audio_queue: asyncio.Queue[bytes] = asyncio.Queue() + self._caller_audio_seen = False + self._caller_playback_pending = False + self._caller_response_active = False + self._caller_response_pending = False + self._caller_response_task: asyncio.Task[None] | None = None + self._assistant_transcript_ready = False + self._end_call_pending = False + self._resampler_state: Any = None + + # ── Role seams (prompt / tools / recording) ─────────────────────── + + def build_prompt(self) -> str: + """Build the user-simulator system prompt from persona + goal.""" + behavior_prompts = load_behavior_prompts() + if self._perturbation_config and self._perturbation_config.behavior: + user_persona = behavior_prompts[self._perturbation_config.behavior.value] + else: + user_persona = behavior_prompts["default"] + user_persona = add_user_language_directive( + self._language, + LANGUAGE_DISPLAY_NAMES.get(Language(self._language), self._language), + user_persona, + ) + domain = self.agent_id.removeprefix("agent_") + return PromptManager().get_prompt( + f"user_simulator.system_prompt_{domain}", + high_level_user_goal=self.goal["high_level_user_goal"], + must_have_criteria=self.goal["decision_tree"]["must_have_criteria"], + escalation_behavior=self.goal["decision_tree"]["escalation_behavior"], + nice_to_have_criteria=self.goal["decision_tree"]["nice_to_have_criteria"], + negotiation_behavior=self.goal["decision_tree"]["negotiation_behavior"], + resolution_condition=self.goal["decision_tree"]["resolution_condition"], + failure_condition=self.goal["decision_tree"]["failure_condition"], + edge_cases=self.goal["decision_tree"]["edge_cases"], + information_required=self.goal["information_required"], + user_persona=user_persona, + starting_utterance=self.goal["starting_utterance"], + current_date_time=self.current_date_time, + ) + + def _end_call_tool_spec(self) -> dict[str, Any]: + """Provider-agnostic ``end_call`` tool spec (backend formats to its schema).""" + return { + "name": "end_call", + "description": END_CALL_DESCRIPTION, + "parameters": {"type": "object", "properties": {}}, + } + + async def handle_tool_call_request(self, request: ToolCallRequest) -> ToolCallResult: + """Handle a caller-side tool call. Only ``end_call`` exists; it arms hang-up. + + The caller never returns a function_call_output to the provider -- + ``end_call`` is terminal intent, consumed locally (the ``run()`` loop + does not relay this result back to the backend). + """ + if request.name == "end_call": + self.event_logger.log_event("tool_call", {"name": "end_call", "arguments": request.arguments}) + self._end_call_pending = True + return ToolCallResult(call_id=request.call_id, result={}) + + def record_audio(self, source: str, audio_data: bytes) -> None: + """Retain only the clean (unperturbed) user track; other sources are the assistant's.""" + if source == "user_clean": + self._user_clean_audio.extend(audio_data) + + # ── Lifecycle ───────────────────────────────────────────────────── + + async def run(self) -> str: + try: + await self._run_conversation() + except Exception as exc: + logger.error(f"User caller simulation error: {exc}", exc_info=True) + self._end_reason = "error" + self.event_logger.log_error(str(exc)) + if self._audio_interface is not None: + with suppress(Exception): + await self._audio_interface.stop_async() + self.event_logger.log_connection_state("session_ended", {"reason": self._end_reason}) + finally: + self.event_logger.save() + return self._end_reason - @abstractmethod def get_end_reason(self) -> str: - """Return the terminal end-reason for this conversation. + return self._end_reason - Mirrors the return value of today's - ``AbstractUserSimulator.run_conversation()`` (``"goodbye"``, - ``"transfer"``, ``"timeout"``, ``"error"``, ...). + def _connection_info(self) -> dict[str, Any]: + """Connection metadata for the ``connected`` event, built from what the role knows. + + Generic caller session facts (transport, sequencing, sample rates) -- + the OpenAI-specific labels the old simulator logged are intentionally + not reproduced here (a sub-macro logging diff). """ - ... + d = self.CALLER_BACKEND_DEFAULTS + return { + "server_url": self.server_url, + "caller_provider": self.provider, + "caller_input_format": d["input_format"], + "caller_turn_detection": {**d["vad_settings"], "manual_turn_taking": d["manual_turn_taking"]}, + "caller_input_sample_rate": self.backend.input_sample_rate, + "caller_output_sample_rate": self.backend.output_sample_rate, + } + + async def _run_conversation(self) -> None: + self._audio_interface = BotToBotAudioBridge( + websocket_uri=self.server_url, + conversation_id=self.output_dir.name, + record_callback=self.record_audio, + event_logger=self.event_logger, + conversation_done_callback=self._on_conversation_end, + perturbator=self._perturbator, + disconnect_reason="assistant_disconnect", + ) + await self._audio_interface.start_async() + self._audio_interface.start(self._on_assistant_audio) + self.event_logger.log_connection_state("connected", self._connection_info()) + + forward_task: asyncio.Task[Any] | None = None + listener_task: asyncio.Task[Any] | None = None + completion_task: asyncio.Task[Any] | None = None + session = await self.backend.open(system_prompt=self.build_prompt(), tools=[self._end_call_tool_spec()]) + self.event_logger.log_connection_state("session_started") + try: + forward_task = asyncio.create_task(self._forward_assistant_audio(session)) + listener_task = asyncio.create_task(self._listen_for_caller_events(session)) + completion_task = asyncio.create_task(self._wait_for_conversation_end()) + + await self._wait_for_session_completion(completion_task, forward_task, listener_task) + # Allow final goodbye audio + transcripts to flush before closing. + await asyncio.sleep(4.0) + finally: + if self._caller_response_task is not None: + await self._cancel_background_task(self._caller_response_task) + for task in (completion_task, forward_task, listener_task): + if task is not None: + await self._cancel_background_task(task) + await self.backend.close(session) + await self._audio_interface.stop_async() + self._save_clean_user_audio(BRIDGE_SAMPLE_RATE) + self.event_logger.log_connection_state("session_ended", {"reason": self._end_reason}) + + @staticmethod + async def _cancel_background_task(task: asyncio.Task[Any]) -> None: + task.cancel() + with suppress(asyncio.CancelledError, Exception): + await task + + async def _wait_for_conversation_end(self) -> None: + try: + await asyncio.wait_for(self._conversation_done.wait(), timeout=self.timeout) + except TimeoutError: + self.event_logger.log_event("timeout", {"duration": self.timeout}) + self._on_conversation_end("timeout") + + async def _wait_for_session_completion( + self, + completion_task: asyncio.Task[Any], + forward_task: asyncio.Task[Any], + listener_task: asyncio.Task[Any], + ) -> None: + done, _ = await asyncio.wait( + {completion_task, forward_task, listener_task}, return_when=asyncio.FIRST_COMPLETED + ) + if completion_task in done: + return + finished_task = next(iter(done)) + if self._conversation_done.is_set(): + await completion_task + return + exception = finished_task.exception() + if exception is not None: + raise exception + task_name = "listener" if finished_task is listener_task else "audio forwarder" + raise RuntimeError(f"Caller {task_name} stopped unexpectedly") + + # ── Messaging hooks ──────────────────────────────────────────────── + + def signal_conversation_ending(self, reason: str | None = None) -> None: + """Advise the assistant the call is over, before transport teardown. Idempotent, never raises.""" + if self._ending_signaled: + return + self._ending_signaled = True + if self.on_conversation_ending is None: + return + try: + self.on_conversation_ending(reason or self._end_reason) + except Exception as e: + logger.warning(f"Failed to signal conversation ending to the assistant: {e}") + + def _on_conversation_end(self, reason: str = "goodbye") -> None: + if not self._conversation_done.is_set(): + self._end_reason = reason + self._conversation_done.set() + logger.info(f"Conversation end signaled: {reason}") + self.signal_conversation_ending(reason) + + def _on_user_speaks(self, response: str) -> None: + current_record_id.set(self._record_id) + self.event_logger.log_event("user_speech", {"text": response, "source": "simulated_user"}) + + def _on_assistant_speaks(self, transcript: str) -> None: + current_record_id.set(self._record_id) + self.event_logger.log_event("assistant_speech", {"text": transcript, "source": "assistant"}) + + def _save_clean_user_audio(self, sample_rate: int) -> None: + if save_audio_track(bytes(self._user_clean_audio), self.output_dir / "audio_user_clean.wav", sample_rate): + logger.info(f"Saved clean user audio to {self.output_dir / 'audio_user_clean.wav'}") + + # ── Assistant audio -> caller backend ───────────────────────────── + + def _on_assistant_audio(self, mulaw_audio: bytes) -> None: + if mulaw_audio and not self._caller_response_active and not self._caller_audio_is_playing(): + self._assistant_audio_queue.put_nowait(mulaw_audio) + + def _caller_audio_is_playing(self) -> bool: + return self._audio_interface is not None and self._audio_interface.is_caller_playing() + + async def _forward_assistant_audio(self, session: Any) -> None: + while True: + mulaw_audio = await self._assistant_audio_queue.get() + if not mulaw_audio: + continue + try: + await self.backend.send(session, audio=mulaw_audio) + except ConnectionClosedOK: + return + + # ── Caller event processing (normalized events only) ────────────── + + async def _listen_for_caller_events(self, session: Any) -> None: + try: + async for event in self.backend.receive(session): + await self._handle_caller_event(event, session) + except asyncio.CancelledError: + raise + except Exception as exc: + logger.error(f"Caller event loop error: {exc}", exc_info=True) + self.event_logger.log_error(str(exc)) + self._on_conversation_end("error") + + async def _handle_caller_event(self, event: BackendEvent, session: Any) -> None: + meta = event.metadata + match event.event_type: + case BackendEventType.INPUT_SPEECH_STOPPED: + self._schedule_caller_response(session, trigger="vad_speech_stopped") + + case BackendEventType.TRANSCRIPT: + if meta.get("stream") == "input": + # From the caller's POV the inbound party is the assistant. + if event.transcript: + self._assistant_transcript_ready = True + self._on_assistant_speaks(event.transcript) + elif meta.get("stream") == "output" and event.transcript: + self._on_user_speaks(event.transcript) + + case BackendEventType.AUDIO_OUTPUT: + if event.audio and self._audio_interface is not None: + pcm16_16k, self._resampler_state = audioop.ratecv( + event.audio, 2, 1, self.backend.output_sample_rate, BRIDGE_SAMPLE_RATE, self._resampler_state + ) + self._audio_interface.output(pcm16_16k) + self._caller_audio_seen = True + self._caller_playback_pending = True + + case BackendEventType.OUTPUT_TURN_STARTED: + self._caller_response_active = True + + case BackendEventType.OUTPUT_AUDIO_DONE: + self._flush_caller_output() + self._resampler_state = None + + case BackendEventType.TOOL_CALL_REQUEST: + if event.tool_call_request is not None: + await self.handle_tool_call_request(event.tool_call_request) + + case BackendEventType.TURN_END: + self._flush_caller_output() + await self._finish_caller_response(session) + + case BackendEventType.ERROR: + if meta.get("code") == "conversation_already_has_active_response": + self._caller_response_active = True + self._caller_response_pending = True + self.event_logger.log_event( + "caller_response_coalesced", {"trigger": "active_response_error", "error": event.error} + ) + return + self.event_logger.log_error(str(event.error)) + self._on_conversation_end("error") + + def _schedule_caller_response(self, session: Any, *, trigger: str, require_settled_turn: bool = True) -> None: + if self._conversation_done.is_set(): + return + if self._caller_response_task is not None and not self._caller_response_task.done(): + self.event_logger.log_event("caller_response_coalesced", {"trigger": trigger}) + return + self._caller_response_task = asyncio.create_task( + self._create_caller_response_when_ready(session, trigger, require_settled_turn=require_settled_turn) + ) + + async def _create_caller_response_when_ready( + self, session: Any, trigger: str, *, require_settled_turn: bool = True + ) -> None: + try: + while not self._conversation_done.is_set(): + turn_ready = not require_settled_turn or self._assistant_turn_is_settled() + if not self._caller_response_active and turn_ready: + self._caller_response_active = True + self._assistant_transcript_ready = False + self.event_logger.log_event("caller_response_created", {"trigger": trigger}) + try: + await self.backend.trigger_response(session) + except Exception as exc: + self._caller_response_active = False + self.event_logger.log_error( + "Failed to request caller response", {"trigger": trigger, "error": str(exc)} + ) + self._on_conversation_end("error") + return + await asyncio.sleep(CALLER_RESPONSE_POLL_SECONDS) + finally: + self._caller_response_task = None + + def _assistant_turn_is_settled(self) -> bool: + if not self._assistant_transcript_ready: + return False + if self._audio_interface is None or self._audio_interface.is_assistant_playing(): + return False + ended_time = self._audio_interface.assistant_audio_ended_at + if ended_time is None: + return False + return asyncio.get_running_loop().time() - ended_time >= CALLER_RESPONSE_SETTLE_SECONDS + + async def _wait_for_caller_playback_complete(self) -> None: + if self._audio_interface is None or not self._caller_playback_pending: + return + while True: + if not self._audio_interface.is_caller_playing(): + await asyncio.sleep(0.7) + if not self._audio_interface.is_caller_playing(): + self._caller_playback_pending = False + return + await asyncio.sleep(0.05) + + async def _finish_caller_response(self, session: Any) -> None: + self._caller_response_active = False + with suppress(TimeoutError): + await asyncio.wait_for(self._wait_for_caller_playback_complete(), timeout=CALLER_PLAYBACK_DRAIN_SECONDS) + if self._end_call_pending: + self._end_call_pending = False + self._on_conversation_end("goodbye") + elif self._caller_response_pending: + self._caller_response_pending = False + self._schedule_caller_response(session, trigger="pending_after_response_done", require_settled_turn=False) + + def _flush_caller_output(self) -> None: + if self._caller_audio_seen and self._audio_interface is not None: + self._audio_interface.output(b"\x00\x00") + self._caller_audio_seen = False diff --git a/src/eva/user_simulator/factory.py b/src/eva/user_simulator/factory.py index d2cef7e2..059b08cf 100644 --- a/src/eva/user_simulator/factory.py +++ b/src/eva/user_simulator/factory.py @@ -4,7 +4,7 @@ from typing import Any -from eva.models.config import ElevenLabsSimulatorConfig, OpenAIRealtimeSimulatorConfig, UserSimulatorConfig +from eva.models.config import ElevenLabsSimulatorConfig, S2SSimulatorConfig, UserSimulatorConfig from eva.user_simulator.base import AbstractUserSimulator @@ -17,7 +17,7 @@ def create_user_simulator( from eva.user_simulator.elevenlabs import ElevenLabsUserSimulator return ElevenLabsUserSimulator(**kwargs) - if isinstance(simulator_config, OpenAIRealtimeSimulatorConfig): + if isinstance(simulator_config, S2SSimulatorConfig): from eva.user_simulator.openai_realtime import OpenAIRealtimeUserSimulator return OpenAIRealtimeUserSimulator(simulator_config=simulator_config, **kwargs) diff --git a/src/eva/user_simulator/openai_realtime.py b/src/eva/user_simulator/openai_realtime.py index 71c1c60b..595b7b70 100644 --- a/src/eva/user_simulator/openai_realtime.py +++ b/src/eva/user_simulator/openai_realtime.py @@ -17,7 +17,7 @@ except ImportError: import audioop_lts as audioop -from eva.models.config import OpenAIRealtimeSimulatorConfig, PerturbationConfig +from eva.models.config import PerturbationConfig, S2SSimulatorConfig from eva.user_simulator.audio_bridge import BotToBotAudioBridge from eva.user_simulator.base import AbstractUserSimulator from eva.utils.logging import get_logger @@ -71,7 +71,7 @@ def __init__( perturbation_config: PerturbationConfig | None = None, language: str = "en", *, - simulator_config: OpenAIRealtimeSimulatorConfig, + simulator_config: S2SSimulatorConfig, ) -> None: super().__init__( current_date_time=current_date_time, diff --git a/tests/unit/models/test_config_models.py b/tests/unit/models/test_config_models.py index d797dbcd..0a5a6ff9 100644 --- a/tests/unit/models/test_config_models.py +++ b/tests/unit/models/test_config_models.py @@ -13,9 +13,9 @@ from eva.models.config import ( ElevenLabsSimulatorConfig, ModelConfig, - OpenAIRealtimeSimulatorConfig, PipelineType, RunConfig, + S2SSimulatorConfig, ) MODEL_LIST = [ @@ -1168,7 +1168,7 @@ def test_defaults_preserve_elevenlabs(self): assert config.provider == "elevenlabs" def test_openai_realtime_defaults(self): - config = OpenAIRealtimeSimulatorConfig() + config = S2SSimulatorConfig() assert config.model == "gpt-realtime-1.5" assert config.female_voice == "marin" @@ -1186,7 +1186,7 @@ def test_nested_environment_configuration(self): } ) - assert config.user_simulator == OpenAIRealtimeSimulatorConfig( + assert config.user_simulator == S2SSimulatorConfig( model="gpt-realtime-2", female_voice="coral", male_voice="verse", diff --git a/tests/unit/orchestrator/test_preflight.py b/tests/unit/orchestrator/test_preflight.py index 1510583a..3fd5d3fd 100644 --- a/tests/unit/orchestrator/test_preflight.py +++ b/tests/unit/orchestrator/test_preflight.py @@ -11,7 +11,7 @@ from pipecat.services.stt_service import STTService from pipecat.services.tts_service import TTSService -from eva.models.config import ModelConfig, RunConfig +from eva.models.config import ModelConfig, RunConfig, S2SSimulatorConfig from eva.orchestrator import preflight from eva.orchestrator.preflight import ( PreflightError, @@ -120,6 +120,59 @@ async def test_s2s_is_skipped(tmp_path): assert results == [] +# ── Cheap backend-construction validation (_preflight_backends) ────────────── + + +def test_backend_construction_validates_s2s_assistant_missing_key(tmp_path): + # S2S key is no longer validated at config load (removed); construction catches it. + with patch.dict(os.environ, _BASE_ENV, clear=True): # no OPENAI_API_KEY + cfg = RunConfig( + model=ModelConfig(s2s="gpt-realtime", s2s_params={"model": "gpt-realtime"}), + framework="openai_realtime", + output_dir=tmp_path / "out", + run_id="r", + ) + with pytest.raises(PreflightError, match="assistant framework 'openai_realtime'"): + preflight._preflight_backends(cfg) + + +def test_backend_construction_passes_with_key(tmp_path): + with patch.dict(os.environ, _BASE_ENV | {"OPENAI_API_KEY": "k"}, clear=True): + cfg = RunConfig( + model=ModelConfig(s2s="gpt-realtime", s2s_params={"api_key": "k", "model": "gpt-realtime"}), + framework="openai_realtime", + output_dir=tmp_path / "out", + run_id="r", + ) + preflight._preflight_backends(cfg) # no raise + + +def test_backend_construction_skips_legacy_user_sim(tmp_path): + # ElevenLabs (the default user sim) isn't factory-backed -> skipped, no raise. + with patch.dict(os.environ, _BASE_ENV, clear=True): + preflight._preflight_backends(_cascade_config(tmp_path)) + + +def test_backend_construction_validates_factory_user_sim(tmp_path): + with patch.dict(os.environ, _BASE_ENV, clear=True): # no OPENAI_API_KEY + cfg = _cascade_config(tmp_path) + cfg.user_simulator = S2SSimulatorConfig(provider="openai_realtime") + with pytest.raises(PreflightError, match="user simulator 'openai_realtime'"): + preflight._preflight_backends(cfg) + + +@pytest.mark.asyncio +async def test_backend_construction_runs_even_when_preflight_disabled(tmp_path): + # --no-preflight skips only the live model probes; the cheap construction check still + # runs. Use the user-sim provider, whose key isn't validated at config load. + with patch.dict(os.environ, _BASE_ENV, clear=True): # no OPENAI_API_KEY + cfg = _cascade_config(tmp_path) + cfg.user_simulator = S2SSimulatorConfig(provider="openai_realtime") + cfg.preflight = False + with pytest.raises(PreflightError, match="user simulator 'openai_realtime'"): + await run_preflight(cfg) + + @pytest.mark.asyncio async def test_guard_times_out(): async def hang(): diff --git a/tests/unit/orchestrator/test_worker.py b/tests/unit/orchestrator/test_worker.py index e5a0366d..7c2d45df 100644 --- a/tests/unit/orchestrator/test_worker.py +++ b/tests/unit/orchestrator/test_worker.py @@ -6,6 +6,7 @@ import pytest +from eva.models.config import ElevenLabsSimulatorConfig from eva.orchestrator.worker import USER_SIMULATOR_SHUTDOWN_GRACE_SECONDS, ConversationWorker, _percentile @@ -222,8 +223,10 @@ async def test_raises_when_simulator_not_initialized(self, tmp_path): class TestUserSimulatorSelection: @pytest.mark.asyncio async def test_worker_uses_configured_factory_and_timeout(self, tmp_path, monkeypatch): + # A non-factory (legacy) provider falls through to create_user_simulator; a + # factory-backed provider (openai_realtime/grok_voice) takes the Role/Backend path. worker = _make_worker(tmp_path) - worker.config.user_simulator = MagicMock(provider="openai_realtime") + worker.config.user_simulator = ElevenLabsSimulatorConfig() worker.config.perturbation = None worker.config.language = "en" worker.config.conversation_time_limit_seconds = 60 diff --git a/tests/unit/test_grok_voice_backend.py b/tests/unit/test_grok_voice_backend.py new file mode 100644 index 00000000..a71c96ff --- /dev/null +++ b/tests/unit/test_grok_voice_backend.py @@ -0,0 +1,111 @@ +"""Unit tests for GrokVoiceBackend (no network). + +Covers what Grok changes vs the OpenAI Realtime backend it subclasses: xAI +defaults, the required api key, factory dispatch, and the buffered/deferred +input-transcription behavior. Everything else is inherited and covered by +``test_openai_realtime_backend``. +""" + +from __future__ import annotations + +from types import SimpleNamespace + +import pytest + +from eva.backend.base import BackendEventType +from eva.backend.factory import BackendFactory +from eva.backend.grok_voice import DEFAULT_VOICE, XAI_REALTIME_BASE_URL, GrokVoiceBackend, GrokVoiceSession +from eva.backend.openai_realtime import OpenAIRealtimeSession + + +def _backend(**overrides) -> GrokVoiceBackend: + return GrokVoiceBackend(config={"model": "grok-voice", "api_key": "xai-key", **overrides}) + + +def _session() -> GrokVoiceSession: + return GrokVoiceSession(client=None, conn_cm=None, conn=None) # type: ignore[arg-type] + + +def _map(session, event): + return GrokVoiceBackend._map_event(session, event) + + +def test_api_key_falls_back_to_xai_env(monkeypatch): + monkeypatch.setenv("XAI_API_KEY", "xai-from-env") + b = GrokVoiceBackend(config={"model": "grok-voice"}) + assert b._api_key == "xai-from-env" + + +def test_openai_env_does_not_satisfy_grok(monkeypatch): + # An OpenAI key is the wrong key for x.ai: it must NOT be used as a fallback. + monkeypatch.delenv("XAI_API_KEY", raising=False) + monkeypatch.setenv("OPENAI_API_KEY", "sk-should-not-be-used") + with pytest.raises(ValueError, match="XAI_API_KEY"): + GrokVoiceBackend(config={"model": "grok-voice"}) + + +def test_xai_defaults_applied(): + b = _backend() + assert b._base_url == XAI_REALTIME_BASE_URL + assert b._session_config["audio"]["output"]["voice"] == DEFAULT_VOICE + + +def test_explicit_config_wins_over_defaults(): + b = _backend(base_url="https://custom/v1", voice="ara") + assert b._base_url == "https://custom/v1" + assert b._session_config["audio"]["output"]["voice"] == "ara" + + +def test_open_uses_grok_session_class(): + assert GrokVoiceBackend._SESSION_CLS is GrokVoiceSession + assert issubclass(GrokVoiceSession, OpenAIRealtimeSession) + + +def test_factory_dispatch(): + b = BackendFactory().create("grok_voice", {"model": "grok-voice", "api_key": "xai-key"}) + assert isinstance(b, GrokVoiceBackend) + + +def test_incremental_transcription_is_buffered_not_emitted(): + s = _session() + # Progressive completed events accumulate but emit nothing. + assert _map(s, SimpleNamespace(type="conversation.item.input_audio_transcription.completed", transcript="I")) == [] + assert ( + _map(s, SimpleNamespace(type="conversation.item.input_audio_transcription.completed", transcript="I want")) + == [] + ) + assert s.pending_input_transcript == "I want" + + +def test_buffered_transcript_flushes_on_speech_started(): + s = _session() + _map(s, SimpleNamespace(type="conversation.item.input_audio_transcription.completed", transcript="hello there")) + events = _map(s, SimpleNamespace(type="input_audio_buffer.speech_started")) + # First event is the flushed final input transcript, then the inherited speech-start. + assert events[0].event_type == BackendEventType.TRANSCRIPT + assert events[0].transcript == "hello there" + assert events[0].metadata == {"stream": "input", "final": True} + assert events[-1].event_type == BackendEventType.INPUT_SPEECH_STARTED + assert s.pending_input_transcript == "" + + +def test_buffered_transcript_flushes_on_response_done(): + s = _session() + _map(s, SimpleNamespace(type="conversation.item.input_audio_transcription.completed", transcript="final text")) + events = _map(s, SimpleNamespace(type="response.done", response=None)) + assert events[0].event_type == BackendEventType.TRANSCRIPT + assert events[0].transcript == "final text" + assert any(e.event_type == BackendEventType.TURN_END for e in events) + + +def test_no_pending_flush_is_noop(): + s = _session() + # speech_started with nothing buffered -> just the inherited event, no TRANSCRIPT. + events = _map(s, SimpleNamespace(type="input_audio_buffer.speech_started")) + assert all(e.event_type != BackendEventType.TRANSCRIPT for e in events) + + +def test_other_events_delegate_to_parent(): + # AUDIO_OUTPUT still normalized by the inherited handler. + (be,) = _map(_session(), SimpleNamespace(type="response.output_audio.delta", delta="AQIDBA==")) + assert be.event_type == BackendEventType.AUDIO_OUTPUT diff --git a/tests/unit/test_openai_realtime_backend.py b/tests/unit/test_openai_realtime_backend.py new file mode 100644 index 00000000..389c1e3b --- /dev/null +++ b/tests/unit/test_openai_realtime_backend.py @@ -0,0 +1,296 @@ +"""Unit tests for the OpenAI Realtime ``Backend`` (no network). + +Covers the pure surfaces: ``session.update`` assembly + tool translation, the +stateful provider-event -> normalized ``BackendEvent`` mapping (including +final-transcript selection and interruption), capability flags, sample-rate +exposure, factory dispatch, and ``send()`` validation. The live session +(open/receive over a real connection) is out of scope here. +""" + +from __future__ import annotations + +from types import SimpleNamespace + +import pytest + +from eva.backend.base import BackendEventType, ToolCallResult +from eva.backend.factory import BackendFactory +from eva.backend.openai_realtime import OpenAIRealtimeBackend, OpenAIRealtimeSession + + +def _backend(**overrides) -> OpenAIRealtimeBackend: + config = {"model": "gpt-realtime", "api_key": "test-key", "input_format": "pcm", **overrides} + return OpenAIRealtimeBackend(config=config) + + +def _session() -> OpenAIRealtimeSession: + return OpenAIRealtimeSession(client=None, conn_cm=None, conn=None) # type: ignore[arg-type] + + +def _map(session, event): + return OpenAIRealtimeBackend._map_event(session, event) + + +def test_requires_api_key(monkeypatch): + monkeypatch.delenv("OPENAI_API_KEY", raising=False) + with pytest.raises(ValueError): + OpenAIRealtimeBackend(config={"model": "gpt-realtime"}) + + +def test_api_key_falls_back_to_env(monkeypatch): + monkeypatch.setenv("OPENAI_API_KEY", "env-key") + # No api_key in config -> backend picks it up from the environment (no raise). + backend = OpenAIRealtimeBackend(config={"model": "gpt-realtime"}) + assert backend.output_sample_rate == 24000 + + +def test_accent_is_rejected(): + with pytest.raises(ValueError): + OpenAIRealtimeBackend(config={"model": "gpt-realtime", "api_key": "k", "accent": "british"}) + + +def test_capabilities(): + caps = _backend().capabilities + assert caps.emits_continuous_audio is True + assert caps.supports_streaming_interruption is True + assert caps.owns_playout_clock is False + + +def test_sample_rates_from_input_format(): + b = _backend() # pcm + assert b.output_sample_rate == 24000 + assert b.input_sample_rate == 24000 + b2 = _backend(input_format="pcmu") # telephony/caller input + assert b2.output_sample_rate == 24000 + assert b2.input_sample_rate == 8000 + + +def test_assemble_assistant_session_defaults(): + # pcm input, auto turn-taking: no manual create_response fields; whisper, no language. + sc = _backend(voice="marin", vad_settings={})._session_config + assert sc["output_modalities"] == ["audio"] + assert sc["audio"]["output"] == {"voice": "marin", "format": {"type": "audio/pcm", "rate": 24000}} + assert sc["audio"]["input"]["format"] == {"type": "audio/pcm", "rate": 24000} + assert sc["audio"]["input"]["turn_detection"] == { + "type": "server_vad", + "threshold": 0.5, + "prefix_padding_ms": 300, + "silence_duration_ms": 200, + } + assert sc["audio"]["input"]["transcription"] == {"model": "whisper-1"} + + +def test_assemble_caller_session_manual_turn_taking(): + sc = _backend( + input_format="pcmu", + voice="ballad", + vad_settings={"threshold": 0.5, "prefix_padding_ms": 300, "silence_duration_ms": 500}, + manual_turn_taking=True, + transcription_language="en", + parallel_tool_calls=False, + )._session_config + assert sc["audio"]["input"]["format"] == {"type": "audio/pcmu"} + assert sc["audio"]["input"]["turn_detection"] == { + "type": "server_vad", + "threshold": 0.5, + "prefix_padding_ms": 300, + "silence_duration_ms": 500, + "create_response": False, + "interrupt_response": False, + "idle_timeout_ms": 15000, + } + assert sc["audio"]["input"]["transcription"] == {"model": "whisper-1", "language": "en"} + assert sc["parallel_tool_calls"] is False + + +def test_build_session_update_stamps_owned_fields_and_translates_tools(): + b = _backend() + tools = [{"name": "get_reservation", "description": "look up", "parameters": {"type": "object", "properties": {}}}] + session = b._build_session_update("SYSTEM PROMPT", tools) + + assert session["type"] == "realtime" + assert session["instructions"] == "SYSTEM PROMPT" + assert session["output_modalities"] == ["audio"] + # Generic tool spec -> OpenAI schema (type: function added). + assert session["tools"] == [ + { + "type": "function", + "name": "get_reservation", + "description": "look up", + "parameters": {"type": "object", "properties": {}}, + } + ] + + +def test_build_session_update_none_tools_becomes_empty_list(): + assert _backend()._build_session_update("p", None)["tools"] == [] + + +def test_map_audio_delta(): + (be,) = _map(_session(), SimpleNamespace(type="response.output_audio.delta", delta="AQIDBA==")) + assert be.event_type == BackendEventType.AUDIO_OUTPUT + assert be.audio == b"\x01\x02\x03\x04" + + +def test_map_empty_audio_delta_dropped(): + assert _map(_session(), SimpleNamespace(type="response.output_audio.delta", delta="")) == [] + + +def test_map_output_transcript_delta_accumulates_silently_then_done_emits(): + s = _session() + assert _map(s, SimpleNamespace(type="response.output_audio_transcript.delta", delta="hel")) == [] + assert _map(s, SimpleNamespace(type="response.output_audio_transcript.delta", delta="lo")) == [] + (be,) = _map(s, SimpleNamespace(type="response.output_audio_transcript.done", transcript="hello")) + assert be.event_type == BackendEventType.TRANSCRIPT + assert be.transcript == "hello" + assert be.metadata == {"stream": "output", "final": True} + + +def test_map_input_transcription_completed_and_failed(): + (done,) = _map( + _session(), SimpleNamespace(type="conversation.item.input_audio_transcription.completed", transcript="hi") + ) + assert done.event_type == BackendEventType.TRANSCRIPT + assert done.transcript == "hi" and done.metadata == {"stream": "input", "final": True} + + # Empty completed transcription is dropped. + assert ( + _map(_session(), SimpleNamespace(type="conversation.item.input_audio_transcription.completed", transcript="")) + == [] + ) + + (failed,) = _map(_session(), SimpleNamespace(type="conversation.item.input_audio_transcription.failed", error="x")) + assert failed.event_type == BackendEventType.TRANSCRIPT + assert failed.metadata == {"stream": "input", "failed": True} + + +def test_map_speech_boundaries_no_interruption(): + (be,) = _map(_session(), SimpleNamespace(type="input_audio_buffer.speech_started")) + assert be.event_type == BackendEventType.INPUT_SPEECH_STARTED + (be2,) = _map(_session(), SimpleNamespace(type="input_audio_buffer.speech_stopped")) + assert be2.event_type == BackendEventType.INPUT_SPEECH_STOPPED + + +def test_map_interruption_flushes_partial_turn_then_speech_started(): + s = _session() + _map(s, SimpleNamespace(type="response.created")) + _map(s, SimpleNamespace(type="response.output_audio_transcript.delta", delta="I was sa")) + events = _map(s, SimpleNamespace(type="input_audio_buffer.speech_started")) + # Interrupted TURN_END (with the partial) precedes the speech-started signal. + assert [e.event_type for e in events] == [BackendEventType.TURN_END, BackendEventType.INPUT_SPEECH_STARTED] + turn_end = events[0] + assert turn_end.transcript == "I was sa" + assert turn_end.metadata["interrupted"] is True and turn_end.metadata["cancelled"] is False + # State reset: a second speech_started does not re-flush. + assert [e.event_type for e in _map(s, SimpleNamespace(type="input_audio_buffer.speech_started"))] == [ + BackendEventType.INPUT_SPEECH_STARTED + ] + + +def test_map_function_call(): + s = _session() + (be,) = _map( + s, + SimpleNamespace( + type="response.function_call_arguments.done", + call_id="c1", + name="get_reservation", + arguments='{"n": "ABC"}', + ), + ) + assert be.event_type == BackendEventType.TOOL_CALL_REQUEST + assert be.tool_call_request.call_id == "c1" + assert be.tool_call_request.arguments == {"n": "ABC"} + assert s.has_function_calls is True + + +def test_map_function_call_bad_arguments_becomes_empty(): + (be,) = _map( + _session(), + SimpleNamespace(type="response.function_call_arguments.done", call_id="c", name="f", arguments="nope"), + ) + assert be.tool_call_request.arguments == {} + + +def test_map_output_audio_done(): + (be,) = _map(_session(), SimpleNamespace(type="response.output_audio.done")) + assert be.event_type == BackendEventType.OUTPUT_AUDIO_DONE + + +def test_map_turn_started(): + (be,) = _map(_session(), SimpleNamespace(type="response.created")) + assert be.event_type == BackendEventType.OUTPUT_TURN_STARTED + + +def test_map_response_done_selects_final_transcript_and_usage(): + s = _session() + _map(s, SimpleNamespace(type="response.created")) + _map(s, SimpleNamespace(type="response.output_audio_transcript.done", transcript="all done")) + usage = SimpleNamespace(input_tokens=11, output_tokens=7) + response = SimpleNamespace(status="completed", usage=usage, output=[]) + (be,) = _map(s, SimpleNamespace(type="response.done", response=response)) + assert be.event_type == BackendEventType.TURN_END + assert be.transcript == "all done" + assert be.metadata["cancelled"] is False and be.metadata["interrupted"] is False + assert be.metadata["usage"] == {"prompt_tokens": 11, "completion_tokens": 7} + + +def test_map_response_done_cancelled(): + response = SimpleNamespace(status="cancelled", usage=None, output=[]) + (be,) = _map(_session(), SimpleNamespace(type="response.done", response=response)) + assert be.event_type == BackendEventType.TURN_END + assert be.metadata["cancelled"] is True and be.metadata["usage"] is None + + +def test_map_response_done_has_function_calls_from_output_items(): + response = SimpleNamespace(status="completed", usage=None, output=[SimpleNamespace(type="function_call")]) + (be,) = _map(_session(), SimpleNamespace(type="response.done", response=response)) + assert be.metadata["has_function_calls"] is True + + +def test_map_error_carries_code(): + (be,) = _map( + _session(), SimpleNamespace(type="error", error=SimpleNamespace(code="rate_limit", message="slow down")) + ) + assert be.event_type == BackendEventType.ERROR + assert be.metadata["code"] == "rate_limit" + + +def test_map_unhandled_event_dropped(): + assert _map(_session(), SimpleNamespace(type="session.updated")) == [] + assert _map(_session(), SimpleNamespace(type="conversation.item.created")) == [] + + +@pytest.mark.asyncio +async def test_send_requires_exactly_one_arg(): + b, s = _backend(), _session() + with pytest.raises(ValueError): + await b.send(s) + with pytest.raises(ValueError): + await b.send(s, audio=b"x", text="y") + + +@pytest.mark.asyncio +async def test_send_wrong_session_type_raises(): + with pytest.raises(TypeError): + await _backend().send(object(), tool_result=ToolCallResult(call_id="c", result={})) # type: ignore[arg-type] + + +@pytest.mark.asyncio +async def test_close_is_idempotent_and_network_free(): + b, s = _backend(), _session() + await b.close(s) + await b.close(s) + + +def test_factory_dispatch_builds_backend_from_flat_config(): + backend = BackendFactory().create( + "openai_realtime", {"model": "gpt-realtime", "api_key": "k", "input_format": "pcmu"} + ) + assert isinstance(backend, OpenAIRealtimeBackend) + assert backend.output_sample_rate == 24000 + assert backend.input_sample_rate == 8000 + + +def test_factory_unknown_provider_returns_none(): + assert BackendFactory().create("does_not_exist", {}) is None diff --git a/tests/unit/user_simulator/test_factory.py b/tests/unit/user_simulator/test_factory.py index d47f1abe..39ee2265 100644 --- a/tests/unit/user_simulator/test_factory.py +++ b/tests/unit/user_simulator/test_factory.py @@ -1,6 +1,6 @@ from pathlib import Path -from eva.models.config import ElevenLabsSimulatorConfig, OpenAIRealtimeSimulatorConfig +from eva.models.config import ElevenLabsSimulatorConfig, S2SSimulatorConfig from eva.user_simulator.elevenlabs import ElevenLabsUserSimulator from eva.user_simulator.factory import create_user_simulator from eva.user_simulator.openai_realtime import OpenAIRealtimeUserSimulator @@ -38,7 +38,7 @@ def test_factory_keeps_elevenlabs_as_default(tmp_path): def test_factory_selects_openai_realtime(tmp_path): - config = OpenAIRealtimeSimulatorConfig() + config = S2SSimulatorConfig() simulator = create_user_simulator(config, **_kwargs(tmp_path)) assert isinstance(simulator, OpenAIRealtimeUserSimulator) diff --git a/tests/unit/user_simulator/test_openai_realtime.py b/tests/unit/user_simulator/test_openai_realtime.py index d4af67c8..181ad037 100644 --- a/tests/unit/user_simulator/test_openai_realtime.py +++ b/tests/unit/user_simulator/test_openai_realtime.py @@ -8,7 +8,7 @@ import pytest -from eva.models.config import OpenAIRealtimeSimulatorConfig, PerturbationConfig +from eva.models.config import PerturbationConfig, S2SSimulatorConfig from eva.user_simulator.openai_realtime import OpenAIRealtimeUserSimulator @@ -33,7 +33,7 @@ def _simulator(tmp_path: Path, *, persona_id: int = 1, **config_overrides) -> Op server_url="ws://localhost:9999/ws", output_dir=tmp_path, agent_id="agent_itsm", - simulator_config=OpenAIRealtimeSimulatorConfig(**config_overrides), + simulator_config=S2SSimulatorConfig(**config_overrides), ) @@ -70,7 +70,7 @@ def test_elevenlabs_specific_accent_variants_are_rejected(tmp_path): output_dir=tmp_path, agent_id="agent_itsm", perturbation_config=PerturbationConfig(accent="french"), - simulator_config=OpenAIRealtimeSimulatorConfig(), + simulator_config=S2SSimulatorConfig(), ) @@ -83,7 +83,7 @@ def test_behavior_variant_uses_shared_prompt(tmp_path): output_dir=tmp_path, agent_id="agent_itsm", perturbation_config=PerturbationConfig(behavior="aggressive_impatient"), - simulator_config=OpenAIRealtimeSimulatorConfig(), + simulator_config=S2SSimulatorConfig(), ) prompt = simulator._build_prompt()