Add cascade STT/LLM/TTS user simulator

.env.example

@@ -199,6 +199,15 @@ EVA_MODEL_LIST='[
       "aws_secret_access_key": "os.environ/AWS_SECRET_ACCESS_KEY",
       "max_parallel_requests": 5
     }
+  },
+  {
+    "model_name": "user-llm",
+    "litellm_params": {
+      "model": "openai/gpt-5.5",
+      "api_key": "os.environ/OPENAI_API_KEY",
+      "max_parallel_requests": 5,
+      "reasoning_effort": "low"
+    }
   }
 ]'
 
@@ -304,7 +313,7 @@ EVA_MODEL_LIST='[
 # --- User simulator provider ---
 #i Caller provider. ElevenLabs is the backward-compatible default.
 #d enum
-#e elevenlabs,openai_realtime
+#e elevenlabs,openai_realtime,cascade
 EVA_USER_SIMULATOR__PROVIDER=elevenlabs
 
 #i OpenAI Realtime caller model. Used only when provider=openai_realtime.

README.md

@@ -99,7 +99,7 @@ EVA_DEBUG=false                       # Run only 1 record for testing when enabl
 EVA_RECORD_IDS=1.2.1,1.2.2            # Run specific records only (remove to run all records)
 
 # User Simulator Configuration
-EVA_USER_SIMULATOR__PROVIDER=elevenlabs      # elevenlabs | openai_realtime
+EVA_USER_SIMULATOR__PROVIDER=elevenlabs      # elevenlabs | openai_realtime | cascade
 EVA_USER_SIMULATOR__MODEL=gpt-realtime-1.5   # Used by openai_realtime
 EVA_USER_SIMULATOR__FEMALE_VOICE=marin       # Used by openai_realtime
 EVA_USER_SIMULATOR__MALE_VOICE=cedar         # Used by openai_realtime
@@ -125,6 +125,8 @@ See `.env.example` for the complete list of configuration options.
 
 The OpenAI Realtime caller supports behavior, background-noise, and connection-degradation perturbations. Accent variants currently require the ElevenLabs caller because they select dedicated ElevenLabs agents. Both caller providers write `user_simulator_events.jsonl`.
 
+The [cascade user simulator](docs/cascade_user_simulator.md) runs a separate STT/LLM/TTS caller with ordinary turn-taking. It supports both real-time streaming and tick-driven OpenAI Realtime evaluation.
+
 #### Known limitations of the OpenAI Realtime caller (beta)
 
 Before relying on it for large-scale evaluation, the following should be kept in mind:

docs/README.md

@@ -3,6 +3,7 @@
 | Document | Description |
 |---|---|
 | [Assistant Server Contract](assistant_server_contract.md) | How to implement a new framework integration (s2s model or custom pipeline) |
+| [Cascade User Simulator](cascade_user_simulator.md) | Caller setup, transports, perturbations, and output artifacts |
 | [Metrics](metrics/README.md) | Metric definitions, scoring rubrics, and judge prompts |
 | [MetricContext](metric_context.md) | Data structures passed to metrics during evaluation |
 | [LLM Configuration](llm_configuration.md) | LLM provider setup and LiteLLM routing |

docs/assistant_server_contract.md

@@ -141,8 +141,14 @@ msg = create_twilio_media_message(stream_sid, mulaw_chunk)
 await websocket.send_text(msg)
 ```
 
-Audio output must be 8 kHz mulaw. Send it in 160-byte chunks (20 ms each) at
-real-time pace. Both `openai_realtime_server.py` and `gemini_live_server.py` use a
+Audio output must be 8 kHz mulaw in 160-byte chunks (20 ms each). Keep
+`paced_output=True` by default. The cascade caller uses `paced_output=False` only
+with OpenAI Realtime, whose server declares `supports_unpaced_output=True`; its
+adapter buffers incoming audio and releases one tick at a time. Other servers
+must accept and forward `paced_output` to the base constructor, which rejects
+unpaced output unless the server explicitly supports it.
+
+For paced output, send chunks at real-time pace. Both `openai_realtime_server.py` and `gemini_live_server.py` use a
 dedicated `_pace_audio_output` asyncio task to drain a queue at this rate — copy that
 pattern. If audio is sent too fast or too slow, the user simulator's may incorrectly detect
 turn boundaries.

docs/cascade_user_simulator.md

@@ -0,0 +1,59 @@
+# Cascade user simulator
+
+The cascade caller combines streaming speech recognition, a caller LLM, and speech
+synthesis. It waits for the assistant to greet, then takes ordinary conversational
+turns using the shared persona, goal, and `end_call` rules.
+
+## Setup
+
+Set `ELEVENLABS_API_KEY` for the default streaming recognizer and `CARTESIA_API_KEY`
+for caller speech. Configure a `user-llm` deployment in `EVA_MODEL_LIST`; the
+deployment controls the caller model and its provider credentials independently
+of the assistant being evaluated. See [LLM configuration](llm_configuration.md).
+
+```dotenv
+EVA_USER_SIMULATOR__PROVIDER=cascade
+EVA_USER_SIMULATOR__STT=elevenlabs
+EVA_USER_SIMULATOR__STT_PARAMS={"model":"scribe_v2_realtime"}
+EVA_USER_SIMULATOR__LLM=user-llm
+EVA_USER_SIMULATOR__TTS=cartesia
+EVA_USER_SIMULATOR__TTS_PARAMS={"model":"sonic-3.5"}
+```
+
+`stt_params` supplies keyword arguments to the selected LiveKit STT plugin; the
+ElevenLabs plugin is included. Other STT providers require their corresponding
+LiveKit plugin. Caller TTS currently uses Cartesia and accepts `model`, `api_key`,
+`female_voice`, and `male_voice` in `tts_params`.
+
+Run a single debug record with your existing assistant configuration:
+
+```bash
+uv run eva --debug --user-simulator.provider cascade
+```
+
+The caller supports the existing behavior-prompt, background-noise, and
+connection-degradation perturbations. Accent perturbations require the ElevenLabs
+Agents caller.
+
+## Transport and turn-taking
+
+Each tick exchanges 200 ms of audio. The caller waits for assistant silence and
+for a reply to its previous turn before generating another utterance. It waits
+briefly for finalized transcripts, then falls back to available partial text.
+
+OpenAI Realtime uses a tick-driven adapter: the server sends unpaced audio and the
+caller releases one tick at a time. Quiet ticks still wait briefly for incoming
+audio and send silence so provider turn detection can finish the caller's turn.
+Other frameworks use the real-time adapter and retain output pacing.
+
+## Artifacts and endings
+
+- `user_simulator_events.jsonl` records speech, audio boundaries, and the terminal
+  connection state.
+- `audio_user_clean.wav` stores synthesized caller audio before perturbation.
+- `user_simulator_decisions.jsonl` records each tick's speech state and audio level.
+
+The caller ends with `goodbye` when its LLM calls `end_call`, or `timeout` when the
+conversation budget expires. Sustained assistant silence ends with
+`inactivity_timeout`. A stalled tick-driven provider ends with the distinct
+`provider_stalled` reason and preserves partial artifacts for diagnosis.

pyproject.toml

@@ -35,6 +35,8 @@ dependencies = [
     "azure-cognitiveservices-speech>=1.31.0",
     "cartesia>=1.0.0",
     "assemblyai>=0.17.0",
+    "livekit-agents>=1.6.8",
+    "livekit-plugins-elevenlabs>=1.6.8",
     "setuptools>=65.0.0",
     "fastapi>=0.100.0",
     "uvicorn>=0.23.0",

src/eva/__init__.py

@@ -7,7 +7,7 @@
 
 # Bump simulation_version when changes affect benchmark outputs (agent code,
 # user simulator, orchestrator, simulation prompts, agent configs, tool mocks).
-simulation_version = "2.0.3"
+simulation_version = "2.0.34"
 
 # Bump metrics_version when changes affect metric computation (metrics code,
 # judge prompts, pricing tables, postprocessor).

src/eva/assistant/base_server.py

@@ -41,6 +41,15 @@ class AbstractAssistantServer(ABC):
     5. Populate the AuditLog with conversation events
     """
 
+    supports_unpaced_output: bool = False
+    """Whether this server can honor ``paced_output=False``.
+
+    Only a server whose outbound relay throttle we own can drop it. A server whose
+    pacing lives inside a third-party runtime cannot, and must reject the request
+    rather than accept it and keep pacing, which would leave a tick-driven caller
+    believing the assistant was unpaced.
+    """
+
     def __init__(
         self,
         current_date_time: str,
@@ -52,6 +61,7 @@ def __init__(
         port: int,
         conversation_id: str,
         language: str = "en",
+        paced_output: bool = True,
     ):
         """Initialize the assistant server.
 
@@ -65,10 +75,20 @@ def __init__(
             port: Port to listen on
             conversation_id: Unique ID for this conversation
             language: BCP 47 language tag for STT/TTS/S2S services (e.g. 'en', 'fr', 'es-MX')
+            paced_output: Whether to emit audio at real-time cadence. True for callers
+                that infer turn boundaries from silence timing (the ElevenLabs simulator).
+                False for a tick-driven caller, which buffers whatever arrives and
+                releases it one tick at a time, so pacing here would only add latency.
         """
         self.current_date_time = current_date_time
         self.pipeline_config = pipeline_config
         self.language = language
+        if not paced_output and not self.supports_unpaced_output:
+            raise ValueError(
+                f"{type(self).__name__} cannot honor paced_output=False: its output pacing is not ours to remove. "
+                "Tick-driving this framework requires leaving pacing on."
+            )
+        self.paced_output = paced_output
         self.initial_message = get_initial_message(language)
         self.agent: AgentConfig = agent
         self.agent_config_path = agent_config_path

src/eva/assistant/elevenlabs_server.py

@@ -134,6 +134,7 @@ def __init__(
         port: int,
         conversation_id: str,
         language: str = "en",
+        paced_output: bool = True,
     ):
         super().__init__(
             current_date_time=current_date_time,
@@ -145,6 +146,7 @@ def __init__(
             port=port,
             conversation_id=conversation_id,
             language=language,
+            paced_output=paced_output,
         )
 
         # Recording sample rate (ElevenLabs operates at 16 kHz)

src/eva/assistant/gemini_live_server.py

@@ -166,6 +166,7 @@ def __init__(
         port: int,
         conversation_id: str,
         language: str = "en",
+        paced_output: bool = True,
     ):
         super().__init__(
             current_date_time=current_date_time,
@@ -177,6 +178,7 @@ def __init__(
             port=port,
             conversation_id=conversation_id,
             language=language,
+            paced_output=paced_output,
         )
 
         # Recording sample rate (Gemini outputs 24 kHz)

src/eva/assistant/openai_realtime_server.py

@@ -38,12 +38,11 @@
 MULAW_CHUNK_SIZE = 160  # bytes per chunk (20ms at 8kHz, 1 byte per sample)
 MULAW_CHUNK_DURATION_S = 0.02  # 20ms per chunk
 # 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
-# event-loop jitter, and padding then injects silence into an active user
-# utterance (the choppy-audio bug). This guard only ever *skips* a pad, so it
-# can never add silence or worsen alignment — during genuine user silence the
-# timestamp is old and padding proceeds normally.
-USER_ACTIVE_GUARD_S = 0.3
+# arrived recently. The speaking-state flag can go stale under event-loop
+# jitter, and padding then injects silence into an active user utterance (the
+# choppy-audio bug). This guard only ever *skips* a pad, so it can never add
+# silence or worsen alignment — during genuine user silence the counter has run
+# past the threshold and padding proceeds normally.
 
 
 def _wall_ms() -> str:
@@ -84,13 +83,22 @@ class OpenAIRealtimeAssistantServer(AbstractAssistantServer):
     _service_name: str = "OpenAI Realtime"
     _metrics_processor_name: str = "openai_realtime"
 
+    USER_ACTIVE_GUARD_DELTAS = 25
+    """Assistant audio deltas after the last user audio before the user counts as idle.
+
+    Counted in deltas rather than wall-clock seconds so a tick-driven caller that
+    pauses to think is not mistaken for a caller who stopped talking.
+    """
+
+    supports_unpaced_output = True
+
     def __init__(self, **kwargs: Any) -> None:
         super().__init__(**kwargs)
 
         self._audio_sample_rate = OPENAI_SAMPLE_RATE
-        # Monotonic time of the last real user-audio frame appended; used to
-        # guard against padding the user track mid-utterance (see USER_ACTIVE_GUARD_S).
-        self._last_user_audio_mono: float = 0.0
+        # Assistant audio deltas seen since the last real user-audio frame; used to
+        # guard against padding the user track mid-utterance (see USER_ACTIVE_GUARD_DELTAS).
+        self._deltas_since_user_audio: int = self.USER_ACTIVE_GUARD_DELTAS
 
         self._system_prompt: str = self._build_system_prompt()
 
@@ -308,6 +316,20 @@ async def _handle_session(self, websocket: WebSocket) -> None:
         finally:
             logger.info(f"Client disconnected from {self._service_name} server")
 
+    # ── User activity tracking (tick-safe, no wall clock) ────────────
+
+    def note_user_audio(self) -> None:
+        """Record that user audio just arrived."""
+        self._deltas_since_user_audio = 0
+
+    def note_assistant_delta(self) -> None:
+        """Record one assistant audio delta since the last user audio."""
+        self._deltas_since_user_audio += 1
+
+    def user_recently_active(self) -> bool:
+        """Whether user audio arrived recently enough to skip padding its track."""
+        return self._deltas_since_user_audio < self.USER_ACTIVE_GUARD_DELTAS
+
     # ── Audio output pacer (OpenAI -> Twilio WS at real-time rate) ───
 
     async def _pace_audio_output(self, websocket: WebSocket, audio_output_queue: asyncio.Queue[bytes]) -> None:
@@ -331,6 +353,12 @@ async def _pace_audio_output(self, websocket: WebSocket, audio_output_queue: asy
                     logger.error(f"Error sending audio to Twilio WS: {e}")
                     return
 
+                if not self.paced_output:
+                    # Tick-driven caller: it buffers what arrives and releases one tick
+                    # per tick, so pacing here would only add latency without changing
+                    # what the caller hears.
+                    continue
+
                 now = time.monotonic()
                 if next_send_time <= now:
                     next_send_time = now
@@ -385,7 +413,7 @@ async def _forward_user_audio(self, websocket: WebSocket, conn: Any) -> None:
                     sync_buffer_to_position(self.assistant_audio_buffer, sync_target)
                     synced = len(self.assistant_audio_buffer) - asst_before
                 self.user_audio_buffer.extend(pcm16_24k)
-                self._last_user_audio_mono = time.monotonic()
+                self.note_user_audio()
                 self._user_frame_count += 1
                 if self._user_frame_count % 50 == 0:
                     diff = len(self.user_audio_buffer) - len(self.assistant_audio_buffer)
@@ -572,7 +600,6 @@ async def _on_audio_delta(self, event: Any, audio_output_queue: asyncio.Queue[by
             return
 
         pcm16_bytes = base64.b64decode(delta_b64)
-
         if self._assistant_state.first_audio_wall_ms is None:
             self._assistant_state.first_audio_wall_ms = _wall_ms()
             self._assistant_state.responding = True
@@ -587,12 +614,13 @@ async def _on_audio_delta(self, event: Any, audio_output_queue: asyncio.Queue[by
                 if 0 < latency_ms < 30_000:
                     self._metrics_log.write_latency("model_response", latency_ms / 1000, self._model)
 
+        self.note_assistant_delta()
+
         user_before = len(self.user_audio_buffer)
         synced = 0
         # Skip the pad if the user track is actively receiving audio (flag may be
         # stale under jitter) — padding then would inject a mid-utterance chop.
-        user_recently_active = (time.monotonic() - self._last_user_audio_mono) <= USER_ACTIVE_GUARD_S
-        if not self._user_speaking and not user_recently_active:
+        if not self._user_speaking and not self.user_recently_active():
             sync_buffer_to_position(self.user_audio_buffer, len(self.assistant_audio_buffer))
             synced = len(self.user_audio_buffer) - user_before
         self.assistant_audio_buffer.extend(pcm16_bytes)

src/eva/assistant/pipecat_server.py

@@ -99,6 +99,7 @@ def __init__(
         port: int,
         conversation_id: str,
         language: str = "en",
+        paced_output: bool = True,
         turn_end_fallback_time: int | None = None,
     ):
         """Initialize the assistant server.
@@ -113,6 +114,8 @@ def __init__(
             port: Port to listen on
             conversation_id: Unique ID for this conversation
             language: BCP 47 language tag for STT/TTS services (e.g. 'en', 'fr', 'es-MX')
+            paced_output: Accepted for interface parity; must be True, since Pipecat's output
+                pacing lives in its own transport and is not ours to remove.
             turn_end_fallback_time: Seconds of user-turn silence after the assistant stops
                 speaking before nudging it to retry. ``None`` disables the fallback.
         """
@@ -126,6 +129,7 @@ def __init__(
             port=port,
             conversation_id=conversation_id,
             language=language,
+            paced_output=paced_output,
         )
 
         self.agentic_system = None  # Will be set in _handle_session

src/eva/assistant/smallest_hydra_server.py

@@ -129,6 +129,7 @@ def __init__(
         port: int,
         conversation_id: str,
         language: str = "en",
+        paced_output: bool = True,
     ):
         super().__init__(
             current_date_time=current_date_time,
@@ -140,6 +141,7 @@ def __init__(
             port=port,
             conversation_id=conversation_id,
             language=language,
+            paced_output=paced_output,
         )
 
         self._audio_sample_rate = _RECORDING_SAMPLE_RATE