Add preflight

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.

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).

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)

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)

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

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}")

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]] = {}

src/eva/run_benchmark.py

@@ -45,12 +45,14 @@ 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
         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