STT-Assistant/tests/test_enroll_assist.py

"""Tests for seed-based assisted speaker enrollment.

Hardware- and network-free: WAVs are synthesized with stdlib ``wave`` (a DC
level encodes the "speaker id"), the embedder is a fake, reviews are scripted,
and playback is a recording stub. The real sherpa-onnx package, questionary,
and audio players are never required.
"""

import argparse
import io
import shutil
import subprocess
import sys
import wave
from contextlib import redirect_stderr, redirect_stdout
from pathlib import Path

import numpy as np
import pytest

import stt_daemon.cli as cli_module
import stt_daemon.enroll_assist as ea
from stt_daemon.config import load_config
from stt_daemon.enroll import enroll_from_files
from stt_daemon.enroll_assist import (
    Candidate,
    EnrollmentAborted,
    QuestionaryReview,
    play_wav,
    rank_candidates,
    review_candidates,
    seed_enroll,
)
from stt_daemon.speakers import SpeakerStore

A = 0.30
B = -0.30


def write_const_wav(path: Path, seconds: float, value: float,
                    sample_rate: int = 16000, nchannels: int = 1) -> Path:
    """Write a constant-level int16 PCM WAV; the level encodes the speaker."""
    count = int(seconds * sample_rate)
    samples = np.full(count, value, dtype=np.float32)
    frames = (samples * 32767.0).astype(np.int16)
    if nchannels == 2:
        frames = np.stack([frames, frames], axis=1).reshape(-1)
    with wave.open(str(path), "wb") as handle:
        handle.setnchannels(nchannels)
        handle.setsampwidth(2)
        handle.setframerate(sample_rate)
        handle.writeframes(frames.tobytes())
    return path


class ConstEmbedder:
    """Deterministic embedding from the clip DC level (the fake speaker id)."""

    def __init__(self, dim: int = 2) -> None:
        self.dim = dim
        self.calls = 0

    def embed(self, audio: np.ndarray) -> np.ndarray:
        self.calls += 1
        mean = float(np.mean(np.asarray(audio, dtype=np.float32)))
        if self.dim == 2:
            if mean > 0.15:
                vector = [1.0, 0.0]
            elif mean < -0.15:
                vector = [0.0, 1.0]
            else:
                vector = [0.7071, 0.7071]
        else:
            vector = [1.0, 0.0, 0.0] if mean > 0.15 else [0.0, 1.0, 0.0]
        return np.array(vector, dtype=np.float32)


class VariedEmbedder:
    """Same-speaker clips with distinct-but-similar embeddings.

    Rotates the A-speaker vector 20 degrees per call, so same-speaker clips
    score high (neighbors at cosine 0.94: above ``min_score``, below the
    ``dup_threshold`` duplicate floor) instead of collapsing to exact
    duplicates the way ConstEmbedder does. B-speaker clips map opposite the
    seed so they are always filtered by ``min_score``. Tracks ``calls`` like
    ConstEmbedder.
    """

    def __init__(self, dim: int = 2) -> None:
        self.dim = dim
        self.calls = 0

    def embed(self, audio: np.ndarray) -> np.ndarray:
        self.calls += 1
        mean = float(np.mean(np.asarray(audio, dtype=np.float32)))
        if mean < -0.15:
            return np.array([-1.0, 0.0], dtype=np.float32)
        angle = np.deg2rad(20.0 * self.calls)
        vector = np.array([np.cos(angle), np.sin(angle)], dtype=np.float32)
        return vector / np.linalg.norm(vector)


class ScriptReview:
    """Scripted ClipReview; records every choose() call."""

    def __init__(self, actions) -> None:
        self.actions = list(actions)
        self.seen: list = []

    def choose(self, candidate, accepted, target) -> str:
        self.seen.append((candidate, accepted, target))
        if not self.actions:
            return "skip"
        return self.actions.pop(0)


def write_cfg(tmp_path: Path, *, min_enroll_s: float = 2.0,
              max_prototypes: int = 8, model_ok: bool = True):
    """Write a temp config; return (cfg, audio_dir, store_dir, model)."""
    model = tmp_path / "model.onnx"
    if model_ok:
        model.write_bytes(b"fake-model")
    session_dir = tmp_path / "sessions"
    audio_dir = session_dir / "2026-09-30-12-00-00" / "audio"
    audio_dir.mkdir(parents=True, exist_ok=True)
    store_dir = tmp_path / "speakers"
    cfg_path = tmp_path / "config.toml"
    cfg_path.write_text(
        "[speaker]\n"
        f'model_path = "{model}"\n'
        f"min_enroll_s = {min_enroll_s}\n"
        f"max_prototypes = {max_prototypes}\n"
        f'store_dir = "{store_dir}"\n'
        "[output]\n"
        f'session_dir = "{session_dir}"\n'
        "[events]\n"
        f'socket = "{tmp_path / "bus.sock"}"\n',
        encoding="utf-8",
    )
    return load_config(str(cfg_path)), audio_dir, store_dir


def make_store(cfg) -> SpeakerStore:
    """Build a SpeakerStore honoring cfg (mirrors the CLI handler)."""
    speaker_cfg = cfg.speaker
    return SpeakerStore(
        Path(speaker_cfg.store_dir),
        threshold_id=speaker_cfg.threshold_id,
        threshold_new=speaker_cfg.threshold_new,
        min_enroll_s=speaker_cfg.min_enroll_s,
        rms_floor=speaker_cfg.rms_floor,
        margin=speaker_cfg.margin,
        mint_cooldown_s=speaker_cfg.mint_cooldown_s,
        confirm_s=speaker_cfg.confirm_s,
        max_prototypes=speaker_cfg.max_prototypes,
        provisional_ttl_h=speaker_cfg.provisional_ttl_h,
    )


def record_enroll(monkeypatch):
    """Wrap ea.enroll_from_files to record enrolled paths; return the log."""
    called: list = []
    real = ea.enroll_from_files

    def fake(store, embedder, name, paths, min_s):
        called.append(list(paths))
        return real(store, embedder, name, paths, min_s)

    monkeypatch.setattr(ea, "enroll_from_files", fake)
    return called


def store_files(store_dir: Path) -> bool:
    return (store_dir / "speakers.json").exists()


def fresh_speakers(store_dir: Path) -> dict:
    store = SpeakerStore(store_dir)
    store.load()
    return dict(store.speakers)


# -- rank_candidates ------------------------------------------------------------


def test_rank_orders_by_score_and_counts_skipped(tmp_path):
    seed = write_const_wav(tmp_path / "seed.wav", 2.5, A)
    good_a = write_const_wav(tmp_path / "a.wav", 2.5, A)
    good_b = write_const_wav(tmp_path / "b.wav", 2.5, B)
    (tmp_path / "notes.wav").write_text("not audio", encoding="utf-8")
    write_const_wav(tmp_path / "stereo.wav", 2.5, A, nchannels=2)
    write_const_wav(tmp_path / "lowrate.wav", 2.5, A, sample_rate=8000)
    from stt_daemon.enroll import read_wav
    seed_audio, _ = read_wav(seed)

    ranked, skipped = rank_candidates(
        ConstEmbedder(), seed_audio,
        [good_a, good_b, tmp_path / "notes.wav", tmp_path / "stereo.wav",
         tmp_path / "lowrate.wav", tmp_path / "missing.wav"],
        min_enroll_s=2.0, min_score=0.55)

    assert [c.path for c in ranked] == [good_a]
    assert skipped == 4
    assert ranked[0].score == pytest.approx(1.0)
    assert ranked[0].duration_s == pytest.approx(2.5)


def test_rank_excludes_seed_and_short_clips(tmp_path):
    seed = write_const_wav(tmp_path / "seed.wav", 2.5, A)
    short = write_const_wav(tmp_path / "short.wav", 0.5, A)
    long_a = write_const_wav(tmp_path / "long.wav", 2.5, A)
    from stt_daemon.enroll import read_wav
    seed_audio, _ = read_wav(seed)

    ranked, _ = rank_candidates(
        ConstEmbedder(), seed_audio, [seed, short, long_a],
        min_enroll_s=2.0, min_score=0.55, exclude=(seed.resolve(),))

    assert [c.path for c in ranked] == [long_a]


def test_rank_min_score_floor(tmp_path):
    seed = write_const_wav(tmp_path / "seed.wav", 2.5, A)
    mid = write_const_wav(tmp_path / "mid.wav", 2.5, 0.0)
    other = write_const_wav(tmp_path / "other.wav", 2.5, B)
    from stt_daemon.enroll import read_wav
    seed_audio, _ = read_wav(seed)

    ranked, _ = rank_candidates(
        ConstEmbedder(), seed_audio, [other, mid],
        min_enroll_s=2.0, min_score=0.55)

    assert [c.path for c in ranked] == [mid]
    assert ranked[0].score == pytest.approx(0.7071, abs=1e-3)


# -- review_candidates ----------------------------------------------------------


def test_review_accept_skip_play_quit(tmp_path):
    a = Candidate(Path("a.wav"), 0.9, 2.5)
    b = Candidate(Path("b.wav"), 0.8, 2.5)
    c = Candidate(Path("c.wav"), 0.7, 2.5)
    played: list = []
    review = ScriptReview(["accept", "skip", "play", "accept"])

    accepted = review_candidates(
        [a, b, c], review, 5, play=lambda p: played.append(p))

    assert accepted == [a, c]
    assert played == [c.path]
    assert [s[0] for s in review.seen] == [a, b, c, c]


def test_review_play_does_not_consume_candidate():
    a = Candidate(Path("a.wav"), 0.9, 2.5)
    review = ScriptReview(["play", "play", "accept"])

    accepted = review_candidates([a], review, 1, play=lambda p: None)

    assert accepted == [a]
    assert len(review.seen) == 3


def test_review_quit_returns_accepted_so_far():
    a = Candidate(Path("a.wav"), 0.9, 2.5)
    b = Candidate(Path("b.wav"), 0.8, 2.5)
    review = ScriptReview(["accept", "quit"])

    assert review_candidates([a, b], review, 5, play=lambda p: None) == [a]


def test_review_count_stops_accepting():
    cands = [Candidate(Path(f"{i}.wav"), 0.9, 2.5) for i in range(4)]

    accepted = review_candidates(
        cands, ScriptReview(["accept"] * 4), 2, play=lambda p: None)

    assert len(accepted) == 2


def test_review_rejects_unknown_action():
    with pytest.raises(ValueError, match="unknown review action"):
        review_candidates(
            [Candidate(Path("a.wav"), 0.9, 2.5)], ScriptReview(["dance"]),
            1, play=lambda p: None)


# -- play_wav -------------------------------------------------------------------


def test_play_wav_no_players_returns_none(monkeypatch, tmp_path):
    monkeypatch.setattr(shutil, "which", lambda _name: None)

    assert play_wav(tmp_path / "clip.wav") is None


def test_play_wav_uses_first_available(monkeypatch, tmp_path):
    monkeypatch.setattr(shutil, "which",
                        lambda name: "/usr/bin/paplay" if name == "paplay" else None)
    ran: list = []
    monkeypatch.setattr(subprocess, "run",
                        lambda *a, **k: ran.append(a[0]) or
                        subprocess.CompletedProcess(a[0], 0))

    assert play_wav(tmp_path / "clip.wav") == "paplay"
    assert ran[0][0] == "paplay"


# -- seed_enroll ----------------------------------------------------------------


def seed_and_clips(audio_dir: Path, *, seed_value=A, seed_s=2.5,
                   clips=((("a1.wav", A), ("a2.wav", A), ("b1.wav", B)))):
    seed = write_const_wav(audio_dir / "seed.wav", seed_s, seed_value)
    paths = [write_const_wav(audio_dir / name, 2.5, value)
             for name, value in clips]
    return seed, paths


def test_seed_enroll_auto_enrolls_seed_plus_top_count(monkeypatch, tmp_path, capsys):
    cfg, audio_dir, store_dir = write_cfg(tmp_path)
    seed, _ = seed_and_clips(audio_dir)
    enrolled_log = record_enroll(monkeypatch)

    accepted = seed_enroll(cfg, VariedEmbedder(), make_store(cfg), "Efren",
                           seed, count=2, auto=True)

    assert [c.path.name for c in accepted] == ["a1.wav", "a2.wav"]
    assert [p.name for p in enrolled_log[0]] == ["seed.wav", "a1.wav", "a2.wav"]
    assert store_files(store_dir)
    assert fresh_speakers(store_dir)["spk_0001"]["name"] == "Efren"


def test_seed_enroll_short_seed_is_query_only(monkeypatch, tmp_path, capsys):
    cfg, audio_dir, store_dir = write_cfg(tmp_path)
    seed, _ = seed_and_clips(audio_dir, seed_s=0.5)
    enrolled_log = record_enroll(monkeypatch)

    accepted = seed_enroll(cfg, VariedEmbedder(), make_store(cfg), "Efren",
                           seed, count=2, auto=True)

    assert len(accepted) == 2
    assert [p.name for p in enrolled_log[0]] == ["a1.wav", "a2.wav"]
    assert "query only" in capsys.readouterr().out


def test_seed_enroll_seed_excluded_from_candidates(tmp_path):
    cfg, audio_dir, _ = write_cfg(tmp_path)
    seed, _ = seed_and_clips(audio_dir)

    accepted = seed_enroll(cfg, VariedEmbedder(), make_store(cfg), "Efren",
                           seed, count=10, auto=True)

    assert seed not in [c.path for c in accepted]
    assert len(accepted) == 2


def test_seed_enroll_fewer_candidates_than_count(monkeypatch, tmp_path):
    cfg, audio_dir, store_dir = write_cfg(tmp_path)
    seed, _ = seed_and_clips(audio_dir, clips=(("a1.wav", A),))
    enrolled_log = record_enroll(monkeypatch)

    accepted = seed_enroll(cfg, VariedEmbedder(), make_store(cfg), "Efren",
                           seed, count=5, auto=True)

    assert len(accepted) == 1
    assert len(enrolled_log[0]) == 2
    assert store_files(store_dir)


def test_seed_enroll_all_skipped_writes_nothing(tmp_path, capsys):
    cfg, audio_dir, store_dir = write_cfg(tmp_path)
    seed, _ = seed_and_clips(audio_dir)

    accepted = seed_enroll(cfg, ConstEmbedder(), make_store(cfg), "Efren",
                           seed, review=ScriptReview(["skip"] * 3))

    assert accepted == []
    assert not store_files(store_dir)
    assert "nothing enrolled" in capsys.readouterr().out


def test_seed_enroll_empty_dir_writes_nothing(tmp_path, capsys):
    cfg, audio_dir, store_dir = write_cfg(tmp_path)
    seed = write_const_wav(tmp_path / "seed.wav", 2.5, A)

    accepted = seed_enroll(cfg, ConstEmbedder(), make_store(cfg), "Efren",
                           seed, count=3, auto=True)

    assert accepted == []
    assert not store_files(store_dir)
    assert "nothing enrolled" in capsys.readouterr().out


def test_seed_enroll_missing_session_root_writes_nothing(tmp_path, capsys):
    cfg, _, store_dir = write_cfg(tmp_path)
    missing = tmp_path / "sessions" / "no-such-session"
    cfg.output.session_dir = str(missing)
    seed = write_const_wav(tmp_path / "seed.wav", 2.5, A)

    accepted = seed_enroll(cfg, ConstEmbedder(), make_store(cfg), "Efren",
                           seed, count=3, auto=True)

    assert accepted == []
    assert not store_files(store_dir)
    assert "nothing enrolled" in capsys.readouterr().out


def test_seed_enroll_quit_aborts_without_writing(tmp_path, capsys):
    cfg, audio_dir, store_dir = write_cfg(tmp_path)
    seed, _ = seed_and_clips(audio_dir)

    with pytest.raises(EnrollmentAborted):
        seed_enroll(cfg, ConstEmbedder(), make_store(cfg), "Efren", seed,
                    review=ScriptReview(["accept", "quit"]))

    assert not store_files(store_dir)
    out = capsys.readouterr().out
    assert "nothing enrolled" in out
    assert "speaker-enroll" in out


def test_seed_enroll_keyboard_interrupt_aborts_without_writing(tmp_path):
    cfg, audio_dir, store_dir = write_cfg(tmp_path)
    seed, _ = seed_and_clips(audio_dir)

    class ExplodingReview:
        def choose(self, candidate, accepted, target):
            raise KeyboardInterrupt

    with pytest.raises(EnrollmentAborted):
        seed_enroll(cfg, ConstEmbedder(), make_store(cfg), "Efren", seed,
                    review=ExplodingReview())

    assert not store_files(store_dir)


def test_seed_enroll_dim_mismatch_errors_before_writing(tmp_path):
    cfg, audio_dir, store_dir = write_cfg(tmp_path)
    seed, _ = seed_and_clips(audio_dir)
    old = make_store(cfg)
    enroll_from_files(old, ConstEmbedder(dim=3), "Efren", [seed], 2.0)
    store = make_store(cfg)
    store.load()

    with pytest.raises(ValueError, match="different model"):
        seed_enroll(cfg, ConstEmbedder(dim=2), store, "Efren", seed, auto=True)

    assert fresh_speakers(store_dir)["spk_0001"]["prototypes"] == 1


def test_seed_enroll_max_prototypes_warns_and_evicts(tmp_path, capsys):
    cfg, audio_dir, store_dir = write_cfg(tmp_path, max_prototypes=2)
    seed, _ = seed_and_clips(audio_dir)

    seed_enroll(cfg, VariedEmbedder(), make_store(cfg), "Efren", seed,
                count=2, auto=True)

    assert "evict" in capsys.readouterr().out
    assert fresh_speakers(store_dir)["spk_0001"]["prototypes"] == 2


def test_seed_enroll_max_scan_caps_embedding(tmp_path, capsys):
    cfg, audio_dir, _ = write_cfg(tmp_path)
    seed = write_const_wav(audio_dir / "seed.wav", 2.5, A)
    for i in range(5):
        write_const_wav(audio_dir / f"clip{i}.wav", 2.5, A)
    embedder = VariedEmbedder()

    accepted = seed_enroll(cfg, embedder, make_store(cfg), "Efren", seed,
                           count=5, max_scan=2, auto=True)

    assert len(accepted) == 2
    # seed embed (dim check) + seed embed (rank) + 2 candidates + enroll of 3
    assert embedder.calls == 7
    assert "max-scan 2" in capsys.readouterr().out


# -- CLI handler ----------------------------------------------------------------


def run_seed_handler(monkeypatch, tmp_path, *, name="Efren", seed=None,
                     count=5, min_score=0.55, audio_dir=None, max_scan=0,
                     yes=False, no_play=False, force=False, model_ok=True,
                     tty=True, review=None, min_enroll_s=2.0,
                     max_prototypes=8, dim=2, pre_enroll_dim=None,
                     embedder_cls: type = ConstEmbedder):
    """Drive ``_handle_speaker_enroll_seed`` with fakes; return (rc, out, err)."""
    cfg, default_audio, store_dir = write_cfg(
        tmp_path, min_enroll_s=min_enroll_s, max_prototypes=max_prototypes,
        model_ok=model_ok)
    if audio_dir is None:
        audio_dir = default_audio
    if seed is None:
        seed = write_const_wav(Path(audio_dir) / "seed.wav", 2.5, A)
    if pre_enroll_dim is not None:
        old = SpeakerStore(store_dir)
        enroll_from_files(old, ConstEmbedder(dim=pre_enroll_dim), name,
                          [seed], min_enroll_s)
    fake = embedder_cls(dim=dim)

    class FakeSherpa:
        def __init__(self, model_path, provider="cpu"):
            self._embedder = fake

        def embed(self, audio):
            return fake.embed(audio)

    monkeypatch.setattr(cli_module, "SherpaEmbedder", FakeSherpa)
    if review is not None:
        monkeypatch.setattr(cli_module, "QuestionaryReview", lambda **k: review)
    if tty:
        monkeypatch.setattr(sys, "stdin", _TtyStdin())
    else:
        monkeypatch.setattr(sys, "stdin", io.StringIO())
    args = argparse.Namespace(
        config=str(tmp_path / "config.toml"), name=name, seed=str(seed),
        count=count, min_score=min_score, audio_dir=str(audio_dir),
        max_scan=max_scan, yes=yes, no_play=no_play, force=force,
    )
    out, err = io.StringIO(), io.StringIO()
    with redirect_stdout(out), redirect_stderr(err):
        rc = cli_module._handle_speaker_enroll_seed(args)
    return rc, out.getvalue(), err.getvalue()


class _TtyStdin(io.StringIO):
    def isatty(self):
        return True


def _audio_with_clips(tmp_path: Path, sub="audio") -> Path:
    audio_dir = tmp_path / sub
    audio_dir.mkdir(exist_ok=True)
    write_const_wav(audio_dir / "seed.wav", 2.5, A)
    write_const_wav(audio_dir / "a1.wav", 2.5, A)
    write_const_wav(audio_dir / "a2.wav", 2.5, A)
    write_const_wav(audio_dir / "b1.wav", 2.5, B)
    return audio_dir


def test_cli_auto_success(monkeypatch, tmp_path):
    audio_dir = _audio_with_clips(tmp_path)

    rc, out, _ = run_seed_handler(monkeypatch, tmp_path, audio_dir=audio_dir,
                                  count=2, yes=True,
                                  embedder_cls=VariedEmbedder)

    assert rc == 0
    assert "enrolled 'Efren' as spk_0001 from seed" in out
    assert "3 clip(s) (seed + 2 reviewed" in out


def test_cli_short_seed_query_only(monkeypatch, tmp_path):
    audio_dir = tmp_path / "audio"
    audio_dir.mkdir(exist_ok=True)
    seed = write_const_wav(audio_dir / "seed.wav", 0.5, A)
    write_const_wav(audio_dir / "a1.wav", 2.5, A)

    rc, out, _ = run_seed_handler(monkeypatch, tmp_path, audio_dir=audio_dir,
                                  seed=seed, count=1, yes=True)

    assert rc == 0
    assert "query-only" in out


def test_cli_non_tty_without_yes_errors(monkeypatch, tmp_path):
    audio_dir = _audio_with_clips(tmp_path)

    rc, _, err = run_seed_handler(monkeypatch, tmp_path, audio_dir=audio_dir,
                                  tty=False, yes=False)

    assert rc == 1
    assert "--yes" in err


def test_cli_non_tty_with_yes_auto_accepts(monkeypatch, tmp_path):
    audio_dir = _audio_with_clips(tmp_path)

    rc, out, _ = run_seed_handler(monkeypatch, tmp_path, audio_dir=audio_dir,
                                  tty=False, yes=True, count=1,
                                  embedder_cls=VariedEmbedder)

    assert rc == 0
    assert "from seed" in out


def test_cli_daemon_guard_and_force(monkeypatch, tmp_path):
    audio_dir = _audio_with_clips(tmp_path)
    (tmp_path / "bus.sock").write_text("", encoding="utf-8")

    rc, _, err = run_seed_handler(monkeypatch, tmp_path, audio_dir=audio_dir,
                                  yes=True)
    assert rc == 1
    assert "daemon" in err.lower()

    rc, _, _ = run_seed_handler(monkeypatch, tmp_path, audio_dir=audio_dir,
                                yes=True, force=True,
                                embedder_cls=VariedEmbedder)
    assert rc == 0


def test_cli_missing_model(monkeypatch, tmp_path):
    audio_dir = _audio_with_clips(tmp_path)

    rc, _, err = run_seed_handler(monkeypatch, tmp_path, audio_dir=audio_dir,
                                  yes=True, model_ok=False)

    assert rc == 1
    assert "model" in err.lower()


def test_cli_dim_mismatch_no_write(monkeypatch, tmp_path):
    audio_dir = _audio_with_clips(tmp_path)
    seed = audio_dir / "seed.wav"

    rc, _, err = run_seed_handler(monkeypatch, tmp_path, audio_dir=audio_dir,
                                  seed=seed, yes=True, dim=2, pre_enroll_dim=3)

    assert rc == 1
    assert "different model" in err
    assert fresh_speakers(tmp_path / "speakers")["spk_0001"]["prototypes"] == 1


def test_cli_quit_no_write(monkeypatch, tmp_path):
    audio_dir = _audio_with_clips(tmp_path)

    rc, out, _ = run_seed_handler(
        monkeypatch, tmp_path, audio_dir=audio_dir,
        review=ScriptReview(["accept", "quit"]))

    assert rc == 1
    assert not (tmp_path / "speakers" / "speakers.json").exists()
    assert "speaker-enroll" in out


def test_cli_no_play_disables_playback(monkeypatch, tmp_path):
    audio_dir = _audio_with_clips(tmp_path)
    seen: dict = {}

    class SpyReviewFactory:
        def __init__(self, **kwargs):
            seen.update(kwargs)

        def choose(self, candidate, accepted, target):
            return "accept" if accepted < 1 else "quit"

    monkeypatch.setattr(cli_module, "QuestionaryReview", SpyReviewFactory)
    monkeypatch.setattr(ea, "play_wav", lambda p: (_ for _ in ()).throw(
        AssertionError("playback attempted")))

    rc, _, _ = run_seed_handler(monkeypatch, tmp_path, audio_dir=audio_dir,
                                count=2, no_play=True)

    assert seen.get("allow_play") is False
    assert rc == 1


def test_cli_no_candidates_no_write(monkeypatch, tmp_path):
    audio_dir = tmp_path / "audio"
    audio_dir.mkdir(exist_ok=True)
    seed = write_const_wav(audio_dir / "seed.wav", 2.5, A)

    rc, out, _ = run_seed_handler(monkeypatch, tmp_path, audio_dir=audio_dir,
                                  seed=seed, yes=True)

    assert rc == 1
    assert "nothing enrolled" in out
    assert not (tmp_path / "speakers" / "speakers.json").exists()


def test_questionary_review_defaults():
    import dataclasses

    review = QuestionaryReview()

    assert review.allow_play is True
    candidate = Candidate(Path("x.wav"), 0.5, 1.0)
    with pytest.raises(dataclasses.FrozenInstanceError):
        candidate.score = 0.1  # type: ignore[misc]


# -- duplicate protection -------------------------------------------------------


def _store_with(root: Path, profiles) -> SpeakerStore:
    store = SpeakerStore(root)
    for name, vecs in profiles:
        store.enroll(name, [np.array(v, dtype=np.float32) for v in vecs])
    return store


def test_candidate_equality_ignores_embedding_and_flags():
    first = Candidate(Path("a.wav"), 0.9, 2.5,
                      embedding=np.array([1.0, 0.0], dtype=np.float32),
                      flags=("duplicate",))
    second = Candidate(Path("a.wav"), 0.9, 2.5,
                       embedding=np.array([0.0, 1.0], dtype=np.float32),
                       flags=())

    assert first == second
    assert hash(first) == hash(second)
    assert first != Candidate(Path("b.wav"), 0.9, 2.5)


def test_rank_sets_unit_embeddings_without_store(tmp_path):
    seed = write_const_wav(tmp_path / "seed.wav", 2.5, A)
    clip = write_const_wav(tmp_path / "a.wav", 2.5, A)
    from stt_daemon.enroll import read_wav
    seed_audio, _ = read_wav(seed)

    ranked, _ = rank_candidates(ConstEmbedder(), seed_audio, [clip],
                                min_enroll_s=2.0, min_score=0.55)

    assert len(ranked) == 1
    embedding = ranked[0].embedding
    assert embedding is not None
    assert embedding.dtype == np.float32
    assert float(np.linalg.norm(embedding)) == pytest.approx(1.0)
    assert ranked[0].flags == ()


def test_rank_flags_duplicate_against_target(tmp_path):
    seed = write_const_wav(tmp_path / "seed.wav", 2.5, A)
    clip = write_const_wav(tmp_path / "a.wav", 2.5, A)
    from stt_daemon.enroll import read_wav
    seed_audio, _ = read_wav(seed)
    store = _store_with(tmp_path / "store", [("Efren", [[1.0, 0.0]])])

    ranked, _ = rank_candidates(ConstEmbedder(), seed_audio, [clip],
                                min_enroll_s=2.0, min_score=0.55,
                                store=store, speaker_name="Efren")

    assert ranked[0].flags == ("duplicate",)


def test_rank_flags_best_other_speaker_by_display_name(tmp_path):
    seed = write_const_wav(tmp_path / "seed.wav", 2.5, A)
    clip = write_const_wav(tmp_path / "b.wav", 2.5, B)
    from stt_daemon.enroll import read_wav
    seed_audio, _ = read_wav(seed)
    store = _store_with(tmp_path / "store", [("Efren", [[1.0, 0.0]]),
                                             ("Rival", [[0.0, 1.0]]),
                                             ("Mid", [[0.7071, 0.7071]])])

    ranked, _ = rank_candidates(ConstEmbedder(), seed_audio, [clip],
                                min_enroll_s=2.0, min_score=-1.0,
                                store=store, speaker_name="Efren")

    assert ranked[0].flags == ("other-speaker:Rival",)


def test_rank_no_flags_without_store_or_name(tmp_path):
    seed = write_const_wav(tmp_path / "seed.wav", 2.5, A)
    clip = write_const_wav(tmp_path / "a.wav", 2.5, A)
    from stt_daemon.enroll import read_wav
    seed_audio, _ = read_wav(seed)
    store = _store_with(tmp_path / "store", [("Efren", [[1.0, 0.0]])])

    ranked, _ = rank_candidates(ConstEmbedder(), seed_audio, [clip],
                                min_enroll_s=2.0, min_score=0.55,
                                store=store)

    assert ranked[0].flags == ()


def test_review_refuses_duplicate_accept_without_consuming_budget(capsys):
    dup = Candidate(Path("dup.wav"), 0.99, 2.5, flags=("duplicate",))
    good = Candidate(Path("good.wav"), 0.9, 2.5)
    review = ScriptReview(["accept", "accept"])

    accepted = review_candidates([dup, good], review, 1, play=lambda p: None)

    assert accepted == [good]
    out = capsys.readouterr().out
    assert "dup.wav" in out
    assert "already covered" in out


def test_seed_enroll_auto_dedupes_incoming_set(monkeypatch, tmp_path, capsys):
    cfg, audio_dir, _ = write_cfg(tmp_path)
    seed, _ = seed_and_clips(audio_dir)
    enrolled_log = record_enroll(monkeypatch)

    accepted = seed_enroll(cfg, ConstEmbedder(), make_store(cfg), "Efren",
                           seed, count=2, auto=True)

    assert accepted == []
    assert [p.name for p in enrolled_log[0]] == ["seed.wav"]
    assert "duplicate" in capsys.readouterr().out


def test_seed_enroll_review_path_dedupes_incoming_set(monkeypatch, tmp_path, capsys):
    cfg, audio_dir, _ = write_cfg(tmp_path)
    seed, _ = seed_and_clips(audio_dir)
    enrolled_log = record_enroll(monkeypatch)

    accepted = seed_enroll(cfg, ConstEmbedder(), make_store(cfg), "Efren",
                           seed, review=ScriptReview(["accept", "accept"]))

    assert accepted == []
    assert [p.name for p in enrolled_log[0]] == ["seed.wav"]
    assert "duplicate" in capsys.readouterr().out


def test_seed_enroll_review_vetoes_store_duplicate(tmp_path, capsys):
    cfg, audio_dir, store_dir = write_cfg(tmp_path)
    seed, _ = seed_and_clips(audio_dir)
    store = make_store(cfg)
    enroll_from_files(store, ConstEmbedder(), "Efren", [seed], 2.0)

    accepted = seed_enroll(cfg, ConstEmbedder(), store, "Efren", seed,
                           review=ScriptReview(["accept"] * 5))

    assert accepted == []
    out = capsys.readouterr().out
    assert "already covered" in out
    assert "nothing enrolled" in out
    assert fresh_speakers(store_dir)["spk_0001"]["prototypes"] == 1


def test_seed_enroll_removes_store_duplicates_and_saves(tmp_path, capsys):
    cfg, audio_dir, store_dir = write_cfg(tmp_path)
    seed, _ = seed_and_clips(audio_dir)
    store = make_store(cfg)
    enroll_from_files(store, ConstEmbedder(), "Efren",
                      [seed, audio_dir / "a1.wav"], 2.0)
    saves: list = []
    real_save = store.save

    def counting_save():
        saves.append(1)
        return real_save()

    store.save = counting_save  # type: ignore[method-assign]

    accepted = seed_enroll(cfg, VariedEmbedder(), store, "Efren", seed,
                           count=2, auto=True)

    assert [c.path.name for c in accepted] == ["a1.wav", "a2.wav"]
    out = capsys.readouterr().out
    assert "removed 1 duplicate" in out
    assert len(saves) == 2  # enroll_from_files save + post-dedupe save
    assert (store_dir / "speakers.json").exists()
    assert fresh_speakers(store_dir)["spk_0001"]["prototypes"] == 4


def test_cli_forwards_dedupe_and_other_thresholds(monkeypatch, tmp_path):
    audio_dir = _audio_with_clips(tmp_path)
    seed = audio_dir / "seed.wav"
    write_cfg(tmp_path)
    seen: dict = {}

    def fake_seed_enroll(cfg, embedder, store, name, seed_path, **kwargs):
        seen.update(kwargs)
        return [Candidate(audio_dir / "a1.wav", 0.99, 2.5)]

    monkeypatch.setattr(cli_module, "seed_enroll", fake_seed_enroll)
    fake = ConstEmbedder()

    class FakeSherpa:
        def __init__(self, model_path, provider="cpu"):
            pass

        def embed(self, audio):
            return fake.embed(audio)

    monkeypatch.setattr(cli_module, "SherpaEmbedder", FakeSherpa)
    args = argparse.Namespace(
        config=str(tmp_path / "config.toml"), name="Efren", seed=str(seed),
        count=2, min_score=0.55, audio_dir=str(audio_dir),
        max_scan=0, yes=True, no_play=True, force=True,
        dedupe_threshold=0.97, other_threshold=0.44,
    )
    out, err = io.StringIO(), io.StringIO()
    with redirect_stdout(out), redirect_stderr(err):
        rc = cli_module._handle_speaker_enroll_seed(args)

    assert rc == 0
    assert seen.get("dup_threshold") == pytest.approx(0.97)
    assert seen.get("other_threshold") == pytest.approx(0.44)