Add SmallestAI Hydra S2S framework support

.env.example

@@ -125,6 +125,14 @@ EVA_MODEL__TTS_PARAMS='{"api_key": "your_cartesia_api_key", "model": "sonic"}'
 #x pipeline_mode=S2S
 #v EVA_MODEL__S2S_PARAMS='{"model": "gpt-realtime-mini", "api_key": ""}'
 
+# Smallest Hydra S2S (set EVA_FRAMEWORK=smallest_hydra). Hydra emits no transcript
+# on the wire, so a batch STT transcribes each turn for the audit log; it defaults
+# to Smallest Pulse keyed on the same api_key. Override via the "transcription" block
+# (provider: smallest|openai|deepgram). voice: wren|sloane|marlowe|reed|knox|tate.
+#x pipeline_mode=S2S
+#v EVA_MODEL__S2S=hydra
+#v EVA_MODEL__S2S_PARAMS='{"model": "hydra", "api_key": "", "voice": "wren", "generate_initial_response": true, "transcription": {"provider": "smallest", "model": "pulse-pro", "language": "en"}}'
+
 # --- AudioLLM mode ---
 #i Audio-input LLM model name.
 #d string
@@ -139,7 +147,7 @@ EVA_MODEL__TTS_PARAMS='{"api_key": "your_cartesia_api_key", "model": "sonic"}'
 # --- Framework (S2S / AudioLLM) ---
 #i Base framework for S2S or AudioLLM pipelines.
 #d enum
-#e pipecat,openai_realtime,gemini_live,elevenlabs,grok_voice
+#e pipecat,openai_realtime,gemini_live,elevenlabs,grok_voice,smallest_hydra
 #v EVA_FRAMEWORK=openai_realtime
 
 # ==============================================

docs/assistant_server_contract.md

@@ -535,3 +535,42 @@ the run to fail or produce `None` latency fields in the result.
 | `audio_assistant.wav` | Yes | TTS quality metrics |
 | `framework_logs.jsonl` | Yes | Turn boundary metrics |
 | `pipecat_metrics.jsonl` | Yes | `model_response_latency` in `ConversationResult` |
+
+---
+
+## 13. Reference implementation: Smallest Hydra (transcript-less S2S)
+
+`src/eva/assistant/smallest_hydra_server.py` (`framework: smallest_hydra`) bridges to Smallest's
+**Hydra** speech-to-speech model over a raw WebSocket (no SDK; uses `websockets`). It is scored as a
+true **S2S** pipeline (`s2s: hydra`, so `get_pipeline_type → S2S`) and follows the Gemini Live
+three-task structure (forward user audio, process model events, pace audio output).
+
+Notable points specific to Hydra:
+
+- **Config.** `framework: smallest_hydra`, `model: {s2s: hydra, s2s_params: {...}}`. Recognised
+  `s2s_params`: `api_key` (**required**, Smallest key), `model` (default `hydra`), `voice`
+  (`wren`|`sloane`|`marlowe`|`reed`|`knox`|`tate`, default `wren`, frozen at handshake),
+  `generate_initial_response` (default `true` — Hydra speaks first; the system prompt is steered to
+  EVA's canned greeting), and `transcription` (see below).
+- **Protocol.** OpenAI-Realtime-shaped JSON over one WebSocket: `session.configure` handshake;
+  `input_audio_buffer.append` (base64 PCM16 in); `response.output_audio.delta` (base64 PCM16 out);
+  client-side tools via `response.function_call_arguments.done` → reply with
+  `conversation.item.create` (`function_call_output`) then `response.create`. Audio is **16 kHz in /
+  48 kHz out**; recording rate is 48 kHz (assistant native; user track upsampled to match).
+- **Transcripts.** Hydra streams a native **assistant** transcript
+  (`response.output_audio_transcript.delta`, accumulated per response and finalized on
+  `response.done`) — used directly, since it is more accurate than re-transcribing the audio.
+  It sends **no user transcript**, so each completed user utterance is batch-transcribed by a
+  configurable STT (`s2s_transcription.py`) and appended to the audit log; the log's
+  timestamp-sort keeps these asynchronous writes in chronological order. The `transcription`
+  block selects the provider (`smallest` default — Pulse-pro for English, Pulse otherwise, keyed
+  on the Hydra `api_key`; or `openai`/`deepgram` with their own `api_key`). User transcription is
+  **fail-soft**: an STT error drops that turn's text but never aborts the conversation.
+- **Latency / tokens.** `model_response` latency is emitted on the first audio delta per turn,
+  anchored on Hydra's `speech_stopped` (the simulator's `user_speech_stop` races), with a 50 ms
+  floor; token usage is read from the `response.done` `usage` block.
+- **Payload ceiling (Smallest-side).** Hydra caps the `session.configure` payload (~32 KB as of
+  2026-07); above it, it silently drops audio output. Instructions are truncated to fit, but for
+  **tool-heavy domains the tool-schema JSON alone can exceed the ceiling** (e.g. ITSM's 59 tools ≈
+  45 KB), which no instruction truncation can fix — those domains stay degraded until the limit
+  clears their tool schemas. Airline (~22 KB total) fits comfortably.

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.3"
 
 # Bump metrics_version when changes affect metric computation (metrics code,
 # judge prompts, pricing tables, postprocessor).

src/eva/assistant/audio_bridge.py

@@ -10,9 +10,11 @@
 
 import audioop
 import base64
+import io
 import json
 import struct
 import time
+import wave
 from pathlib import Path
 
 import numpy as np
@@ -51,6 +53,56 @@ def mulaw_8k_to_pcm16_24k(mulaw_bytes: bytes) -> bytes:
     return pcm_24k
 
 
+def mulaw_8k_to_pcm16_48k(mulaw_bytes: bytes) -> bytes:
+    """Convert 8kHz mu-law audio to 48kHz 16-bit PCM.
+
+    Used by the Smallest Hydra S2S server, which records at 48 kHz (its native
+    output rate) and must upsample the 8 kHz user track to match.
+    """
+    pcm_8k = audioop.ulaw2lin(mulaw_bytes, 2)
+    pcm_48k, _ = audioop.ratecv(pcm_8k, 2, 1, 8000, 48000, None)
+    # Clamp to exactly 6× input length so tracks stay positionally aligned.
+    expected_bytes = len(pcm_8k) * 6
+    if len(pcm_48k) < expected_bytes:
+        pcm_48k = pcm_48k + b"\x00" * (expected_bytes - len(pcm_48k))
+    elif len(pcm_48k) > expected_bytes:
+        pcm_48k = pcm_48k[:expected_bytes]
+    return pcm_48k
+
+
+def pcm16_48k_to_mulaw_8k(pcm_bytes: bytes) -> bytes:
+    """Convert 48kHz 16-bit PCM to 8kHz mu-law.
+
+    Uses soxr VHQ resampling for proper anti-aliasing during the 6:1
+    downsampling (audioop.ratecv lacks an anti-aliasing filter and sounds
+    muffled). Mirrors ``pcm16_24k_to_mulaw_8k`` for Hydra's 48 kHz output.
+    """
+    audio_data = np.frombuffer(pcm_bytes, dtype=np.int16)
+    resampled = soxr.resample(audio_data, 48000, 8000, quality="VHQ")
+    expected_samples = round(len(audio_data) * 8000 / 48000)
+    if len(resampled) < expected_samples:
+        resampled = np.pad(resampled, (0, expected_samples - len(resampled)))
+    elif len(resampled) > expected_samples:
+        resampled = resampled[:expected_samples]
+    pcm_8k = resampled.astype(np.int16).tobytes()
+    return audioop.lin2ulaw(pcm_8k, 2)
+
+
+def pcm16_to_wav_bytes(pcm_bytes: bytes, sample_rate: int) -> bytes:
+    """Wrap raw 16-bit mono PCM in an in-memory WAV container.
+
+    Used to hand a recorded utterance to a batch STT endpoint without touching
+    disk. 16-bit mono => 2 bytes/sample.
+    """
+    buf = io.BytesIO()
+    with wave.open(buf, "wb") as wav:
+        wav.setnchannels(1)
+        wav.setsampwidth(2)
+        wav.setframerate(sample_rate)
+        wav.writeframes(pcm_bytes)
+    return buf.getvalue()
+
+
 def pcm16_24k_to_mulaw_8k(pcm_bytes: bytes) -> bytes:
     """Convert 24kHz 16-bit PCM to 8kHz mu-law.
 
@@ -72,6 +124,19 @@ def pcm16_24k_to_mulaw_8k(pcm_bytes: bytes) -> bytes:
     return audioop.lin2ulaw(pcm_8k, 2)
 
 
+def resample_pcm16_soxr(pcm_bytes: bytes, from_rate: int, to_rate: int) -> bytes:
+    """Resample 16-bit mono PCM between arbitrary rates using soxr VHQ.
+
+    Anti-aliased (unlike audioop.ratecv), so it is safe for both up- and
+    down-sampling. Returns the input unchanged when the rates already match.
+    """
+    if from_rate == to_rate or not pcm_bytes:
+        return pcm_bytes
+    audio = np.frombuffer(pcm_bytes, dtype=np.int16)
+    resampled = soxr.resample(audio, from_rate, to_rate, quality="VHQ")
+    return bytes(resampled.astype(np.int16).tobytes())
+
+
 def sync_buffer_to_position(buffer: bytearray, target_position: int) -> None:
     """Pad *buffer* with silence bytes so it reaches *target_position*.
 

src/eva/assistant/s2s_transcription.py

@@ -0,0 +1,178 @@
+"""Pluggable batch transcription for S2S frameworks that emit no transcript.
+
+Some speech-to-speech models (notably Smallest Hydra) stream audio in both
+directions but never put text on the wire. EVA's audit log — which nearly all
+accuracy/experience metrics read — still needs a per-turn transcript, so the
+server transcribes each completed utterance (the PCM it sent or received) with a
+separate batch STT call.
+
+The provider is configurable via the ``transcription`` block inside
+``s2s_params``::
+
+    "transcription": {
+        "provider": "smallest" | "openai" | "deepgram",
+        "model": "...",          # provider-specific; sensible default per provider
+        "api_key": "...",        # falls back to the S2S api_key (smallest) / env
+        "language": "en",        # falls back to the run language
+        "base_url": "..."        # optional override
+    }
+
+All providers are called over a single shared async ``httpx`` client so the
+module has no heavy SDK import surface and is straightforward to mock in tests.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any
+
+import httpx
+
+from eva.assistant.audio_bridge import pcm16_to_wav_bytes
+from eva.utils.logging import get_logger
+
+logger = get_logger(__name__)
+
+_SMALLEST_STT_URL = "https://api.smallest.ai/waves/v1/stt/"
+_OPENAI_STT_URL = "https://api.openai.com/v1/audio/transcriptions"
+_DEEPGRAM_STT_URL = "https://api.deepgram.com/v1/listen"
+
+# Default model per provider. Smallest Pulse-pro is English-only; for any other
+# language we fall back to the multilingual Pulse model.
+_DEFAULT_MODELS = {
+    "smallest": "pulse-pro",
+    "openai": "whisper-1",
+    "deepgram": "nova-3",
+}
+
+
+class BatchTranscriber:
+    """Transcribe completed utterances via a configurable batch STT provider."""
+
+    def __init__(
+        self,
+        provider: str,
+        model: str,
+        api_key: str,
+        language: str,
+        base_url: str | None = None,
+        timeout: float = 30.0,
+    ) -> None:
+        self.provider = provider
+        self.model = model
+        self.api_key = api_key
+        self.language = language
+        self.base_url = base_url
+        self._client = httpx.AsyncClient(timeout=timeout)
+
+    async def transcribe(self, pcm: bytes, sample_rate: int) -> str:
+        """Transcribe raw 16-bit mono PCM. Returns "" on any failure (fail-soft).
+
+        Transcription errors must never abort a conversation — a dropped
+        transcript degrades that turn's text metrics but the audio recording and
+        tool calls are preserved.
+        """
+        if not pcm:
+            return ""
+        wav = pcm16_to_wav_bytes(pcm, sample_rate)
+        try:
+            if self.provider == "smallest":
+                return await self._transcribe_smallest(wav)
+            if self.provider == "openai":
+                return await self._transcribe_openai(wav)
+            if self.provider == "deepgram":
+                return await self._transcribe_deepgram(wav)
+            logger.error(f"Unknown transcription provider: {self.provider!r}")
+            return ""
+        except Exception as e:  # noqa: BLE001 - fail-soft by design
+            logger.error(f"{self.provider} transcription failed: {e}", exc_info=True)
+            return ""
+
+    async def _transcribe_smallest(self, wav: bytes) -> str:
+        url = self.base_url or _SMALLEST_STT_URL
+        resp = await self._client.post(
+            url,
+            params={"model": self.model, "language": self.language},
+            headers={
+                "Authorization": f"Bearer {self.api_key}",
+                "Content-Type": "application/octet-stream",
+            },
+            content=wav,
+        )
+        resp.raise_for_status()
+        return (resp.json().get("transcription") or "").strip()
+
+    async def _transcribe_openai(self, wav: bytes) -> str:
+        url = self.base_url or _OPENAI_STT_URL
+        resp = await self._client.post(
+            url,
+            headers={"Authorization": f"Bearer {self.api_key}"},
+            files={"file": ("audio.wav", wav, "audio/wav")},
+            data={"model": self.model, "language": self.language},
+        )
+        resp.raise_for_status()
+        return (resp.json().get("text") or "").strip()
+
+    async def _transcribe_deepgram(self, wav: bytes) -> str:
+        url = self.base_url or _DEEPGRAM_STT_URL
+        resp = await self._client.post(
+            url,
+            params={"model": self.model, "language": self.language, "smart_format": "true"},
+            headers={
+                "Authorization": f"Token {self.api_key}",
+                "Content-Type": "audio/wav",
+            },
+            content=wav,
+        )
+        resp.raise_for_status()
+        alternatives = resp.json()["results"]["channels"][0]["alternatives"]
+        return (alternatives[0]["transcript"] if alternatives else "").strip()
+
+    async def aclose(self) -> None:
+        await self._client.aclose()
+
+
+def create_transcriber(s2s_params: dict[str, Any], language: str) -> BatchTranscriber:
+    """Build a :class:`BatchTranscriber` from ``s2s_params['transcription']``.
+
+    Defaults to the Smallest Pulse provider keyed on the S2S ``api_key`` so an
+    in-stack transcript works with no extra configuration. The Pulse-pro model is
+    English-only; for any non-English run the default model falls back to the
+    multilingual ``pulse`` model.
+    """
+    cfg = dict(s2s_params.get("transcription") or {})
+    provider = cfg.get("provider", "smallest")
+    lang = cfg.get("language") or language or "en"
+
+    model = cfg.get("model")
+    if not model:
+        model = _DEFAULT_MODELS.get(provider, "")
+        # Smallest Pulse-pro is English-only; use multilingual Pulse otherwise.
+        if provider == "smallest" and not lang.startswith("en"):
+            model = "pulse"
+
+    # API key resolution: explicit transcription key, else a sensible fallback —
+    # the shared S2S key for Smallest, or the provider's standard env var.
+    api_key = cfg.get("api_key")
+    if not api_key:
+        if provider == "smallest":
+            api_key = s2s_params.get("api_key", "")
+        elif provider == "openai":
+            api_key = os.environ.get("OPENAI_API_KEY", "")
+        elif provider == "deepgram":
+            api_key = os.environ.get("DEEPGRAM_API_KEY", "")
+    if not api_key:
+        raise ValueError(
+            f"No API key for transcription provider {provider!r}. "
+            f"Set s2s_params.transcription.api_key (or s2s_params.api_key for 'smallest', "
+            f"OPENAI_API_KEY for 'openai', DEEPGRAM_API_KEY for 'deepgram')."
+        )
+
+    logger.info(f"S2S transcription: provider={provider} model={model} language={lang}")
+    return BatchTranscriber(
+        provider=provider,
+        model=model,
+        api_key=api_key,
+        language=lang,
+        base_url=cfg.get("base_url"),
+    )