tests/test_quality_runtime.py

import json
import math
from types import SimpleNamespace

import numpy as np
import pytest
import soundfile as sf
import torch

from production import (
    count_network_endpoint_duration_units,
    count_speech_units,
    normalize_spoken_forms,
    split_text_for_tts,
)
from quality_runtime import (
    ADAPTIVE_CASCADE_STAGE_LIMITS,
    BASE_GENERATION_POLICY,
    COMPLETION_HEADROOM_GENERATION_POLICY,
    CASCADE_EVIDENCE_LOG_PREFIX,
    CASCADE_EVIDENCE_SCHEMA_VERSION,
    ENDPOINT_ENERGY_WEIGHT,
    ENDPOINT_HARD_STOP_PENALTY,
    ENDPOINT_PREFERRED_SQUIM_PESQ_SLACK,
    ENDPOINT_PREFERRED_SQUIM_STOI_SLACK,
    SAFE_DURATION_GENERATION_POLICY,
    SQUIM_OBJECTIVE_ASSET_PATH,
    SQUIM_OBJECTIVE_SAMPLE_RATE,
    SQUIM_OBJECTIVE_WEIGHT_SHA256,
    VERIFICATION_WHISPER_MODEL_ID,
    VERIFICATION_WHISPER_REVISION,
    WHISPER_ATTENTION_IMPLEMENTATION,
    WHISPER_MODEL_ID,
    WHISPER_RETURN_ATTENTION_MASK,
    WHISPER_REVISION,
    CandidateGenerationEvidence,
    CandidateObservation,
    CandidateVerification,
    CascadeResult,
    ChunkCandidateArtifact,
    FinalOutputRejectedError,
    LazySquimObjective,
    LazyWhisperASR,
    LocalIndependentGateEvidence,
    NoQualifiedCandidateError,
    RELEASE_SPEAKER_TRIGGER_SECONDS,
    SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP,
    TrajectoryGateResult,
    SquimObjectiveRuntime,
    WholeWaveformVerificationCache,
    _sequence_fallback_candidate_result,
    WhisperRuntime,
    _split_whisper_audio,
    active_audio_median_f0_hz,
    active_voiced_duration_seconds,
    candidate_gate_evidence,
    candidate_chunk_transition_score,
    candidate_endpoint_selection_cost,
    candidate_limit_for_chunk_budget,
    cosine_similarity,
    exact_waveform_sha256,
    endpoint_tail_energy_ratio,
    format_cascade_evidence_log,
    generation_cfg_for_candidate_offset,
    generation_policy_for_candidate_offset,
    intersect_local_semantic_verification,
    local_candidate_has_coverage_eligibility,
    load_pinned_whisper_runtime,
    load_pinned_verification_whisper_runtime,
    load_pinned_squim_objective_runtime,
    prepare_candidate_audio,
    qualify_trajectory_with_joined_output,
    release_speaker_evidence_from_audio,
    release_speaker_measurement_required,
    require_verified_final_output,
    resolve_request_seed,
    run_adaptive_cascade,
    run_coverage_adaptive_cascade,
    select_k_candidate_sequences,
    speaker_embedding_from_audio,
    speaker_evidence_from_audio,
    squim_objective_evidence_from_audio,
    trim_release_speaker_activity,
    transcribe_whisper,
    transcribe_verification_whisper,
    trajectory_gate_evidence,
    verify_candidate,
    verify_trajectory,
)


class _Factory:
    def __init__(self, value):
        self.value = value
        self.calls = []

    def from_pretrained(self, *args, **kwargs):
        self.calls.append((args, kwargs))
        return self.value


class _FakeSquimModel:
    def __init__(self, outputs=None):
        self.outputs = outputs
        self.loaded = []
        self.devices = []
        self.eval_calls = 0
        self.input_shapes = []

    def load_state_dict(self, state_dict, strict):
        self.loaded.append((state_dict, strict))

    def to(self, device):
        self.devices.append(torch.device(device))
        return self

    def eval(self):
        self.eval_calls += 1
        return self

    def __call__(self, waveform):
        self.input_shapes.append(tuple(waveform.shape))
        batch = waveform.shape[0]
        if self.outputs is None:
            return (
                torch.full((batch,), 0.80),
                torch.full((batch,), 1.50),
                torch.full((batch,), 5.0),
            )
        return tuple(torch.as_tensor(row) for row in self.outputs)


def test_pinned_squim_hash_is_checked_before_safe_state_loading(tmp_path):
    weight = tmp_path / "squim.pth"
    weight.write_bytes(b"test-only-state")
    events = []
    model = _FakeSquimModel()

    runtime = load_pinned_squim_objective_runtime(
        asset_fetcher=lambda asset: events.append(("fetch", asset)) or weight,
        file_hasher=lambda path: events.append(("hash", path))
        or SQUIM_OBJECTIVE_WEIGHT_SHA256,
        state_loader=lambda path, **kwargs: events.append(
            ("load", path, kwargs)
        )
        or {"weight": torch.tensor(1.0)},
        model_factory=lambda: model,
    )

    assert events[0] == ("fetch", SQUIM_OBJECTIVE_ASSET_PATH)
    assert events[1] == ("hash", weight)
    assert events[2][0] == "load"
    assert events[2][2] == {"map_location": "cpu", "weights_only": True}
    assert runtime.device == torch.device("cpu")
    assert runtime.sample_rate == SQUIM_OBJECTIVE_SAMPLE_RATE
    assert runtime.weight_sha256 == SQUIM_OBJECTIVE_WEIGHT_SHA256
    assert model.loaded == [({"weight": torch.tensor(1.0)}, True)]
    assert model.devices == [torch.device("cpu")]
    assert model.eval_calls == 1


def test_pinned_squim_hash_mismatch_fails_before_deserialization(tmp_path):
    weight = tmp_path / "corrupt.pth"
    weight.write_bytes(b"corrupt")
    state_loads = []

    with pytest.raises(RuntimeError, match="SHA-256 mismatch"):
        load_pinned_squim_objective_runtime(
            asset_fetcher=lambda _asset: weight,
            file_hasher=lambda _path: "0" * 64,
            state_loader=lambda *args, **kwargs: state_loads.append((args, kwargs)),
            model_factory=_FakeSquimModel,
        )

    assert state_loads == []


def test_lazy_squim_loads_once_and_rejects_contract_substitution():
    model = _FakeSquimModel()
    calls = []

    def loader():
        calls.append(1)
        return SquimObjectiveRuntime(
            model=model,
            device=torch.device("cpu"),
            sample_rate=SQUIM_OBJECTIVE_SAMPLE_RATE,
            weight_sha256=SQUIM_OBJECTIVE_WEIGHT_SHA256,
        )

    lazy = LazySquimObjective(loader)
    assert lazy.get_runtime() is lazy.get_runtime()
    assert calls == [1]

    substituted = LazySquimObjective(
        lambda: SquimObjectiveRuntime(
            model=model,
            device=torch.device("cpu"),
            sample_rate=SQUIM_OBJECTIVE_SAMPLE_RATE,
            weight_sha256="0" * 64,
        )
    )
    with pytest.raises(RuntimeError, match="pinned contract"):
        substituted.get_runtime()


def test_squim_scoring_is_cpu_bounded_and_pools_long_audio_conservatively():
    model = _FakeSquimModel(
        outputs=(
            [0.90, 0.75, 0.82],
            [2.00, 1.25, 1.60],
            [8.0, -2.0, 3.0],
        )
    )
    runtime = SquimObjectiveRuntime(
        model=model,
        device=torch.device("cpu"),
        sample_rate=SQUIM_OBJECTIVE_SAMPLE_RATE,
        weight_sha256=SQUIM_OBJECTIVE_WEIGHT_SHA256,
    )
    audio = np.linspace(-0.1, 0.1, 48_000 * 40, dtype=np.float32)

    evidence = squim_objective_evidence_from_audio(
        audio,
        48_000,
        runtime=runtime,
    )

    assert evidence.window_count == 3
    assert evidence.stoi == pytest.approx(0.75)
    assert evidence.pesq == pytest.approx(1.25)
    assert evidence.si_sdr == pytest.approx(3.0)
    assert model.input_shapes == [(3, 160_000)]


def test_squim_scoring_rejects_nonfinite_model_evidence():
    model = _FakeSquimModel(outputs=([float("nan")], [1.5], [2.0]))
    runtime = SquimObjectiveRuntime(
        model=model,
        device=torch.device("cpu"),
        sample_rate=SQUIM_OBJECTIVE_SAMPLE_RATE,
        weight_sha256=SQUIM_OBJECTIVE_WEIGHT_SHA256,
    )

    with pytest.raises(ValueError, match="non-finite"):
        squim_objective_evidence_from_audio(
            np.ones(16_000, dtype=np.float32),
            16_000,
            runtime=runtime,
        )


def test_release_speaker_trigger_covers_wav_roundtrip_eligibility_boundary(tmp_path):
    assert RELEASE_SPEAKER_TRIGGER_SECONDS == 1.48
    assert not release_speaker_measurement_required(1.479)
    assert release_speaker_measurement_required(1.48)

    sample_rate = 48_000
    for nominal_seconds in (1.49, 1.50, 1.51):
        active_samples = int(round(nominal_seconds * sample_rate))
        phase = np.arange(active_samples, dtype=np.float32) / sample_rate
        active = 0.2 * np.sin(2.0 * np.pi * 220.0 * phase)
        waveform = np.concatenate(
            (
                np.zeros(sample_rate // 10, dtype=np.float32),
                active.astype(np.float32),
                np.zeros(sample_rate // 10, dtype=np.float32),
            )
        )
        before = active_voiced_duration_seconds(waveform, sample_rate)
        path = tmp_path / f"boundary-{nominal_seconds:.2f}.wav"
        sf.write(path, waveform, sample_rate, subtype="PCM_16")
        decoded, decoded_rate = sf.read(path, dtype="float32")
        after = active_voiced_duration_seconds(decoded, decoded_rate)
        assert not (after >= 1.50 and not release_speaker_measurement_required(before))


def test_release_trigger_threshold_rejects_bad_measured_speaker_at_1_49_seconds():
    result = verify_trajectory(
        [
            CandidateObservation(
                target_text="完整內容",
                transcript_text="完整內容",
                audio_duration_seconds=1.49,
                speaker_similarity=0.0,
                begin_speaker_similarity=0.0,
                end_speaker_similarity=0.0,
                pace_cps=3.0,
            )
        ],
        short_audio_seconds=RELEASE_SPEAKER_TRIGGER_SECONDS,
        min_speaker_similarity=0.105,
        max_boundary_speaker_drop=0.095,
    )

    assert not result.passed
    assert "chunk_0:speaker_similarity" in result.rejection_reasons


def test_candidate_generation_policy_mapping_uses_sparse_completion_headroom():
    assert ADAPTIVE_CASCADE_STAGE_LIMITS == (1, 5, 10, 15, 20, 24, 28, 32)
    base = generation_policy_for_candidate_offset(0)
    assert base is BASE_GENERATION_POLICY
    assert (
        base.name,
        base.cjk_cps,
        base.ascii_cps,
        base.hard_stop_margin_steps,
        base.short_headroom_max_units,
        base.short_hard_stop_floor_steps,
    ) == (
        "base",
        5.2,
        4.6,
        1,
        0,
        0,
    )

    for offset in (1, 2, 5, 9):
        safe = generation_policy_for_candidate_offset(offset)
        assert safe is SAFE_DURATION_GENERATION_POLICY
        assert (
            safe.name,
            safe.cjk_cps,
            safe.ascii_cps,
            safe.hard_stop_margin_steps,
            safe.short_headroom_max_units,
            safe.short_hard_stop_floor_steps,
        ) == ("safe_duration", 4.6, 4.0, 1, 0, 0)

    for offset in (4, 8, 12, 16, 20, 24, 28):
        headroom = generation_policy_for_candidate_offset(offset)
        assert headroom is COMPLETION_HEADROOM_GENERATION_POLICY
        assert (
            headroom.name,
            headroom.cjk_cps,
            headroom.ascii_cps,
            headroom.hard_stop_margin_steps,
            headroom.short_headroom_max_units,
            headroom.short_hard_stop_floor_steps,
        ) == ("completion_headroom", 4.2, 3.6, 1, 2, 5)


def test_endpoint_tail_energy_and_selection_cost_prefer_natural_quiet_endings():
    waveform = np.concatenate(
        (
            np.ones(950, dtype=np.float32),
            np.full(50, 0.25, dtype=np.float32),
        )
    )
    energy = endpoint_tail_energy_ratio(
        waveform,
        10_000,
        window_ms=5.0,
    )
    threshold = ChunkCandidateArtifact(
        stop_reason="stop_threshold",
        endpoint_energy_ratio=energy,
        generated_steps=4,
        hard_stop_steps=5,
    )
    threshold_at_cap = ChunkCandidateArtifact(
        stop_reason="stop_threshold",
        endpoint_energy_ratio=energy,
        generated_steps=5,
        hard_stop_steps=5,
    )
    forced = ChunkCandidateArtifact(
        stop_reason="hard_stop",
        endpoint_energy_ratio=energy,
        generated_steps=5,
        hard_stop_steps=5,
    )

    assert energy == pytest.approx(0.25)
    assert candidate_endpoint_selection_cost(threshold) == pytest.approx(
        ENDPOINT_ENERGY_WEIGHT * 0.25
    )
    assert candidate_endpoint_selection_cost(threshold_at_cap) == pytest.approx(
        ENDPOINT_HARD_STOP_PENALTY + ENDPOINT_ENERGY_WEIGHT * 0.25
    )
    assert candidate_endpoint_selection_cost(forced) == pytest.approx(
        ENDPOINT_HARD_STOP_PENALTY + ENDPOINT_ENERGY_WEIGHT * 0.25
    )
    assert ENDPOINT_PREFERRED_SQUIM_STOI_SLACK == pytest.approx(0.02)
    assert ENDPOINT_PREFERRED_SQUIM_PESQ_SLACK == pytest.approx(0.03)


@pytest.mark.parametrize("offset", [-1, 0.5, 1.0, True, None, "one"])
def test_candidate_generation_policy_mapping_rejects_invalid_offsets(offset):
    with pytest.raises(ValueError, match="non-negative integer"):
        generation_policy_for_candidate_offset(offset)


def test_mixed_cfg_interleave_keeps_zero_and_even_primary_with_odd_alternate():
    values = [
        generation_cfg_for_candidate_offset(
            offset,
            primary_cfg=3.0,
            alternate_cfg=2.0,
        )
        for offset in range(20)
    ]

    assert values == [3.0 if offset % 2 == 0 else 2.0 for offset in range(20)]


@pytest.mark.parametrize("offset", [-1, 0.5, True, None, "one"])
def test_mixed_cfg_mapping_rejects_invalid_offsets(offset):
    with pytest.raises(ValueError, match="non-negative integer"):
        generation_cfg_for_candidate_offset(offset)


@pytest.mark.parametrize(
    "kwargs",
    (
        {"primary_cfg": float("nan")},
        {"alternate_cfg": 0.0},
        {"primary_cfg": 4.1},
    ),
)
def test_mixed_cfg_mapping_rejects_values_outside_release_range(kwargs):
    with pytest.raises(ValueError, match=r"\[1\.0, 4\.0\]"):
        generation_cfg_for_candidate_offset(0, **kwargs)


def test_explicit_request_seed_is_forwarded_without_calling_random_factory():
    calls = []

    def factory(limit):
        calls.append(limit)
        return 456

    assert resolve_request_seed(123, factory) == 123
    assert calls == []
    assert resolve_request_seed(None, factory) == 456
    assert calls == [2**31]


@pytest.mark.parametrize("seed", [-1, 2**31, 1.0, True, "123", None])
def test_request_seed_validation_fails_closed(seed):
    factory = lambda _limit: 2**31 if seed is None else 0
    with pytest.raises(ValueError, match=r"\[0, 2147483648\)"):
        resolve_request_seed(seed, factory)


def test_quality_gate_accepts_only_eval_canonicalized_pronoun_homophones():
    result = verify_candidate(
        CandidateObservation(
            target_text="她提醒我",
            transcript_text="他提醒我",
            audio_duration_seconds=1.0,
        )
    )

    assert result.passed
    assert result.comparison.cer == 0.0
    assert result.comparison.prefix_cer == 0.0
    assert result.comparison.suffix_cer == 0.0


class _FakeWhisperModel:
    def __init__(self):
        self.device = None
        self.evaluated = False
        self.generate_calls = []

    def to(self, device):
        self.device = torch.device(device)
        return self

    def eval(self):
        self.evaluated = True
        return self

    def generate(self, features, **kwargs):
        self.generate_calls.append((features, kwargs))
        return torch.tensor([[1, 2, 3]]).repeat(features.shape[0], 1)


class _FakeProcessor:
    def __init__(self):
        self.calls = []
        self.attention_masks = []

    def __call__(self, waveform, **kwargs):
        copied = (
            [segment.copy() for segment in waveform]
            if isinstance(waveform, list)
            else waveform.copy()
        )
        self.calls.append((copied, kwargs))
        batch_size = len(waveform) if isinstance(waveform, list) else 1
        attention_mask = torch.arange(10).remainder(2).repeat(batch_size, 1)
        self.attention_masks.append(attention_mask)
        return SimpleNamespace(
            input_features=torch.ones(batch_size, 80, 10),
            attention_mask=attention_mask,
        )

    def batch_decode(self, token_ids, **kwargs):
        assert all(row == [1, 2, 3] for row in token_ids.tolist())
        assert kwargs == {"skip_special_tokens": True}
        return ["  合成內容完整。  "] * token_ids.shape[0]


class _StatsEncoder:
    def __init__(self, *, invalid=False):
        self.lengths = []
        self.wav_lens = []
        self.invalid = invalid

    def encode_batch(self, tensor, wav_lens=None):
        self.lengths.append(tensor.shape[-1])
        if wav_lens is None:
            wav_lens = torch.ones(tensor.shape[0], device=tensor.device)
        self.wav_lens.append(wav_lens.detach().cpu().numpy())
        if self.invalid:
            return torch.full(
                (tensor.shape[0], 1, 2),
                math.nan,
                device=tensor.device,
            )
        rows = []
        for row, relative_length in zip(tensor, wav_lens, strict=True):
            sample_count = max(1, int(round(float(relative_length) * tensor.shape[-1])))
            active = row[:sample_count]
            rows.append(
                torch.stack(
                    (
                        active.mean(),
                        active.std(unbiased=False),
                        active.abs().amax(),
                    )
                )
            )
        return torch.stack(rows)


def _tone(sample_rate=16_000, seconds=1.0, amplitude=0.2, frequency=220.0):
    timeline = np.arange(round(sample_rate * seconds), dtype=np.float32) / sample_rate
    return amplitude * np.sin(2.0 * np.pi * frequency * timeline).astype(np.float32)


def _gate_result(passed, score=0.0):
    return TrajectoryGateResult(
        passed=passed,
        candidate_results=(),
        score=score if passed else math.inf,
        rejection_reasons=() if passed else ("rejected",),
    )


def _chunk_verification(target, transcript, *, embedding, rms_db):
    observation = CandidateObservation(
        target_text=target,
        transcript_text=transcript,
        audio_duration_seconds=2.0,
        speaker_similarity=0.8,
        begin_speaker_similarity=0.8,
        end_speaker_similarity=0.8,
    )
    return observation, ChunkCandidateArtifact(
        speaker_embedding=(
            None if embedding is None else np.asarray(embedding, dtype=np.float32)
        ),
        rms_db=rms_db,
    )


def _whole_speaker_verification(*, similarity, boundary_drop, passed=True):
    target = "完整而且穩定的候選內容"
    observation = CandidateObservation(
        target_text=target,
        transcript_text=target if passed else "錯誤內容",
        audio_duration_seconds=2.5,
        speaker_similarity=similarity,
        begin_speaker_similarity=0.60,
        end_speaker_similarity=0.60 - boundary_drop,
    )
    artifact = ChunkCandidateArtifact(
        speaker_embedding=np.array([1.0, 0.0], dtype=np.float32),
        rms_db=-20.0,
    )
    return verify_trajectory(
        [observation],
        chunk_artifacts=[artifact],
        min_speaker_similarity=0.10,
        max_boundary_speaker_drop=0.10,
    )


def _joined_verification(target, transcript):
    return verify_trajectory(
        [
            CandidateObservation(
                target_text=target,
                transcript_text=transcript,
                audio_duration_seconds=3.0,
                speaker_similarity=0.8,
                begin_speaker_similarity=0.8,
                end_speaker_similarity=0.8,
            )
        ]
    )


def test_pinned_whisper_loader_uses_exact_revision_without_global_download():
    processor = _FakeProcessor()
    model = _FakeWhisperModel()
    processor_factory = _Factory(processor)
    model_factory = _Factory(model)

    runtime = load_pinned_whisper_runtime(
        device="cpu",
        processor_factory=processor_factory,
        model_factory=model_factory,
    )

    assert runtime.processor is processor
    assert runtime.model is model
    assert runtime.device == torch.device("cpu")
    assert runtime.dtype == torch.float32
    assert model.evaluated
    assert processor_factory.calls == [
        ((WHISPER_MODEL_ID,), {"revision": WHISPER_REVISION})
    ]
    model_args, model_kwargs = model_factory.calls[0]
    assert model_args == (WHISPER_MODEL_ID,)
    assert model_kwargs["revision"] == WHISPER_REVISION
    assert model_kwargs["attn_implementation"] == WHISPER_ATTENTION_IMPLEMENTATION
    assert model_kwargs["torch_dtype"] == torch.float32
    assert model_kwargs["use_safetensors"] is True
    assert WHISPER_ATTENTION_IMPLEMENTATION == "eager"
    assert WHISPER_RETURN_ATTENTION_MASK is True


def test_pinned_verification_whisper_loader_is_separate_and_exact():
    processor = _FakeProcessor()
    model = _FakeWhisperModel()
    processor_factory = _Factory(processor)
    model_factory = _Factory(model)

    runtime = load_pinned_verification_whisper_runtime(
        device="cpu",
        processor_factory=processor_factory,
        model_factory=model_factory,
    )

    assert runtime.processor is processor
    assert runtime.model is model
    assert processor_factory.calls == [
        (
            (VERIFICATION_WHISPER_MODEL_ID,),
            {"revision": VERIFICATION_WHISPER_REVISION},
        )
    ]
    model_args, model_kwargs = model_factory.calls[0]
    assert model_args == (VERIFICATION_WHISPER_MODEL_ID,)
    assert model_kwargs["revision"] == VERIFICATION_WHISPER_REVISION
    assert model_kwargs["attn_implementation"] == WHISPER_ATTENTION_IMPLEMENTATION
    assert model_kwargs["torch_dtype"] == torch.float32
    assert model_kwargs["use_safetensors"] is True


def test_lazy_whisper_runtime_loads_once_and_transcription_is_deterministic():
    processor = _FakeProcessor()
    model = _FakeWhisperModel()
    runtime = WhisperRuntime(processor, model, torch.device("cpu"), torch.float32)
    loads = []
    lazy_asr = LazyWhisperASR(lambda: loads.append("load") or runtime)
    audio_8khz = _tone(sample_rate=8_000, seconds=0.5)

    first = transcribe_whisper(audio_8khz, 8_000, lazy_asr=lazy_asr)
    second = transcribe_whisper(audio_8khz, 8_000, lazy_asr=lazy_asr)

    assert first == second == "合成內容完整。"
    assert loads == ["load"]
    assert processor.calls[0][0].shape == (8_000,)
    assert processor.calls[0][1] == {
        "sampling_rate": 16_000,
        "return_tensors": "pt",
        "return_attention_mask": True,
    }
    _, generation_kwargs = model.generate_calls[0]
    attention_mask = generation_kwargs.pop("attention_mask")
    assert torch.equal(attention_mask, processor.attention_masks[0])
    assert attention_mask.device == runtime.device
    assert generation_kwargs == {
        "language": "zh",
        "task": "transcribe",
        "do_sample": False,
        "num_beams": 1,
        "max_new_tokens": 128,
    }


def test_verification_transcriber_uses_injected_full_runtime_without_turbo_state():
    processor = _FakeProcessor()
    model = _FakeWhisperModel()
    runtime = WhisperRuntime(processor, model, torch.device("cpu"), torch.float32)
    loads = []
    lazy_asr = LazyWhisperASR(lambda: loads.append("verification") or runtime)

    transcript = transcribe_verification_whisper(
        _tone(seconds=0.5),
        16_000,
        lazy_asr=lazy_asr,
        max_new_tokens=440,
    )

    assert transcript == "合成內容完整。"
    assert loads == ["verification"]
    assert model.generate_calls[0][1]["max_new_tokens"] == 440


def test_long_whisper_audio_is_batched_below_thirty_second_limit():
    processor = _FakeProcessor()
    model = _FakeWhisperModel()
    runtime = WhisperRuntime(processor, model, torch.device("cpu"), torch.float32)
    audio = np.concatenate(
        (
            _tone(seconds=21.0),
            np.zeros(16_000, dtype=np.float32),
            _tone(seconds=21.0, frequency=240.0),
            np.zeros(16_000, dtype=np.float32),
            _tone(seconds=21.0, frequency=260.0),
        )
    )

    transcript = transcribe_whisper(audio, 16_000, runtime=runtime, max_new_tokens=440)

    segments = processor.calls[0][0]
    assert isinstance(segments, list)
    assert len(segments) == 3
    assert all(segment.size <= 30 * 16_000 for segment in segments)
    assert transcript == "合成內容完整。 合成內容完整。 合成內容完整。"
    assert len(model.generate_calls) == 1
    assert model.generate_calls[0][1]["max_new_tokens"] == 440


def test_transcription_rejects_nonfinite_audio_and_conflicting_injection():
    processor = _FakeProcessor()
    model = _FakeWhisperModel()
    runtime = WhisperRuntime(processor, model, torch.device("cpu"), torch.float32)
    with pytest.raises(ValueError, match="non-finite"):
        transcribe_whisper(np.array([0.0, math.nan]), 16_000, runtime=runtime)
    with pytest.raises(ValueError, match="either runtime or lazy_asr"):
        transcribe_whisper(
            np.ones(20),
            16_000,
            runtime=runtime,
            lazy_asr=LazyWhisperASR(lambda: runtime),
        )


def test_candidate_audio_data_or_asr_value_error_becomes_rejection_evidence():
    calls = []

    def transcriber(waveform, sample_rate):
        calls.append((waveform.copy(), sample_rate))
        return "  內容完整  "

    prepared = prepare_candidate_audio(_tone(seconds=0.25), 16_000, transcriber=transcriber)
    assert prepared is not None
    assert prepared.transcript_text == "內容完整"
    assert prepared.duration_seconds == pytest.approx(0.25)
    assert len(calls) == 1

    assert prepare_candidate_audio(
        [0.0, math.nan],
        16_000,
        transcriber=transcriber,
    ) is None
    assert prepare_candidate_audio(
        _tone(seconds=0.25),
        16_000,
        transcriber=lambda *_: (_ for _ in ()).throw(ValueError("bad candidate")),
    ) is None
    assert len(calls) == 1


def test_candidate_audio_infrastructure_error_propagates_fail_closed():
    with pytest.raises(RuntimeError, match="ASR unavailable"):
        prepare_candidate_audio(
            _tone(seconds=0.25),
            16_000,
            transcriber=lambda *_: (_ for _ in ()).throw(RuntimeError("ASR unavailable")),
        )


def test_ecapa_ndarray_helper_trims_resamples_normalizes_and_handles_channels():
    encoder = _StatsEncoder()
    tone = _tone(sample_rate=8_000, seconds=1.0)
    padded = np.concatenate((np.zeros(4_000), tone, np.zeros(4_000))).astype(np.float32)
    stereo = np.stack((padded, padded), axis=0)

    embedding = speaker_embedding_from_audio(stereo, 8_000, encoder)

    assert embedding.shape == (3,)
    assert np.isfinite(embedding).all()
    assert np.linalg.norm(embedding) == pytest.approx(1.0, abs=1e-6)
    assert 16_000 <= encoder.lengths[0] < 32_000
    assert len(encoder.lengths) == 1


def test_ecapa_ndarray_helper_fails_closed_on_silence_nan_and_bad_embedding():
    with pytest.raises(ValueError, match="active speech"):
        speaker_embedding_from_audio(np.zeros(16_000), 16_000, _StatsEncoder())
    with pytest.raises(ValueError, match="non-finite samples"):
        speaker_embedding_from_audio(
            np.array([0.0, math.nan], dtype=np.float32),
            16_000,
            _StatsEncoder(),
        )
    with pytest.raises(ValueError, match="invalid embedding"):
        speaker_embedding_from_audio(_tone(), 16_000, _StatsEncoder(invalid=True))


def test_cosine_and_speaker_evidence_are_finite_and_directional():
    encoder = _StatsEncoder()
    audio = np.concatenate((_tone(seconds=1.0), _tone(seconds=1.0, amplitude=0.1)))
    anchor = speaker_embedding_from_audio(audio, 16_000, encoder)

    evidence = speaker_evidence_from_audio(
        audio,
        16_000,
        encoder,
        anchor,
        edge_seconds=0.75,
    )

    # Anchor extraction is one call; whole/begin/end evidence is one batched
    # call instead of three repeated ECAPA forwards for the same candidate.
    assert len(encoder.lengths) == 2
    assert encoder.wav_lens[-1].shape == (3,)
    assert evidence.similarity == pytest.approx(1.0, abs=1e-5)
    assert -1.0 <= evidence.begin_similarity <= 1.0
    assert -1.0 <= evidence.end_similarity <= 1.0
    assert evidence.boundary_drop == max(
        0.0,
        evidence.begin_similarity - evidence.end_similarity,
    )
    assert evidence.active_duration_seconds > 1.5
    assert cosine_similarity(anchor, anchor) == pytest.approx(1.0)
    with pytest.raises(ValueError):
        cosine_similarity([0.0, 0.0], [0.0, 0.0])


def test_long_speaker_evidence_uses_one_bounded_window_batch():
    encoder = _StatsEncoder()
    audio = _tone(seconds=20.0)
    anchor = np.array([0.0, 0.0, 1.0], dtype=np.float32)

    evidence = speaker_evidence_from_audio(audio, 16_000, encoder, anchor)

    assert len(encoder.lengths) == 1
    assert encoder.lengths[0] <= 3 * 16_000
    assert encoder.wav_lens[0].shape == (6,)
    assert evidence.active_duration_seconds == pytest.approx(20.0, abs=0.05)
    assert evidence.speaker_embedding.shape == (3,)
    assert math.isfinite(evidence.active_rms_db)


def test_release_speaker_evidence_uses_independent_trim_full_audio_and_thirds():
    encoder = _StatsEncoder()
    audio = np.concatenate(
        (
            np.zeros(16_000, dtype=np.float32),
            _tone(seconds=3.0),
            np.zeros(16_000, dtype=np.float32),
        )
    )
    active = trim_release_speaker_activity(audio, 16_000)
    assert 3.0 <= active.size / 16_000 <= 3.2

    evidence = release_speaker_evidence_from_audio(
        audio,
        16_000,
        encoder,
        np.array([0.0, 0.0, 1.0], dtype=np.float32),
    )

    # Full active audio and each disjoint third are independently encoded,
    # matching the external promotion verifier instead of a padded batch.
    assert len(encoder.lengths) == 4
    assert encoder.lengths[0] == active.size
    assert max(encoder.lengths[1:]) - min(encoder.lengths[1:]) <= 1
    assert evidence.active_duration_seconds == pytest.approx(3.0, abs=0.05)
    assert evidence.boundary_drop == max(
        0.0,
        evidence.begin_similarity - evidence.end_similarity,
    )
    assert np.linalg.norm(evidence.speaker_embedding) == pytest.approx(1.0, abs=1e-6)


def test_active_union_pace_ignores_one_or_three_second_internal_silence():
    def waveform(pause_seconds):
        return np.concatenate(
            (
                _tone(seconds=0.75),
                np.zeros(round(pause_seconds * 16_000), dtype=np.float32),
                _tone(seconds=0.75, frequency=260.0),
            )
        )

    one_second = active_voiced_duration_seconds(waveform(1.0), 16_000)
    three_seconds = active_voiced_duration_seconds(waveform(3.0), 16_000)

    assert one_second == pytest.approx(three_seconds, abs=1.0e-9)
    assert 8 / one_second == pytest.approx(8 / three_seconds, abs=1.0e-9)
    assert one_second < 1.6


def test_active_audio_median_f0_tracks_tone_and_fails_closed_on_silence():
    measured = active_audio_median_f0_hz(
        _tone(seconds=1.0, frequency=220.0),
        16_000,
    )

    assert measured == pytest.approx(220.0, abs=3.0)
    assert active_audio_median_f0_hz(np.zeros(16_000), 16_000) is None
    assert active_audio_median_f0_hz(_tone(seconds=0.02), 16_000) is None


def test_long_whisper_split_prefers_sustained_production_pauses():
    sample_rate = 1_000
    tone = np.full(20 * sample_rate, 0.2, dtype=np.float32)
    pause = np.zeros(round(0.30 * sample_rate), dtype=np.float32)
    waveform = np.concatenate((tone, pause, tone, pause, tone))

    segments = _split_whisper_audio(
        waveform,
        sample_rate=sample_rate,
        max_segment_seconds=28.0,
    )

    durations = [segment.size / sample_rate for segment in segments]
    assert len(segments) == 3
    assert all(duration <= 28.0 for duration in durations)
    assert durations[0] == pytest.approx(20.15, abs=0.08)
    assert durations[1] == pytest.approx(20.30, abs=0.12)


def test_long_whisper_split_fails_closed_without_qualified_silence():
    waveform = np.full(61_000, 0.2, dtype=np.float32)

    with pytest.raises(ValueError, match="no qualified 250ms pause"):
        _split_whisper_audio(
            waveform,
            sample_rate=1_000,
            max_segment_seconds=28.0,
        )


def test_active_union_duration_controls_speaker_eligibility_at_1_5_seconds():
    below = active_voiced_duration_seconds(_tone(seconds=1.49), 16_000)
    boundary = active_voiced_duration_seconds(_tone(seconds=1.50), 16_000)
    assert below == pytest.approx(1.49, abs=1.0e-9)
    assert boundary == pytest.approx(1.50, abs=1.0e-9)

    common = {
        "target_text": "這是一段完整而且穩定的合成內容",
        "transcript_text": "這是一段完整而且穩定的合成內容",
    }
    short_result = verify_candidate(
        CandidateObservation(**common, audio_duration_seconds=below)
    )
    boundary_result = verify_candidate(
        CandidateObservation(**common, audio_duration_seconds=boundary)
    )
    assert short_result.passed
    assert not short_result.speaker_gate_applied
    assert not boundary_result.passed
    assert boundary_result.speaker_gate_applied
    assert "missing_speaker_evidence" in boundary_result.rejection_reasons


def test_short_candidate_uses_exact_semantics_but_bypasses_speaker_hard_gate():
    passed = verify_candidate(
        CandidateObservation(
            target_text="謝謝你",
            transcript_text="謝謝你",
            audio_duration_seconds=1.49,
        )
    )
    failed = verify_candidate(
        CandidateObservation(
            target_text="謝謝你",
            transcript_text="謝謝",
            audio_duration_seconds=1.0,
        )
    )

    assert passed.passed
    assert not passed.speaker_gate_applied
    assert passed.speaker_similarity is None
    assert passed.score == 0.0
    assert not failed.passed
    assert failed.rejection_reasons == ("semantic_gate",)


def test_semantic_only_gate_never_requires_speaker_evidence_but_stays_strict():
    long_target = "這是一段超過短音訊門檻而且內容完整的驗證文字"
    passed = verify_candidate(
        CandidateObservation(
            target_text=long_target,
            transcript_text=long_target,
            audio_duration_seconds=10.0,
        ),
        speaker_gate_enabled=False,
    )
    failed = verify_candidate(
        CandidateObservation(
            target_text=long_target,
            transcript_text=long_target[:-2],
            audio_duration_seconds=10.0,
        ),
        speaker_gate_enabled=False,
    )
    invalid = verify_candidate(
        CandidateObservation(
            target_text=long_target,
            transcript_text=long_target,
            audio_duration_seconds=10.0,
        ),
        speaker_gate_enabled="no",
    )

    assert passed.passed
    assert not passed.speaker_gate_applied
    assert passed.score == 0.0
    assert not failed.passed
    assert failed.rejection_reasons == ("semantic_gate",)
    assert not invalid.passed
    assert "invalid_gate_config" in invalid.rejection_reasons


def test_quality_gate_rejects_inexact_network_span_below_whole_cer_limit():
    target = normalize_spoken_forms(
        "請先閱讀 https://museum.example.tw/path，確認展覽時間與集合位置後再回覆。"
    )
    transcript = target.replace(
        "museum",
        "museuman",
    )
    result = verify_candidate(
        CandidateObservation(
            target_text=target,
            transcript_text=transcript,
            audio_duration_seconds=5.0,
        ),
        speaker_gate_enabled=False,
        max_prefix_cer=1.0,
        max_suffix_cer=1.0,
    )

    assert result.comparison.cer <= 0.20
    assert result.comparison.network_protected_spans == 1
    assert result.comparison.network_protected_spans_passed is False
    assert result.rejection_reasons == (
        "semantic_gate",
        "network_protected_span_mismatch",
    )

    adjusted = _sequence_fallback_candidate_result(
        result,
        max_local_boundary_speaker_drop=0.15,
    )
    assert adjusted is result
    assert not adjusted.passed


def test_candidate_pace_gate_is_optional_and_fails_closed_when_enabled():
    base = CandidateObservation(
        target_text="內容完整",
        transcript_text="內容完整",
        audio_duration_seconds=1.0,
    )

    assert verify_candidate(base).passed
    assert verify_candidate(
        CandidateObservation(**{**base.__dict__, "pace_cps": 4.2}),
        max_pace_cps=4.3,
    ).passed
    missing = verify_candidate(base, max_pace_cps=4.3)
    fast = verify_candidate(
        CandidateObservation(**{**base.__dict__, "pace_cps": 4.31}),
        max_pace_cps=4.3,
    )

    assert missing.rejection_reasons == ("missing_pace_evidence",)
    assert fast.rejection_reasons == ("pace_too_fast",)


def test_candidate_squim_gate_rejects_low_or_missing_quality_evidence():
    base = {
        "target_text": "內容完整",
        "transcript_text": "內容完整",
        "audio_duration_seconds": 1.0,
        "squim_stoi": 0.80,
        "squim_pesq": 1.50,
        "squim_si_sdr": -12.0,
    }

    passed = verify_candidate(
        CandidateObservation(**base),
        squim_gate_enabled=True,
        min_squim_stoi=0.72,
        min_squim_pesq=1.20,
    )
    low_stoi = verify_candidate(
        CandidateObservation(**{**base, "squim_stoi": 0.599}),
        squim_gate_enabled=True,
    )
    low_pesq = verify_candidate(
        CandidateObservation(**{**base, "squim_pesq": 1.119}),
        squim_gate_enabled=True,
    )
    missing = verify_candidate(
        CandidateObservation(
            target_text="內容完整",
            transcript_text="內容完整",
            audio_duration_seconds=1.0,
        ),
        squim_gate_enabled=True,
    )

    assert passed.passed
    assert passed.squim_gate_applied
    assert passed.squim_stoi == pytest.approx(0.80)
    assert passed.squim_pesq == pytest.approx(1.50)
    assert passed.squim_si_sdr == pytest.approx(-12.0)
    expected_cost = (
        0.03 * (1.0 - 0.80)
        + 0.02 * ((4.5 - 1.50) / 3.5)
        + 0.01 * ((20.0 - -12.0) / 40.0)
    )
    assert passed.squim_quality_cost == pytest.approx(expected_cost)
    assert passed.score == pytest.approx(expected_cost)
    assert low_stoi.rejection_reasons == ("squim_stoi_too_low",)
    assert low_pesq.rejection_reasons == ("squim_pesq_too_low",)
    assert missing.rejection_reasons == ("missing_squim_evidence",)


@pytest.mark.parametrize(
    "kwargs",
    [
        {"squim_gate_enabled": "yes"},
        {"squim_gate_enabled": True, "min_squim_stoi": float("nan")},
        {"squim_gate_enabled": True, "min_squim_stoi": 1.01},
        {"squim_gate_enabled": True, "min_squim_pesq": 5.01},
        {"squim_gate_enabled": True, "squim_stoi_weight": -0.01},
        {"squim_gate_enabled": True, "squim_pesq_weight": float("nan")},
        {"squim_gate_enabled": True, "squim_si_sdr_weight": -0.01},
    ],
)
def test_candidate_squim_gate_rejects_invalid_config(kwargs):
    observation = CandidateObservation(
        target_text="內容完整",
        transcript_text="內容完整",
        audio_duration_seconds=1.0,
        squim_stoi=0.8,
        squim_pesq=1.5,
        squim_si_sdr=2.0,
    )

    result = verify_candidate(observation, **kwargs)

    assert not result.passed
    assert "invalid_gate_config" in result.rejection_reasons


def test_long_candidate_requires_speaker_evidence_and_clamps_negative_drop():
    observation = CandidateObservation(
        target_text="這是一段完整而且穩定的合成內容",
        transcript_text="這是一段完整而且穩定的合成內容",
        audio_duration_seconds=2.0,
        speaker_similarity=0.70,
        begin_speaker_similarity=0.50,
        end_speaker_similarity=0.60,
    )
    result = verify_candidate(observation)

    assert result.passed
    assert result.speaker_gate_applied
    assert result.boundary_speaker_drop == 0.0
    assert result.score == pytest.approx(0.05 * 0.30)

    missing = verify_candidate(
        CandidateObservation(
            target_text=observation.target_text,
            transcript_text=observation.transcript_text,
            audio_duration_seconds=2.0,
        )
    )
    assert not missing.passed
    assert "missing_speaker_evidence" in missing.rejection_reasons


def test_candidate_rejects_boundary_drop_low_similarity_truncation_and_tail():
    base = dict(
        target_text="這是一段完整而且穩定的合成內容",
        transcript_text="這是一段完整而且穩定的合成內容",
        audio_duration_seconds=2.0,
        speaker_similarity=0.7,
        begin_speaker_similarity=0.5,
        end_speaker_similarity=0.5,
    )
    boundary = verify_candidate(
        CandidateObservation(**{**base, "begin_speaker_similarity": 0.8})
    )
    similarity = verify_candidate(
        CandidateObservation(**{**base, "speaker_similarity": 0.09})
    )
    truncation = verify_candidate(CandidateObservation(**{**base, "truncated": True}))
    tail = verify_candidate(
        CandidateObservation(**{**base, "transcript_text": base["target_text"] + "多講"})
    )

    assert boundary.rejection_reasons == ("boundary_speaker_drop",)
    assert similarity.rejection_reasons == ("speaker_similarity",)
    assert truncation.rejection_reasons == ("truncated",)
    assert "semantic_gate" in tail.rejection_reasons
    # ASR comparison canonicalizes both sides to Simplified Chinese.
    assert tail.comparison.extra_tail == "多讲"


def test_candidate_and_trajectory_fail_closed_for_bad_values_or_any_bad_chunk():
    good_short = CandidateObservation("內容完整", "內容完整", 1.0)
    bad_short = CandidateObservation("句尾完整", "句尾", 1.0)
    invalid = verify_candidate(
        CandidateObservation("內容完整", "內容完整", math.nan),
        max_cer=math.nan,
    )
    passed_trajectory = verify_trajectory([good_short, good_short])
    failed_trajectory = verify_trajectory([good_short, bad_short])

    assert not invalid.passed
    assert "invalid_audio_duration" in invalid.rejection_reasons
    assert "invalid_gate_config" in invalid.rejection_reasons
    assert passed_trajectory.passed
    assert passed_trajectory.score == 0.0
    assert not failed_trajectory.passed
    assert math.isinf(failed_trajectory.score)
    assert failed_trajectory.rejection_reasons == ("chunk_1:semantic_gate",)
    assert verify_trajectory([]).rejection_reasons == ("empty_trajectory",)


def test_role_aware_local_endpoints_allow_only_internal_substitutions():
    target = "甲乙丙丁戊己庚辛壬癸子丑"
    internal_substitutions = CandidateObservation(
        target,
        "寅乙丙丁戊己庚辛壬癸子卯",
        1.0,
    )
    exact = CandidateObservation(target, target, 1.0)
    common_gate = {
        "speaker_gate_enabled": False,
        "short_text_units": 0,
        "max_cer": 0.20,
        "max_prefix_cer": 1.0 / 6.0,
        "max_suffix_cer": 1.0 / 6.0,
        "max_prefix_deletions": 0,
        "max_suffix_deletions": 0,
    }

    internal = verify_trajectory(
        (exact, internal_substitutions, exact),
        candidate_gate_kwargs_by_index=(
            {"max_prefix_cer": 0.0},
            {},
            {"max_suffix_cer": 0.0},
        ),
        **common_gate,
    )
    absolute_first = verify_trajectory(
        (internal_substitutions, exact),
        candidate_gate_kwargs_by_index=(
            {"max_prefix_cer": 0.0},
            {"max_suffix_cer": 0.0},
        ),
        **common_gate,
    )
    absolute_last = verify_trajectory(
        (exact, internal_substitutions),
        candidate_gate_kwargs_by_index=(
            {"max_prefix_cer": 0.0},
            {"max_suffix_cer": 0.0},
        ),
        **common_gate,
    )
    deleted_prefix = verify_trajectory(
        (
            exact,
            CandidateObservation(target, target[1:], 1.0),
            exact,
        ),
        candidate_gate_kwargs_by_index=(
            {"max_prefix_cer": 0.0},
            {},
            {"max_suffix_cer": 0.0},
        ),
        **common_gate,
    )

    assert internal.passed
    assert absolute_first.rejection_reasons == ("chunk_0:semantic_gate",)
    assert absolute_last.rejection_reasons == ("chunk_1:semantic_gate",)
    assert deleted_prefix.rejection_reasons == ("chunk_1:semantic_gate",)


@pytest.mark.parametrize("invalid_limit", [True, np.bool_(False), 0.0, "0", -1])
def test_deletion_limits_reject_bool_noninteger_and_negative_values(invalid_limit):
    result = verify_candidate(
        CandidateObservation("內容完整", "內容完整", 1.0),
        speaker_gate_enabled=False,
        max_prefix_deletions=invalid_limit,
    )

    assert not result.passed
    assert result.rejection_reasons == ("invalid_gate_config",)


@pytest.mark.parametrize(
    "overrides",
    [
        ({"speaker_gate_enabled": False},),
        ({"max_prefix_cer": 0.0, "max_cer": 1.0},),
        ({"max_prefix_deletions": None},),
        ({"max_prefix_cer": 0.0}, {}),
        ([],),
    ],
)
def test_per_index_gate_overrides_are_aligned_dicts_with_endpoint_keys_only(
    overrides,
):
    result = verify_trajectory(
        (CandidateObservation("內容完整", "內容完整", 1.0),),
        candidate_gate_kwargs_by_index=overrides,
        speaker_gate_enabled=False,
    )

    assert not result.passed
    assert result.rejection_reasons == ("invalid_candidate_gate_overrides",)


def test_final_whole_output_gate_rejects_post_join_regression_without_fallback():
    passed = verify_trajectory(
        [CandidateObservation("整段內容完整", "整段內容完整", 1.0)]
    )
    rejected = verify_trajectory(
        [CandidateObservation("整段內容完整", "整段內容", 1.0)]
    )

    assert require_verified_final_output(passed) is passed
    with pytest.raises(FinalOutputRejectedError, match="final output rejected"):
        require_verified_final_output(rejected)
    with pytest.raises(FinalOutputRejectedError, match="invalid result"):
        require_verified_final_output(None)


def test_whole_waveform_cache_reuses_only_exact_audio_target_rate_and_profile():
    cache = WholeWaveformVerificationCache()
    waveform = _tone(seconds=0.25)
    calls = []

    def verifier(exact_waveform, sample_rate, target_text):
        calls.append((exact_waveform.copy(), sample_rate, target_text))
        return _joined_verification(target_text, target_text)

    first = cache.verify(waveform, 16_000, "內容完整", "large-v3@one", verifier)
    second = cache.verify(
        waveform.copy(),
        16_000,
        "內容完整",
        "large-v3@one",
        verifier,
    )
    changed = waveform.copy()
    changed[-1] = np.nextafter(changed[-1], np.float32(1.0))
    cache.verify(changed, 16_000, "內容完整", "large-v3@one", verifier)
    cache.verify(waveform, 48_000, "內容完整", "large-v3@one", verifier)
    cache.verify(waveform, 16_000, "另一內容", "large-v3@one", verifier)
    cache.verify(waveform, 16_000, "內容完整", "large-v3@two", verifier)

    assert first is second
    assert len(calls) == 5
    assert cache.entry_count == 5
    assert exact_waveform_sha256(waveform, 16_000) == exact_waveform_sha256(
        waveform.copy(),
        16_000,
    )
    assert exact_waveform_sha256(waveform, 16_000) != exact_waveform_sha256(
        waveform,
        48_000,
    )


def test_whole_waveform_cache_caches_rejections_but_never_errors_or_bad_results():
    waveform = _tone(seconds=0.25)
    rejected = _joined_verification("完整內容", "錯誤內容")
    cache = WholeWaveformVerificationCache()
    rejection_calls = []

    def reject_verifier(*args):
        rejection_calls.append(args)
        return rejected

    assert not cache.verify(
        waveform,
        16_000,
        "完整內容",
        "large-v3@one",
        reject_verifier,
    ).passed
    assert not cache.verify(
        waveform.copy(),
        16_000,
        "完整內容",
        "large-v3@one",
        reject_verifier,
    ).passed
    assert len(rejection_calls) == 1
    assert cache.entry_count == 1

    error_cache = WholeWaveformVerificationCache()
    error_calls = []

    def fail_verifier(*args):
        error_calls.append(args)
        raise RuntimeError("verification OOM")

    for _ in range(2):
        with pytest.raises(RuntimeError, match="OOM"):
            error_cache.verify(
                waveform,
                16_000,
                "完整內容",
                "large-v3@one",
                fail_verifier,
            )
    assert len(error_calls) == 2
    assert error_cache.entry_count == 0

    with pytest.raises(RuntimeError, match="invalid result"):
        error_cache.verify(
            waveform,
            16_000,
            "完整內容",
            "large-v3@one",
            lambda *args: True,
        )
    assert error_cache.entry_count == 0


@pytest.mark.parametrize(
    ("audio", "sample_rate", "target", "profile", "match"),
    [
        ([0.0, math.nan], 16_000, "內容", "profile", "non-finite"),
        ([0.0], 0, "內容", "profile", "sample_rate"),
        ([0.0], True, "內容", "profile", "sample_rate"),
        ([0.0], 16_000, "", "profile", "target_text"),
        ([0.0], 16_000, "內容", "", "verifier_profile"),
    ],
)
def test_whole_waveform_cache_rejects_malformed_keys(
    audio,
    sample_rate,
    target,
    profile,
    match,
):
    cache = WholeWaveformVerificationCache()
    with pytest.raises(ValueError, match=match):
        cache.verify(
            audio,
            sample_rate,
            target,
            profile,
            lambda *args: _joined_verification("內容", "內容"),
        )
    assert cache.entry_count == 0


def test_joined_output_rejection_preserves_local_results_for_sequence_dp():
    observations = []
    artifacts = []
    for target in ("第一段完整", "第二段完整"):
        observation, artifact = _chunk_verification(
            target,
            target,
            embedding=[1.0, 0.0],
            rms_db=-20.0,
        )
        observations.append(observation)
        artifacts.append(artifact)
    local = verify_trajectory(observations, chunk_artifacts=artifacts)
    joined_failed = _joined_verification("第一段完整第二段完整", "第一段完整")

    qualified = qualify_trajectory_with_joined_output(local, joined_failed)

    assert local.passed
    assert not qualified.passed
    assert math.isinf(qualified.score)
    assert qualified.candidate_results is local.candidate_results
    assert qualified.chunk_artifacts is local.chunk_artifacts
    assert qualified.rejection_reasons == ("joined_output:chunk_0:semantic_gate",)
    assert all(result.passed for result in qualified.candidate_results)


def test_joined_output_qualification_returns_local_result_only_when_joined_passes():
    local = verify_trajectory(
        [CandidateObservation("第一段完整", "第一段完整", 1.0)]
    )
    joined = verify_trajectory(
        [CandidateObservation("第一段完整", "第一段完整", 1.0)]
    )

    assert qualify_trajectory_with_joined_output(local, joined) is local


def test_independent_local_semantic_reject_preserves_primary_acoustic_evidence():
    target = normalize_spoken_forms(
        "https://museum.example.tw/path"
    )
    artifact = ChunkCandidateArtifact(
        speaker_embedding=np.asarray([1.0, 0.0], dtype=np.float32),
        rms_db=-20.0,
        median_f0_hz=180.0,
    )
    primary = verify_trajectory(
        [
            CandidateObservation(
                target_text=target,
                transcript_text=target,
                audio_duration_seconds=2.0,
                speaker_similarity=0.70,
                begin_speaker_similarity=0.60,
                end_speaker_similarity=0.55,
                pace_cps=4.0,
                squim_stoi=0.80,
                squim_pesq=1.50,
                squim_si_sdr=4.0,
            )
        ],
        chunk_artifacts=(artifact,),
        max_pace_cps=4.3,
        squim_gate_enabled=True,
        min_squim_stoi=0.60,
        min_squim_pesq=1.12,
        max_boundary_speaker_drop=0.10,
    )
    independent = verify_trajectory(
        [
            CandidateObservation(
                target_text=target,
                transcript_text=target.replace("museum", "museuman"),
                audio_duration_seconds=2.0,
            )
        ],
        speaker_gate_enabled=False,
    )

    combined = intersect_local_semantic_verification(
        primary,
        independent,
        (0,),
    )

    assert primary.passed
    assert independent.rejection_reasons == (
        "chunk_0:semantic_gate",
        "chunk_0:network_protected_span_mismatch",
    )
    assert not combined.passed
    assert math.isinf(combined.score)
    result = combined.candidate_results[0]
    primary_result = primary.candidate_results[0]
    assert result.comparison is primary_result.comparison
    assert result.speaker_similarity == primary_result.speaker_similarity
    assert result.boundary_speaker_drop == primary_result.boundary_speaker_drop
    assert result.squim_stoi == primary_result.squim_stoi
    assert result.squim_pesq == primary_result.squim_pesq
    assert result.squim_si_sdr == primary_result.squim_si_sdr
    assert result.rejection_reasons == (
        "semantic_gate",
        "network_protected_span_mismatch",
    )
    assert combined.chunk_artifacts is primary.chunk_artifacts
    # Boundary-only fallback cannot erase an independent semantic rejection.
    assert (
        _sequence_fallback_candidate_result(
            result,
            max_local_boundary_speaker_drop=0.15,
        )
        is result
    )


def test_independent_local_semantic_pass_preserves_boundary_proxy_unchanged():
    target = "這段網路片段內容完整"
    primary = verify_trajectory(
        [
            CandidateObservation(
                target_text=target,
                transcript_text=target,
                audio_duration_seconds=2.0,
                speaker_similarity=0.70,
                begin_speaker_similarity=0.80,
                end_speaker_similarity=0.68,
            )
        ],
    )
    independent = verify_trajectory(
        [CandidateObservation(target, target, 2.0)],
        speaker_gate_enabled=False,
    )

    combined = intersect_local_semantic_verification(
        primary,
        independent,
        (0,),
    )

    assert not primary.passed
    assert primary.rejection_reasons == ("chunk_0:boundary_speaker_drop",)
    assert independent.passed
    assert combined is primary
    relaxed = _sequence_fallback_candidate_result(
        combined.candidate_results[0],
        max_local_boundary_speaker_drop=0.15,
    )
    assert relaxed.passed
    assert relaxed.speaker_similarity == primary.candidate_results[0].speaker_similarity


@pytest.mark.parametrize(
    "indices",
    [(), (0, 0), (1, 0), (2,), (True,)],
)
def test_independent_local_semantic_intersection_rejects_unbound_indices(
    indices,
):
    primary = verify_trajectory(
        [
            CandidateObservation("第一段", "第一段", 1.0),
            CandidateObservation("第二段", "第二段", 1.0),
        ],
        speaker_gate_enabled=False,
    )
    independent = verify_trajectory(
        [CandidateObservation("第一段", "第一段", 1.0)],
        speaker_gate_enabled=False,
    )

    with pytest.raises(ValueError, match="primary indices"):
        intersect_local_semantic_verification(
            primary,
            independent,
            indices,
        )


def test_local_coverage_eligibility_matches_the_frozen_boundary_proxy():
    target = "網路片段內容完整"
    strict = verify_candidate(
        CandidateObservation(target, target, 1.0),
        speaker_gate_enabled=False,
    )
    boundary = verify_candidate(
        CandidateObservation(
            target,
            target,
            2.0,
            speaker_similarity=0.70,
            begin_speaker_similarity=0.80,
            end_speaker_similarity=0.68,
        )
    )
    semantic = verify_candidate(
        CandidateObservation(target, target[:-1], 1.0),
        speaker_gate_enabled=False,
    )

    assert local_candidate_has_coverage_eligibility(
        strict,
        max_local_boundary_speaker_drop=0.15,
    )
    assert local_candidate_has_coverage_eligibility(
        boundary,
        max_local_boundary_speaker_drop=(
            SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP
        ),
    )
    assert not local_candidate_has_coverage_eligibility(
        boundary,
        max_local_boundary_speaker_drop=0.10,
    )
    assert not local_candidate_has_coverage_eligibility(
        semantic,
        max_local_boundary_speaker_drop=0.15,
    )


def test_canonical_attempt_evidence_distinguishes_local_v3_pass_and_not_run():
    chunks = ("普通內容完整", "網路內容完整")
    primary = verify_trajectory(
        [
            CandidateObservation(chunks[0], chunks[0], 1.0),
            CandidateObservation(chunks[1], chunks[1], 1.0),
        ],
        speaker_gate_enabled=False,
    )
    bounded_result = candidate_gate_evidence(primary.candidate_results[1])
    local_independent = (
        LocalIndependentGateEvidence(False, None, 0, None),
        LocalIndependentGateEvidence(True, True, 1, bounded_result),
    )

    result = run_adaptive_cascade(
        chunks,
        900,
        lambda candidate_chunks, seed: tuple(candidate_chunks),
        lambda trajectory, candidate_chunks, seed: CandidateVerification(
            primary,
            independent_local_results=local_independent,
        ),
        max_candidates=1,
    )
    payload = json.loads(
        format_cascade_evidence_log(
            result.diagnostics,
            outcome="returned",
            generated_chunk_limit=32,
            selection=result,
        ).removeprefix(CASCADE_EVIDENCE_LOG_PREFIX)
    )
    attempt = payload["attempts"][0]

    assert payload["schema_version"] == 6
    assert attempt["independent_local_evidence_complete"] is True
    assert attempt["independent_local_results"] == [
        {
            "attempted": False,
            "passed": None,
            "proof_count": 0,
            "result": None,
        },
        {
            "attempted": True,
            "passed": True,
            "proof_count": 1,
            "result": attempt["local_results"][1],
        },
    ]


@pytest.mark.parametrize(
    "bad_evidence",
    [
        (LocalIndependentGateEvidence(False, None, 1.0, None),),
        (LocalIndependentGateEvidence(False, None, 33, None),),
        (LocalIndependentGateEvidence(True, None, 1, None),),
    ],
)
def test_candidate_verification_rejects_malformed_local_v3_attestation(
    bad_evidence,
):
    verification = verify_trajectory(
        [CandidateObservation("內容", "內容", 1.0)],
        speaker_gate_enabled=False,
    )

    with pytest.raises(TypeError, match="independent-local"):
        run_adaptive_cascade(
            ("內容",),
            901,
            lambda chunks, seed: tuple(chunks),
            lambda trajectory, chunks, seed: CandidateVerification(
                verification,
                independent_local_results=bad_evidence,
            ),
            max_candidates=1,
        )


def test_cascade_evidence_success_is_content_free_canonical_and_behavior_golden():
    private_target = "PRIVATE_TARGET_ALPHA"
    private_transcript = "PRIVATE_TRANSCRIPT_OMEGA"
    calls = []

    def generator(chunks, seed):
        calls.append(seed)
        return (f"audio-{seed}",)

    def verifier(trajectory, chunks, seed):
        passed = seed == 103
        verification = verify_trajectory(
            [
                CandidateObservation(
                    target_text=chunks[0],
                    transcript_text=(chunks[0] if passed else private_transcript),
                    audio_duration_seconds=1.0,
                )
            ]
        )
        if seed == 100:
            verification = TrajectoryGateResult(
                passed=False,
                candidate_results=verification.candidate_results,
                score=math.inf,
                rejection_reasons=("PRIVATE_REASON_MUST_NOT_LEAK",),
                chunk_artifacts=verification.chunk_artifacts,
            )
        joined = (
            trajectory_gate_evidence(_joined_verification(chunks[0], chunks[0]))
            if passed
            else None
        )
        return CandidateVerification(
            verification,
            joined_output=joined,
            independent_output=joined,
        )

    result = run_adaptive_cascade(
        (private_target,),
        100,
        generator,
        verifier,
        max_candidates=5,
    )
    final_output = trajectory_gate_evidence(
        _joined_verification(private_target, private_target)
    )
    line = format_cascade_evidence_log(
        result.diagnostics,
        outcome="returned",
        generated_chunk_limit=5,
        selection=result,
        final_output=final_output,
        independent_final_output=final_output,
    )

    # Frozen pre-diagnostics behavior: the whole stage is generated, then the
    # only passing trajectory at offset three wins with unchanged seed fields.
    assert calls == [100, 101, 102, 103, 104]
    assert (
        result.selection_mode,
        result.candidate_index,
        result.seed,
        result.attempted_seeds,
        result.chunk_candidate_indices,
        result.chunk_seeds,
    ) == (
        "whole_trajectory",
        3,
        103,
        (100, 101, 102, 103, 104),
        (3,),
        (103,),
    )
    assert len(result.diagnostics.attempts) == 5
    assert result.diagnostics.attempts[3].joined_output is not None
    assert result.diagnostics.attempts[3].joined_output.passed
    assert result.diagnostics.attempts[3].independent_output is not None
    assert result.diagnostics.attempts[3].independent_output.passed

    assert line.startswith(CASCADE_EVIDENCE_LOG_PREFIX)
    assert "\n" not in line
    assert private_target not in line
    assert private_transcript not in line
    assert "PRIVATE_REASON_MUST_NOT_LEAK" not in line
    for forbidden_key in (
        '"target_text"',
        '"transcript_text"',
        '"extra_tail"',
        '"audio"',
        '"waveform"',
        '"speaker_embedding"',
    ):
        assert forbidden_key not in line
    payload_text = line.removeprefix(CASCADE_EVIDENCE_LOG_PREFIX)
    payload = json.loads(payload_text)
    assert payload["schema_version"] == CASCADE_EVIDENCE_SCHEMA_VERSION
    assert payload["outcome"] == "returned"
    assert payload["attempt_count"] == 5
    assert payload["attempts"][0]["trajectory_reasons"] == [
        "unknown_rejection"
    ]
    assert payload["attempts"][0]["policy"] == "base"
    assert payload["attempts"][1]["policy"] == "safe_duration"
    assert payload["selection"]["candidate_index"] == 3
    assert payload["final_output"]["passed"] is True
    assert payload["independent_final_output"]["passed"] is True
    assert payload["attempts"][3]["independent_output"]["passed"] is True
    assert payload_text == json.dumps(
        payload,
        allow_nan=False,
        ensure_ascii=True,
        separators=(",", ":"),
        sort_keys=True,
    )


def test_no_qualified_diagnostics_identify_zero_eligible_chunk_row():
    chunks = ("第一段完整", "第二段完整")
    callback_paths = []

    def verifier(trajectory, candidate_chunks, seed):
        observations = []
        artifacts = []
        for index, target in enumerate(candidate_chunks):
            observation, artifact = _chunk_verification(
                target,
                target if index == 0 else "錯誤內容",
                embedding=[1.0, 0.0],
                rms_db=-20.0,
            )
            observations.append(observation)
            artifacts.append(artifact)
        return verify_trajectory(observations, chunk_artifacts=artifacts)

    with pytest.raises(NoQualifiedCandidateError) as caught:
        run_adaptive_cascade(
            chunks,
            30,
            lambda candidate_chunks, seed: tuple(candidate_chunks),
            verifier,
            max_candidates=2,
            sequence_final_verifier=lambda result, candidate_chunks: (
                callback_paths.append(result.chunk_candidate_indices)
            ),
        )

    diagnostics = caught.value.diagnostics
    assert len(diagnostics.attempts) == 2
    assert diagnostics.sequence_search is not None
    assert diagnostics.sequence_search.eligible_candidate_counts == (2, 0)
    assert diagnostics.sequence_search.finite_transition_counts == (0,)
    assert diagnostics.sequence_search.ranked_path_count == 0
    assert diagnostics.sequence_search.checked_paths == ()
    assert callback_paths == []
    line = format_cascade_evidence_log(
        diagnostics,
        outcome="no_qualified_candidate",
        generated_chunk_limit=2,
    )
    payload = json.loads(line.removeprefix(CASCADE_EVIDENCE_LOG_PREFIX))
    assert payload["sequence_search"]["zero_eligible_rows"] == [1]
    assert payload["sequence_search"]["zero_transition_edges"] == [0]
    assert payload["selection"] is None
    assert payload["final_output"] is None


def test_no_qualified_diagnostics_distinguish_zero_transition_edge():
    chunks = ("第一段完整", "第二段完整")

    def verifier(trajectory, candidate_chunks, seed):
        observations = []
        artifacts = []
        for target in candidate_chunks:
            observation, artifact = _chunk_verification(
                target,
                target,
                embedding=[1.0, 0.0],
                rms_db=None,
            )
            observations.append(observation)
            artifacts.append(artifact)
        local = verify_trajectory(observations, chunk_artifacts=artifacts)
        joined = _joined_verification("".join(candidate_chunks), candidate_chunks[0])
        return CandidateVerification(
            qualify_trajectory_with_joined_output(local, joined),
            joined_output=trajectory_gate_evidence(joined),
        )

    with pytest.raises(NoQualifiedCandidateError) as caught:
        run_adaptive_cascade(
            chunks,
            40,
            lambda candidate_chunks, seed: tuple(candidate_chunks),
            verifier,
            max_candidates=2,
            sequence_final_verifier=lambda *args: pytest.fail(
                "a zero-edge lattice must not reach the final verifier"
            ),
        )

    search = caught.value.diagnostics.sequence_search
    assert search is not None
    assert search.eligible_candidate_counts == (2, 2)
    assert search.finite_transition_counts == (0,)
    assert search.ranked_path_count == 0
    assert all(
        attempt.joined_output is not None
        and not attempt.joined_output.passed
        and "chunk_0:semantic_gate"
        in attempt.joined_output.rejection_reasons
        for attempt in caught.value.diagnostics.attempts
    )


def test_k_best_sequence_paths_are_cost_ranked_stable_distinct_and_bounded():
    ranked = select_k_candidate_sequences(
        [[0.0, 0.2], [0.0, 0.1]],
        [[[0.4, 0.0], [0.0, 0.5]]],
        max_paths=3,
    )

    assert [selection.candidate_indices for selection in ranked] == [
        (0, 1),
        (1, 0),
        (0, 0),
    ]
    assert [selection.total_score for selection in ranked] == pytest.approx(
        [0.1, 0.2, 0.4]
    )
    assert len({selection.candidate_indices for selection in ranked}) == 3

    tied = select_k_candidate_sequences(
        [[0.0, 0.0], [0.0, 0.0]],
        [[[0.0, 0.0], [0.0, 0.0]]],
        max_paths=3,
    )
    assert [selection.candidate_indices for selection in tied] == [
        (0, 0),
        (0, 1),
        (1, 0),
    ]


def test_k_best_sequence_selection_fails_closed_for_malformed_or_single_chunk_graphs():
    assert select_k_candidate_sequences([[0.0, 0.1]], (), max_paths=3) == ()
    assert select_k_candidate_sequences([[0.0], [0.0]], (), max_paths=3) == ()
    assert (
        select_k_candidate_sequences(
            [[0.0, 0.1], [0.0, 0.1]],
            [[[0.0]]],
            max_paths=3,
        )
        == ()
    )
    for invalid in (0, 4, 1.0, True):
        with pytest.raises(ValueError, match="between 1 and 3"):
            select_k_candidate_sequences(
                [[0.0], [0.0]],
                [[[0.0]]],
                max_paths=invalid,
            )


def _locally_safe_join_rejected_verification(chunks, seed):
    # Crossed embeddings make the two mixed paths cheaper than either
    # same-candidate path, giving deterministic k-best callback order.
    first_candidate = seed % 2 == 0
    similarities = (0.9, 0.2) if first_candidate else (0.2, 0.9)
    embeddings = (
        ([1.0, 0.0], [0.0, 1.0])
        if first_candidate
        else ([0.0, 1.0], [1.0, 0.0])
    )
    observations = []
    artifacts = []
    for target, similarity, embedding in zip(
        chunks,
        similarities,
        embeddings,
        strict=True,
    ):
        observations.append(
            CandidateObservation(
                target_text=target,
                transcript_text=target,
                audio_duration_seconds=2.0,
                speaker_similarity=similarity,
                begin_speaker_similarity=similarity,
                end_speaker_similarity=similarity,
            )
        )
        artifacts.append(
            ChunkCandidateArtifact(
                speaker_embedding=np.asarray(embedding, dtype=np.float32),
                rms_db=-20.0,
            )
        )
    local = verify_trajectory(observations, chunk_artifacts=artifacts)
    joined_failed = _joined_verification("".join(chunks), chunks[0])
    return qualify_trajectory_with_joined_output(local, joined_failed)


def test_k_best_sequence_final_callback_uses_rank_order_and_first_passing_path():
    chunks = ("第一段完整", "第二段完整")
    callback_paths = []

    def generator(candidate_chunks, seed):
        return tuple(f"seed-{seed}-chunk-{index}" for index in range(len(candidate_chunks)))

    def verifier(trajectory, candidate_chunks, seed):
        return _locally_safe_join_rejected_verification(candidate_chunks, seed)

    def final_verifier(sequence_result, candidate_chunks):
        callback_paths.append(sequence_result.chunk_candidate_indices)
        transcript = (
            "".join(candidate_chunks)
            if sequence_result.chunk_candidate_indices == (1, 0)
            else candidate_chunks[0]
        )
        return _joined_verification("".join(candidate_chunks), transcript)

    result = run_adaptive_cascade(
        chunks,
        20,
        generator,
        verifier,
        max_candidates=2,
        sequence_final_verifier=final_verifier,
        max_sequence_paths=3,
    )

    assert callback_paths == [(0, 1), (1, 0)]
    assert result.selection_mode == "sequence_dp"
    assert result.chunk_candidate_indices == (1, 0)
    assert result.sequence_path_rank == 2
    assert result.sequence_paths_checked == 2
    assert result.diagnostics.sequence_search is not None
    assert result.diagnostics.sequence_search.eligible_candidate_counts == (2, 2)
    assert result.diagnostics.sequence_search.finite_transition_counts == (4,)
    assert result.diagnostics.sequence_search.ranked_path_count == 3
    assert [
        path.chunk_candidate_indices
        for path in result.diagnostics.sequence_search.checked_paths
    ] == [(0, 1), (1, 0)]
    assert [
        path.final_output.passed
        for path in result.diagnostics.sequence_search.checked_paths
    ] == [False, True]


def test_k_best_sequence_all_paths_rejected_is_no_qualified_candidate():
    chunks = ("第一段完整", "第二段完整")
    callback_paths = []

    def generator(candidate_chunks, seed):
        return tuple(f"seed-{seed}-chunk-{index}" for index in range(len(candidate_chunks)))

    def final_verifier(sequence_result, candidate_chunks):
        callback_paths.append(sequence_result.chunk_candidate_indices)
        return _joined_verification("".join(candidate_chunks), candidate_chunks[0])

    with pytest.raises(
        NoQualifiedCandidateError,
        match="after 2 candidates",
    ) as caught:
        run_adaptive_cascade(
            chunks,
            20,
            generator,
            lambda trajectory, candidate_chunks, seed: (
                _locally_safe_join_rejected_verification(candidate_chunks, seed)
            ),
            max_candidates=2,
            sequence_final_verifier=final_verifier,
            max_sequence_paths=3,
        )

    assert callback_paths == [(0, 1), (1, 0), (0, 0)]
    assert len(set(callback_paths)) == 3
    search = caught.value.diagnostics.sequence_search
    assert search is not None
    assert search.ranked_path_count == 3
    assert [path.rank for path in search.checked_paths] == [1, 2, 3]
    assert [
        path.chunk_candidate_indices for path in search.checked_paths
    ] == callback_paths
    assert all(not path.final_output.passed for path in search.checked_paths)
    assert all(
        "chunk_0:semantic_gate" in path.final_output.rejection_reasons
        for path in search.checked_paths
    )


def test_k_best_sequence_callback_exception_aborts_fail_closed():
    chunks = ("第一段完整", "第二段完整")
    callback_paths = []

    def callback(sequence_result, candidate_chunks):
        callback_paths.append(sequence_result.chunk_candidate_indices)
        raise ValueError("ASR backend disappeared")

    with pytest.raises(RuntimeError, match="refusing unverified audio"):
        run_adaptive_cascade(
            chunks,
            20,
            lambda candidate_chunks, seed: tuple(candidate_chunks),
            lambda trajectory, candidate_chunks, seed: (
                _locally_safe_join_rejected_verification(candidate_chunks, seed)
            ),
            max_candidates=2,
            sequence_final_verifier=callback,
        )

    assert callback_paths == [(0, 1)]


def test_k_best_sequence_invalid_callback_result_aborts_fail_closed():
    chunks = ("第一段完整", "第二段完整")

    with pytest.raises(RuntimeError, match="invalid result"):
        run_adaptive_cascade(
            chunks,
            20,
            lambda candidate_chunks, seed: tuple(candidate_chunks),
            lambda trajectory, candidate_chunks, seed: (
                _locally_safe_join_rejected_verification(candidate_chunks, seed)
            ),
            max_candidates=2,
            sequence_final_verifier=lambda *args: None,
        )


def test_k_best_sequence_inconsistent_callback_evidence_never_passes():
    chunks = ("第一段完整", "第二段完整")
    failed_candidate = verify_candidate(
        CandidateObservation("完整內容", "錯誤內容", 1.0)
    )

    with pytest.raises(NoQualifiedCandidateError, match="after 2 candidates"):
        run_adaptive_cascade(
            chunks,
            20,
            lambda candidate_chunks, seed: tuple(candidate_chunks),
            lambda trajectory, candidate_chunks, seed: (
                _locally_safe_join_rejected_verification(candidate_chunks, seed)
            ),
            max_candidates=2,
            sequence_final_verifier=lambda *args: TrajectoryGateResult(
                passed=True,
                candidate_results=(failed_candidate,),
                score=0.0,
                rejection_reasons=(),
            ),
        )


def test_sequence_final_callback_is_not_used_for_single_chunk_or_safe_whole():
    callback_calls = []

    with pytest.raises(NoQualifiedCandidateError):
        run_adaptive_cascade(
            ("單一完整段落",),
            20,
            lambda candidate_chunks, seed: tuple(candidate_chunks),
            lambda trajectory, candidate_chunks, seed: TrajectoryGateResult(
                passed=False,
                candidate_results=(
                    verify_candidate(
                        CandidateObservation(
                            candidate_chunks[0],
                            candidate_chunks[0],
                            1.0,
                        )
                    ),
                ),
                score=math.inf,
                rejection_reasons=("joined_output:semantic_gate",),
                chunk_artifacts=(ChunkCandidateArtifact(rms_db=-20.0),),
            ),
            max_candidates=2,
            sequence_final_verifier=lambda *args: callback_calls.append(args),
        )
    assert callback_calls == []

    result = run_adaptive_cascade(
        ("第一段", "第二段"),
        30,
        lambda candidate_chunks, seed: tuple(candidate_chunks),
        lambda *args: _gate_result(True, 0.1),
        max_candidates=2,
        sequence_final_verifier=lambda *args: callback_calls.append(args),
    )
    assert result.selection_mode == "whole_trajectory"
    assert callback_calls == []


def test_local_pass_join_fail_expands_and_selects_next_joined_safe_whole():
    calls = []

    def generator(chunks, seed):
        calls.append(seed)
        return tuple(f"seed-{seed}-{chunk}" for chunk in chunks)

    def verifier(trajectory, chunks, seed):
        observations = []
        artifacts = []
        for target in chunks:
            observation, artifact = _chunk_verification(
                target,
                target,
                embedding=[1.0, 0.0],
                rms_db=-20.0,
            )
            observations.append(observation)
            artifacts.append(artifact)
        local = verify_trajectory(observations, chunk_artifacts=artifacts)
        joined_transcript = "".join(chunks) if seed == 11 else chunks[0]
        joined = _joined_verification("".join(chunks), joined_transcript)
        return qualify_trajectory_with_joined_output(local, joined)

    result = run_adaptive_cascade(
        ("第一段完整", "第二段完整"),
        10,
        generator,
        verifier,
        max_candidates=2,
    )

    assert calls == [10, 11]
    assert result.selection_mode == "whole_trajectory"
    assert result.candidate_index == 1
    assert result.seed == 11


def test_all_local_pass_join_fail_can_only_return_as_sequence_dp():
    chunks = ("第一段完整", "第二段完整")

    def generator(candidate_chunks, seed):
        return tuple(f"seed-{seed}-{chunk}" for chunk in candidate_chunks)

    def verifier(trajectory, candidate_chunks, seed):
        similarities = (0.9, 0.2) if seed == 20 else (0.2, 0.9)
        observations = []
        artifacts = []
        for target, similarity in zip(candidate_chunks, similarities, strict=True):
            observations.append(
                CandidateObservation(
                    target_text=target,
                    transcript_text=target,
                    audio_duration_seconds=2.0,
                    speaker_similarity=similarity,
                    begin_speaker_similarity=similarity,
                    end_speaker_similarity=similarity,
                )
            )
            artifacts.append(
                ChunkCandidateArtifact(
                    speaker_embedding=np.array([1.0, 0.0], dtype=np.float32),
                    rms_db=-20.0,
                )
            )
        local = verify_trajectory(observations, chunk_artifacts=artifacts)
        joined = _joined_verification("".join(candidate_chunks), candidate_chunks[0])
        return qualify_trajectory_with_joined_output(local, joined)

    result = run_adaptive_cascade(
        chunks,
        20,
        generator,
        verifier,
        max_candidates=2,
    )

    assert result.selection_mode == "sequence_dp"
    assert result.candidate_index is None
    assert result.seed is None
    assert result.chunk_candidate_indices == (0, 1)
    assert result.attempted_seeds == (20, 21)


def test_adaptive_cascade_returns_first_candidate_immediately():
    generation_calls = []
    verification_calls = []

    def generator(chunks, seed):
        generation_calls.append((chunks, seed))
        return tuple((chunk, seed) for chunk in chunks)

    def verifier(trajectory, chunks, seed):
        verification_calls.append((trajectory, chunks, seed))
        assert all(chunk_seed == seed for _, chunk_seed in trajectory)
        return _gate_result(True, 0.2)

    result = run_adaptive_cascade(
        ["第一段", "第二段"],
        12345,
        generator,
        verifier,
    )

    assert isinstance(result, CascadeResult)
    assert result.seed == 12345
    assert result.candidate_index == 0
    assert result.attempted_seeds == (12345,)
    assert generation_calls == [(("第一段", "第二段"), 12345)]
    assert len(verification_calls) == 1


def test_marginal_stage_one_speaker_pass_expands_to_preferred_candidate():
    calls = []

    def generator(chunks, seed):
        calls.append(seed)
        return (chunks[0], seed)

    def verifier(trajectory, chunks, seed):
        if seed == 100:
            return _whole_speaker_verification(
                similarity=0.20,
                boundary_drop=0.04,
            )
        if seed == 105:
            return _whole_speaker_verification(
                similarity=0.35,
                boundary_drop=0.02,
            )
        return _whole_speaker_verification(
            similarity=0.30,
            boundary_drop=0.02,
            passed=False,
        )

    result = run_adaptive_cascade(
        ["完整而且穩定的候選內容"],
        100,
        generator,
        verifier,
        max_candidates=10,
    )

    assert calls == list(range(100, 110))
    assert result.selection_mode == "whole_trajectory"
    assert result.candidate_index == 5
    assert result.seed == 105


def test_final_stage_can_return_marginal_hard_pass_to_preserve_availability():
    calls = []

    def generator(chunks, seed):
        calls.append(seed)
        return (chunks[0], seed)

    def verifier(trajectory, chunks, seed):
        return _whole_speaker_verification(
            similarity=0.20,
            boundary_drop=0.08,
            passed=seed == 200,
        )

    result = run_adaptive_cascade(
        ["完整而且穩定的候選內容"],
        200,
        generator,
        verifier,
        max_candidates=2,
    )

    assert calls == [200, 201]
    assert result.seed == 200
    assert result.candidate_index == 0
    assert result.verification.passed


def test_adaptive_cascade_expands_to_five_and_selects_lowest_verified_score():
    calls = []
    scores = {
        100: None,
        101: 0.4,
        102: None,
        103: 0.1,
        104: 0.1,
    }

    def generator(chunks, seed):
        calls.append((chunks, seed))
        return {"chunks": tuple((chunk, seed) for chunk in chunks), "seed": seed}

    def verifier(trajectory, chunks, seed):
        assert trajectory["seed"] == seed
        assert all(chunk_seed == seed for _, chunk_seed in trajectory["chunks"])
        score = scores[seed]
        return _gate_result(score is not None, score or 0.0)

    result = run_adaptive_cascade(
        ["第一段", "第二段"],
        100,
        generator,
        verifier,
    )

    assert [seed for _, seed in calls] == [100, 101, 102, 103, 104]
    assert result.seed == 103
    assert result.candidate_index == 3
    assert result.verification.score == 0.1
    assert result.attempted_seeds == (100, 101, 102, 103, 104)


def test_hard_secondary_rejection_continues_cascade_to_later_candidate():
    calls = []

    def generator(chunks, seed):
        calls.append(seed)
        return (chunks[0], seed)

    def dual_verifier(trajectory, chunks, seed):
        # This models a turbo-passing candidate rejected by the independent
        # whole-output ASR.  It remains a normal rejected gate result so the
        # deterministic cascade must continue instead of returning it.
        return _gate_result(seed == 101, score=0.0)

    result = run_adaptive_cascade(
        ["完整內容"],
        100,
        generator,
        dual_verifier,
        max_candidates=5,
    )

    assert calls == [100, 101, 102, 103, 104]
    assert result.seed == 101
    assert result.candidate_index == 1
    assert result.attempted_seeds == (100, 101, 102, 103, 104)


def test_adaptive_cascade_uses_ten_only_when_stage_five_has_no_verified_candidate():
    calls = []

    def generator(chunks, seed):
        calls.append(seed)
        return (chunks, seed)

    def verifier(trajectory, chunks, seed):
        return _gate_result(seed in {107, 109}, score=float(110 - seed))

    result = run_adaptive_cascade(
        ["完整內容"],
        100,
        generator,
        verifier,
        max_candidates=10,
    )

    assert result.seed == 109
    assert result.candidate_index == 9
    assert result.attempted_seeds == tuple(range(100, 110))
    assert calls == list(range(100, 110))


def test_single_chunk_emergency_stage_expands_from_ten_to_twenty():
    calls = []

    def generator(chunks, seed):
        calls.append(seed)
        return (chunks, seed)

    def verifier(trajectory, chunks, seed):
        return _gate_result(seed == 117, score=0.1)

    result = run_adaptive_cascade(
        ["完整內容"],
        100,
        generator,
        verifier,
        max_candidates=20,
    )

    assert result.seed == 117
    assert result.candidate_index == 17
    assert result.attempted_seeds == tuple(range(100, 120))
    assert calls == list(range(100, 120))


def test_single_chunk_preferred_candidate_stops_at_fifteen_stage():
    calls = []

    def generator(chunks, seed):
        calls.append(seed)
        return (chunks, seed)

    def verifier(trajectory, chunks, seed):
        return _gate_result(seed == 111, score=0.1)

    result = run_adaptive_cascade(
        ["完整內容"],
        100,
        generator,
        verifier,
        max_candidates=20,
    )

    assert result.seed == 111
    assert result.candidate_index == 11
    assert result.attempted_seeds == tuple(range(100, 115))
    assert calls == list(range(100, 115))


def test_sequence_fallback_can_take_a_prefix_from_a_and_suffix_from_b():
    chunks = ("第一段完整", "第二段完整")
    trajectories = {
        10: ("audio-a0", "audio-a1"),
        11: ("audio-b0", "audio-b1"),
    }

    def generator(candidate_chunks, seed):
        assert candidate_chunks == chunks
        return trajectories[seed]

    def verifier(trajectory, candidate_chunks, seed):
        transcripts = (
            candidate_chunks
            if seed == 999
            else (
                (candidate_chunks[0], "錯誤內容")
                if seed == 10
                else ("錯誤內容", candidate_chunks[1])
            )
        )
        observations = []
        artifacts = []
        for target, transcript in zip(candidate_chunks, transcripts, strict=True):
            observation, artifact = _chunk_verification(
                target,
                transcript,
                embedding=[1.0, 0.0],
                rms_db=-20.0,
            )
            observations.append(observation)
            artifacts.append(artifact)
        return verify_trajectory(observations, chunk_artifacts=artifacts)

    result = run_adaptive_cascade(
        chunks,
        10,
        generator,
        verifier,
        max_candidates=2,
    )

    assert result.selection_mode == "sequence_dp"
    assert result.seed is None
    assert result.candidate_index is None
    assert result.trajectory == ("audio-a0", "audio-b1")
    assert result.chunk_candidate_indices == (0, 1)
    assert result.chunk_seeds == (10, 11)
    assert result.attempted_seeds == (10, 11)
    assert result.verification.passed
    assert len(result.verification.candidate_results) == 2
    assert len(result.verification.chunk_artifacts) == 2


def _boundary_only_sequence_verification(chunks, seed, boundary_drop):
    selected_chunk = 0 if seed % 2 == 0 else 1
    observations = []
    artifacts = []
    for chunk_index, target in enumerate(chunks):
        selected = chunk_index == selected_chunk
        observations.append(
            CandidateObservation(
                target_text=target,
                transcript_text=target if selected else "錯誤內容",
                audio_duration_seconds=2.0,
                speaker_similarity=0.8,
                begin_speaker_similarity=0.8,
                end_speaker_similarity=(
                    0.8 - boundary_drop if selected else 0.8
                ),
            )
        )
        artifacts.append(
            ChunkCandidateArtifact(
                speaker_embedding=np.array([1.0, 0.0], dtype=np.float32),
                rms_db=-20.0,
            )
        )
    return verify_trajectory(
        observations,
        chunk_artifacts=artifacts,
        min_speaker_similarity=0.10,
        max_boundary_speaker_drop=0.10,
    )


def test_sequence_fallback_recovers_boundary_only_chunk_below_frozen_cap():
    chunks = ("第一段完整", "第二段完整")
    callback_paths = []

    def final_verifier(sequence_result, candidate_chunks):
        callback_paths.append(sequence_result.chunk_candidate_indices)
        return _joined_verification("".join(candidate_chunks), "".join(candidate_chunks))

    result = run_adaptive_cascade(
        chunks,
        10,
        lambda candidate_chunks, seed: tuple(
            f"seed-{seed}-chunk-{index}" for index in range(len(candidate_chunks))
        ),
        lambda trajectory, candidate_chunks, seed: (
            _boundary_only_sequence_verification(
                candidate_chunks,
                seed,
                boundary_drop=0.149,
            )
        ),
        max_candidates=2,
        sequence_final_verifier=final_verifier,
        sequence_fallback_max_local_boundary_speaker_drop=(
            SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP
        ),
    )

    assert SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP == 0.15
    assert result.selection_mode == "sequence_dp"
    assert result.chunk_candidate_indices == (0, 1)
    assert callback_paths == [(0, 1)]
    assert all(candidate.passed for candidate in result.verification.candidate_results)
    assert [
        candidate.boundary_speaker_drop
        for candidate in result.verification.candidate_results
    ] == pytest.approx([0.149, 0.149])
    assert [candidate.score for candidate in result.verification.candidate_results] == (
        pytest.approx([0.0249, 0.0249])
    )


def test_sequence_fallback_does_not_relax_boundary_above_frozen_cap():
    chunks = ("第一段完整", "第二段完整")
    callback_paths = []

    with pytest.raises(NoQualifiedCandidateError, match="after 2 candidates"):
        run_adaptive_cascade(
            chunks,
            10,
            lambda candidate_chunks, seed: tuple(candidate_chunks),
            lambda trajectory, candidate_chunks, seed: (
                _boundary_only_sequence_verification(
                    candidate_chunks,
                    seed,
                    boundary_drop=0.151,
                )
            ),
            max_candidates=2,
            sequence_final_verifier=lambda sequence_result, candidate_chunks: (
                callback_paths.append(sequence_result.chunk_candidate_indices)
            ),
            sequence_fallback_max_local_boundary_speaker_drop=0.15,
        )

    assert callback_paths == []


def test_sequence_fallback_boundary_cap_is_inclusive():
    observation = CandidateObservation(
        target_text="第一段完整",
        transcript_text="第一段完整",
        audio_duration_seconds=2.0,
        speaker_similarity=0.8,
        begin_speaker_similarity=0.8,
        end_speaker_similarity=0.65,
    )
    rejected = verify_candidate(
        observation,
        max_boundary_speaker_drop=0.10,
    )

    adjusted = _sequence_fallback_candidate_result(
        rejected,
        max_local_boundary_speaker_drop=0.15,
    )

    assert rejected.rejection_reasons == ("boundary_speaker_drop",)
    assert adjusted.passed
    assert adjusted.boundary_speaker_drop == pytest.approx(0.15)
    assert adjusted.score == pytest.approx(0.025)


def test_sequence_fallback_boundary_relaxation_never_relaxes_semantic_failure():
    chunks = ("第一段完整", "第二段完整")
    callback_paths = []

    def verifier(trajectory, candidate_chunks, seed):
        verification = _boundary_only_sequence_verification(
            candidate_chunks,
            seed,
            boundary_drop=0.149,
        )
        selected_chunk = 0 if seed % 2 == 0 else 1
        observations = []
        for chunk_index, target in enumerate(candidate_chunks):
            observations.append(
                CandidateObservation(
                    target_text=target,
                    transcript_text="錯誤內容",
                    audio_duration_seconds=2.0,
                    speaker_similarity=0.8,
                    begin_speaker_similarity=0.8,
                    end_speaker_similarity=(
                        0.651 if chunk_index == selected_chunk else 0.8
                    ),
                )
            )
        return verify_trajectory(
            observations,
            chunk_artifacts=verification.chunk_artifacts,
            min_speaker_similarity=0.10,
            max_boundary_speaker_drop=0.10,
        )

    with pytest.raises(NoQualifiedCandidateError, match="after 2 candidates"):
        run_adaptive_cascade(
            chunks,
            10,
            lambda candidate_chunks, seed: tuple(candidate_chunks),
            verifier,
            max_candidates=2,
            sequence_final_verifier=lambda sequence_result, candidate_chunks: (
                callback_paths.append(sequence_result.chunk_candidate_indices)
            ),
            sequence_fallback_max_local_boundary_speaker_drop=0.15,
        )

    assert callback_paths == []


def test_sequence_fallback_boundary_relaxation_keeps_every_other_hard_reject():
    common = dict(
        target_text="第一段完整",
        transcript_text="第一段完整",
        audio_duration_seconds=2.0,
        speaker_similarity=0.8,
        begin_speaker_similarity=0.8,
        end_speaker_similarity=0.651,
    )
    rejected = (
        verify_candidate(
            CandidateObservation(**common, pace_cps=4.5),
            max_boundary_speaker_drop=0.10,
            max_pace_cps=4.30,
        ),
        verify_candidate(
            CandidateObservation(**{**common, "speaker_similarity": 0.05}),
            min_speaker_similarity=0.10,
            max_boundary_speaker_drop=0.10,
        ),
        verify_candidate(
            CandidateObservation(**{**common, "speaker_similarity": None}),
            max_boundary_speaker_drop=0.10,
        ),
        verify_candidate(
            CandidateObservation(**common, truncated=True),
            max_boundary_speaker_drop=0.10,
        ),
    )

    for result in rejected:
        adjusted = _sequence_fallback_candidate_result(
            result,
            max_local_boundary_speaker_drop=0.15,
        )
        assert adjusted is result
        assert not adjusted.passed
        assert math.isinf(adjusted.score)


def test_sequence_fallback_boundary_relaxation_requires_and_obeys_final_verifier():
    chunks = ("第一段完整", "第二段完整")
    generator = lambda candidate_chunks, seed: tuple(candidate_chunks)
    verifier = lambda trajectory, candidate_chunks, seed: (
        _boundary_only_sequence_verification(
            candidate_chunks,
            seed,
            boundary_drop=0.149,
        )
    )

    with pytest.raises(ValueError, match="requires a final verifier"):
        run_adaptive_cascade(
            chunks,
            10,
            generator,
            verifier,
            max_candidates=2,
            sequence_fallback_max_local_boundary_speaker_drop=0.15,
        )

    with pytest.raises(NoQualifiedCandidateError, match="after 2 candidates"):
        run_adaptive_cascade(
            chunks,
            10,
            generator,
            verifier,
            max_candidates=2,
            sequence_final_verifier=lambda sequence_result, candidate_chunks: (
                _joined_verification("".join(candidate_chunks), candidate_chunks[0])
            ),
            sequence_fallback_max_local_boundary_speaker_drop=0.15,
        )


def test_sequence_fallback_boundary_relaxation_never_applies_to_single_chunk():
    chunks = ("單一完整段落",)

    def verifier(trajectory, candidate_chunks, seed):
        observation = CandidateObservation(
            target_text=candidate_chunks[0],
            transcript_text=candidate_chunks[0],
            audio_duration_seconds=2.0,
            speaker_similarity=0.8,
            begin_speaker_similarity=0.8,
            end_speaker_similarity=0.651,
        )
        return verify_trajectory(
            [observation],
            chunk_artifacts=[
                ChunkCandidateArtifact(
                    speaker_embedding=np.array([1.0, 0.0], dtype=np.float32),
                    rms_db=-20.0,
                )
            ],
            max_boundary_speaker_drop=0.10,
        )

    with pytest.raises(NoQualifiedCandidateError, match="after 2 candidates"):
        run_adaptive_cascade(
            chunks,
            10,
            lambda candidate_chunks, seed: tuple(candidate_chunks),
            verifier,
            max_candidates=2,
            sequence_final_verifier=lambda *args: _joined_verification(
                chunks[0], chunks[0]
            ),
            sequence_fallback_max_local_boundary_speaker_drop=0.15,
        )

def test_same_seed_whole_trajectory_wins_before_sequence_fallback():
    chunks = ("第一段完整", "第二段完整")

    def generator(candidate_chunks, seed):
        return tuple(f"seed-{seed}-chunk-{index}" for index in range(len(candidate_chunks)))

    def verifier(trajectory, candidate_chunks, seed):
        transcripts = (
            (candidate_chunks[0], "錯誤內容")
            if seed == 20
            else candidate_chunks
        )
        observations = []
        artifacts = []
        for target, transcript in zip(candidate_chunks, transcripts, strict=True):
            observation, artifact = _chunk_verification(
                target,
                transcript,
                embedding=[1.0, 0.0],
                rms_db=-20.0,
            )
            observations.append(observation)
            artifacts.append(artifact)
        return verify_trajectory(observations, chunk_artifacts=artifacts)

    result = run_adaptive_cascade(
        chunks,
        20,
        generator,
        verifier,
        max_candidates=2,
    )

    assert result.selection_mode == "whole_trajectory"
    assert result.seed == 21
    assert result.candidate_index == 1
    assert result.chunk_candidate_indices == (1, 1)
    assert result.chunk_seeds == (21, 21)
    assert result.trajectory == ("seed-21-chunk-0", "seed-21-chunk-1")


def test_sequence_fallback_rejects_unsafe_transition_edge():
    chunks = ("第一段完整", "第二段完整")

    def generator(candidate_chunks, seed):
        return tuple(f"seed-{seed}-chunk-{index}" for index in range(len(candidate_chunks)))

    def verifier(trajectory, candidate_chunks, seed):
        transcripts = (
            (candidate_chunks[0], "錯誤內容")
            if seed == 30
            else ("錯誤內容", candidate_chunks[1])
        )
        embeddings = (
            ([1.0, 0.0], [1.0, 0.0])
            if seed == 30
            else ([1.0, 0.0], [1.0, 0.0, 0.0])
        )
        observations = []
        artifacts = []
        for target, transcript, embedding in zip(
            candidate_chunks,
            transcripts,
            embeddings,
            strict=True,
        ):
            observation, artifact = _chunk_verification(
                target,
                transcript,
                embedding=embedding,
                rms_db=-20.0,
            )
            observations.append(observation)
            artifacts.append(artifact)
        return verify_trajectory(observations, chunk_artifacts=artifacts)

    with pytest.raises(NoQualifiedCandidateError, match="after 2 candidates"):
        run_adaptive_cascade(
            chunks,
            30,
            generator,
            verifier,
            max_candidates=2,
        )


def test_transition_score_requires_safe_evidence_and_penalizes_rms_delta():
    first_observation, first_artifact = _chunk_verification(
        "第一段完整",
        "第一段完整",
        embedding=[1.0, 0.0],
        rms_db=-20.0,
    )
    second_observation, second_artifact = _chunk_verification(
        "第二段完整",
        "第二段完整",
        embedding=[1.0, 0.0],
        rms_db=-16.0,
    )
    first = verify_trajectory([first_observation], chunk_artifacts=[first_artifact])
    second = verify_trajectory([second_observation], chunk_artifacts=[second_artifact])

    score = candidate_chunk_transition_score(
        first.candidate_results[0],
        first.chunk_artifacts[0],
        second.candidate_results[0],
        second.chunk_artifacts[0],
    )
    assert score == pytest.approx(0.20)

    unsafe = ChunkCandidateArtifact(speaker_embedding=None, rms_db=-16.0)
    assert math.isinf(
        candidate_chunk_transition_score(
            first.candidate_results[0],
            first.chunk_artifacts[0],
            second.candidate_results[0],
            unsafe,
        )
    )


@pytest.mark.parametrize(
    ("chunk_count", "expected_limit"),
    [(1, 32), (2, 16), (3, 10), (4, 8), (5, 6), (6, 5), (10, 3), (20, 1), (32, 1)],
)
def test_candidate_limit_obeys_thirty_two_generated_chunk_budget(
    chunk_count,
    expected_limit,
):
    limit = candidate_limit_for_chunk_budget(chunk_count)
    assert limit == expected_limit
    assert limit * chunk_count <= 32


def test_candidate_limit_rejects_a_trajectory_that_already_exceeds_budget():
    with pytest.raises(ValueError, match="one trajectory exceeds"):
        candidate_limit_for_chunk_budget(33)
    with pytest.raises(ValueError, match="between 1 and 32"):
        candidate_limit_for_chunk_budget(1, max_candidates=33)


@pytest.mark.parametrize(
    ("text_units", "expected_limit"),
    [(8, 32), (40, 20), (80, 10), (360, 2)],
)
def test_candidate_limit_obeys_generated_text_unit_budget(text_units, expected_limit):
    limit = candidate_limit_for_chunk_budget(
        1 if text_units <= 80 else 5,
        total_text_units=text_units,
        max_generated_text_units=800,
    )
    assert limit == expected_limit
    assert limit * text_units <= 800


def test_network_endpoint_units_do_not_change_k32_or_800_work_budget():
    text = " coastwatch 點 example 點 T W 斜線 tide。"
    public_units = count_speech_units(text)
    endpoint_units = count_network_endpoint_duration_units(text)

    assert (public_units, endpoint_units) == (12, 17)
    assert candidate_limit_for_chunk_budget(
        1,
        total_text_units=public_units,
        max_generated_text_units=800,
    ) == 32
    assert candidate_limit_for_chunk_budget(
        1,
        total_text_units=endpoint_units,
        max_generated_text_units=800,
    ) == 32
    evidence = CandidateGenerationEvidence(
        chunk_indices=(0,),
        chunk_text_units=(public_units,),
        scheduled_cfg=3.0,
        effective_cfgs=(3.0,),
        floor_reasons=((),),
    )
    result = run_coverage_adaptive_cascade(
        (text,),
        902,
        lambda chunks, seed: tuple(chunks),
        lambda trajectory, chunks, seed: verify_trajectory(
            [CandidateObservation(text, text, 1.0)],
            chunk_artifacts=[ChunkCandidateArtifact(rms_db=-20.0)],
            speaker_gate_enabled=False,
        ),
        lambda trajectory, chunks, seed: pytest.fail("no refill expected"),
        sequence_final_verifier=lambda result, chunks: pytest.fail(
            "no sequence expected"
        ),
        generation_evidence_factory=(
            lambda candidate_index, seed, chunk_indices, chunks: evidence
        ),
        max_generated_text_units=800,
    )
    assert result.generated_text_units == public_units
    assert result.diagnostics.attempts[0].chunk_text_units == (public_units,)


def test_transformed_initial_text_budget_is_checked_before_generation():
    generated = []
    verified = []
    contexts = []

    def transform(chunks, generation_context):
        contexts.append(generation_context)
        return ("甲" * 801,)

    with pytest.raises(
        ValueError,
        match="initial trajectory exceeds the generated-text-unit budget",
    ):
        run_coverage_adaptive_cascade(
            ("甲",),
            902,
            lambda chunks, seed: generated.append((chunks, seed)),
            lambda *args: verified.append(("whole", args)),
            lambda *args: verified.append(("refill", args)),
            sequence_final_verifier=lambda *args: verified.append(
                ("final", args)
            ),
            candidate_generation_text_transform=transform,
            max_generated_chunks=32,
            max_generated_text_units=800,
        )

    assert len(contexts) == 1
    assert contexts[0].candidate_index == 0
    assert contexts[0].chunk_indices == (0,)
    assert generated == []
    assert verified == []


def test_candidate_limit_requires_complete_positive_text_unit_budget():
    with pytest.raises(ValueError, match="provided together"):
        candidate_limit_for_chunk_budget(1, total_text_units=40)
    with pytest.raises(ValueError, match="positive"):
        candidate_limit_for_chunk_budget(
            1,
            total_text_units=0,
            max_generated_text_units=800,
        )


def test_360_character_profiles_stay_inside_generated_chunk_budget():
    profiles = [
        "甲" * 360,
        ("甲" * 11 + "。") * 30,
        ("甲" * 40 + "。") * 8 + "乙" * 32,
    ]
    for text in profiles:
        chunks = split_text_for_tts(text, max_chars=80, min_chunk_chars=12)
        total_units = sum(count_speech_units(chunk) for chunk in chunks)
        limit = candidate_limit_for_chunk_budget(
            len(chunks),
            total_text_units=total_units,
            max_generated_text_units=800,
        )
        assert limit * len(chunks) <= 20
        assert limit * total_units <= 800
        assert limit <= 2


def test_candidate_asr_value_error_rejects_seed_and_cascade_continues():
    calls = []

    def generator(chunks, seed):
        calls.append(seed)
        return _tone(seconds=0.25)

    def verifier(trajectory, chunks, seed):
        prepared = prepare_candidate_audio(
            trajectory,
            16_000,
            transcriber=(
                (lambda *_: (_ for _ in ()).throw(ValueError("candidate ASR failure")))
                if seed == 50
                else (lambda *_: chunks[0])
            ),
        )
        if prepared is None:
            return _gate_result(False)
        return verify_trajectory(
            [
                CandidateObservation(
                    target_text=chunks[0],
                    transcript_text=prepared.transcript_text,
                    audio_duration_seconds=prepared.duration_seconds,
                )
            ]
        )

    result = run_adaptive_cascade(
        ["完整內容"],
        50,
        generator,
        verifier,
        max_candidates=2,
    )

    assert result.seed == 51
    assert result.candidate_index == 1
    assert calls == [50, 51]


def test_adaptive_cascade_raises_without_qualified_candidate_and_never_falls_back():
    generated = []

    def generator(chunks, seed):
        generated.append(seed)
        return (chunks, seed)

    with pytest.raises(NoQualifiedCandidateError, match="after 5 candidates"):
        run_adaptive_cascade(
            ["必須完整"],
            7,
            generator,
            lambda trajectory, chunks, seed: _gate_result(False),
        )
    assert generated == [7, 8, 9, 10, 11]


def test_fail_closed_evidence_is_bounded_to_frozen_candidate_limit():
    private_target = "PRIVATE_BOUNDED_TARGET"
    private_transcript = "PRIVATE_BOUNDED_TRANSCRIPT"

    def verifier(trajectory, chunks, seed):
        return verify_trajectory(
            [
                CandidateObservation(
                    target_text=chunks[0],
                    transcript_text=private_transcript,
                    audio_duration_seconds=1.0,
                )
            ]
        )

    with pytest.raises(NoQualifiedCandidateError) as caught:
        run_adaptive_cascade(
            (private_target,),
            70,
            lambda chunks, seed: seed,
            verifier,
            max_candidates=32,
        )

    diagnostics = caught.value.diagnostics
    assert len(diagnostics.attempts) == 32
    assert [attempt.candidate_index for attempt in diagnostics.attempts] == list(
        range(32)
    )
    assert all(attempt.local_result_count == 1 for attempt in diagnostics.attempts)
    assert all(len(attempt.local_results) == 1 for attempt in diagnostics.attempts)
    line = format_cascade_evidence_log(
        diagnostics,
        outcome="no_qualified_candidate",
        generated_chunk_limit=32,
    )
    payload = json.loads(line.removeprefix(CASCADE_EVIDENCE_LOG_PREFIX))
    assert payload["attempt_count"] == 32
    assert len(payload["attempts"]) == 32
    assert len(line.encode("utf-8")) < 65_536
    assert private_target not in line
    assert private_transcript not in line


def test_adaptive_cascade_validates_contract_and_rejects_non_gate_verifier():
    generator = lambda chunks, seed: (chunks, seed)
    with pytest.raises(ValueError, match="start with exactly one"):
        run_adaptive_cascade(
            ["內容"],
            1,
            generator,
            lambda *args: _gate_result(True),
            initial_candidates=2,
        )
    with pytest.raises(ValueError, match="between 1 and 32"):
        run_adaptive_cascade(
            ["內容"],
            1,
            generator,
            lambda *args: _gate_result(True),
            max_candidates=33,
        )
    with pytest.raises(TypeError, match="TrajectoryGateResult"):
        run_adaptive_cascade(["內容"], 1, generator, lambda *args: True)