From cf40f51f9655a296b35cff71a0aad106a85e8bd4 Mon Sep 17 00:00:00 2001 From: "joseph.marinier" Date: Mon, 15 Jun 2026 10:08:42 -0400 Subject: [PATCH 1/4] Fix strict LLM check See 6886ad60 for more details. --- src/eva/run_benchmark.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/eva/run_benchmark.py b/src/eva/run_benchmark.py index 0fb6df9a..1ea66887 100644 --- a/src/eva/run_benchmark.py +++ b/src/eva/run_benchmark.py @@ -45,8 +45,8 @@ async def run_benchmark(config: RunConfig) -> int: return 1 # Apply env-dependent values (secrets, urls) from live env onto saved config. - # Skip strict LLM check when force-rerunning metrics only — the LLM is not needed. - runner.config.apply_env_overrides(config, strict_llm=not config.force_rerun_metrics) + # Skip strict LLM check when no conversations can run (max_rerun_attempts == 0). + runner.config.apply_env_overrides(config, strict_llm=config.max_rerun_attempts != 0) # Apply CLI overrides runner.config.max_rerun_attempts = config.max_rerun_attempts From d7df899cf02a9a25f0157e0e35cd39d22e849d2f Mon Sep 17 00:00:00 2001 From: "joseph.marinier" Date: Mon, 15 Jun 2026 12:14:13 -0400 Subject: [PATCH 2/4] Suppress `usage:` block in `eva --help` as it is completely redundant with the options bellow. --- src/eva/cli.py | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/src/eva/cli.py b/src/eva/cli.py index 518fe0e5..ef641ecc 100644 --- a/src/eva/cli.py +++ b/src/eva/cli.py @@ -4,6 +4,7 @@ Used by both the `eva` console script (installed via pip/uv) and `python main.py`. """ +import argparse import asyncio import faulthandler import sys @@ -13,14 +14,25 @@ faulthandler.enable() # Print Python stack trace on segfaults (exit 139) +class _NoUsageHelpFormatter(argparse.RawDescriptionHelpFormatter): + """Suppress the long, auto-generated `usage:` block in `--help` output.""" + + def _format_usage(self, *args, **kwargs): + return "" + + def main(): """Entry point for the `eva` console script.""" # Import config first (lightweight) for fast --help and validation errors. # Heavy deps (pipecat, litellm, etc.) are imported only in run_benchmark. + from pydantic_settings import CliSettingsSource + from eva.models.config import RunConfig + cli_source = CliSettingsSource(RunConfig, cli_parse_args=True, formatter_class=_NoUsageHelpFormatter) + try: - config = RunConfig(_cli_parse_args=True, _env_file=".env") + config = RunConfig(_cli_settings_source=cli_source, _env_file=".env") except ValidationError as e: print(e, file=sys.stderr) sys.exit(1) From 0e732b62e513e46adfa11e175b74025b26aad914 Mon Sep 17 00:00:00 2001 From: "joseph.marinier" Date: Wed, 29 Jul 2026 11:28:21 -0400 Subject: [PATCH 3/4] Migrate deprecated pipecat runner to worker --- src/eva/__init__.py | 2 +- src/eva/assistant/pipecat_server.py | 15 ++++++++------- 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/src/eva/__init__.py b/src/eva/__init__.py index 23260438..1995823a 100644 --- a/src/eva/__init__.py +++ b/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.1" +simulation_version = "2.0.2" # Bump metrics_version when changes affect metric computation (metrics code, # judge prompts, pricing tables, postprocessor). diff --git a/src/eva/assistant/pipecat_server.py b/src/eva/assistant/pipecat_server.py index dc8fa542..695db83b 100644 --- a/src/eva/assistant/pipecat_server.py +++ b/src/eva/assistant/pipecat_server.py @@ -19,8 +19,7 @@ from pipecat.observers.loggers.metrics_log_observer import MetricsLogObserver from pipecat.pipeline.parallel_pipeline import ParallelPipeline from pipecat.pipeline.pipeline import Pipeline -from pipecat.pipeline.runner import PipelineRunner -from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.pipeline.worker import PipelineParams, PipelineWorker from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.llm_response_universal import ( AssistantTurnStoppedMessage, @@ -37,6 +36,7 @@ from pipecat.turns.user_stop import TurnAnalyzerUserTurnStopStrategy from pipecat.turns.user_turn_strategies import UserTurnStrategies from pipecat.utils.time import time_now_iso8601 +from pipecat.workers.runner import WorkerRunner from eva.assistant.agentic.audit_log import convert_to_epoch_ms, current_timestamp_ms from eva.assistant.audio_buffer import ContiguousAudioBufferProcessor @@ -145,8 +145,8 @@ def __init__( self._app = None self._server = None self._server_task = None - self._runner: PipelineRunner | None = None - self._task: PipelineTask | None = None + self._runner: WorkerRunner | None = None + self._task: PipelineWorker | None = None self._running = False self.num_seconds = 0 self._metrics_observer: MetricsFileObserver | None = None @@ -488,7 +488,7 @@ async def on_assistant_response(msg: str) -> None: self._metrics_observer, # Write metrics to JSONL file ] - self._task = PipelineTask( + self._task = PipelineWorker( pipeline, params=PipelineParams( enable_metrics=True, # Enable TTFB and ProcessingMetricsData @@ -512,8 +512,9 @@ async def on_assistant_response(msg: str) -> None: ) # Run the pipeline - self._runner = PipelineRunner(handle_sigint=False, force_gc=True) - await self._runner.run(self._task) + self._runner = WorkerRunner(handle_sigint=False, force_gc=True) + await self._runner.add_workers(self._task) + await self._runner.run() except Exception as e: logger.error(f"Session error: {e}", exc_info=True) From 5620ab0a005ed159c509b342bca9b61431677ac7 Mon Sep 17 00:00:00 2001 From: "joseph.marinier" Date: Tue, 28 Jul 2026 13:33:48 -0400 Subject: [PATCH 4/4] Add preflight MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Probe each configured model before a run starts and abort early on a credential/connectivity failure, instead of discovering it only after every record has exhausted its attempts. - eva/orchestrator/preflight.py: run_preflight probes the run's models by reusing the real client/factory paths (LiteLLMClient for the LLM, the pipecat STT/TTS service factories, the audio-LLM client). It is conservative, so only a definitive error fails a probe. S2S has no live probe yet (logged). - Wired into BenchmarkRunner._run() — the single point both a fresh run and a rerun funnel through — gated on "there's something to run" (pending records + attempts remaining), so it fires exactly once per invocation and only when conversations will actually run. - Added configs `--no-preflight` and `--preflight-timeout-seconds`. --- README.md | 4 + src/eva/models/config.py | 20 ++- src/eva/orchestrator/preflight.py | 195 ++++++++++++++++++++++ src/eva/orchestrator/runner.py | 6 + src/eva/run_benchmark.py | 2 + tests/integration/test_evaluation_mode.py | 20 +++ tests/unit/models/test_config_models.py | 12 ++ tests/unit/orchestrator/test_preflight.py | 168 +++++++++++++++++++ 8 files changed, 420 insertions(+), 7 deletions(-) create mode 100644 src/eva/orchestrator/preflight.py create mode 100644 tests/unit/orchestrator/test_preflight.py diff --git a/README.md b/README.md index 5008b164..ae86d0b6 100644 --- a/README.md +++ b/README.md @@ -136,6 +136,10 @@ Before relying on it for large-scale evaluation, the following should be kept in ### Running EVA +#### Pre-flight model check + +Before a run starts, EVA sends one cheap real request to each configured model (LLM / STT / TTS / audio-LLM) and aborts immediately if any is unreachable or has bad credentials. Pass `--no-preflight` or `EVA_PREFLIGHT=false` to bypass it, and `--preflight-timeout-seconds` or `EVA_PREFLIGHT_TIMEOUT_SECONDS` to tune the per-model timeout (default 20). + #### OpenAI Realtime Caller Smoke Test A smoke test is easier to perform with the OpenAI Realtime caller, as it does not require an ElevenLabs agent ID. diff --git a/src/eva/models/config.py b/src/eva/models/config.py index c2c0b6fc..a83c8f9a 100644 --- a/src/eva/models/config.py +++ b/src/eva/models/config.py @@ -51,7 +51,7 @@ def _get_all_metrics() -> list[str]: return [m for m in get_global_registry().list_metrics() if m not in _VALIDATION_METRIC_NAMES] -def _param_alias(params: dict[str, Any]) -> str: +def get_model_alias_from_params(params: dict[str, Any]) -> str: """Return the display alias from a params dict.""" return params.get("alias") or params["model"] @@ -228,22 +228,22 @@ def pipeline_parts(self) -> dict[str, str]: match self.pipeline_type: case PipelineType.AUDIO_LLM: return { - "audio_llm": _param_alias(self.audio_llm_params), - "tts": _param_alias(self.tts_params), + "audio_llm": get_model_alias_from_params(self.audio_llm_params), + "tts": get_model_alias_from_params(self.tts_params), } case PipelineType.S2S: if self.s2s == "elevenlabs": # hardcoded for now. Models are set on the agent UI return { - "s2s": _param_alias(self.s2s_params) or self.s2s, + "s2s": get_model_alias_from_params(self.s2s_params) or self.s2s, **_fetch_elevenlabs_agent_models(self.s2s_params), } - return {"s2s": _param_alias(self.s2s_params)} + return {"s2s": get_model_alias_from_params(self.s2s_params)} case PipelineType.CASCADE: return { - "stt": _param_alias(self.stt_params), + "stt": get_model_alias_from_params(self.stt_params), "llm": self.llm, - "tts": _param_alias(self.tts_params), + "tts": get_model_alias_from_params(self.tts_params), } @model_validator(mode="before") @@ -669,6 +669,12 @@ class ModelDeployment(DeploymentTypedDict): description="Logging level", ) dry_run: bool = Field(False, description="Validate configuration without running") + preflight: bool = Field( + True, + description="Skip the pre-flight model check. By default, probe each configured model " + "(STT/LLM/TTS/audio-LLM) before the run starts, aborting early on credential/connectivity failures.", + ) + preflight_timeout_seconds: float = Field(20.0, gt=0, description="Per-model timeout for the pre-flight probe") @computed_field @property diff --git a/src/eva/orchestrator/preflight.py b/src/eva/orchestrator/preflight.py new file mode 100644 index 00000000..5f26af58 --- /dev/null +++ b/src/eva/orchestrator/preflight.py @@ -0,0 +1,195 @@ +"""Pre-flight model checks — exercise each configured provider before a run starts. + +A run can otherwise burn through every record and attempt before discovering that, say, +an LLM API key is expired or an STT endpoint is unreachable. ``run_preflight`` catches +those credential/connectivity failures up front by sending the cheapest possible real +request to each model the run will use. + +Design notes: +- Probes reuse the same factories/clients the real run uses (``LiteLLMClient``, + ``create_stt_service``/``create_tts_service``, ``create_audio_llm_client``), so a pass + means the run's own construction path works. +- 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``. +""" + +import asyncio +from collections.abc import Awaitable, Coroutine +from dataclasses import dataclass +from typing import Any + +from pipecat.frames.frames import ( + EndFrame, + ErrorFrame, + Frame, + InputAudioRawFrame, + TTSSpeakFrame, +) +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.worker import PipelineWorker +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.workers.runner import WorkerRunner + +from eva.assistant.pipeline.services import ( + create_audio_llm_client, + create_stt_service, + create_tts_service, +) +from eva.assistant.services.llm import LiteLLMClient +from eva.models.config import PipelineType, RunConfig, get_model_alias_from_params +from eva.utils.logging import get_logger + +logger = get_logger(__name__) + +# 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 +_SILENCE_PCM = b"\x00\x00" * (_SILENCE_SAMPLE_RATE // 3) + + +class PreflightError(RuntimeError): + """Raised when a pre-flight probe fails, to abort before any simulations run.""" + + +@dataclass +class ProbeResult: + """Outcome of probing a single model/provider.""" + + model_type: str # "LLM", "STT", "TTS", "AUDIO_LLM" + model_alias: str + ok: bool + detail: str = "" + + +class _FrameCapture(FrameProcessor): + """Pass-through processor that records every frame it sees.""" + + def __init__(self) -> None: + super().__init__() + self.frames: list[Frame] = [] + + async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: + await super().process_frame(frame, direction) + self.frames.append(frame) + await self.push_frame(frame, direction) + + +async def _drive_pipecat_service(service: FrameProcessor, input_frames: list[Frame]) -> tuple[bool, str]: + """Run a pipecat STT/TTS service through a minimal pipeline and report success. + + The service is wrapped between two capture processors: errors travel upstream and + output (audio/transcripts) travels downstream, so both are observed. The pipeline + auto-sends a ``StartFrame`` (which is when streaming services connect and surface + auth failures as an ``ErrorFrame``); the trailing ``EndFrame`` shuts it down cleanly. + + Returns ``(ok, detail)`` — ``ok=False`` only when an ``ErrorFrame`` is seen. + """ + head, tail = _FrameCapture(), _FrameCapture() + task = PipelineWorker(Pipeline([head, service, tail])) + runner = WorkerRunner(handle_sigint=False, force_gc=True) + + async def feed() -> None: + await task.queue_frames([*input_frames, EndFrame()]) + + try: + await runner.add_workers(task) + await asyncio.gather(runner.run(), feed()) + finally: + if not task.has_finished(): + try: + await task.cancel() + except Exception: + pass + + errors = [f for f in head.frames + tail.frames if isinstance(f, ErrorFrame)] + if errors: + return False, str(errors[0].error) + return True, "" + + +async def _probe_llm(config: RunConfig) -> tuple[bool, str]: + client = LiteLLMClient(model=config.model.llm) + await client.complete([{"role": "user", "content": "ping"}], max_retries=0) + return True, "" + + +async def _probe_audio_llm(config: RunConfig) -> tuple[bool, str]: + client = create_audio_llm_client( + config.model.audio_llm, config.model.audio_llm_params, language=config.language.value + ) + # Send a real (silent) audio turn so the probe exercises the actual ALM path. + message = client.build_audio_user_message(_SILENCE_PCM, source_sample_rate=_SILENCE_SAMPLE_RATE) + await client.complete([message]) + return True, "" + + +async def _probe_stt(config: RunConfig) -> tuple[bool, str]: + service = create_stt_service(config.model.stt, config.model.stt_params, language_code=config.language.value) + if service is None: + return True, "" + audio = InputAudioRawFrame(audio=_SILENCE_PCM, sample_rate=_SILENCE_SAMPLE_RATE, num_channels=1) + return await _drive_pipecat_service(service, [audio]) + + +async def _probe_tts(config: RunConfig) -> tuple[bool, str]: + service = create_tts_service(config.model.tts, config.model.tts_params, language_code=config.language.value) + if service is None: + return True, "" + return await _drive_pipecat_service(service, [TTSSpeakFrame("ok")]) + + +async def _guard( + model_type: str, model_alias: str, probe: Coroutine[Any, Any, tuple[bool, str]], timeout: float +) -> ProbeResult: + """Run one probe under a timeout, converting any failure into a ``ProbeResult``.""" + try: + ok, detail = await asyncio.wait_for(probe, timeout) + return ProbeResult(model_type, model_alias, ok, detail) + except TimeoutError: + return ProbeResult(model_type, model_alias, False, f"timed out after {timeout:.0f}s") + except Exception as e: + msg = str(e).strip() or type(e).__name__ + return ProbeResult(model_type, model_alias, False, msg[:200]) + + +async def _run_preflight(config: RunConfig) -> list[ProbeResult]: + """Probe every model the run will use; return one ``ProbeResult`` per component. + + S2S runs are skipped (no live probe yet) — only a log line is emitted. + """ + timeout = config.preflight_timeout_seconds + model = config.model + probes: list[tuple[str, str, Awaitable[tuple[bool, str]]]] = [] + + match model.pipeline_type: + case PipelineType.CASCADE: + probes.append(("LLM", model.llm, _probe_llm(config))) + probes.append(("STT", get_model_alias_from_params(model.stt_params), _probe_stt(config))) + probes.append(("TTS", get_model_alias_from_params(model.tts_params), _probe_tts(config))) + case PipelineType.AUDIO_LLM: + 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") + 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)) + + +async def run_preflight(config: RunConfig) -> None: + """Probe models and raise ``PreflightError`` if any required component fails. + + No-op when ``config.preflight`` is set or there are no probes to run + (e.g. S2S). Call this immediately before launching simulations. + """ + if not config.preflight: + return + results = await _run_preflight(config) + failed = [r for r in results if not r.ok] + if failed: + detail = "\n".join(f"{r.model_type} ({r.model_alias}): {r.detail}" for r in failed) + raise PreflightError(f"pre-flight check failed:\n{detail}") diff --git a/src/eva/orchestrator/runner.py b/src/eva/orchestrator/runner.py index 6282876e..57921c9f 100644 --- a/src/eva/orchestrator/runner.py +++ b/src/eva/orchestrator/runner.py @@ -18,6 +18,7 @@ from eva.models.record import EvaluationRecord from eva.models.results import ConversationResult, RunResult from eva.orchestrator.port_pool import PortPool +from eva.orchestrator.preflight import run_preflight from eva.orchestrator.validation_runner import ValidationResult, ValidationRunner from eva.orchestrator.worker import ConversationWorker from eva.utils.conversation_checks import check_conversation_finished, find_records_with_llm_generic_error @@ -219,6 +220,11 @@ async def _run( """ max_attempts = self.config.max_rerun_attempts pending_output_ids = list(pending_ids) + + # If there are conversations to run, probe models before running any conversation. + if pending_output_ids and max_attempts > 0: + await run_preflight(self.config) + rerun_history: dict[str, list[dict]] = {} time_limit_attempt_counts: dict[str, int] = {} time_limit_validation_cache: dict[str, dict[int, ValidationResult]] = {} diff --git a/src/eva/run_benchmark.py b/src/eva/run_benchmark.py index 1ea66887..478d0fbc 100644 --- a/src/eva/run_benchmark.py +++ b/src/eva/run_benchmark.py @@ -51,6 +51,8 @@ async def run_benchmark(config: RunConfig) -> int: # Apply CLI overrides runner.config.max_rerun_attempts = config.max_rerun_attempts runner.config.force_rerun_metrics = config.force_rerun_metrics + runner.config.preflight = config.preflight + runner.config.preflight_timeout_seconds = config.preflight_timeout_seconds if config.metrics is not None: runner.config.metrics = config.metrics diff --git a/tests/integration/test_evaluation_mode.py b/tests/integration/test_evaluation_mode.py index 362646e1..12965dcb 100644 --- a/tests/integration/test_evaluation_mode.py +++ b/tests/integration/test_evaluation_mode.py @@ -18,6 +18,7 @@ from eva.models.config import ModelConfig, RunConfig from eva.models.record import EvaluationRecord, GroundTruth from eva.models.results import ConversationResult +from eva.orchestrator.preflight import PreflightError, ProbeResult from eva.orchestrator.runner import BenchmarkRunner from eva.orchestrator.validation_runner import ValidationResult @@ -86,6 +87,7 @@ def eval_config(tmp_path): }, max_concurrent_conversations=2, output_dir=tmp_path / "output", + preflight=False, # These tests mock conversations with fake keys, so let's skip preflight. ) @@ -421,3 +423,21 @@ def completed_fn(record_id, per_record_attempt): # Should reach max attempts assert call_counts.get("fail_record_1") == 3 + + +@pytest.mark.asyncio +async def test_preflight_failure_aborts_run_before_conversations(eval_config, mock_dataset): + """A failing pre-flight check aborts the run before any conversation is launched.""" + eval_config.preflight = True # exercise the real gate (fixture default skips it) + runner = BenchmarkRunner(eval_config) + + failed = [ProbeResult("LLM", "gpt-4", ok=False, detail="401 invalid api key")] + with ( + patch("eva.orchestrator.preflight.run_preflight", new=AsyncMock(return_value=failed)), + patch.object(runner, "_run_conversation") as mock_conversation, + ): + with pytest.raises(PreflightError, match="LLM"): + await runner.run(mock_dataset) + + # Preflight ran (via the _run gate) and aborted before any conversation started. + mock_conversation.assert_not_called() diff --git a/tests/unit/models/test_config_models.py b/tests/unit/models/test_config_models.py index 4d96402d..d797dbcd 100644 --- a/tests/unit/models/test_config_models.py +++ b/tests/unit/models/test_config_models.py @@ -685,6 +685,18 @@ def test_num_trials(self): c = _config(env_vars=_BASE_ENV | {"EVA_NUM_TRIALS": "10"}) assert c.num_trials == 10 + def test_preflight_default(self): + c = _config(env_vars=_BASE_ENV) + assert c.preflight is True + + def test_preflight_disabled_via_env(self): + c = _config(env_vars=_BASE_ENV | {"EVA_PREFLIGHT": "false"}) + assert c.preflight is False + + def test_preflight_timeout_seconds(self): + c = _config(env_vars=_BASE_ENV | {"EVA_PREFLIGHT_TIMEOUT_SECONDS": "5"}) + assert c.preflight_timeout_seconds == 5.0 + def test_validation_thresholds(self): thresholds = {"conversation_valid_end": 0.9, "user_behavioral_fidelity": 0.8} c = _config(env_vars=_BASE_ENV | {"EVA_VALIDATION_THRESHOLDS": json.dumps(thresholds)}) diff --git a/tests/unit/orchestrator/test_preflight.py b/tests/unit/orchestrator/test_preflight.py new file mode 100644 index 00000000..292a52b8 --- /dev/null +++ b/tests/unit/orchestrator/test_preflight.py @@ -0,0 +1,168 @@ +"""Tests for orchestrator/preflight.py — pre-flight model probes.""" + +import asyncio +import json +import os +from collections.abc import AsyncGenerator +from unittest.mock import AsyncMock, patch + +import pytest +from pipecat.frames.frames import ErrorFrame, Frame, TTSAudioRawFrame +from pipecat.services.stt_service import STTService +from pipecat.services.tts_service import TTSService + +from eva.models.config import ModelConfig, RunConfig +from eva.orchestrator import preflight +from eva.orchestrator.preflight import ( + PreflightError, + _run_preflight, + run_preflight, +) + +_MODEL_LIST = [{"model_name": "test-model", "litellm_params": {"model": "test"}}] +# Preserve PATH so pydub's lazy ffmpeg lookup doesn't KeyError under clear=True. +_BASE_ENV = {"EVA_MODEL_LIST": json.dumps(_MODEL_LIST), "PATH": os.environ.get("PATH", "")} + + +class _GoodTTS(TTSService): + async def run_tts(self, text, context_id=None) -> AsyncGenerator[Frame | None, None]: + yield TTSAudioRawFrame(audio=b"\x00\x00", sample_rate=24000, num_channels=1) + + +class _BadTTS(TTSService): + async def run_tts(self, text, context_id=None) -> AsyncGenerator[Frame | None, None]: + await self.push_error(ErrorFrame("401 Unauthorized", fatal=True)) + return + yield # pragma: no cover — makes this an async generator + + +class _GoodSTT(STTService): + """STT that connects fine and emits nothing (silence in → no transcript).""" + + async def run_stt(self, audio) -> AsyncGenerator[Frame | None, None]: + yield None + + +def _cascade_config(tmp_path) -> RunConfig: + with patch.dict(os.environ, _BASE_ENV, clear=True): + return RunConfig( + model=ModelConfig( + llm="test-model", + stt="deepgram", + tts="cartesia", + stt_params={"api_key": "k", "model": "nova-2", "alias": "deepgram-prod"}, + tts_params={"api_key": "k", "model": "sonic"}, + ), + output_dir=tmp_path / "out", + run_id="r", + ) + + +@pytest.mark.asyncio +async def test_cascade_all_pass(tmp_path): + cfg = _cascade_config(tmp_path) + with ( + patch.object(preflight, "create_stt_service", return_value=_GoodSTT()), + patch.object(preflight, "create_tts_service", return_value=_GoodTTS()), + patch.object(preflight, "LiteLLMClient") as llm, + ): + llm.return_value.complete = AsyncMock(return_value=("pong", {})) + results = await _run_preflight(cfg) + + assert {r.model_type for r in results} == {"LLM", "STT", "TTS"} + assert all(r.ok for r in results) + # Display name uses the params alias when present. + assert next(r for r in results if r.model_type == "STT").model_alias == "deepgram-prod" + + +@pytest.mark.asyncio +async def test_llm_failure_is_reported(tmp_path): + cfg = _cascade_config(tmp_path) + with ( + patch.object(preflight, "create_stt_service", return_value=_GoodSTT()), + patch.object(preflight, "create_tts_service", return_value=_GoodTTS()), + patch.object(preflight, "LiteLLMClient") as llm, + ): + llm.return_value.complete = AsyncMock(side_effect=Exception("401 invalid api key")) + results = await _run_preflight(cfg) + + llm_result = next(r for r in results if r.model_type == "LLM") + assert not llm_result.ok + assert "401" in llm_result.detail + + +@pytest.mark.asyncio +async def test_tts_error_frame_fails(tmp_path): + cfg = _cascade_config(tmp_path) + with ( + patch.object(preflight, "create_stt_service", return_value=_GoodSTT()), + patch.object(preflight, "create_tts_service", return_value=_BadTTS()), + patch.object(preflight, "LiteLLMClient") as llm, + ): + llm.return_value.complete = AsyncMock(return_value=("pong", {})) + results = await _run_preflight(cfg) + + tts_result = next(r for r in results if r.model_type == "TTS") + assert not tts_result.ok + assert "Unauthorized" in tts_result.detail + + +@pytest.mark.asyncio +async def test_s2s_is_skipped(tmp_path): + with patch.dict(os.environ, _BASE_ENV, 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", + ) + results = await _run_preflight(cfg) + assert results == [] + + +@pytest.mark.asyncio +async def test_guard_times_out(): + async def hang(): + await asyncio.sleep(10) + return True, "" + + result = await preflight._guard("LLM", "m", hang(), timeout=0.05) + assert not result.ok + assert "timed out" in result.detail + + +@pytest.mark.asyncio +async def test_or_raise_passes_when_all_ok(tmp_path): + cfg = _cascade_config(tmp_path) + with ( + patch.object(preflight, "create_stt_service", return_value=_GoodSTT()) as stt, + patch.object(preflight, "create_tts_service", return_value=_GoodTTS()) as tts, + patch.object(preflight, "LiteLLMClient") as llm, + ): + llm.return_value.complete = AsyncMock(return_value=("pong", {})) + await run_preflight(cfg) + stt.assert_called_once() + tts.assert_called_once() + llm.assert_called_once() + + +@pytest.mark.asyncio +async def test_or_raise_raises_on_failure(tmp_path): + cfg = _cascade_config(tmp_path) + with ( + patch.object(preflight, "create_stt_service", return_value=_GoodSTT()), + patch.object(preflight, "create_tts_service", return_value=_GoodTTS()), + patch.object(preflight, "LiteLLMClient") as llm, + ): + llm.return_value.complete = AsyncMock(side_effect=Exception("401")) + with pytest.raises(PreflightError, match="LLM"): + await run_preflight(cfg) + + +@pytest.mark.asyncio +async def test_or_raise_noop_when_disabled(tmp_path): + cfg = _cascade_config(tmp_path) + cfg.preflight = False + with patch.object(preflight, "run_preflight") as probe: + await run_preflight(cfg) + probe.assert_not_called()