How to Test a Voice AI Agent

"""Transcribe-then-eval core: drive a call fixture through an agent, get turns back.

The point of this module is that you never assert on audio. You capture the call,
transcribe both sides, and assert on a *structured* transcript: one record per
turn, with what the caller said, what ASR heard, what the agent replied, and the
per-stage latency for that turn. A flat string blob throws away exactly the
information you need to debug a failure.

Everything a vendor could own sits behind an interface:

* :class:`ASRProvider`  - Whisper, Deepgram, a vendor SDK, or a stub.
* :class:`TTSProvider`   - anything that turns text into an :class:`AudioClip`.
* :class:`AgentAdapter`  - your agent, reached however you reach it.

No vendor SDK is imported anywhere in this package.
"""

from __future__ import annotations

import json
import wave
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, List, Optional, Protocol

from voice_testing.latency import DEFAULT_BUDGET_MS, LatencyBudget

DEFAULT_SAMPLE_RATE = 16_000

#: Fixtures ship with the repo; resolved relative to this file, never absolute.
FIXTURE_DIR = Path(__file__).resolve().parent.parent / "fixtures"


# ---------------------------------------------------------------------------
# Audio
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class AudioClip:
    """A chunk of mono/stereo PCM audio plus just enough metadata to decode it.

    ``source_text`` is populated only for *synthesized* audio (the text a TTS
    provider was asked to speak). Real telephony captures leave it ``None``.
    Stub ASR providers use it; production ASR providers must ignore it.
    """

    pcm: bytes
    sample_rate: int = DEFAULT_SAMPLE_RATE
    channels: int = 1
    sample_width: int = 2
    source_text: Optional[str] = None
    label: str = ""

    def __post_init__(self) -> None:
        if self.sample_rate <= 0:
            raise ValueError(f"sample_rate must be positive, got {self.sample_rate}")
        if self.channels < 1:
            raise ValueError(f"channels must be >= 1, got {self.channels}")
        if self.sample_width < 1:
            raise ValueError(f"sample_width must be >= 1, got {self.sample_width}")

    @property
    def bytes_per_frame(self) -> int:
        return self.channels * self.sample_width

    @property
    def frame_count(self) -> int:
        return len(self.pcm) // self.bytes_per_frame

    @property
    def duration_ms(self) -> float:
        return 1000.0 * self.frame_count / self.sample_rate

    @classmethod
    def from_wav(
        cls, path: "str | Path", *, source_text: Optional[str] = None, label: str = ""
    ) -> "AudioClip":
        wav_path = Path(path)
        if not wav_path.is_file():
            raise FileNotFoundError(f"no audio file at {wav_path}")
        with wave.open(str(wav_path), "rb") as wav:
            return cls(
                pcm=wav.readframes(wav.getnframes()),
                sample_rate=wav.getframerate(),
                channels=wav.getnchannels(),
                sample_width=wav.getsampwidth(),
                source_text=source_text,
                label=label or wav_path.name,
            )

    def write_wav(self, path: "str | Path") -> Path:
        wav_path = Path(path)
        wav_path.parent.mkdir(parents=True, exist_ok=True)
        with wave.open(str(wav_path), "wb") as wav:
            wav.setnchannels(self.channels)
            wav.setsampwidth(self.sample_width)
            wav.setframerate(self.sample_rate)
            wav.writeframes(self.pcm)
        return wav_path


# ---------------------------------------------------------------------------
# Provider interfaces
# ---------------------------------------------------------------------------


class ASRProvider(Protocol):
    """Anything that turns caller or agent audio into text."""

    def transcribe(self, clip: AudioClip) -> str:  # pragma: no cover - interface
        ...


class TTSProvider(Protocol):
    """Anything that turns text into audio the agent can hear."""

    def synthesize(self, text: str) -> AudioClip:  # pragma: no cover - interface
        ...


class AgentAdapter(Protocol):
    """Your voice agent, reduced to one call: audio in, audio out.

    Implementations should mark the supplied :class:`LatencyBudget` at each
    pipeline stage boundary they can observe. A budget of ``None`` means the
    caller is not measuring latency for this turn.
    """

    def respond(
        self, clip: AudioClip, budget: Optional[LatencyBudget] = None
    ) -> AudioClip:  # pragma: no cover - interface
        ...


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class CallerTurn:
    """One thing the caller says, optionally on top of the agent still talking."""

    text: str
    audio: Optional[str] = None
    barge_in: bool = False
    note: str = ""


@dataclass
class CallFixture:
    """A scripted call: caller audio (or text to synthesize) plus a reference.

    The reference transcript holds one expected agent reply per caller turn. It
    is a *reference*, not a golden string: assertions compare against it
    semantically (see :mod:`voice_testing.assertions`) or with word error rate
    (see :mod:`voice_testing.wer`).
    """

    id: str
    description: str
    caller_turns: List[CallerTurn]
    reference_transcript: List[str]
    base_dir: Optional[Path] = None

    def __post_init__(self) -> None:
        if not self.id:
            raise ValueError("fixture id must not be empty")
        if not self.caller_turns:
            raise ValueError(f"fixture {self.id!r} has no caller turns")
        if len(self.reference_transcript) != len(self.caller_turns):
            raise ValueError(
                f"fixture {self.id!r} has {len(self.caller_turns)} caller turns but "
                f"{len(self.reference_transcript)} reference replies; there must be "
                "exactly one expected agent reply per caller turn"
            )

    # -- loading -----------------------------------------------------------

    @classmethod
    def load(cls, path: "str | Path") -> "CallFixture":
        """Load a fixture from ``fixture.json`` (or a directory containing one)."""
        target = Path(path)
        if target.is_dir():
            target = target / "fixture.json"
        if not target.is_file():
            raise FileNotFoundError(f"no fixture manifest at {target}")
        try:
            data = json.loads(target.read_text(encoding="utf-8"))
        except json.JSONDecodeError as exc:
            raise ValueError(f"fixture manifest {target} is not valid JSON: {exc}") from exc
        return cls.from_dict(data, base_dir=target.parent)

    @classmethod
    def load_named(
        cls, name: str, fixture_dir: "str | Path" = FIXTURE_DIR
    ) -> "CallFixture":
        """Load ``<fixture_dir>/<name>.json``."""
        directory = Path(fixture_dir)
        candidate = directory / f"{name}.json"
        if not candidate.is_file():
            available = sorted(p.stem for p in directory.glob("*.json"))
            raise FileNotFoundError(
                f"no fixture named {name!r} in {directory}; available: "
                f"{', '.join(available) or '(none)'}"
            )
        data = json.loads(candidate.read_text(encoding="utf-8"))
        return cls.from_dict(data, base_dir=directory)

    @classmethod
    def from_dict(
        cls, data: Dict[str, object], *, base_dir: Optional[Path] = None
    ) -> "CallFixture":
        for required in ("id", "caller_turns", "reference_transcript"):
            if required not in data:
                raise ValueError(f"fixture manifest is missing {required!r}")
        raw_turns = data["caller_turns"]
        if not isinstance(raw_turns, list):
            raise ValueError("caller_turns must be a list")
        turns: List[CallerTurn] = []
        for index, raw in enumerate(raw_turns):
            if isinstance(raw, str):
                turns.append(CallerTurn(text=raw))
                continue
            if not isinstance(raw, dict) or "text" not in raw:
                raise ValueError(
                    f"caller_turns[{index}] must be a string or an object with a "
                    "'text' key"
                )
            turns.append(
                CallerTurn(
                    text=str(raw["text"]),
                    audio=raw.get("audio"),
                    barge_in=bool(raw.get("barge_in", False)),
                    note=str(raw.get("note", "")),
                )
            )
        return cls(
            id=str(data["id"]),
            description=str(data.get("description", "")),
            caller_turns=turns,
            reference_transcript=[str(t) for t in data["reference_transcript"]],
            base_dir=base_dir,
        )

    # -- audio -------------------------------------------------------------

    def caller_clip(
        self, turn: CallerTurn, tts: Optional[TTSProvider] = None
    ) -> AudioClip:
        """Get audio for one caller turn: from disk if present, else synthesized."""
        if turn.audio:
            audio_path = Path(turn.audio)
            if not audio_path.is_absolute():
                audio_path = (self.base_dir or Path.cwd()) / audio_path
            return AudioClip.from_wav(audio_path, source_text=turn.text)
        if tts is None:
            raise ValueError(
                f"caller turn {turn.text!r} in fixture {self.id!r} has no 'audio' "
                "file, so a TTSProvider is required to synthesize it. Pass "
                "tts=... to run_call_fixture()."
            )
        return tts.synthesize(turn.text)


# ---------------------------------------------------------------------------
# Transcripts
# ---------------------------------------------------------------------------


@dataclass
class Turn:
    """One caller/agent exchange, with what each side actually said."""

    index: int
    caller_text: str
    caller_asr_text: str
    agent_text: str
    reference_agent_text: str
    barge_in: bool = False
    caller_audio_ms: float = 0.0
    agent_audio_ms: float = 0.0
    latency: Optional[LatencyBudget] = None

    @property
    def number(self) -> int:
        """1-based turn number, for human-readable output."""
        return self.index + 1


@dataclass
class TurnTranscript:
    """The structured result of one call: turns, not one flat string."""

    fixture_id: str
    turns: List[Turn] = field(default_factory=list)

    def __len__(self) -> int:
        return len(self.turns)

    def __iter__(self):
        return iter(self.turns)

    def __getitem__(self, index: int) -> Turn:
        return self.turns[index]

    def agent_texts(self) -> List[str]:
        return [t.agent_text for t in self.turns]

    def caller_texts(self) -> List[str]:
        return [t.caller_text for t in self.turns]

    def reference_texts(self) -> List[str]:
        return [t.reference_agent_text for t in self.turns]

    def agent_flat(self) -> str:
        return " ".join(self.agent_texts())

    def reference_flat(self) -> str:
        return " ".join(self.reference_texts())

    def barge_in_turns(self) -> List[Turn]:
        return [t for t in self.turns if t.barge_in]

    def as_dict(self) -> Dict[str, object]:
        return {
            "fixture_id": self.fixture_id,
            "turns": [
                {
                    "turn": t.number,
                    "caller": t.caller_text,
                    "caller_as_heard": t.caller_asr_text,
                    "agent": t.agent_text,
                    "reference": t.reference_agent_text,
                    "barge_in": t.barge_in,
                    "latency": t.latency.as_dict() if t.latency else None,
                }
                for t in self.turns
            ],
        }

    def report(self) -> str:
        lines = [f"transcript: {self.fixture_id} ({len(self.turns)} turns)"]
        for turn in self.turns:
            flag = " [barge-in]" if turn.barge_in else ""
            lines.append(f"  turn {turn.number}{flag}")
            lines.append(f"    caller        : {turn.caller_text}")
            if turn.caller_asr_text != turn.caller_text:
                lines.append(f"    caller (heard): {turn.caller_asr_text}")
            lines.append(f"    agent         : {turn.agent_text}")
            lines.append(f"    reference     : {turn.reference_agent_text}")
        return "\n".join(lines)


# ---------------------------------------------------------------------------
# The driver
# ---------------------------------------------------------------------------


def run_call_fixture(
    fixture: CallFixture,
    agent: AgentAdapter,
    asr: ASRProvider,
    *,
    tts: Optional[TTSProvider] = None,
    thresholds_ms: Optional[Dict[str, float]] = None,
    measure_latency: bool = True,
) -> TurnTranscript:
    """Drive ``fixture`` through ``agent``, transcribe both sides, return turns.

    Args:
        fixture: the scripted call to replay.
        agent: the agent under test.
        asr: the ASR provider used to transcribe both sides of the call.
        tts: required only for fixture turns that carry no audio file.
        thresholds_ms: per-stage latency budget; defaults to
            :data:`voice_testing.latency.DEFAULT_BUDGET_MS`.
        measure_latency: set ``False`` to skip budget creation entirely.

    Returns:
        A :class:`TurnTranscript` with one :class:`Turn` per caller turn.
    """
    transcript = TurnTranscript(fixture_id=fixture.id)
    for index, caller_turn in enumerate(fixture.caller_turns):
        caller_clip = fixture.caller_clip(caller_turn, tts=tts)
        budget: Optional[LatencyBudget] = None
        if measure_latency:
            budget = LatencyBudget(
                thresholds_ms=dict(thresholds_ms or DEFAULT_BUDGET_MS),
                label=f"{fixture.id} turn {index + 1}",
            )
        agent_clip = agent.respond(caller_clip, budget=budget)
        if not isinstance(agent_clip, AudioClip):
            raise TypeError(
                f"{type(agent).__name__}.respond() must return an AudioClip, got "
                f"{type(agent_clip).__name__}"
            )
        transcript.turns.append(
            Turn(
                index=index,
                caller_text=caller_turn.text,
                caller_asr_text=asr.transcribe(caller_clip),
                agent_text=asr.transcribe(agent_clip),
                reference_agent_text=fixture.reference_transcript[index],
                barge_in=caller_turn.barge_in,
                caller_audio_ms=caller_clip.duration_ms,
                agent_audio_ms=agent_clip.duration_ms,
                latency=budget,
            )
        )
    return transcript

"""Assertions for output that is never byte-identical twice.

A voice agent is non-deterministic twice over: ASR transcribes the same audio
slightly differently, and the model words the same answer slightly differently.
`assert agent_text == "Your appointment is confirmed"` is therefore a flaky test
generator, not a test.

Three tools replace it:

* :func:`semantic_similarity` - cosine similarity between embeddings, for
  "did it say the same thing?"
* :func:`llm_judge_verdict` - a second model scores the turn against a
  :class:`Rubric`, for "did it behave correctly?"
* :func:`run_n_times` - repeat the scenario N times and gate on a pass rate,
  because a single green run of a stochastic system proves nothing.

The embedding and judge backends are interfaces. A deterministic, offline
:class:`HashingEmbedder` and a rule-based judge (see
:mod:`voice_testing.mocks`) ship with the repo so the suite runs with no API key.
"""

from __future__ import annotations

import hashlib
import re
from dataclasses import dataclass, field
from typing import Callable, Dict, Iterable, Iterator, List, Optional, Protocol, Sequence

import numpy as np

_WORD_RE = re.compile(r"[a-z0-9']+")


# ---------------------------------------------------------------------------
# Embeddings
# ---------------------------------------------------------------------------


class EmbeddingProvider(Protocol):
    """Anything that maps text to a fixed-length vector."""

    def embed(self, text: str) -> Sequence[float]:  # pragma: no cover - interface
        ...


def _normalize(text: str) -> List[str]:
    return _WORD_RE.findall(text.lower())


class HashingEmbedder:
    """Deterministic, dependency-free stand-in for a real embedding model.

    It hashes word unigrams, word bigrams and character n-grams into a fixed
    number of dimensions and L2-normalizes the result. That makes it a *lexical*
    similarity measure: good enough to prove the assertion plumbing works
    offline and in CI, and good enough to catch a wholly wrong answer, but it
    does not understand meaning.

    Swap it for a real embedding model in staging::

        class OpenAIEmbedder:
            def embed(self, text): return client.embeddings.create(...).data[0].embedding

    Hashing uses blake2b rather than ``hash()`` because ``hash()`` on ``str`` is
    salted per process, which would make vectors differ between runs.
    """

    def __init__(self, dims: int = 512, char_ngram: int = 4) -> None:
        if dims < 8:
            raise ValueError(f"dims must be >= 8, got {dims}")
        if char_ngram < 2:
            raise ValueError(f"char_ngram must be >= 2, got {char_ngram}")
        self.dims = dims
        self.char_ngram = char_ngram

    def _features(self, text: str) -> Iterator[str]:
        words = _normalize(text)
        for word in words:
            yield f"w:{word}"
        for left, right in zip(words, words[1:]):
            yield f"b:{left}_{right}"
        joined = " ".join(words)
        n = self.char_ngram
        for start in range(max(0, len(joined) - n + 1)):
            yield f"c:{joined[start:start + n]}"

    def _bucket(self, feature: str) -> int:
        digest = hashlib.blake2b(feature.encode("utf-8"), digest_size=8).digest()
        return int.from_bytes(digest, "big") % self.dims

    def embed(self, text: str) -> Sequence[float]:
        vector = np.zeros(self.dims, dtype=np.float64)
        for feature in self._features(text):
            vector[self._bucket(feature)] += 1.0
        norm = float(np.linalg.norm(vector))
        if norm == 0.0:
            return vector.tolist()
        return (vector / norm).tolist()


_DEFAULT_EMBEDDER = HashingEmbedder()


def cosine_similarity(left: Sequence[float], right: Sequence[float]) -> float:
    """Cosine similarity of two vectors, clamped to ``[-1, 1]``."""
    a = np.asarray(left, dtype=np.float64)
    b = np.asarray(right, dtype=np.float64)
    if a.shape != b.shape:
        raise ValueError(f"vector shapes differ: {a.shape} vs {b.shape}")
    denominator = float(np.linalg.norm(a) * np.linalg.norm(b))
    if denominator == 0.0:
        return 0.0
    return float(np.clip(np.dot(a, b) / denominator, -1.0, 1.0))


def semantic_similarity(
    actual: str, expected: str, embedder: Optional[EmbeddingProvider] = None
) -> float:
    """How close two utterances are in meaning, as a number in ``[-1, 1]``."""
    provider = embedder or _DEFAULT_EMBEDDER
    return cosine_similarity(provider.embed(actual), provider.embed(expected))


class SemanticMismatch(AssertionError):
    """Raised by :func:`assert_semantic_match`."""


def assert_semantic_match(
    actual: str,
    expected: str,
    *,
    threshold: float = 0.75,
    embedder: Optional[EmbeddingProvider] = None,
    context: str = "",
) -> float:
    """Assert ``actual`` means roughly what ``expected`` means."""
    if not 0.0 < threshold <= 1.0:
        raise ValueError(f"threshold must be in (0, 1], got {threshold}")
    score = semantic_similarity(actual, expected, embedder=embedder)
    if score < threshold:
        where = f" [{context}]" if context else ""
        raise SemanticMismatch(
            f"semantic similarity {score:.3f} < {threshold:.3f}{where}\n"
            f"  expected (reference): {expected!r}\n"
            f"  actual   (agent)    : {actual!r}"
        )
    return score


# ---------------------------------------------------------------------------
# LLM as judge
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class Rubric:
    """What "correct" means for one turn, in a form a model or a stub can use.

    ``instructions`` is the prose a real judge model gets. ``must_mention`` and
    ``must_not_mention`` are the machine-checkable core, which lets the bundled
    offline judge grade the same rubric without a model call.
    """

    name: str
    instructions: str = ""
    must_mention: Sequence[str] = field(default_factory=tuple)
    must_not_mention: Sequence[str] = field(default_factory=tuple)

    def as_prompt(self, transcript_text: str) -> str:
        """Render the rubric as a judge prompt. Useful for real judge backends."""
        lines = [
            "You are grading one turn of a phone call handled by a voice AI agent.",
            f"Rubric: {self.name}",
        ]
        if self.instructions:
            lines.append(f"Criteria: {self.instructions}")
        if self.must_mention:
            lines.append(f"The reply must convey: {', '.join(self.must_mention)}")
        if self.must_not_mention:
            lines.append(f"The reply must not claim: {', '.join(self.must_not_mention)}")
        lines += [
            "",
            "Transcript:",
            transcript_text,
            "",
            'Answer with JSON: {"passed": bool, "score": 0..1, "reason": str}',
        ]
        return "\n".join(lines)


@dataclass(frozen=True)
class JudgeVerdict:
    """A judge's answer for one turn."""

    passed: bool
    reason: str = ""
    score: Optional[float] = None

    def __post_init__(self) -> None:
        if self.score is not None and not 0.0 <= self.score <= 1.0:
            raise ValueError(f"score must be in [0, 1], got {self.score}")


class LLMJudge(Protocol):
    """Anything that grades a transcript against a rubric."""

    def verdict(
        self, rubric: Rubric, transcript_text: str
    ) -> JudgeVerdict:  # pragma: no cover - interface
        ...


class JudgeRejected(AssertionError):
    """Raised by :func:`assert_judge_passes`."""


def llm_judge_verdict(
    judge: LLMJudge, rubric: Rubric, transcript_text: str
) -> JudgeVerdict:
    """Ask ``judge`` to grade ``transcript_text``, validating what comes back.

    A judge that returns a malformed verdict is a bug in the harness, not a
    failing test, so it raises ``TypeError`` rather than failing the assertion.
    """
    if not transcript_text.strip():
        raise ValueError("transcript_text is empty; nothing to judge")
    verdict = judge.verdict(rubric, transcript_text)
    if not isinstance(verdict, JudgeVerdict):
        raise TypeError(
            f"{type(judge).__name__}.verdict() must return a JudgeVerdict, got "
            f"{type(verdict).__name__}"
        )
    return verdict


def assert_judge_passes(
    judge: LLMJudge,
    rubric: Rubric,
    transcript_text: str,
    *,
    min_score: Optional[float] = None,
) -> JudgeVerdict:
    """Assert the judge passed the turn, and optionally cleared ``min_score``."""
    verdict = llm_judge_verdict(judge, rubric, transcript_text)
    if not verdict.passed:
        raise JudgeRejected(
            f"judge failed rubric {rubric.name!r}: {verdict.reason}\n"
            f"  transcript: {transcript_text!r}"
        )
    if min_score is not None:
        if verdict.score is None:
            raise JudgeRejected(
                f"rubric {rubric.name!r} requires min_score={min_score} but the "
                f"judge returned no score"
            )
        if verdict.score < min_score:
            raise JudgeRejected(
                f"judge score {verdict.score:.2f} < {min_score:.2f} for rubric "
                f"{rubric.name!r}: {verdict.reason}"
            )
    return verdict


# ---------------------------------------------------------------------------
# The non-determinism gate
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class Attempt:
    """One run of a repeated scenario."""

    index: int
    passed: bool
    detail: str = ""


class PassRateTooLow(AssertionError):
    """Raised by :meth:`RepeatResult.assert_pass_rate`."""


@dataclass
class RepeatResult:
    """Outcome of running a scenario N times."""

    label: str
    attempts: List[Attempt] = field(default_factory=list)

    @property
    def total(self) -> int:
        return len(self.attempts)

    @property
    def passes(self) -> int:
        return sum(1 for a in self.attempts if a.passed)

    @property
    def failures(self) -> List[Attempt]:
        return [a for a in self.attempts if not a.passed]

    @property
    def pass_rate(self) -> float:
        if not self.attempts:
            return 0.0
        return self.passes / self.total

    def assert_pass_rate(self, threshold: float) -> "RepeatResult":
        if not 0.0 < threshold <= 1.0:
            raise ValueError(f"threshold must be in (0, 1], got {threshold}")
        if self.pass_rate < threshold:
            detail = "\n".join(
                f"  - attempt {a.index + 1}: {a.detail or 'failed'}"
                for a in self.failures
            )
            raise PassRateTooLow(
                f"{self.label}: pass rate {self.pass_rate:.0%} "
                f"({self.passes}/{self.total}) below threshold {threshold:.0%}\n"
                f"{detail}"
            )
        return self

    def report(self) -> str:
        return (
            f"{self.label}: {self.passes}/{self.total} passed "
            f"({self.pass_rate:.0%})"
        )


def run_n_times(
    scenario: Callable[[int], object],
    *,
    n: int = 5,
    label: str = "scenario",
    stop_early: bool = False,
) -> RepeatResult:
    """Run ``scenario(attempt_index)`` ``n`` times and collect pass/fail.

    ``scenario`` may signal failure two ways: return a falsy value, or raise
    ``AssertionError``. Any other exception propagates, because a crash in the
    harness is not a flaky agent.

    Gate the result with :meth:`RepeatResult.assert_pass_rate`::

        run_n_times(one_call, n=10, label="reschedule").assert_pass_rate(0.9)
    """
    if n < 1:
        raise ValueError(f"n must be >= 1, got {n}")
    result = RepeatResult(label=label)
    for index in range(n):
        try:
            outcome = scenario(index)
        except AssertionError as exc:
            result.attempts.append(Attempt(index=index, passed=False, detail=str(exc)))
        else:
            passed = bool(outcome) if outcome is not None else True
            detail = "" if passed else f"scenario returned {outcome!r}"
            result.attempts.append(
                Attempt(index=index, passed=passed, detail=detail)
            )
        if stop_early and not result.attempts[-1].passed:
            break
    return result


def turn_similarity_report(
    actual_turns: Iterable[str],
    reference_turns: Iterable[str],
    *,
    embedder: Optional[EmbeddingProvider] = None,
) -> List[Dict[str, object]]:
    """Per-turn similarity scores, handy for debugging a failed call."""
    rows: List[Dict[str, object]] = []
    for index, (actual, reference) in enumerate(zip(actual_turns, reference_turns)):
        rows.append(
            {
                "turn": index + 1,
                "similarity": round(
                    semantic_similarity(actual, reference, embedder=embedder), 4
                ),
                "agent": actual,
                "reference": reference,
            }
        )
    return rows

"""Word error rate: the one number that tells you the ears are still working.

Semantic assertions test what the agent decided. WER tests what the agent
*heard*. Track it separately, because an agent that answers a question nobody
asked is an ASR bug wearing a reasoning bug's clothes.

Run it as a script in CI::

    python voice_testing/wer.py \\
        --reference examples/reference_transcript.txt \\
        --hypothesis examples/asr_output.txt \\
        --threshold 0.10

Exit code is 0 when WER <= threshold and 1 when it is not, so the command drops
straight into a pipeline step.

``jiwer`` is used when installed (it is the reference implementation for this
metric). When it is not, an equivalent Levenshtein alignment in the standard
library takes over, so the check still runs in a bare environment. Both paths
produce the same substitution/deletion/insertion counts.
"""

from __future__ import annotations

import argparse
import json
import re
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Dict, List, Optional, Sequence, Tuple

try:  # pragma: no cover - depends on the environment
    import jiwer

    HAS_JIWER = True
except ImportError:  # pragma: no cover - depends on the environment
    jiwer = None
    HAS_JIWER = False

_PUNCTUATION_RE = re.compile(r"[^\w\s']", re.UNICODE)
_WHITESPACE_RE = re.compile(r"\s+")

#: Rewritten before comparison so "3 pm" and "three pm" are not a substitution.
_NUMBER_WORDS: Dict[str, str] = {
    "zero": "0",
    "one": "1",
    "two": "2",
    "three": "3",
    "four": "4",
    "five": "5",
    "six": "6",
    "seven": "7",
    "eight": "8",
    "nine": "9",
    "ten": "10",
    "eleven": "11",
    "twelve": "12",
}


def normalize(text: str, *, map_numbers: bool = True) -> List[str]:
    """Lowercase, strip punctuation, collapse whitespace, tokenize.

    ASR output and human reference transcripts disagree about punctuation and
    casing constantly. Comparing raw strings measures your transcription style,
    not your recognizer.
    """
    lowered = text.lower().replace("-", " ")
    stripped = _PUNCTUATION_RE.sub(" ", lowered)
    tokens = _WHITESPACE_RE.sub(" ", stripped).strip().split(" ")
    tokens = [t for t in tokens if t]
    if map_numbers:
        tokens = [_NUMBER_WORDS.get(t, t) for t in tokens]
    return tokens


@dataclass(frozen=True)
class WERResult:
    """Word error rate plus the edit counts that explain it."""

    wer: float
    substitutions: int
    deletions: int
    insertions: int
    hits: int
    reference_words: int
    backend: str

    @property
    def errors(self) -> int:
        return self.substitutions + self.deletions + self.insertions

    def as_dict(self) -> Dict[str, object]:
        return {
            "wer": round(self.wer, 4),
            "substitutions": self.substitutions,
            "deletions": self.deletions,
            "insertions": self.insertions,
            "hits": self.hits,
            "reference_words": self.reference_words,
            "errors": self.errors,
            "backend": self.backend,
        }

    def report(self) -> str:
        return (
            f"WER {self.wer:.2%}  "
            f"(S={self.substitutions} D={self.deletions} I={self.insertions} "
            f"H={self.hits} over {self.reference_words} reference words, "
            f"backend={self.backend})"
        )


def _levenshtein_counts(
    reference: Sequence[str], hypothesis: Sequence[str]
) -> Tuple[int, int, int, int]:
    """Return ``(substitutions, deletions, insertions, hits)`` via edit alignment.

    Standard Wagner-Fischer with backtracking, kept in the standard library so
    the check works without ``jiwer`` installed.
    """
    rows, cols = len(reference) + 1, len(hypothesis) + 1
    cost = [[0] * cols for _ in range(rows)]
    for i in range(rows):
        cost[i][0] = i
    for j in range(cols):
        cost[0][j] = j
    for i in range(1, rows):
        for j in range(1, cols):
            if reference[i - 1] == hypothesis[j - 1]:
                cost[i][j] = cost[i - 1][j - 1]
            else:
                cost[i][j] = 1 + min(
                    cost[i - 1][j - 1],  # substitution
                    cost[i - 1][j],  # deletion
                    cost[i][j - 1],  # insertion
                )

    substitutions = deletions = insertions = hits = 0
    i, j = len(reference), len(hypothesis)
    while i > 0 or j > 0:
        if i > 0 and j > 0 and reference[i - 1] == hypothesis[j - 1] and cost[i][j] == cost[i - 1][j - 1]:
            hits += 1
            i, j = i - 1, j - 1
        elif i > 0 and j > 0 and cost[i][j] == cost[i - 1][j - 1] + 1:
            substitutions += 1
            i, j = i - 1, j - 1
        elif i > 0 and cost[i][j] == cost[i - 1][j] + 1:
            deletions += 1
            i -= 1
        else:
            insertions += 1
            j -= 1
    return substitutions, deletions, insertions, hits


def measure(
    reference: str,
    hypothesis: str,
    *,
    map_numbers: bool = True,
    force_pure_python: bool = False,
) -> WERResult:
    """Compute WER for one reference/hypothesis pair."""
    ref_tokens = normalize(reference, map_numbers=map_numbers)
    hyp_tokens = normalize(hypothesis, map_numbers=map_numbers)
    if not ref_tokens:
        raise ValueError("reference transcript is empty after normalization")

    use_jiwer = HAS_JIWER and not force_pure_python
    if use_jiwer:
        output = jiwer.process_words([" ".join(ref_tokens)], [" ".join(hyp_tokens)])
        return WERResult(
            wer=float(output.wer),
            substitutions=int(output.substitutions),
            deletions=int(output.deletions),
            insertions=int(output.insertions),
            hits=int(output.hits),
            reference_words=len(ref_tokens),
            backend="jiwer",
        )

    substitutions, deletions, insertions, hits = _levenshtein_counts(
        ref_tokens, hyp_tokens
    )
    errors = substitutions + deletions + insertions
    return WERResult(
        wer=errors / len(ref_tokens),
        substitutions=substitutions,
        deletions=deletions,
        insertions=insertions,
        hits=hits,
        reference_words=len(ref_tokens),
        backend="pure-python",
    )


def word_error_rate(reference: str, hypothesis: str, **kwargs: object) -> float:
    """Convenience wrapper returning just the rate."""
    return measure(reference, hypothesis, **kwargs).wer  # type: ignore[arg-type]


class WERTooHigh(AssertionError):
    """Raised by :func:`assert_wer_within`."""


def assert_wer_within(
    reference: str,
    hypothesis: str,
    *,
    threshold: float = 0.10,
    context: str = "",
    map_numbers: bool = True,
) -> WERResult:
    """Assert transcription error stays under ``threshold`` (0.10 == 10%)."""
    if not 0.0 <= threshold <= 1.0:
        raise ValueError(f"threshold must be in [0, 1], got {threshold}")
    result = measure(reference, hypothesis, map_numbers=map_numbers)
    if result.wer > threshold:
        where = f" [{context}]" if context else ""
        raise WERTooHigh(
            f"{result.report()} exceeds threshold {threshold:.2%}{where}\n"
            f"  reference : {reference!r}\n"
            f"  hypothesis: {hypothesis!r}"
        )
    return result


def _read_transcript(path: str) -> str:
    file_path = Path(path)
    if not file_path.is_file():
        raise SystemExit(f"error: no transcript file at {file_path}")
    text = file_path.read_text(encoding="utf-8")
    if not text.strip():
        raise SystemExit(f"error: transcript file {file_path} is empty")
    return text


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="wer",
        description=(
            "Compute word error rate between a known-correct reference "
            "transcript and ASR output. Exits 1 when WER exceeds --threshold."
        ),
    )
    parser.add_argument(
        "--reference", required=True, help="path to the known-correct transcript"
    )
    parser.add_argument(
        "--hypothesis", required=True, help="path to the ASR output transcript"
    )
    parser.add_argument(
        "--threshold",
        type=float,
        default=0.10,
        help="maximum acceptable WER as a fraction (default: 0.10)",
    )
    parser.add_argument(
        "--json", action="store_true", help="emit the result as JSON on stdout"
    )
    parser.add_argument(
        "--no-number-mapping",
        action="store_true",
        help="do not rewrite number words as digits before comparing",
    )
    return parser


def main(argv: Optional[Sequence[str]] = None) -> int:
    args = build_parser().parse_args(argv)
    if not 0.0 <= args.threshold <= 1.0:
        raise SystemExit(
            f"error: --threshold must be between 0 and 1, got {args.threshold}"
        )
    reference = _read_transcript(args.reference)
    hypothesis = _read_transcript(args.hypothesis)
    result = measure(reference, hypothesis, map_numbers=not args.no_number_mapping)
    within = result.wer <= args.threshold

    if args.json:
        payload = result.as_dict()
        payload["threshold"] = args.threshold
        payload["passed"] = within
        print(json.dumps(payload, indent=2))
    else:
        print(result.report())
        verdict = "PASS" if within else "FAIL"
        print(f"{verdict}: threshold {args.threshold:.2%}")
    return 0 if within else 1


if __name__ == "__main__":
    sys.exit(main())

"""Per-stage latency instrumentation for one voice agent turn.

A voice agent feels broken long before it answers wrong. The failure mode is
almost always latency, and a single end-to-end number hides which stage caused
it. This module records a mark per pipeline stage and asserts each stage against
its own millisecond threshold, so a regression points at the guilty component
instead of at "the call felt slow".

Thresholds are cumulative: the value for a stage is the maximum time allowed
between the start of the turn (the moment caller audio stops arriving, or
whatever boundary you choose in ``start()``) and that stage's mark.

Typical wiring inside a call handler::

    budget = LatencyBudget().start()
    text = await asr.finalize(stream);        budget.mark("asr_finalize")
    async for token in llm.stream(text):
        budget.mark_once("first_token")
        tts.feed(token)
    budget.mark("generation_complete")
    ...
    budget.mark("tts_first_byte")             # from the TTS output callback
    budget.assert_within_budget()
"""

from __future__ import annotations

import time
from dataclasses import dataclass, field
from typing import Callable, Dict, List, Optional, Tuple

#: The four stage boundaries worth instrumenting in a streaming voice pipeline.
STAGES: Tuple[str, ...] = (
    "asr_finalize",
    "first_token",
    "generation_complete",
    "tts_first_byte",
)

#: Cumulative milliseconds allowed from turn start to each stage mark.
#: These are deliberately tight defaults for a phone call; override per agent.
DEFAULT_BUDGET_MS: Dict[str, float] = {
    "asr_finalize": 300.0,
    "first_token": 600.0,
    "tts_first_byte": 800.0,
    "generation_complete": 1400.0,
}


def monotonic_clock() -> float:
    """Default clock. Seconds, monotonic, not wall-clock."""
    return time.perf_counter()


@dataclass(frozen=True)
class StageViolation:
    """One stage that blew its share of the budget."""

    stage: str
    elapsed_ms: float
    threshold_ms: float

    @property
    def over_by_ms(self) -> float:
        return self.elapsed_ms - self.threshold_ms

    def __str__(self) -> str:
        return (
            f"{self.stage}: {self.elapsed_ms:.1f}ms "
            f"(budget {self.threshold_ms:.1f}ms, over by {self.over_by_ms:.1f}ms)"
        )


class BudgetExceeded(AssertionError):
    """Raised by :meth:`LatencyBudget.assert_within_budget`."""

    def __init__(self, label: str, violations: List[StageViolation]) -> None:
        self.violations = list(violations)
        header = f"latency budget exceeded for {label or 'turn'}"
        body = "\n".join(f"  - {v}" for v in self.violations)
        super().__init__(f"{header}:\n{body}")


class MissingStageMarks(AssertionError):
    """Raised when a required stage was never marked."""

    def __init__(self, label: str, missing: List[str]) -> None:
        self.missing = list(missing)
        super().__init__(
            f"stages never marked for {label or 'turn'}: {', '.join(self.missing)}. "
            "Call budget.mark(stage) at each boundary, or pass "
            "require_all=False to assert only on what was measured."
        )


@dataclass
class LatencyBudget:
    """Collects stage marks for a single turn and checks them against thresholds.

    ``clock`` exists so tests can inject a scripted clock and stay deterministic.
    ``record`` exists so replayed traces (or a simulator) can supply timings that
    were measured elsewhere, without pretending to own the wall clock.
    """

    thresholds_ms: Dict[str, float] = field(
        default_factory=lambda: dict(DEFAULT_BUDGET_MS)
    )
    clock: Callable[[], float] = monotonic_clock
    label: str = ""
    marks_ms: Dict[str, float] = field(default_factory=dict)
    _start: Optional[float] = field(default=None, repr=False)

    def __post_init__(self) -> None:
        if not self.thresholds_ms:
            raise ValueError("thresholds_ms must not be empty")
        for stage, threshold in self.thresholds_ms.items():
            if threshold <= 0:
                raise ValueError(
                    f"threshold for {stage!r} must be positive, got {threshold}"
                )

    # -- lifecycle ---------------------------------------------------------

    def start(self) -> "LatencyBudget":
        """Mark the start of the turn. Clears any previous marks."""
        self._start = self.clock()
        self.marks_ms.clear()
        return self

    def __enter__(self) -> "LatencyBudget":
        return self.start()

    def __exit__(self, exc_type, exc, tb) -> bool:
        return False

    @property
    def started(self) -> bool:
        return self._start is not None

    # -- recording ---------------------------------------------------------

    def _check_stage(self, stage: str) -> None:
        if stage not in self.thresholds_ms:
            known = ", ".join(sorted(self.thresholds_ms))
            raise KeyError(f"unknown stage {stage!r}; budgeted stages are: {known}")

    def mark(self, stage: str) -> float:
        """Record the elapsed time for ``stage`` using the clock."""
        self._check_stage(stage)
        if self._start is None:
            raise RuntimeError(
                "LatencyBudget.start() must be called before mark(); "
                "start it when the turn begins."
            )
        elapsed_ms = (self.clock() - self._start) * 1000.0
        self.marks_ms[stage] = elapsed_ms
        return elapsed_ms

    def mark_once(self, stage: str) -> float:
        """Mark ``stage`` only if it has not been marked yet.

        Useful for ``first_token``, which lives inside a streaming loop.
        """
        self._check_stage(stage)
        if stage in self.marks_ms:
            return self.marks_ms[stage]
        return self.mark(stage)

    def record(self, stage: str, elapsed_ms: float) -> float:
        """Record a stage timing measured somewhere else (replay, simulator)."""
        self._check_stage(stage)
        if elapsed_ms < 0:
            raise ValueError(f"elapsed_ms for {stage!r} must be >= 0")
        if self._start is None:
            self._start = self.clock()
        self.marks_ms[stage] = float(elapsed_ms)
        return self.marks_ms[stage]

    # -- reading -----------------------------------------------------------

    def elapsed_ms(self, stage: str) -> float:
        self._check_stage(stage)
        if stage not in self.marks_ms:
            raise KeyError(f"stage {stage!r} was never marked")
        return self.marks_ms[stage]

    def ordered_marks(self) -> List[Tuple[str, float]]:
        """Marks sorted chronologically."""
        return sorted(self.marks_ms.items(), key=lambda kv: kv[1])

    def stage_deltas_ms(self) -> List[Tuple[str, float]]:
        """Time spent *inside* each stage, i.e. delta from the previous mark."""
        deltas: List[Tuple[str, float]] = []
        previous = 0.0
        for stage, elapsed in self.ordered_marks():
            deltas.append((stage, elapsed - previous))
            previous = elapsed
        return deltas

    def violations(self) -> List[StageViolation]:
        found: List[StageViolation] = []
        for stage, elapsed in self.ordered_marks():
            threshold = self.thresholds_ms[stage]
            if elapsed > threshold:
                found.append(StageViolation(stage, elapsed, threshold))
        return found

    def missing_stages(self) -> List[str]:
        return [s for s in self.thresholds_ms if s not in self.marks_ms]

    # -- asserting ---------------------------------------------------------

    def assert_within_budget(self, *, require_all: bool = True) -> "LatencyBudget":
        """Raise if any marked stage exceeded its threshold.

        With ``require_all`` (the default) an unmarked budgeted stage is also a
        failure, because a stage that silently never fires is the bug you most
        want to catch.
        """
        if require_all:
            missing = self.missing_stages()
            if missing:
                raise MissingStageMarks(self.label, missing)
        violations = self.violations()
        if violations:
            raise BudgetExceeded(self.label, violations)
        return self

    # -- reporting ---------------------------------------------------------

    def as_dict(self) -> Dict[str, object]:
        return {
            "label": self.label,
            "marks_ms": {s: round(v, 2) for s, v in self.ordered_marks()},
            "thresholds_ms": dict(self.thresholds_ms),
            "violations": [str(v) for v in self.violations()],
            "missing_stages": self.missing_stages(),
        }

    def report(self) -> str:
        lines = [f"latency: {self.label or 'turn'}"]
        deltas = dict(self.stage_deltas_ms())
        for stage, elapsed in self.ordered_marks():
            threshold = self.thresholds_ms[stage]
            status = "OK " if elapsed <= threshold else "OVER"
            lines.append(
                f"  [{status}] {stage:<20} {elapsed:8.1f}ms cumulative "
                f"(+{deltas[stage]:7.1f}ms in stage, budget {threshold:.0f}ms)"
            )
        for stage in self.missing_stages():
            lines.append(f"  [MISS] {stage:<20} never marked")
        return "\n".join(lines)

"""A simulated caller: scripted personas, synthesized speech, real barge-in timing.

Manual dialing does not scale and does not regress. This module replaces the
human on the other end of the line with a scripted persona: each turn is
synthesized through a :class:`~voice_testing.transcribe_eval.TTSProvider` and
injected into an :class:`AgentAudioChannel`, and every turn is timed.

The interesting measurement is barge-in. A caller who interrupts is the single
most common way a voice agent falls apart, and the number that matters is how
long the agent keeps talking after the caller starts. Because only the audio
channel can observe when the agent's output actually stopped (in production, RTP
packet arrival), the channel reports timings and this module aggregates them.
That keeps wall-clock measurement where it belongs and makes the harness fully
deterministic under a mock.

Run the bundled demo::

    python voice_testing/simulated_caller.py                 # one persona, verbose
    python voice_testing/simulated_caller.py --persona all --repeat 20
"""

from __future__ import annotations

import argparse
import json
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, List, Optional, Protocol, Sequence

if __package__ in (None, ""):  # pragma: no cover - only when run as a file path
    # Executed as `python voice_testing/simulated_caller.py`, which puts this
    # file's directory on sys.path instead of the repo root. Add the repo root so
    # the absolute package imports below resolve either way.
    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from voice_testing.assertions import run_n_times
from voice_testing.transcribe_eval import ASRProvider, AudioClip, TTSProvider

#: Milliseconds the agent may take to emit its first audio byte of a reply.
DEFAULT_FIRST_BYTE_BUDGET_MS = 800.0

#: Milliseconds the agent may keep talking after the caller starts speaking.
#: Past roughly half a second, humans perceive the agent as talking over them.
DEFAULT_BARGE_IN_BUDGET_MS = 500.0


# ---------------------------------------------------------------------------
# Persona
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class CallerPersona:
    """A scripted caller: what they say, and whether they interrupt.

    ``barge_in_turn`` is the 0-based index of the turn the caller delivers while
    the agent is still speaking.
    """

    name: str
    turns: Sequence[str]
    description: str = ""
    barge_in_turn: Optional[int] = None

    def __post_init__(self) -> None:
        if not self.name:
            raise ValueError("persona name must not be empty")
        if not self.turns:
            raise ValueError(f"persona {self.name!r} has no turns")
        if self.barge_in_turn is not None:
            if not 0 <= self.barge_in_turn < len(self.turns):
                raise ValueError(
                    f"persona {self.name!r} sets barge_in_turn="
                    f"{self.barge_in_turn}, which is outside 0..{len(self.turns) - 1}"
                )

    def barges_in_on(self, index: int) -> bool:
        return self.barge_in_turn is not None and self.barge_in_turn == index


# ---------------------------------------------------------------------------
# Channel interface
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class AgentResponse:
    """What the channel observed after caller audio was injected.

    Attributes:
        clip: the agent's reply audio.
        first_byte_ms: from end of caller audio to the agent's first audio byte.
        complete_ms: from end of caller audio to the end of the agent's reply.
        interrupted_after_ms: for a barge-in turn, how long the agent kept
            talking after the caller started. ``None`` when the caller did not
            interrupt on this turn.
    """

    clip: AudioClip
    first_byte_ms: float
    complete_ms: float
    interrupted_after_ms: Optional[float] = None

    def __post_init__(self) -> None:
        if self.first_byte_ms < 0 or self.complete_ms < 0:
            raise ValueError("response timings must be >= 0")
        if self.complete_ms < self.first_byte_ms:
            raise ValueError(
                f"complete_ms ({self.complete_ms}) cannot precede first_byte_ms "
                f"({self.first_byte_ms})"
            )


class AgentAudioChannel(Protocol):
    """The transport between the simulated caller and the agent.

    In production this wraps a SIP/WebRTC leg. In CI it is a mock. Either way it
    owns timing, because only the transport can see when audio actually moved.
    """

    def open(self) -> None:  # pragma: no cover - interface
        ...

    def speak(
        self, clip: AudioClip, *, barge_in: bool = False
    ) -> AgentResponse:  # pragma: no cover - interface
        ...

    def close(self) -> None:  # pragma: no cover - interface
        ...


# ---------------------------------------------------------------------------
# Results
# ---------------------------------------------------------------------------


@dataclass
class TurnTiming:
    """One caller turn and everything measured about the agent's answer."""

    index: int
    caller_text: str
    agent_text: str
    caller_audio_ms: float
    agent_first_byte_ms: float
    agent_complete_ms: float
    barge_in: bool = False
    barge_in_response_ms: Optional[float] = None

    @property
    def number(self) -> int:
        return self.index + 1


class CallHealthError(AssertionError):
    """Raised by :func:`assert_call_healthy`."""


@dataclass
class SimulatedCallResult:
    """Transcript plus timings for one simulated call."""

    persona: str
    turns: List[TurnTiming] = field(default_factory=list)

    def __len__(self) -> int:
        return len(self.turns)

    def __iter__(self):
        return iter(self.turns)

    @property
    def barge_in_turns(self) -> List[TurnTiming]:
        return [t for t in self.turns if t.barge_in]

    @property
    def barge_in_response_ms(self) -> Optional[float]:
        """Worst barge-in response across the call, or ``None`` if none occurred."""
        values = [
            t.barge_in_response_ms
            for t in self.barge_in_turns
            if t.barge_in_response_ms is not None
        ]
        return max(values) if values else None

    @property
    def max_first_byte_ms(self) -> float:
        return max((t.agent_first_byte_ms for t in self.turns), default=0.0)

    def as_dict(self) -> Dict[str, object]:
        return {
            "persona": self.persona,
            "max_first_byte_ms": round(self.max_first_byte_ms, 1),
            "barge_in_response_ms": (
                round(self.barge_in_response_ms, 1)
                if self.barge_in_response_ms is not None
                else None
            ),
            "turns": [
                {
                    "turn": t.number,
                    "caller": t.caller_text,
                    "agent": t.agent_text,
                    "first_byte_ms": round(t.agent_first_byte_ms, 1),
                    "complete_ms": round(t.agent_complete_ms, 1),
                    "barge_in": t.barge_in,
                    "barge_in_response_ms": (
                        round(t.barge_in_response_ms, 1)
                        if t.barge_in_response_ms is not None
                        else None
                    ),
                }
                for t in self.turns
            ],
        }

    def transcript_report(self) -> str:
        lines = [f"call transcript: persona={self.persona}"]
        for turn in self.turns:
            flag = " [caller barges in]" if turn.barge_in else ""
            lines.append(f"  {turn.number}. caller{flag}: {turn.caller_text}")
            lines.append(f"     agent           : {turn.agent_text}")
        return "\n".join(lines)

    def timing_report(self) -> str:
        lines = [
            f"call timing: persona={self.persona}",
            f"  {'turn':<5}{'caller audio':>14}{'first byte':>13}"
            f"{'reply end':>12}{'barge-in':>11}",
        ]
        for turn in self.turns:
            barge = (
                f"{turn.barge_in_response_ms:.0f}ms"
                if turn.barge_in_response_ms is not None
                else "-"
            )
            lines.append(
                f"  {turn.number:<5}{turn.caller_audio_ms:>12.0f}ms"
                f"{turn.agent_first_byte_ms:>11.0f}ms"
                f"{turn.agent_complete_ms:>10.0f}ms{barge:>11}"
            )
        return "\n".join(lines)


def assert_call_healthy(
    result: SimulatedCallResult,
    *,
    first_byte_budget_ms: float = DEFAULT_FIRST_BYTE_BUDGET_MS,
    barge_in_budget_ms: float = DEFAULT_BARGE_IN_BUDGET_MS,
    require_barge_in: bool = False,
) -> SimulatedCallResult:
    """Assert every turn answered fast enough and every interruption was honored."""
    problems: List[str] = []
    for turn in result.turns:
        if turn.agent_first_byte_ms > first_byte_budget_ms:
            problems.append(
                f"turn {turn.number}: first audio byte after "
                f"{turn.agent_first_byte_ms:.0f}ms "
                f"(budget {first_byte_budget_ms:.0f}ms)"
            )
        if not turn.agent_text.strip():
            problems.append(f"turn {turn.number}: agent said nothing")
        if turn.barge_in:
            if turn.barge_in_response_ms is None:
                problems.append(
                    f"turn {turn.number}: caller barged in but the channel "
                    "reported no interruption, so the agent never yielded"
                )
            elif turn.barge_in_response_ms > barge_in_budget_ms:
                problems.append(
                    f"turn {turn.number}: agent kept talking for "
                    f"{turn.barge_in_response_ms:.0f}ms after the caller "
                    f"interrupted (budget {barge_in_budget_ms:.0f}ms)"
                )
    if require_barge_in and not result.barge_in_turns:
        problems.append("no barge-in turn was exercised in this call")
    if problems:
        raise CallHealthError(
            f"simulated call failed for persona {result.persona!r}:\n"
            + "\n".join(f"  - {p}" for p in problems)
        )
    return result


# ---------------------------------------------------------------------------
# The caller
# ---------------------------------------------------------------------------


class SimulatedCaller:
    """Drives a :class:`CallerPersona` through an :class:`AgentAudioChannel`."""

    def __init__(
        self,
        persona: CallerPersona,
        tts: TTSProvider,
        channel: AgentAudioChannel,
        asr: ASRProvider,
    ) -> None:
        self.persona = persona
        self.tts = tts
        self.channel = channel
        self.asr = asr

    def run(self) -> SimulatedCallResult:
        """Place one full scripted call and return its transcript and timings."""
        result = SimulatedCallResult(persona=self.persona.name)
        self.channel.open()
        try:
            for index, text in enumerate(self.persona.turns):
                barge_in = self.persona.barges_in_on(index)
                caller_clip = self.tts.synthesize(text)
                response = self.channel.speak(caller_clip, barge_in=barge_in)
                if not isinstance(response, AgentResponse):
                    raise TypeError(
                        f"{type(self.channel).__name__}.speak() must return an "
                        f"AgentResponse, got {type(response).__name__}"
                    )
                result.turns.append(
                    TurnTiming(
                        index=index,
                        caller_text=text,
                        agent_text=self.asr.transcribe(response.clip),
                        caller_audio_ms=caller_clip.duration_ms,
                        agent_first_byte_ms=response.first_byte_ms,
                        agent_complete_ms=response.complete_ms,
                        barge_in=barge_in,
                        barge_in_response_ms=response.interrupted_after_ms,
                    )
                )
        finally:
            self.channel.close()
        return result


# ---------------------------------------------------------------------------
# CLI demo / nightly sweep
# ---------------------------------------------------------------------------


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="simulated_caller",
        description=(
            "Run scripted personas against the bundled mock agent. With "
            "--repeat > 1 this is the nightly sweep: every persona is replayed "
            "N times and the run fails if the pass rate drops below the "
            "threshold."
        ),
    )
    parser.add_argument(
        "--persona",
        default="reschedule",
        help="persona name, or 'all' for every bundled persona",
    )
    parser.add_argument(
        "--repeat", type=int, default=1, help="repetitions per persona (default: 1)"
    )
    parser.add_argument(
        "--pass-threshold",
        type=float,
        default=0.95,
        help="minimum fraction of repetitions that must pass (default: 0.95)",
    )
    parser.add_argument(
        "--first-byte-budget-ms",
        type=float,
        default=DEFAULT_FIRST_BYTE_BUDGET_MS,
        help=f"time-to-first-audio budget (default: {DEFAULT_FIRST_BYTE_BUDGET_MS:.0f})",
    )
    parser.add_argument(
        "--barge-in-budget-ms",
        type=float,
        default=DEFAULT_BARGE_IN_BUDGET_MS,
        help=f"barge-in response budget (default: {DEFAULT_BARGE_IN_BUDGET_MS:.0f})",
    )
    parser.add_argument("--json", action="store_true", help="emit JSON on stdout")
    parser.add_argument(
        "--list", action="store_true", help="list bundled personas and exit"
    )
    return parser


def main(argv: Optional[Sequence[str]] = None) -> int:
    # Imported here rather than at module scope: the mocks implement the
    # interfaces defined above, so a top-level import would be circular.
    from voice_testing import mocks

    args = build_parser().parse_args(argv)

    if args.list:
        for name, scenario in mocks.SCENARIOS.items():
            print(f"{name:<16} {scenario.persona.description}")
        return 0
    if args.repeat < 1:
        raise SystemExit(f"error: --repeat must be >= 1, got {args.repeat}")

    if args.persona == "all":
        names = list(mocks.SCENARIOS)
    elif args.persona in mocks.SCENARIOS:
        names = [args.persona]
    else:
        raise SystemExit(
            f"error: unknown persona {args.persona!r}; available: "
            f"{', '.join(mocks.SCENARIOS)} (or 'all')"
        )

    summaries: List[Dict[str, object]] = []
    exit_code = 0

    for name in names:
        scenario = mocks.SCENARIOS[name]

        def one_call(attempt: int, scenario=scenario) -> bool:
            result = scenario.build_caller(attempt=attempt).run()
            assert_call_healthy(
                result,
                first_byte_budget_ms=args.first_byte_budget_ms,
                barge_in_budget_ms=args.barge_in_budget_ms,
            )
            return True

        repeat_result = run_n_times(one_call, n=args.repeat, label=f"persona:{name}")
        first_call = scenario.build_caller(attempt=0).run()

        if not args.json:
            print(first_call.transcript_report())
            print()
            print(first_call.timing_report())
            print()
            print(repeat_result.report())
            for failure in repeat_result.failures:
                print(f"  attempt {failure.index + 1} failed: {failure.detail}")
            print()

        summaries.append(
            {
                "persona": name,
                "pass_rate": round(repeat_result.pass_rate, 4),
                "passes": repeat_result.passes,
                "attempts": repeat_result.total,
                "first_call": first_call.as_dict(),
            }
        )
        if repeat_result.pass_rate < args.pass_threshold:
            exit_code = 1

    if args.json:
        print(
            json.dumps(
                {
                    "pass_threshold": args.pass_threshold,
                    "personas": summaries,
                    "passed": exit_code == 0,
                },
                indent=2,
            )
        )
    else:
        verdict = "PASS" if exit_code == 0 else "FAIL"
        print(f"{verdict}: pass-rate threshold {args.pass_threshold:.0%}")
    return exit_code


if __name__ == "__main__":
    sys.exit(main())

"""The suite a voice agent actually needs, running entirely offline.

Four things are tested here, and they map one-to-one onto the four ways a voice
agent fails in production:

1. it mishears the caller            -> word error rate against a reference
2. it answers too slowly            -> per-stage latency budget
3. it answers the wrong thing       -> semantic similarity + LLM-as-judge
4. it only fails sometimes          -> run-N-times pass-rate gate

Everything runs against the bundled mocks, so::

    pytest tests/test_voice_agent.py -v

passes with no API key, no network and no telephony account. Swap the mocks for
real providers and the assertions do not change.
"""

from __future__ import annotations

import pytest

from voice_testing.assertions import (
    HashingEmbedder,
    PassRateTooLow,
    Rubric,
    SemanticMismatch,
    assert_judge_passes,
    assert_semantic_match,
    llm_judge_verdict,
    run_n_times,
    semantic_similarity,
)
from voice_testing.latency import (
    DEFAULT_BUDGET_MS,
    BudgetExceeded,
    LatencyBudget,
    MissingStageMarks,
)
from voice_testing.mocks import (
    NOISY_LINE,
    RESCHEDULE,
    SCENARIOS,
    MockTTSProvider,
    RuleBasedJudge,
    ScriptedAgentAdapter,
    StubASRProvider,
    sample_fixture,
    scripted_clock,
)
from voice_testing.simulated_caller import (
    CallHealthError,
    SimulatedCallResult,
    TurnTiming,
    assert_call_healthy,
)
from voice_testing.transcribe_eval import (
    CallFixture,
    CallerTurn,
    TurnTranscript,
    run_call_fixture,
)
from voice_testing.wer import WERTooHigh, assert_wer_within, measure

# The one place ASR mangles the agent's confirmation in these tests: "Tuesday"
# comes back as "Tuesdays". One substitution in nineteen words.
CONFIRMATION_MISHEARD = {
    "Done. Your appointment is confirmed for Tuesday at three PM. "
    "Is there anything else I can help you with?": (
        "done your appointment is confirmed for tuesdays at 3 pm "
        "is there anything else i can help you with"
    )
}


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------


@pytest.fixture
def tts() -> MockTTSProvider:
    return MockTTSProvider()


@pytest.fixture
def asr() -> StubASRProvider:
    """A stub recognizer that mishears one word of the agent's confirmation."""
    return StubASRProvider(degradations=CONFIRMATION_MISHEARD)


@pytest.fixture
def call_fixture() -> CallFixture:
    return sample_fixture("appointment_reschedule")


@pytest.fixture
def agent(call_fixture: CallFixture, tts: MockTTSProvider) -> ScriptedAgentAdapter:
    """A scripted agent that replies exactly as the fixture's reference expects."""
    return ScriptedAgentAdapter(call_fixture.reference_transcript, tts=tts)


@pytest.fixture
def embedder() -> HashingEmbedder:
    return HashingEmbedder()


@pytest.fixture
def judge() -> RuleBasedJudge:
    return RuleBasedJudge()


@pytest.fixture
def transcript(
    call_fixture: CallFixture,
    agent: ScriptedAgentAdapter,
    asr: StubASRProvider,
    tts: MockTTSProvider,
) -> TurnTranscript:
    """One full call, transcribed turn by turn."""
    return run_call_fixture(call_fixture, agent, asr, tts=tts)


# ---------------------------------------------------------------------------
# The harness itself
# ---------------------------------------------------------------------------


def test_fixture_loads_with_a_reference_per_turn(call_fixture: CallFixture) -> None:
    assert call_fixture.id == "appointment_reschedule"
    assert len(call_fixture.caller_turns) == len(call_fixture.reference_transcript)
    assert [t.barge_in for t in call_fixture.caller_turns] == [False, True, False]


def test_run_call_fixture_returns_structured_turns(transcript: TurnTranscript) -> None:
    assert len(transcript) == 3
    assert transcript.caller_texts()[0].startswith("Hi, I need to move")
    # Turn-level structure is the whole point: one record per exchange, with the
    # caller side and the agent side kept apart.
    for turn in transcript:
        assert turn.caller_text
        assert turn.agent_text
        assert turn.reference_agent_text
        assert turn.caller_audio_ms > 0
        assert turn.agent_audio_ms > 0
    assert [t.number for t in transcript.barge_in_turns()] == [2]


def test_fixture_rejects_a_mismatched_reference() -> None:
    with pytest.raises(ValueError, match="exactly one expected agent reply"):
        CallFixture(
            id="broken",
            description="two caller turns, one reference reply",
            caller_turns=[CallerTurn(text="one"), CallerTurn(text="two")],
            reference_transcript=["only one"],
        )


# ---------------------------------------------------------------------------
# 1. Did it hear the caller? Word error rate.
# ---------------------------------------------------------------------------


def test_wer_passes_when_asr_is_close_enough(transcript: TurnTranscript) -> None:
    confirmation = transcript[1]
    result = assert_wer_within(
        confirmation.reference_agent_text,
        confirmation.agent_text,
        threshold=0.10,
        context="turn 2 confirmation",
    )
    assert result.substitutions == 1
    assert result.deletions == 0
    assert result.insertions == 0
    assert 0.0 < result.wer <= 0.10


def test_wer_normalizes_punctuation_casing_and_number_words() -> None:
    # "three PM" and "3 pm" are the same utterance. Only a broken harness calls
    # that an error.
    result = measure(
        "Your appointment is confirmed for Tuesday at three PM.",
        "your appointment is confirmed for tuesday at 3 pm",
    )
    assert result.wer == 0.0
    assert result.errors == 0


def test_wer_fails_a_real_mishearing() -> None:
    reference = "Order four four seven two ships tomorrow morning."
    misheard = NOISY_LINE.asr_degradations[reference]
    with pytest.raises(WERTooHigh) as excinfo:
        assert_wer_within(reference, misheard, threshold=0.10)
    assert "WER 12.50%" in str(excinfo.value)


def test_wer_backends_agree() -> None:
    """jiwer and the standard-library fallback must produce identical counts."""
    reference = "Done your appointment is confirmed for Tuesday at three PM"
    hypothesis = "done your appointment is confirmed for tuesdays at 3 pm please"
    with_backend = measure(reference, hypothesis)
    pure_python = measure(reference, hypothesis, force_pure_python=True)
    assert with_backend.wer == pytest.approx(pure_python.wer)
    assert with_backend.substitutions == pure_python.substitutions
    assert with_backend.deletions == pure_python.deletions
    assert with_backend.insertions == pure_python.insertions


# ---------------------------------------------------------------------------
# 2. Did it answer fast enough? Per-stage latency budget.
# ---------------------------------------------------------------------------


def test_latency_budget_passes_when_every_stage_is_inside_its_own_threshold() -> None:
    # start, then one clock read per mark, in chronological order.
    budget = LatencyBudget(
        clock=scripted_clock([0.0, 0.220, 0.450, 0.620, 1.050]),
        label="turn 1",
    ).start()
    budget.mark("asr_finalize")
    budget.mark("first_token")
    budget.mark("tts_first_byte")
    budget.mark("generation_complete")

    budget.assert_within_budget()
    assert budget.elapsed_ms("asr_finalize") == pytest.approx(220.0)
    assert budget.elapsed_ms("generation_complete") == pytest.approx(1050.0)
    # Stage deltas isolate which component owns the time.
    assert dict(budget.stage_deltas_ms())["first_token"] == pytest.approx(230.0)


def test_latency_budget_names_the_stage_that_blew_the_budget() -> None:
    budget = LatencyBudget(
        clock=scripted_clock([0.0, 0.250, 0.500, 0.980, 1.300]),
        label="turn 3",
    ).start()
    for stage in ("asr_finalize", "first_token", "tts_first_byte", "generation_complete"):
        budget.mark(stage)

    with pytest.raises(BudgetExceeded) as excinfo:
        budget.assert_within_budget()
    message = str(excinfo.value)
    assert "tts_first_byte" in message
    assert "asr_finalize" not in message
    assert [v.stage for v in excinfo.value.violations] == ["tts_first_byte"]


def test_latency_budget_catches_a_stage_that_never_fired() -> None:
    budget = LatencyBudget(clock=scripted_clock([0.0, 0.100])).start()
    budget.mark("asr_finalize")
    with pytest.raises(MissingStageMarks, match="first_token"):
        budget.assert_within_budget()


def test_agent_turns_report_latency_per_stage(transcript: TurnTranscript) -> None:
    for turn in transcript:
        assert turn.latency is not None
        assert set(turn.latency.marks_ms) == set(DEFAULT_BUDGET_MS)
        turn.latency.assert_within_budget()


# ---------------------------------------------------------------------------
# 3. Did it answer the right thing? Semantics, not string equality.
# ---------------------------------------------------------------------------


def test_semantic_similarity_prefers_a_paraphrase_over_a_wrong_answer(
    embedder: HashingEmbedder,
) -> None:
    reference = "Your appointment is confirmed for Tuesday at three PM."
    paraphrase = "I have confirmed your appointment for Tuesday at 3 PM."
    wrong = "I cannot find any appointment under that phone number."

    paraphrase_score = semantic_similarity(paraphrase, reference, embedder=embedder)
    wrong_score = semantic_similarity(wrong, reference, embedder=embedder)

    assert paraphrase_score > wrong_score
    assert paraphrase_score >= 0.55
    assert wrong_score < 0.35


def test_semantic_assertion_accepts_the_agents_wording(
    transcript: TurnTranscript, embedder: HashingEmbedder
) -> None:
    confirmation = transcript[1]
    # Exact string equality would fail here: ASR lowercased it, dropped the
    # punctuation and wrote "3 pm". The meaning survived.
    assert confirmation.agent_text != confirmation.reference_agent_text
    assert_semantic_match(
        confirmation.agent_text,
        confirmation.reference_agent_text,
        threshold=0.75,
        embedder=embedder,
        context="turn 2 confirmation",
    )


def test_semantic_assertion_rejects_the_wrong_answer(
    embedder: HashingEmbedder,
) -> None:
    with pytest.raises(SemanticMismatch, match="semantic similarity"):
        assert_semantic_match(
            "Sorry, I have no record of your booking.",
            "Your appointment is confirmed for Tuesday at three PM.",
            threshold=0.75,
            embedder=embedder,
        )


def test_llm_judge_grades_the_transcript_against_a_rubric(
    transcript: TurnTranscript, judge: RuleBasedJudge
) -> None:
    rubric = Rubric(
        name="reschedule-confirmed",
        instructions=(
            "The agent must confirm the new appointment slot and must not claim "
            "the original slot is still booked."
        ),
        must_mention=["tuesday", "3 pm", "confirmed"],
        must_not_mention=["thursday is still"],
    )
    verdict = assert_judge_passes(
        judge, rubric, transcript[1].agent_text, min_score=1.0
    )
    assert verdict.passed
    assert verdict.score == 1.0


def test_llm_judge_fails_a_reply_that_omits_the_new_slot(
    judge: RuleBasedJudge,
) -> None:
    rubric = Rubric(
        name="reschedule-confirmed",
        must_mention=["tuesday", "3 pm"],
    )
    verdict = llm_judge_verdict(judge, rubric, "Okay, all set. Anything else?")
    assert not verdict.passed
    assert "tuesday" in verdict.reason


def test_a_malformed_judge_is_a_harness_bug_not_a_test_failure() -> None:
    class BrokenJudge:
        def verdict(self, rubric: Rubric, transcript_text: str) -> str:
            return "looks fine to me"

    with pytest.raises(TypeError, match="must return a JudgeVerdict"):
        llm_judge_verdict(BrokenJudge(), Rubric(name="anything"), "some text")


# ---------------------------------------------------------------------------
# 4. Does it answer the right thing *reliably*? The pass-rate gate.
# ---------------------------------------------------------------------------


def test_run_n_times_gates_on_a_pass_rate() -> None:
    def scenario(attempt: int) -> bool:
        # Deterministic stand-in for a stochastic agent: 4 of 5 attempts pass.
        return attempt != 2

    result = run_n_times(scenario, n=5, label="reschedule")
    assert result.pass_rate == pytest.approx(0.8)
    result.assert_pass_rate(0.8)

    with pytest.raises(PassRateTooLow, match="80%"):
        result.assert_pass_rate(0.9)


def test_run_n_times_records_assertion_failures_without_aborting() -> None:
    def scenario(attempt: int) -> bool:
        if attempt == 1:
            raise AssertionError("agent hung up early")
        return True

    result = run_n_times(scenario, n=3, label="hangup")
    assert result.total == 3
    assert [a.detail for a in result.failures] == ["agent hung up early"]


def test_run_n_times_lets_harness_errors_through() -> None:
    def scenario(attempt: int) -> bool:
        raise RuntimeError("telephony credentials missing")

    with pytest.raises(RuntimeError, match="telephony credentials missing"):
        run_n_times(scenario, n=3)


def test_repeated_calls_pass_the_semantic_and_latency_gates_together(
    call_fixture: CallFixture, embedder: HashingEmbedder, judge: RuleBasedJudge
) -> None:
    """The end-to-end shape of a real voice test: repeat, then gate."""
    rubric = Rubric(name="reschedule-confirmed", must_mention=["tuesday", "3 pm"])

    def one_call(attempt: int) -> bool:
        shared_tts = MockTTSProvider()
        result = run_call_fixture(
            call_fixture,
            ScriptedAgentAdapter(call_fixture.reference_transcript, tts=shared_tts),
            StubASRProvider(degradations=CONFIRMATION_MISHEARD),
            tts=shared_tts,
        )
        confirmation = result[1]
        assert_wer_within(
            confirmation.reference_agent_text, confirmation.agent_text, threshold=0.10
        )
        assert_semantic_match(
            confirmation.agent_text,
            confirmation.reference_agent_text,
            threshold=0.75,
            embedder=embedder,
        )
        assert_judge_passes(judge, rubric, confirmation.agent_text)
        for turn in result:
            assert turn.latency is not None
            turn.latency.assert_within_budget()
        return True

    run_n_times(one_call, n=5, label="reschedule end to end").assert_pass_rate(1.0)


# ---------------------------------------------------------------------------
# The simulated caller, including barge-in
# ---------------------------------------------------------------------------


def test_simulated_caller_records_barge_in_response_time() -> None:
    result = RESCHEDULE.build_caller(attempt=0).run()

    assert len(result) == 3
    assert [t.number for t in result.barge_in_turns] == [2]
    assert result.barge_in_response_ms is not None
    # The agent stopped talking well inside the half-second humans notice.
    assert result.barge_in_response_ms < 500.0
    assert_call_healthy(result, require_barge_in=True)


def test_simulated_caller_is_reproducible_for_a_given_attempt() -> None:
    first = RESCHEDULE.build_caller(attempt=3).run().as_dict()
    second = RESCHEDULE.build_caller(attempt=3).run().as_dict()
    assert first == second


def test_an_agent_that_talks_over_the_caller_fails() -> None:
    result = SimulatedCallResult(
        persona="stubborn",
        turns=[
            TurnTiming(
                index=0,
                caller_text="Actually, make it Tuesday.",
                agent_text="...and your Thursday slot is confirmed at four PM...",
                caller_audio_ms=1200.0,
                agent_first_byte_ms=400.0,
                agent_complete_ms=2600.0,
                barge_in=True,
                barge_in_response_ms=1400.0,
            )
        ],
    )
    with pytest.raises(CallHealthError, match="kept talking for 1400ms"):
        assert_call_healthy(result)


def test_a_barge_in_the_agent_ignored_entirely_fails() -> None:
    result = SimulatedCallResult(
        persona="deaf",
        turns=[
            TurnTiming(
                index=0,
                caller_text="Wait, stop.",
                agent_text="Your total is forty two dollars.",
                caller_audio_ms=800.0,
                agent_first_byte_ms=300.0,
                agent_complete_ms=1900.0,
                barge_in=True,
                barge_in_response_ms=None,
            )
        ],
    )
    with pytest.raises(CallHealthError, match="never yielded"):
        assert_call_healthy(result)


@pytest.mark.parametrize("persona_name", sorted(SCENARIOS))
def test_every_bundled_persona_completes_a_healthy_call(persona_name: str) -> None:
    result = SCENARIOS[persona_name].build_caller(attempt=0).run()
    assert len(result) == len(SCENARIOS[persona_name].persona.turns)
    assert_call_healthy(result)


@pytest.mark.nightly
@pytest.mark.parametrize("persona_name", sorted(SCENARIOS))
def test_persona_sweep_holds_up_over_many_repetitions(persona_name: str) -> None:
    """The nightly job: same personas, many repetitions, gated on pass rate."""
    scenario = SCENARIOS[persona_name]

    def one_call(attempt: int) -> bool:
        assert_call_healthy(scenario.build_caller(attempt=attempt).run())
        return True

    run_n_times(one_call, n=20, label=f"persona:{persona_name}").assert_pass_rate(0.95)