tests/test_coverage_adaptive.py

from dataclasses import replace
import json
import math

import numpy as np
import pytest

from production import count_speech_units
from quality_runtime import (
    CASCADE_EVIDENCE_LOG_PREFIX,
    CandidateGenerationContext,
    CandidateGenerationEvidence,
    CandidateObservation,
    CandidateVerification,
    ChunkCandidateArtifact,
    LocalIndependentGateEvidence,
    NoQualifiedCandidateError,
    SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP,
    TrajectoryGateResult,
    candidate_gate_evidence,
    format_cascade_evidence_log,
    generation_cfg_for_candidate_offset,
    intersect_local_semantic_verification,
    run_coverage_adaptive_cascade,
    select_culprit_diverse_candidate_sequences,
    trajectory_gate_evidence,
    verify_candidate,
    verify_trajectory,
)


def _local_verification(chunks, passing=None, *, scores=None):
    if passing is None:
        passing = (True,) * len(chunks)
    if scores is None:
        scores = (0.8,) * len(chunks)
    observations = []
    artifacts = []
    for chunk, passed, similarity in zip(chunks, passing, scores, strict=True):
        observations.append(
            CandidateObservation(
                target_text=chunk,
                transcript_text=chunk if passed else "完全錯誤",
                audio_duration_seconds=2.0,
                speaker_similarity=similarity,
                begin_speaker_similarity=similarity,
                end_speaker_similarity=similarity,
                pace_cps=2.0,
            )
        )
        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,
        max_pace_cps=4.3,
    )


def _joined_rejected_local(chunks, passing=None):
    local = _local_verification(chunks, passing)
    return TrajectoryGateResult(
        passed=False,
        candidate_results=local.candidate_results,
        score=math.inf,
        rejection_reasons=("joined_output:semantic_gate",),
        chunk_artifacts=local.chunk_artifacts,
    )


def _exact_final(chunks, passed=True):
    target = "".join(chunks)
    transcript = target if passed else "錯誤內容"
    return verify_trajectory(
        [CandidateObservation(target, transcript, 1.0)],
        chunk_artifacts=[ChunkCandidateArtifact(rms_db=-20.0)],
    )


def _squim_verification(
    chunks,
    *,
    stoi,
    pesq,
    si_sdr=0.0,
    speaker_similarity=0.30,
    boundary_drop=0.01,
    audio_duration_seconds=2.0,
    hard_max_boundary_speaker_drop=0.03,
):
    observations = []
    artifacts = []
    for chunk in chunks:
        observations.append(
            CandidateObservation(
                target_text=chunk,
                transcript_text=chunk,
                audio_duration_seconds=audio_duration_seconds,
                speaker_similarity=speaker_similarity,
                begin_speaker_similarity=0.50,
                end_speaker_similarity=0.50 - boundary_drop,
                pace_cps=2.0,
                squim_stoi=stoi,
                squim_pesq=pesq,
                squim_si_sdr=si_sdr,
            )
        )
        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,
        max_pace_cps=4.3,
        max_boundary_speaker_drop=hard_max_boundary_speaker_drop,
        squim_gate_enabled=True,
        min_squim_stoi=0.60,
        min_squim_pesq=1.12,
    )


def test_whole_exact_winner_returns_before_any_refill_or_sequence_check():
    chunks = ("第一段", "第二段")
    generated = []
    refill_calls = []
    final_calls = []
    enrichment_calls = []

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

    def whole_verifier(trajectory, candidate_chunks, seed):
        return CandidateVerification(
            _local_verification(candidate_chunks),
            joined_output=trajectory_gate_evidence(
                _exact_final(candidate_chunks)
            ),
        )

    result = run_coverage_adaptive_cascade(
        chunks,
        80,
        generator,
        whole_verifier,
        lambda *args: refill_calls.append(args),
        sequence_final_verifier=lambda *args: final_calls.append(args),
        transition_artifact_enricher=lambda *args: enrichment_calls.append(
            args
        ),
    )

    assert generated == [(chunks, 80)]
    assert refill_calls == []
    assert final_calls == []
    assert enrichment_calls == []
    assert result.selection_mode == "whole_trajectory"
    assert result.chunk_candidate_indices == (0, 0)
    assert result.chunk_seeds == (80, 80)
    assert result.attempted_seeds == (80,)
    assert result.generated_chunk_count == 2
    assert result.generated_text_units == 6
    assert result.chunk_candidate_counts == (1, 1)
    assert len(result.diagnostics.attempts) == 1
    assert result.diagnostics.attempts[0].joined_output is not None
    assert result.diagnostics.attempts[0].joined_output.passed


def test_single_chunk_initial_exact_joined_preferred_overrides_local_proxy():
    chunk = ("完整內容",)
    generated = []

    def whole_verifier(trajectory, candidate_chunks, seed):
        local = _squim_verification(
            candidate_chunks,
            stoi=0.75,
            pesq=1.25,
            speaker_similarity=0.20,
        )
        exact_joined = _squim_verification(
            candidate_chunks,
            stoi=0.75,
            pesq=1.25,
            speaker_similarity=0.30,
            boundary_drop=0.02,
        )
        return CandidateVerification(
            local,
            joined_output=trajectory_gate_evidence(exact_joined),
            independent_output=trajectory_gate_evidence(
                _exact_final(candidate_chunks)
            ),
        )

    result = run_coverage_adaptive_cascade(
        chunk,
        80,
        lambda candidate_chunks, seed: (
            generated.append(seed) or (f"audio-{seed}",)
        ),
        whole_verifier,
        lambda *args: pytest.fail("preferred exact initial must not refill"),
        sequence_final_verifier=lambda *args: pytest.fail(
            "preferred exact initial must not enter sequence search"
        ),
        preferred_min_speaker_similarity=0.25,
        preferred_max_boundary_speaker_drop=0.05,
        preferred_min_squim_stoi=0.72,
        preferred_min_squim_pesq=1.20,
    )

    assert generated == [80]
    assert result.candidate_index == 0
    assert result.generated_chunk_count == 1


def test_mixed_network_local_v3_failure_cannot_enter_strict_or_proxy_pool():
    chunks = ("普通內容完整", "網路內容完整")
    generated = []

    def primary(candidate_chunks):
        observations = []
        artifacts = []
        for index, chunk in enumerate(candidate_chunks):
            observations.append(
                CandidateObservation(
                    target_text=chunk,
                    transcript_text=chunk,
                    audio_duration_seconds=2.0,
                    speaker_similarity=0.70,
                    begin_speaker_similarity=(0.80 if index == len(candidate_chunks) - 1 else 0.60),
                    end_speaker_similarity=(0.68 if index == len(candidate_chunks) - 1 else 0.60),
                    pace_cps=2.0,
                )
            )
            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,
            max_pace_cps=4.3,
            max_boundary_speaker_drop=0.10,
        )

    def independent_failure(target):
        return verify_trajectory(
            [CandidateObservation(target, "完全錯誤", 2.0)],
            speaker_gate_enabled=False,
        )

    def combined(candidate_chunks):
        primary_verification = primary(candidate_chunks)
        network_local_index = len(candidate_chunks) - 1
        independent = independent_failure(
            candidate_chunks[network_local_index]
        )
        verification = intersect_local_semantic_verification(
            primary_verification,
            independent,
            (network_local_index,),
        )
        local_evidence = tuple(
            (
                LocalIndependentGateEvidence(
                    True,
                    False,
                    1,
                    candidate_gate_evidence(
                        independent.candidate_results[0]
                    ),
                )
                if index == network_local_index
                else LocalIndependentGateEvidence(False, None, 0, None)
            )
            for index in range(len(candidate_chunks))
        )
        return CandidateVerification(
            verification,
            independent_local_results=local_evidence,
        )

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

    with pytest.raises(NoQualifiedCandidateError) as captured:
        run_coverage_adaptive_cascade(
            chunks,
            800,
            generator,
            lambda trajectory, candidate_chunks, seed: combined(
                candidate_chunks
            ),
            lambda trajectory, candidate_chunks, seed: combined(
                candidate_chunks
            ),
            sequence_final_verifier=lambda result, candidate_chunks: pytest.fail(
                "a full path must not exist"
            ),
            max_generated_chunks=3,
            max_generated_text_units=100,
            sequence_fallback_max_local_boundary_speaker_drop=(
                SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP
            ),
        )

    assert generated == [
        (chunks, 800),
        ((chunks[1],), 801),
    ]
    diagnostics = captured.value.diagnostics
    assert len(diagnostics.attempts) == 2
    assert diagnostics.sequence_search is not None
    assert diagnostics.sequence_search.eligible_candidate_counts == (1, 0)
    initial = diagnostics.attempts[0]
    refill = diagnostics.attempts[1]
    assert initial.local_results[0].passed is True
    assert initial.local_results[1].passed is False
    assert initial.local_results[1].rejection_reasons == (
        "boundary_speaker_drop",
        "semantic_gate",
    )
    assert initial.independent_local_results[0] == (
        LocalIndependentGateEvidence(False, None, 0, None)
    )
    assert initial.independent_local_results[1].attempted is True
    assert initial.independent_local_results[1].passed is False
    assert refill.independent_local_results[0].attempted is True
    assert refill.independent_local_results[0].passed is False
    payload = json.loads(
        format_cascade_evidence_log(
            diagnostics,
            outcome="no_qualified_candidate",
            generated_chunk_limit=3,
            generated_text_unit_limit=100,
        ).removeprefix(CASCADE_EVIDENCE_LOG_PREFIX)
    )
    assert payload["attempts"][0]["independent_local_results"][0] == {
        "attempted": False,
        "passed": None,
        "proof_count": 0,
        "result": None,
    }
    assert payload["attempts"][0]["independent_local_results"][1][
        "passed"
    ] is False


def test_single_chunk_requires_joint_preferred_quality_and_speaker_for_early_return():
    chunk = ("完整內容",)
    generated = []

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

    def verification(seed):
        if seed == 10:
            return _squim_verification(chunk, stoi=0.65, pesq=1.15)
        if seed == 11:
            return _squim_verification(
                chunk,
                stoi=0.75,
                pesq=1.25,
                speaker_similarity=0.20,
            )
        return _squim_verification(chunk, stoi=0.76, pesq=1.30)

    def generation_evidence(candidate_index, seed, chunk_indices, candidate_chunks):
        scheduled = generation_cfg_for_candidate_offset(candidate_index)
        effective = max(3.0, scheduled)
        return CandidateGenerationEvidence(
            chunk_indices=chunk_indices,
            chunk_text_units=tuple(
                len(candidate_chunk) for candidate_chunk in candidate_chunks
            ),
            scheduled_cfg=scheduled,
            effective_cfgs=(effective,),
            floor_reasons=(("short_text",),) if effective != scheduled else ((),),
        )

    result = run_coverage_adaptive_cascade(
        chunk,
        10,
        generator,
        lambda trajectory, chunks, seed: verification(seed),
        lambda trajectory, chunks, seed: verification(seed),
        sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
        generation_evidence_factory=generation_evidence,
        max_generated_chunks=5,
        max_generated_text_units=100,
        preferred_min_speaker_similarity=0.25,
        preferred_max_boundary_speaker_drop=0.05,
        preferred_min_squim_stoi=0.72,
        preferred_min_squim_pesq=1.20,
    )

    assert generated == [10, 11, 12]
    assert result.seed == 12
    assert result.candidate_index == 2
    assert result.chunk_candidate_indices == (2,)
    assert result.generated_chunk_count == 3
    line = format_cascade_evidence_log(
        result.diagnostics,
        outcome="returned",
        generated_chunk_limit=5,
        generated_text_unit_limit=100,
        selection=result,
    )
    payload = json.loads(line.removeprefix(CASCADE_EVIDENCE_LOG_PREFIX))
    assert payload["generated_chunk_count"] == 3
    assert payload["selection"]["candidate_index"] == 2
    assert payload["selection"]["generated_chunks"] == 3


def test_single_chunk_uses_exact_joined_preferred_evidence_for_early_return():
    chunk = ("完整內容",)
    generated = []

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

    def refill_verifier(trajectory, candidate_chunks, seed):
        local = _squim_verification(
            candidate_chunks,
            stoi=0.75,
            pesq=1.25,
            speaker_similarity=0.20,
        )
        exact_joined = _squim_verification(
            candidate_chunks,
            stoi=0.75,
            pesq=1.25,
            speaker_similarity=0.30,
            boundary_drop=0.02,
        )
        return CandidateVerification(
            local,
            joined_output=trajectory_gate_evidence(exact_joined),
            independent_output=trajectory_gate_evidence(
                _exact_final(candidate_chunks)
            ),
        )

    result = run_coverage_adaptive_cascade(
        chunk,
        10,
        generator,
        lambda trajectory, candidate_chunks, seed: _joined_rejected_local(
            candidate_chunks,
            passing=(False,),
        ),
        refill_verifier,
        sequence_final_verifier=lambda result, chunks: pytest.fail(
            "exact joined preferred single chunk must return before sequence search"
        ),
        max_generated_chunks=5,
        max_generated_text_units=100,
        preferred_min_speaker_similarity=0.25,
        preferred_max_boundary_speaker_drop=0.05,
        preferred_min_squim_stoi=0.72,
        preferred_min_squim_pesq=1.20,
    )

    assert generated == [10, 11]
    assert result.seed == 11
    assert result.candidate_index == 1
    assert result.chunk_candidate_indices == (1,)
    assert result.generated_chunk_count == 2
    selected = result.diagnostics.attempts[-1]
    assert selected.joined_output is not None
    assert selected.joined_output.result is not None
    assert selected.joined_output.result.speaker_similarity == pytest.approx(0.30)


def test_single_chunk_initial_exact_fast_tier_returns_without_refill():
    chunk = ("完整內容",)
    generated = []

    def whole_verifier(trajectory, candidate_chunks, seed):
        exact = _squim_verification(
            candidate_chunks,
            stoi=0.75,
            pesq=1.25,
            speaker_similarity=0.24,
            boundary_drop=0.075,
            hard_max_boundary_speaker_drop=0.10,
        )
        return CandidateVerification(
            exact,
            joined_output=trajectory_gate_evidence(exact),
            independent_output=trajectory_gate_evidence(
                _exact_final(candidate_chunks)
            ),
        )

    result = run_coverage_adaptive_cascade(
        chunk,
        80,
        lambda candidate_chunks, seed: (
            generated.append(seed) or (f"audio-{seed}",)
        ),
        whole_verifier,
        lambda *args: pytest.fail("exact fast initial must not refill"),
        sequence_final_verifier=lambda *args: pytest.fail(
            "exact fast initial must not enter sequence search"
        ),
        preferred_min_speaker_similarity=0.25,
        preferred_max_boundary_speaker_drop=0.05,
        preferred_min_squim_stoi=0.72,
        preferred_min_squim_pesq=1.20,
        preferred_min_squim_audio_duration_seconds=1.5,
        single_exact_min_speaker_similarity=0.23,
        single_exact_max_boundary_speaker_drop=0.08,
    )

    assert generated == [80]
    assert result.candidate_index == 0
    assert result.generated_chunk_count == 1


def test_single_chunk_refill_exact_fast_tier_returns_before_more_candidates():
    chunk = ("完整內容",)
    generated = []

    def refill_verifier(trajectory, candidate_chunks, seed):
        local = _squim_verification(
            candidate_chunks,
            stoi=0.75,
            pesq=1.25,
            speaker_similarity=0.20,
        )
        exact = _squim_verification(
            candidate_chunks,
            stoi=0.75,
            pesq=1.25,
            speaker_similarity=0.24,
            boundary_drop=0.075,
            hard_max_boundary_speaker_drop=0.10,
        )
        return CandidateVerification(
            local,
            joined_output=trajectory_gate_evidence(exact),
            independent_output=trajectory_gate_evidence(
                _exact_final(candidate_chunks)
            ),
        )

    result = run_coverage_adaptive_cascade(
        chunk,
        10,
        lambda candidate_chunks, seed: (
            generated.append(seed) or (f"audio-{seed}",)
        ),
        lambda trajectory, candidate_chunks, seed: _joined_rejected_local(
            candidate_chunks,
            passing=(False,),
        ),
        refill_verifier,
        sequence_final_verifier=lambda *args: pytest.fail(
            "exact fast refill must return before sequence search"
        ),
        max_generated_chunks=5,
        max_generated_text_units=100,
        preferred_min_speaker_similarity=0.25,
        preferred_max_boundary_speaker_drop=0.05,
        preferred_min_squim_stoi=0.72,
        preferred_min_squim_pesq=1.20,
        preferred_min_squim_audio_duration_seconds=1.5,
        single_exact_min_speaker_similarity=0.23,
        single_exact_max_boundary_speaker_drop=0.08,
    )

    assert generated == [10, 11]
    assert result.seed == 11
    assert result.candidate_index == 1
    assert result.generated_chunk_count == 2


def test_single_exact_fast_tier_never_uses_local_proxy():
    chunk = ("完整內容",)
    generated = []

    def verifier(trajectory, candidate_chunks, seed):
        if seed == 10:
            return _squim_verification(
                candidate_chunks,
                stoi=0.75,
                pesq=1.25,
                speaker_similarity=0.24,
                boundary_drop=0.075,
                hard_max_boundary_speaker_drop=0.10,
            )
        return _squim_verification(
            candidate_chunks,
            stoi=0.75,
            pesq=1.25,
            speaker_similarity=0.30,
            boundary_drop=0.02,
        )

    result = run_coverage_adaptive_cascade(
        chunk,
        10,
        lambda candidate_chunks, seed: (
            generated.append(seed) or (f"audio-{seed}",)
        ),
        verifier,
        verifier,
        sequence_final_verifier=lambda *args: pytest.fail(
            "strict preferred refill must return before sequence search"
        ),
        max_generated_chunks=5,
        max_generated_text_units=100,
        preferred_min_speaker_similarity=0.25,
        preferred_max_boundary_speaker_drop=0.05,
        preferred_min_squim_stoi=0.72,
        preferred_min_squim_pesq=1.20,
        preferred_min_squim_audio_duration_seconds=1.5,
        single_exact_min_speaker_similarity=0.23,
        single_exact_max_boundary_speaker_drop=0.08,
    )

    assert generated == [10, 11]
    assert result.seed == 11


def test_single_exact_fast_tier_requires_whole_semantic_projection_evidence():
    chunk = ("完整內容",)
    generated = []

    def verifier(trajectory, candidate_chunks, seed):
        if seed == 10:
            exact = _squim_verification(
                candidate_chunks,
                stoi=0.75,
                pesq=1.25,
                speaker_similarity=0.24,
                boundary_drop=0.075,
                hard_max_boundary_speaker_drop=0.10,
            )
            return CandidateVerification(
                exact,
                joined_output=trajectory_gate_evidence(exact),
            )
        return _squim_verification(
            candidate_chunks,
            stoi=0.75,
            pesq=1.25,
            speaker_similarity=0.30,
            boundary_drop=0.02,
        )

    result = run_coverage_adaptive_cascade(
        chunk,
        10,
        lambda candidate_chunks, seed: (
            generated.append(seed) or (f"audio-{seed}",)
        ),
        verifier,
        verifier,
        sequence_final_verifier=lambda *args: pytest.fail(
            "strict preferred refill must return before sequence search"
        ),
        max_generated_chunks=5,
        max_generated_text_units=100,
        preferred_min_speaker_similarity=0.25,
        preferred_max_boundary_speaker_drop=0.05,
        preferred_min_squim_stoi=0.72,
        preferred_min_squim_pesq=1.20,
        preferred_min_squim_audio_duration_seconds=1.5,
        single_exact_min_speaker_similarity=0.23,
        single_exact_max_boundary_speaker_drop=0.08,
    )

    assert generated == [10, 11]
    assert result.seed == 11


def test_preferred_refill_selects_prior_natural_endpoint_over_forced_candidate():
    chunk = ("完整內容",)
    generated = []
    profiles = {
        10: (0.65, 1.15),
        11: (0.71, 1.19),
        12: (0.73, 1.21),
    }

    def verifier(trajectory, chunks, seed):
        stoi, pesq = profiles[seed]
        return _squim_verification(
            chunks,
            stoi=stoi,
            pesq=pesq,
            speaker_similarity=0.30,
        )

    def generation_evidence(candidate_index, seed, chunk_indices, candidate_chunks):
        scheduled = generation_cfg_for_candidate_offset(candidate_index)
        effective = max(3.0, scheduled)
        threshold_stop = candidate_index == 1
        return CandidateGenerationEvidence(
            chunk_indices=chunk_indices,
            chunk_text_units=(4,),
            scheduled_cfg=scheduled,
            effective_cfgs=(effective,),
            floor_reasons=(("short_text",),) if effective != scheduled else ((),),
            chunk_stop_reasons=(
                ("stop_threshold",) if threshold_stop else ("hard_stop",)
            ),
            chunk_endpoint_energy_ratios=(
                (0.10,) if threshold_stop else (0.90,)
            ),
            chunk_generated_steps=((4,) if threshold_stop else (5,)),
            chunk_hard_stop_steps=(5,),
        )

    result = run_coverage_adaptive_cascade(
        chunk,
        10,
        lambda chunks, seed: generated.append(seed) or (f"audio-{seed}",),
        verifier,
        verifier,
        sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
        generation_evidence_factory=generation_evidence,
        max_generated_chunks=5,
        max_generated_text_units=100,
        preferred_min_speaker_similarity=0.25,
        preferred_max_boundary_speaker_drop=0.05,
        preferred_min_squim_stoi=0.72,
        preferred_min_squim_pesq=1.20,
        require_endpoint_evidence=True,
    )

    assert generated == [10, 11, 12]
    assert result.candidate_index == 1
    assert result.seed == 11
    artifact = result.verification.chunk_artifacts[0]
    assert artifact.stop_reason == "stop_threshold"
    assert artifact.endpoint_energy_ratio == pytest.approx(0.10)
    payload = json.loads(
        format_cascade_evidence_log(
            result.diagnostics,
            outcome="returned",
            generated_chunk_limit=5,
            generated_text_unit_limit=100,
            selection=result,
        ).removeprefix(CASCADE_EVIDENCE_LOG_PREFIX)
    )
    assert payload["endpoint_evidence_complete"] is True
    assert payload["selection"]["generation"]["endpoint_complete"] is True
    assert payload["selection"]["generation"]["chunk_stop_reasons"] == [
        "stop_threshold"
    ]


def test_endpoint_binding_keeps_preflight_rejection_and_continues_to_refill():
    chunk = ("完整內容",)
    generated = []

    def verifier(trajectory, candidate_chunks, seed):
        if seed == 10:
            return verify_trajectory(
                [
                    CandidateObservation(
                        target_text=candidate_chunks[0],
                        transcript_text=candidate_chunks[0],
                        audio_duration_seconds=0.5,
                        pace_cps=8.0,
                    )
                ],
                chunk_artifacts=(ChunkCandidateArtifact(),),
                max_pace_cps=4.3,
            )
        return _local_verification(candidate_chunks)

    def generation_evidence(candidate_index, seed, chunk_indices, candidate_chunks):
        return CandidateGenerationEvidence(
            chunk_indices=chunk_indices,
            chunk_text_units=(4,),
            scheduled_cfg=generation_cfg_for_candidate_offset(candidate_index),
            effective_cfgs=(
                generation_cfg_for_candidate_offset(candidate_index),
            ),
            floor_reasons=((),),
            chunk_stop_reasons=("stop_threshold",),
            chunk_endpoint_energy_ratios=(0.10,),
            chunk_generated_steps=(4,),
            chunk_hard_stop_steps=(5,),
        )

    result = run_coverage_adaptive_cascade(
        chunk,
        10,
        lambda candidate_chunks, seed: (
            generated.append(seed) or (f"audio-{seed}",)
        ),
        verifier,
        verifier,
        sequence_final_verifier=lambda *args: pytest.fail(
            "passing refill must return before sequence search"
        ),
        generation_evidence_factory=generation_evidence,
        max_generated_chunks=5,
        max_generated_text_units=100,
        preferred_min_speaker_similarity=0.70,
        preferred_max_boundary_speaker_drop=0.10,
        require_endpoint_evidence=True,
    )

    assert generated == [10, 11]
    assert result.seed == 11
    assert result.diagnostics.attempts[0].trajectory_passed is False
    assert (
        "chunk_0:pace_too_fast"
        in result.diagnostics.attempts[0].trajectory_rejection_reasons
    )
    assert (
        result.diagnostics.attempts[0].chunk_stop_reasons
        == ("stop_threshold",)
    )


def test_fast_exact_refill_cannot_select_relaxed_endpoint_waiver_candidate():
    chunk = ("完整內容",)
    generated = []

    def verifier(trajectory, candidate_chunks, seed):
        if seed == 10:
            return _squim_verification(
                candidate_chunks,
                stoi=0.65,
                pesq=1.15,
                speaker_similarity=0.30,
            )
        if seed == 11:
            return _squim_verification(
                candidate_chunks,
                stoi=0.71,
                pesq=1.19,
                speaker_similarity=0.30,
            )
        exact = _squim_verification(
            candidate_chunks,
            stoi=0.75,
            pesq=1.25,
            speaker_similarity=0.24,
            boundary_drop=0.075,
            hard_max_boundary_speaker_drop=0.10,
        )
        return CandidateVerification(
            exact,
            joined_output=trajectory_gate_evidence(exact),
            independent_output=trajectory_gate_evidence(
                _exact_final(candidate_chunks)
            ),
        )

    def generation_evidence(candidate_index, seed, chunk_indices, candidate_chunks):
        threshold_stop = candidate_index == 1
        return CandidateGenerationEvidence(
            chunk_indices=chunk_indices,
            chunk_text_units=(4,),
            scheduled_cfg=generation_cfg_for_candidate_offset(candidate_index),
            effective_cfgs=(
                generation_cfg_for_candidate_offset(candidate_index),
            ),
            floor_reasons=((),),
            chunk_stop_reasons=(
                ("stop_threshold",) if threshold_stop else ("hard_stop",)
            ),
            chunk_endpoint_energy_ratios=(
                (0.10,) if threshold_stop else (0.90,)
            ),
            chunk_generated_steps=((4,) if threshold_stop else (5,)),
            chunk_hard_stop_steps=(5,),
        )

    result = run_coverage_adaptive_cascade(
        chunk,
        10,
        lambda candidate_chunks, seed: (
            generated.append(seed) or (f"audio-{seed}",)
        ),
        verifier,
        verifier,
        sequence_final_verifier=lambda *args: pytest.fail(
            "fast exact refill must return before sequence search"
        ),
        generation_evidence_factory=generation_evidence,
        max_generated_chunks=5,
        max_generated_text_units=100,
        preferred_min_speaker_similarity=0.25,
        preferred_max_boundary_speaker_drop=0.05,
        preferred_min_squim_stoi=0.72,
        preferred_min_squim_pesq=1.20,
        preferred_min_squim_audio_duration_seconds=1.5,
        single_exact_min_speaker_similarity=0.23,
        single_exact_max_boundary_speaker_drop=0.08,
        require_endpoint_evidence=True,
    )

    assert generated == [10, 11, 12]
    assert result.candidate_index == 2
    assert result.seed == 12


@pytest.mark.parametrize(
    ("field", "value"),
    [
        ("chunk_stop_reasons", ("unknown",)),
        ("chunk_endpoint_energy_ratios", (math.nan,)),
        ("chunk_generated_steps", (True,)),
        ("chunk_generated_steps", (6,)),
    ],
)
def test_canonical_log_rejects_invalid_endpoint_rows(field, value):
    chunk = ("完整內容",)

    def evidence(candidate_index, seed, chunk_indices, candidate_chunks):
        return CandidateGenerationEvidence(
            chunk_indices=chunk_indices,
            chunk_text_units=(4,),
            scheduled_cfg=3.0,
            effective_cfgs=(3.0,),
            floor_reasons=((),),
            chunk_stop_reasons=("stop_threshold",),
            chunk_endpoint_energy_ratios=(0.10,),
            chunk_generated_steps=(4,),
            chunk_hard_stop_steps=(5,),
        )

    result = run_coverage_adaptive_cascade(
        chunk,
        10,
        lambda chunks, seed: (f"audio-{seed}",),
        lambda trajectory, chunks, seed: _squim_verification(
            chunks,
            stoi=0.73,
            pesq=1.21,
            speaker_similarity=0.30,
        ),
        lambda trajectory, chunks, seed: _squim_verification(
            chunks,
            stoi=0.73,
            pesq=1.21,
            speaker_similarity=0.30,
        ),
        sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
        generation_evidence_factory=evidence,
        max_generated_chunks=5,
        max_generated_text_units=100,
        preferred_min_speaker_similarity=0.25,
        preferred_max_boundary_speaker_drop=0.05,
        preferred_min_squim_stoi=0.72,
        preferred_min_squim_pesq=1.20,
        require_endpoint_evidence=True,
    )

    invalid_attempt = replace(
        result.diagnostics.attempts[0],
        **{field: value},
    )
    invalid_diagnostics = replace(
        result.diagnostics,
        attempts=(invalid_attempt,),
    )
    with pytest.raises(ValueError, match="canonical endpoint evidence is invalid"):
        format_cascade_evidence_log(
            invalid_diagnostics,
            outcome="returned",
            generated_chunk_limit=5,
            generated_text_unit_limit=100,
            selection=result,
        )


@pytest.mark.parametrize(
    ("natural_stoi", "natural_pesq", "natural_speaker", "natural_steps"),
    [
        (0.69, 1.19, 0.30, 4),
        (0.71, 1.16, 0.30, 4),
        (0.71, 1.19, 0.24, 4),
        (0.71, 1.19, 0.30, 5),
    ],
)
def test_natural_endpoint_waiver_cannot_bypass_bounded_preferred_tier(
    natural_stoi,
    natural_pesq,
    natural_speaker,
    natural_steps,
):
    chunk = ("完整內容",)
    profiles = {
        10: (0.65, 1.15, 0.30),
        11: (natural_stoi, natural_pesq, natural_speaker),
        12: (0.73, 1.21, 0.30),
    }

    def verifier(trajectory, chunks, seed):
        stoi, pesq, speaker = profiles[seed]
        return _squim_verification(
            chunks,
            stoi=stoi,
            pesq=pesq,
            speaker_similarity=speaker,
        )

    def generation_evidence(candidate_index, seed, chunk_indices, candidate_chunks):
        natural = candidate_index == 1
        scheduled = generation_cfg_for_candidate_offset(candidate_index)
        return CandidateGenerationEvidence(
            chunk_indices=chunk_indices,
            chunk_text_units=(4,),
            scheduled_cfg=scheduled,
            effective_cfgs=(scheduled,),
            floor_reasons=((),),
            chunk_stop_reasons=(
                ("stop_threshold",) if natural else ("hard_stop",)
            ),
            chunk_endpoint_energy_ratios=((0.10,) if natural else (0.90,)),
            chunk_generated_steps=((natural_steps,) if natural else (5,)),
            chunk_hard_stop_steps=(5,),
        )

    result = run_coverage_adaptive_cascade(
        chunk,
        10,
        lambda chunks, seed: (f"audio-{seed}",),
        verifier,
        verifier,
        sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
        generation_evidence_factory=generation_evidence,
        max_generated_chunks=5,
        max_generated_text_units=100,
        preferred_min_speaker_similarity=0.25,
        preferred_max_boundary_speaker_drop=0.05,
        preferred_min_squim_stoi=0.72,
        preferred_min_squim_pesq=1.20,
        require_endpoint_evidence=True,
    )

    assert result.candidate_index == 2
    assert result.seed == 12


def test_single_chunk_final_cap_selects_lowest_bounded_soft_quality_cost():
    chunk = ("完整內容",)
    profiles = {
        10: (0.65, 1.15, 0.0),
        11: (0.70, 1.18, 5.0),
        12: (0.68, 1.16, -5.0),
    }
    generated = []
    enrichment_calls = []

    def verifier(trajectory, chunks, seed):
        stoi, pesq, si_sdr = profiles[seed]
        return _squim_verification(
            chunks,
            stoi=stoi,
            pesq=pesq,
            si_sdr=si_sdr,
        )

    result = run_coverage_adaptive_cascade(
        chunk,
        10,
        lambda chunks, seed: generated.append(seed) or (f"audio-{seed}",),
        verifier,
        verifier,
        sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
        transition_artifact_enricher=lambda *args: enrichment_calls.append(
            args
        ),
        max_generated_chunks=3,
        max_generated_text_units=100,
        preferred_min_speaker_similarity=0.25,
        preferred_max_boundary_speaker_drop=0.05,
        preferred_min_squim_stoi=0.72,
        preferred_min_squim_pesq=1.20,
    )

    assert generated == [10, 11, 12]
    assert enrichment_calls == []
    assert result.chunk_candidate_indices == (1,)
    assert result.chunk_seeds == (11,)
    assert result.generated_chunk_count == 3
    selected = result.verification.candidate_results[0]
    assert selected.squim_quality_cost is not None
    assert selected.squim_stoi == pytest.approx(0.70)


def test_multi_chunk_hard_pass_keeps_existing_immediate_return_contract():
    chunks = ("第一段", "第二段")
    generated = []
    result = run_coverage_adaptive_cascade(
        chunks,
        10,
        lambda candidate_chunks, seed: (
            generated.append(seed) or tuple(candidate_chunks)
        ),
        lambda trajectory, candidate_chunks, seed: _squim_verification(
            candidate_chunks,
            stoi=0.65,
            pesq=1.15,
        ),
        lambda trajectory, candidate_chunks, seed: pytest.fail(
            "multi-chunk hard pass must not refill"
        ),
        sequence_final_verifier=lambda result, chunks: pytest.fail(
            "multi-chunk hard pass must not enter sequence search"
        ),
        preferred_min_speaker_similarity=0.25,
        preferred_max_boundary_speaker_drop=0.05,
        preferred_min_squim_stoi=0.72,
        preferred_min_squim_pesq=1.20,
    )

    assert generated == [10]
    assert result.chunk_candidate_indices == (0, 0)


def test_multi_chunk_eager_rank_one_probe_returns_at_first_complete_pool():
    chunks = ("第一段", "第二段")
    generated = []
    final_calls = []
    enrichment_calls = []

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

    def refill_verifier(trajectory, candidate_chunks, seed):
        return _squim_verification(
            candidate_chunks,
            stoi=0.75,
            pesq=1.25,
            speaker_similarity=0.30,
        )

    def final_verifier(sequence_result, candidate_chunks):
        final_calls.append(sequence_result.chunk_candidate_indices)
        return _squim_verification(
            ("".join(candidate_chunks),),
            stoi=0.75,
            pesq=1.25,
            speaker_similarity=0.30,
        )

    result = run_coverage_adaptive_cascade(
        chunks,
        10,
        generator,
        lambda trajectory, candidate_chunks, seed: _joined_rejected_local(
            candidate_chunks,
            passing=(False, False),
        ),
        refill_verifier,
        sequence_final_verifier=final_verifier,
        transition_artifact_enricher=lambda *args: enrichment_calls.append(
            args
        ),
        max_generated_chunks=8,
        max_generated_text_units=100,
        max_sequence_paths=3,
        preferred_min_speaker_similarity=0.25,
        preferred_max_boundary_speaker_drop=0.05,
        preferred_min_squim_stoi=0.72,
        preferred_min_squim_pesq=1.20,
    )

    assert generated == [
        (chunks, 10),
        (("第一段",), 11),
        (("第二段",), 12),
    ]
    assert final_calls == [(1, 2)]
    assert enrichment_calls == []
    assert result.chunk_candidate_indices == (1, 2)
    assert result.chunk_seeds == (11, 12)
    assert result.generated_chunk_count == 4
    assert result.chunk_candidate_counts == (1, 1)
    assert result.sequence_path_rank == 1
    assert result.sequence_paths_checked == 1
    assert result.diagnostics.sequence_search is not None
    assert result.diagnostics.sequence_search.eligible_candidate_counts == (1, 1)
    assert len(result.diagnostics.sequence_search.checked_paths) == 1
    assert result.diagnostics.sequence_search.checked_paths[0].final_output.passed


def test_multi_chunk_eager_preferred_miss_uses_one_bounded_sequence_ledger():
    chunks = ("第一段", "第二段")
    generated = []
    final_calls = []

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

    def refill_verifier(trajectory, candidate_chunks, seed):
        return _squim_verification(
            candidate_chunks,
            stoi=0.75,
            pesq=1.25,
            speaker_similarity=0.30,
        )

    def final_verifier(sequence_result, candidate_chunks):
        final_calls.append(sequence_result.chunk_candidate_indices)
        if len(final_calls) == 1:
            # The eager path clears hard SQUIM but misses the preferred tier.
            return _squim_verification(
                ("".join(candidate_chunks),),
                stoi=0.65,
                pesq=1.15,
                speaker_similarity=0.30,
            )
        return _exact_final(candidate_chunks, passed=False)

    result = run_coverage_adaptive_cascade(
        chunks,
        10,
        generator,
        lambda trajectory, candidate_chunks, seed: _joined_rejected_local(
            candidate_chunks,
            passing=(False, False),
        ),
        refill_verifier,
        sequence_final_verifier=final_verifier,
        max_generated_chunks=8,
        max_generated_text_units=100,
        max_sequence_paths=3,
        preferred_min_speaker_similarity=0.25,
        preferred_max_boundary_speaker_drop=0.05,
        preferred_min_squim_stoi=0.72,
        preferred_min_squim_pesq=1.20,
    )

    assert [seed for _, seed in generated] == list(range(10, 17))
    assert final_calls[0] == (1, 2)
    assert len(final_calls) == 3
    assert len(set(final_calls)) == 3
    assert result.chunk_candidate_indices == (1, 2)
    assert result.generated_chunk_count == 8
    assert result.chunk_candidate_counts == (3, 3)
    assert result.sequence_paths_checked == 3
    search = result.diagnostics.sequence_search
    assert search is not None
    assert search.ranked_path_count == 3
    assert len(search.checked_paths) == 3
    assert [
        path.chunk_candidate_indices for path in search.checked_paths
    ] == final_calls
    assert [
        path.final_output.passed for path in search.checked_paths
    ] == [True, False, False]


def test_multi_chunk_eager_hard_failure_reserves_remaining_completed_search():
    chunks = ("第一段", "第二段")
    generated = []
    final_calls = []

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

    def final_verifier(sequence_result, candidate_chunks):
        final_calls.append(
            (
                sequence_result.chunk_candidate_indices,
                len(generated),
            )
        )
        return _exact_final(
            candidate_chunks,
            passed=len(final_calls) == 3,
        )

    result = run_coverage_adaptive_cascade(
        chunks,
        10,
        generator,
        lambda trajectory, candidate_chunks, seed: _joined_rejected_local(
            candidate_chunks,
            passing=(False, False),
        ),
        lambda trajectory, candidate_chunks, seed: _squim_verification(
            candidate_chunks,
            stoi=0.75,
            pesq=1.25,
            speaker_similarity=0.30,
        ),
        sequence_final_verifier=final_verifier,
        max_generated_chunks=8,
        max_generated_text_units=100,
        max_sequence_paths=3,
        preferred_min_speaker_similarity=0.25,
        preferred_max_boundary_speaker_drop=0.05,
        preferred_min_squim_stoi=0.72,
        preferred_min_squim_pesq=1.20,
    )

    assert [snapshot for _, snapshot in final_calls] == [3, 7, 7]
    assert len({identity for identity, _ in final_calls}) == 3
    assert result.generated_chunk_count == 8
    assert result.sequence_paths_checked == 3


def test_lazy_transition_enrichment_matches_precomputed_multi_path_ranking():
    chunks = ("第一段", "第二段")

    def run_once(*, lazy):
        generated = []
        final_calls = []
        enrichment_calls = []

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

        def with_f0(verification, seed):
            return replace(
                verification,
                chunk_artifacts=tuple(
                    replace(artifact, median_f0_hz=float(seed))
                    for artifact in verification.chunk_artifacts
                ),
            )

        def refill_verifier(trajectory, candidate_chunks, seed):
            verification = _squim_verification(
                candidate_chunks,
                stoi=0.75,
                pesq=1.25,
                speaker_similarity=0.30,
            )
            return verification if lazy else with_f0(verification, seed)

        def enrich(audio, artifact):
            enrichment_calls.append(str(audio))
            seed = int(str(audio).split(":", 1)[0])
            return replace(
                artifact,
                median_f0_hz=np.float32(seed),
                rms_db=999.0,
                speaker_embedding=np.array([0.0, 1.0], dtype=np.float32),
                stop_reason="hard_stop",
                generated_steps=1,
                hard_stop_steps=2,
            )

        def final_verifier(sequence_result, candidate_chunks):
            identity = sequence_result.chunk_candidate_indices
            final_calls.append(identity)
            if identity == (1, 2):
                return _squim_verification(
                    ("".join(candidate_chunks),),
                    stoi=0.65,
                    pesq=1.15,
                    speaker_similarity=0.30,
                )
            return _squim_verification(
                ("".join(candidate_chunks),),
                stoi=0.75,
                pesq=1.25,
                speaker_similarity=0.30,
            )

        result = run_coverage_adaptive_cascade(
            chunks,
            10,
            generator,
            lambda trajectory, candidate_chunks, seed: (
                _joined_rejected_local(
                    candidate_chunks,
                    passing=(False, False),
                )
            ),
            refill_verifier,
            sequence_final_verifier=final_verifier,
            transition_artifact_enricher=(enrich if lazy else None),
            max_generated_chunks=8,
            max_generated_text_units=100,
            max_sequence_paths=3,
            preferred_min_speaker_similarity=0.25,
            preferred_max_boundary_speaker_drop=0.05,
            preferred_min_squim_stoi=0.72,
            preferred_min_squim_pesq=1.20,
        )
        return result, generated, final_calls, enrichment_calls

    eager, eager_generated, eager_finals, eager_enrichment = run_once(
        lazy=False
    )
    lazy, lazy_generated, lazy_finals, lazy_enrichment = run_once(lazy=True)

    assert lazy_generated == eager_generated
    assert lazy_finals == eager_finals
    assert len(lazy_finals) == 2
    assert lazy_finals[0] == (1, 2)
    assert lazy.chunk_candidate_indices == eager.chunk_candidate_indices
    assert lazy.chunk_seeds == eager.chunk_seeds
    assert lazy.verification.score == eager.verification.score
    assert lazy.generated_chunk_count == eager.generated_chunk_count == 5
    assert lazy.generated_text_units == eager.generated_text_units
    assert lazy.attempted_seeds == eager.attempted_seeds
    assert eager_enrichment == []
    assert len(lazy_enrichment) == 3
    assert len(set(lazy_enrichment)) == 3
    assert tuple(
        artifact.median_f0_hz
        for artifact in lazy.verification.chunk_artifacts
    ) == tuple(
        artifact.median_f0_hz
        for artifact in eager.verification.chunk_artifacts
    )
    assert all(
        type(artifact.median_f0_hz) is float
        for artifact in lazy.verification.chunk_artifacts
    )
    for lazy_artifact, eager_artifact in zip(
        lazy.verification.chunk_artifacts,
        eager.verification.chunk_artifacts,
        strict=True,
    ):
        assert lazy_artifact.rms_db == eager_artifact.rms_db
        assert np.array_equal(
            lazy_artifact.speaker_embedding,
            eager_artifact.speaker_embedding,
        )
        assert lazy_artifact.stop_reason == eager_artifact.stop_reason
        assert lazy_artifact.generated_steps == eager_artifact.generated_steps
        assert lazy_artifact.hard_stop_steps == eager_artifact.hard_stop_steps


@pytest.mark.parametrize(
    ("enricher", "expected_error"),
    [
        (1, ValueError),
        (lambda _audio, _artifact: None, RuntimeError),
        (
            lambda _audio, artifact: replace(
                artifact,
                median_f0_hz=float("nan"),
            ),
            RuntimeError,
        ),
        (
            lambda _audio, artifact: replace(
                artifact,
                median_f0_hz=0.0,
            ),
            RuntimeError,
        ),
        (
            lambda _audio, _artifact: (_ for _ in ()).throw(
                OSError("unavailable")
            ),
            RuntimeError,
        ),
    ],
)
def test_transition_artifact_enricher_fails_closed(
    enricher,
    expected_error,
):
    chunks = ("第一段", "第二段")
    with pytest.raises(expected_error, match="transition artifact"):
        run_coverage_adaptive_cascade(
            chunks,
            10,
            lambda candidate_chunks, seed: tuple(
                f"{seed}:{chunk}" for chunk in candidate_chunks
            ),
            lambda trajectory, candidate_chunks, seed: (
                _joined_rejected_local(
                    candidate_chunks,
                    passing=(False, False),
                )
            ),
            lambda trajectory, candidate_chunks, seed: _local_verification(
                candidate_chunks
            ),
            sequence_final_verifier=lambda result, candidate_chunks: (
                _exact_final(candidate_chunks, passed=False)
            ),
            transition_artifact_enricher=enricher,
            max_generated_chunks=8,
            max_generated_text_units=100,
            max_sequence_paths=2,
        )


def test_short_single_chunk_hard_pass_skips_duration_biased_preferred_search():
    generated = []

    def verifier(trajectory, chunks, seed):
        return _squim_verification(
            chunks,
            stoi=0.65,
            pesq=1.15,
            audio_duration_seconds=1.0,
        )

    result = run_coverage_adaptive_cascade(
        ("你好",),
        10,
        lambda chunks, seed: generated.append(seed) or (f"audio-{seed}",),
        verifier,
        verifier,
        sequence_final_verifier=lambda result, chunks: pytest.fail(
            "short hard pass must not enter sequence search"
        ),
        preferred_min_speaker_similarity=0.25,
        preferred_max_boundary_speaker_drop=0.05,
        preferred_min_squim_stoi=0.72,
        preferred_min_squim_pesq=1.20,
        preferred_min_squim_audio_duration_seconds=1.5,
    )

    assert generated == [10]
    assert result.candidate_index == 0
    assert result.generated_chunk_count == 1


@pytest.mark.parametrize(
    "preferred",
    [
        {"preferred_min_speaker_similarity": 0.25},
        {"preferred_max_boundary_speaker_drop": 0.05},
        {"preferred_min_squim_stoi": 0.72},
        {"preferred_min_squim_pesq": 1.20},
        {
            "preferred_min_squim_stoi": float("nan"),
            "preferred_min_squim_pesq": 1.20,
        },
        {
            "preferred_min_squim_stoi": 0.72,
            "preferred_min_squim_pesq": 1.20,
            "preferred_min_squim_audio_duration_seconds": float("nan"),
        },
    ],
)
def test_coverage_preferred_thresholds_must_be_explicit_finite_pairs(preferred):
    with pytest.raises(ValueError, match="preferred"):
        run_coverage_adaptive_cascade(
            ("完整內容",),
            10,
            lambda chunks, seed: tuple(chunks),
            lambda trajectory, chunks, seed: _local_verification(chunks),
            lambda trajectory, chunks, seed: _local_verification(chunks),
            sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
            **preferred,
        )


@pytest.mark.parametrize(
    "single_exact",
    [
        {"single_exact_min_speaker_similarity": 0.23},
        {"single_exact_max_boundary_speaker_drop": 0.08},
        {
            "single_exact_min_speaker_similarity": float("nan"),
            "single_exact_max_boundary_speaker_drop": 0.08,
        },
        {
            "single_exact_min_speaker_similarity": 0.23,
            "single_exact_max_boundary_speaker_drop": 0.08,
        },
    ],
)
def test_single_exact_fast_thresholds_require_pair_and_preferred_squim(
    single_exact,
):
    with pytest.raises(ValueError, match="single exact"):
        run_coverage_adaptive_cascade(
            ("完整內容",),
            10,
            lambda chunks, seed: tuple(chunks),
            lambda trajectory, chunks, seed: _local_verification(chunks),
            lambda trajectory, chunks, seed: _local_verification(chunks),
            sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
            **single_exact,
        )


def test_refill_budget_accounts_exact_generated_chunks_and_text_units():
    chunks = ("甲乙", "丙丁")
    generated = []

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

    def generation_evidence(candidate_index, seed, chunk_indices, candidate_chunks):
        scheduled = 3.0 if candidate_index % 2 == 0 else 2.0
        effective = 3.0 if candidate_index == 1 else scheduled
        reasons = (("short_text",),) if candidate_index == 1 else ((), ())
        return CandidateGenerationEvidence(
            chunk_indices=chunk_indices,
            chunk_text_units=tuple(2 for _ in candidate_chunks),
            scheduled_cfg=scheduled,
            effective_cfgs=tuple(effective for _ in candidate_chunks),
            floor_reasons=reasons,
        )

    result = run_coverage_adaptive_cascade(
        chunks,
        10,
        generator,
        lambda trajectory, candidate_chunks, seed: _joined_rejected_local(
            candidate_chunks
        ),
        lambda trajectory, candidate_chunks, seed: _local_verification(
            candidate_chunks
        ),
        sequence_final_verifier=lambda result, candidate_chunks: _exact_final(
            candidate_chunks
        ),
        generation_evidence_factory=generation_evidence,
        max_generated_chunks=20,
        max_generated_text_units=6,
        max_sequence_paths=3,
    )

    assert generated == [
        (chunks, 10),
        (("甲乙",), 11),
    ]
    assert result.generated_chunk_count == 3
    assert result.generated_text_units == 6
    assert result.chunk_candidate_counts == (2, 1)
    assert result.attempted_seeds == (10, 11)
    assert result.chunk_candidate_indices == (1, 0)
    assert len(result.diagnostics.attempts) == 2
    assert result.diagnostics.sequence_search is not None
    assert result.diagnostics.sequence_search.eligible_candidate_counts == (2, 1)
    assert result.diagnostics.sequence_search.ranked_path_count >= 1
    assert [
        path.final_output.passed
        for path in result.diagnostics.sequence_search.checked_paths
    ] == [True]
    line = format_cascade_evidence_log(
        result.diagnostics,
        outcome="returned",
        generated_chunk_limit=20,
        generated_text_unit_limit=6,
        selection=result,
    )
    payload = json.loads(line.removeprefix(CASCADE_EVIDENCE_LOG_PREFIX))
    assert payload["schema_version"] == 6
    assert payload["generation_evidence_complete"] is True
    assert payload["request_chunk_count"] == 2
    assert payload["generated_chunk_count"] == 3
    assert payload["generated_text_units"] == 6
    assert payload["limits"] == {
        "generated_chunks": 20,
        "generated_text_units": 6,
    }
    assert payload["attempts"][1]["chunk_indices"] == [0]
    assert payload["attempts"][1]["chunk_candidate_ordinals"] == [1]
    assert payload["attempts"][1]["scheduled_cfg"] == 2.0
    assert payload["attempts"][1]["effective_cfgs"] == [3.0]
    assert payload["attempts"][1]["floor_reasons"] == [["short_text"]]
    assert payload["selection"]["generation"] == {
        "complete": True,
        "chunk_scheduled_cfgs": [2.0, 3.0],
        "chunk_effective_cfgs": [3.0, 3.0],
        "chunk_floor_reasons": [["short_text"], []],
        "chunk_candidate_ordinals": [1, 0],
        "chunk_policies": ["safe_duration", "base"],
        "network_conditioned": [False, False],
        "chunk_text_variants": ["base", "base"],
        "endpoint_complete": False,
        "chunk_stop_reasons": [None, None],
        "chunk_endpoint_energy_ratios": [None, None],
        "chunk_generated_steps": [None, None],
        "chunk_hard_stop_steps": [None, None],
    }


def test_one_path_mode_still_creates_one_alternative_to_rejected_initial_audio():
    generated = []

    result = run_coverage_adaptive_cascade(
        ("完整內容",),
        10,
        lambda chunks, seed: (
            generated.append((chunks, seed))
            or tuple(f"{seed}:{chunk}" for chunk in chunks)
        ),
        lambda trajectory, chunks, seed: _joined_rejected_local(chunks),
        lambda trajectory, chunks, seed: _local_verification(chunks),
        sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
        max_generated_chunks=2,
        max_generated_text_units=100,
        max_sequence_paths=1,
    )

    assert generated == [(("完整內容",), 10), (("完整內容",), 11)]
    assert result.chunk_candidate_indices == (1,)
    assert result.generated_chunk_count == 2


def test_generation_evidence_factory_mismatch_fails_closed():
    def mismatched_evidence(candidate_index, seed, chunk_indices, candidate_chunks):
        return CandidateGenerationEvidence(
            chunk_indices=(99,),
            chunk_text_units=(4,),
            scheduled_cfg=3.0,
            effective_cfgs=(3.0,),
            floor_reasons=((),),
        )

    with pytest.raises(RuntimeError, match="candidate generation evidence is invalid"):
        run_coverage_adaptive_cascade(
            ("完整內容",),
            10,
            lambda chunks, seed: tuple(chunks),
            lambda trajectory, chunks, seed: _local_verification(chunks),
            lambda trajectory, chunks, seed: _local_verification(chunks),
            sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
            generation_evidence_factory=mismatched_evidence,
        )


def test_no_qualified_canonical_evidence_retains_complete_cfg_budget_provenance():
    chunk = "完整內容"

    def generation_evidence(candidate_index, seed, chunk_indices, candidate_chunks):
        scheduled = 3.0 if candidate_index % 2 == 0 else 2.0
        effective = 3.0
        reasons = () if scheduled == effective else ("short_text",)
        return CandidateGenerationEvidence(
            chunk_indices=chunk_indices,
            chunk_text_units=(4,),
            scheduled_cfg=scheduled,
            effective_cfgs=(effective,),
            floor_reasons=(reasons,),
        )

    with pytest.raises(NoQualifiedCandidateError) as caught:
        run_coverage_adaptive_cascade(
            (chunk,),
            20,
            lambda chunks, seed: tuple(chunks),
            lambda trajectory, chunks, seed: _joined_rejected_local(chunks),
            lambda trajectory, chunks, seed: _local_verification(chunks),
            sequence_final_verifier=lambda result, chunks: _exact_final(
                chunks,
                passed=False,
            ),
            generation_evidence_factory=generation_evidence,
            max_generated_chunks=2,
            max_generated_text_units=8,
            max_sequence_paths=1,
        )

    line = format_cascade_evidence_log(
        caught.value.diagnostics,
        outcome="no_qualified_candidate",
        generated_chunk_limit=2,
        generated_text_unit_limit=8,
    )
    payload = json.loads(line.removeprefix(CASCADE_EVIDENCE_LOG_PREFIX))
    assert payload["outcome"] == "no_qualified_candidate"
    assert payload["generation_evidence_complete"] is True
    assert payload["generated_chunk_count"] == 2
    assert payload["generated_text_units"] == 8
    assert payload["attempts"][1]["chunk_indices"] == [0]
    assert payload["attempts"][1]["scheduled_cfg"] == 2.0
    assert payload["attempts"][1]["effective_cfgs"] == [3.0]
    assert payload["selection"] is None

@pytest.mark.parametrize(
    ("budget_name", "value", "frozen_cap"),
    [
        ("max_generated_chunks", 33, 32),
        ("max_generated_text_units", 801, 800),
    ],
)
def test_generation_budgets_cannot_expand_beyond_frozen_caps(
    budget_name,
    value,
    frozen_cap,
):
    with pytest.raises(ValueError, match=f"frozen cap of {frozen_cap}"):
        run_coverage_adaptive_cascade(
            ("完整內容",),
            10,
            lambda chunks, seed: tuple(chunks),
            lambda trajectory, chunks, seed: _local_verification(chunks),
            lambda trajectory, chunks, seed: _local_verification(chunks),
            sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
            **{budget_name: value},
        )


def test_refill_schedule_and_selected_path_are_deterministic():
    chunks = ("甲乙", "丙丁", "戊己")

    def run_once():
        generated = []
        checked = []

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

        result = run_coverage_adaptive_cascade(
            chunks,
            20,
            generator,
            lambda trajectory, candidate_chunks, seed: _joined_rejected_local(
                candidate_chunks
            ),
            lambda trajectory, candidate_chunks, seed: _local_verification(
                candidate_chunks
            ),
            sequence_final_verifier=lambda result, candidate_chunks: (
                checked.append(result.chunk_candidate_indices)
                or _exact_final(candidate_chunks)
            ),
            max_generated_chunks=8,
            max_generated_text_units=100,
            max_sequence_paths=2,
        )
        return generated, checked, result

    first_generated, first_checked, first = run_once()
    second_generated, second_checked, second = run_once()

    assert first_generated == second_generated
    assert first_checked == second_checked
    assert first.chunk_candidate_indices == second.chunk_candidate_indices
    assert first.chunk_seeds == second.chunk_seeds
    assert first.generated_chunk_count == second.generated_chunk_count == 6
    assert first.chunk_candidate_counts == second.chunk_candidate_counts == (2, 2, 2)


def test_context_aware_refills_rotate_cfg_and_endpoint_policy_per_chunk():
    chunks = ("第一段", "第二段")
    contexts = []

    def generator(candidate_chunks, seed, *, generation_context):
        assert isinstance(generation_context, CandidateGenerationContext)
        contexts.append(generation_context)
        return generation_context.chunk_candidate_ordinals

    def generation_evidence(
        candidate_index,
        seed,
        chunk_indices,
        candidate_chunks,
        *,
        generation_context,
    ):
        assert candidate_index == generation_context.candidate_index
        assert seed == generation_context.seed
        assert chunk_indices == generation_context.chunk_indices
        ordinal = generation_context.chunk_candidate_ordinals[0]
        assert all(
            candidate_ordinal == ordinal
            for candidate_ordinal in generation_context.chunk_candidate_ordinals
        )
        scheduled = generation_cfg_for_candidate_offset(ordinal)
        return CandidateGenerationEvidence(
            chunk_indices=chunk_indices,
            chunk_text_units=tuple(3 for _ in candidate_chunks),
            scheduled_cfg=scheduled,
            effective_cfgs=tuple(scheduled for _ in candidate_chunks),
            floor_reasons=tuple(() for _ in candidate_chunks),
            chunk_candidate_ordinals=(ordinal,) * len(candidate_chunks),
        )

    result = run_coverage_adaptive_cascade(
        chunks,
        100,
        generator,
        lambda trajectory, candidate_chunks, seed: _joined_rejected_local(
            candidate_chunks,
            passing=(False, False),
        ),
        lambda trajectory, candidate_chunks, seed: _local_verification(
            candidate_chunks,
            passing=(trajectory[0] >= 4,),
        ),
        sequence_final_verifier=lambda result, candidate_chunks: _exact_final(
            candidate_chunks
        ),
        generation_evidence_factory=generation_evidence,
        max_generated_chunks=10,
        max_generated_text_units=100,
        max_sequence_paths=1,
    )

    assert [context.candidate_index for context in contexts] == list(range(9))
    assert [context.seed for context in contexts] == list(range(100, 109))
    assert [context.chunk_indices for context in contexts] == [
        (0, 1),
        (0,),
        (1,),
        (0,),
        (1,),
        (0,),
        (1,),
        (0,),
        (1,),
    ]
    assert [context.chunk_candidate_ordinals for context in contexts] == [
        (0, 0),
        (1,),
        (1,),
        (2,),
        (2,),
        (3,),
        (3,),
        (4,),
        (4,),
    ]
    assert result.attempted_seeds == tuple(range(100, 109))
    assert result.chunk_candidate_indices == (7, 8)
    assert result.chunk_seeds == (107, 108)

    line = format_cascade_evidence_log(
        result.diagnostics,
        outcome="returned",
        generated_chunk_limit=10,
        generated_text_unit_limit=100,
        selection=result,
    )
    payload = json.loads(line.removeprefix(CASCADE_EVIDENCE_LOG_PREFIX))
    assert payload["generation_evidence_complete"] is True
    assert payload["attempts"][7]["candidate_index"] == 7
    assert payload["attempts"][7]["seed"] == 107
    assert payload["attempts"][7]["chunk_candidate_ordinals"] == [4]
    assert payload["attempts"][7]["scheduled_cfg"] == 3.0
    assert payload["attempts"][7]["policy"] == "completion_headroom"
    assert payload["attempts"][7]["chunk_policies"] == [
        "completion_headroom"
    ]
    assert payload["selection"]["generation"]["chunk_candidate_ordinals"] == [
        4,
        4,
    ]
    assert payload["selection"]["generation"]["chunk_policies"] == [
        "completion_headroom",
        "completion_headroom",
    ]


def test_candidate_local_generation_text_is_separate_from_canonical_verification():
    chunks = ("甲", "乙")
    transformed = []
    generated = []
    evidenced = []
    verified = []

    def transform(candidate_chunks, generation_context):
        assert isinstance(generation_context, CandidateGenerationContext)
        transformed.append((candidate_chunks, generation_context))
        if generation_context.candidate_index == 0:
            return ("甲甲", "乙乙")
        if generation_context.chunk_indices == (0,):
            return ("甲甲甲甲",)
        return ("乙乙乙",)

    def generator(candidate_chunks, seed, *, generation_context):
        generated.append((candidate_chunks, seed, generation_context))
        return candidate_chunks

    def generation_evidence(
        candidate_index,
        seed,
        chunk_indices,
        candidate_chunks,
        *,
        generation_context,
    ):
        evidenced.append((candidate_chunks, generation_context))
        ordinal = generation_context.chunk_candidate_ordinals[0]
        assert all(
            value == ordinal
            for value in generation_context.chunk_candidate_ordinals
        )
        scheduled = generation_cfg_for_candidate_offset(ordinal)
        return CandidateGenerationEvidence(
            chunk_indices=chunk_indices,
            chunk_text_units=tuple(
                count_speech_units(chunk) for chunk in candidate_chunks
            ),
            scheduled_cfg=scheduled,
            effective_cfgs=tuple(scheduled for _ in candidate_chunks),
            floor_reasons=tuple(() for _ in candidate_chunks),
            chunk_candidate_ordinals=(
                generation_context.chunk_candidate_ordinals
            ),
        )

    def whole_verifier(trajectory, candidate_chunks, seed):
        verified.append(("whole", candidate_chunks))
        return _joined_rejected_local(candidate_chunks)

    def refill_verifier(trajectory, candidate_chunks, seed):
        verified.append(("refill", candidate_chunks))
        return _local_verification(candidate_chunks)

    def final_verifier(result, candidate_chunks):
        verified.append(("final", candidate_chunks))
        return _exact_final(candidate_chunks)

    result = run_coverage_adaptive_cascade(
        chunks,
        10,
        generator,
        whole_verifier,
        refill_verifier,
        sequence_final_verifier=final_verifier,
        generation_evidence_factory=generation_evidence,
        candidate_generation_text_transform=transform,
        max_generated_chunks=3,
        max_generated_text_units=7,
        max_sequence_paths=1,
    )

    assert [(values, seed) for values, seed, _ in generated] == [
        (("甲甲", "乙乙"), 10),
        (("乙乙乙",), 11),
    ]
    assert [values for values, _ in evidenced] == [
        ("甲甲", "乙乙"),
        ("乙乙乙",),
    ]
    assert verified == [
        ("whole", chunks),
        ("refill", ("乙",)),
        ("final", chunks),
    ]
    assert [context.chunk_indices for _, context in transformed] == [
        (0, 1),
        (0,),
        (1,),
    ]
    assert result.trajectory == ("甲甲", "乙乙乙")
    assert result.chunk_candidate_indices == (0, 1)
    assert result.generated_chunk_count == 3
    assert result.generated_text_units == 7
    assert result.diagnostics.attempts[0].chunk_text_units == (2, 2)
    assert result.diagnostics.attempts[1].chunk_text_units == (3,)


def test_generation_evidence_must_report_transformed_text_units():
    def evidence(candidate_index, seed, chunk_indices, candidate_chunks):
        return CandidateGenerationEvidence(
            chunk_indices=chunk_indices,
            chunk_text_units=(1,),
            scheduled_cfg=3.0,
            effective_cfgs=(3.0,),
            floor_reasons=((),),
        )

    with pytest.raises(
        RuntimeError,
        match="candidate generation evidence is invalid",
    ):
        run_coverage_adaptive_cascade(
            ("甲",),
            10,
            lambda candidate_chunks, seed: candidate_chunks,
            lambda trajectory, candidate_chunks, seed: _local_verification(
                candidate_chunks
            ),
            lambda trajectory, candidate_chunks, seed: _local_verification(
                candidate_chunks
            ),
            sequence_final_verifier=lambda result, candidate_chunks: (
                _exact_final(candidate_chunks)
            ),
            generation_evidence_factory=evidence,
            candidate_generation_text_transform=(
                lambda candidate_chunks, context: ("甲甲",)
            ),
        )


@pytest.mark.parametrize(
    "transform",
    [
        lambda chunks, context: "甲",
        lambda chunks, context: ("甲", "乙"),
        lambda chunks, context: ("",),
        lambda chunks, context: (object(),),
    ],
)
def test_candidate_generation_text_transform_must_return_aligned_speech_chunks(
    transform,
):
    generated = []
    with pytest.raises(RuntimeError, match="transform returned invalid chunks"):
        run_coverage_adaptive_cascade(
            ("甲",),
            10,
            lambda chunks, seed: generated.append((chunks, seed)),
            lambda trajectory, chunks, seed: _local_verification(chunks),
            lambda trajectory, chunks, seed: _local_verification(chunks),
            sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
            candidate_generation_text_transform=transform,
        )
    assert generated == []


def test_context_generation_and_evidence_must_opt_in_together_and_match():
    def context_generator(chunks, seed, *, generation_context):
        return tuple(chunks)

    with pytest.raises(ValueError, match="must opt in together"):
        run_coverage_adaptive_cascade(
            ("完整內容",),
            10,
            context_generator,
            lambda trajectory, chunks, seed: _local_verification(chunks),
            lambda trajectory, chunks, seed: _local_verification(chunks),
            sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
        )

    def mismatched_evidence(
        candidate_index,
        seed,
        chunk_indices,
        candidate_chunks,
        *,
        generation_context,
    ):
        return CandidateGenerationEvidence(
            chunk_indices=chunk_indices,
            chunk_text_units=(4,),
            scheduled_cfg=2.0,
            effective_cfgs=(2.0,),
            floor_reasons=((),),
            chunk_candidate_ordinals=(1,),
        )

    with pytest.raises(RuntimeError, match="generation evidence is invalid"):
        run_coverage_adaptive_cascade(
            ("完整內容",),
            10,
            context_generator,
            lambda trajectory, chunks, seed: _local_verification(chunks),
            lambda trajectory, chunks, seed: _local_verification(chunks),
            sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
            generation_evidence_factory=mismatched_evidence,
        )


def test_context_generation_records_and_enforces_network_cfg_provenance():
    def generator(chunks, seed, *, generation_context):
        return tuple(chunks)

    def network_evidence(
        candidate_index,
        seed,
        chunk_indices,
        candidate_chunks,
        *,
        generation_context,
    ):
        ordinal = generation_context.chunk_candidate_ordinals[0]
        scheduled = generation_cfg_for_candidate_offset(ordinal)
        return CandidateGenerationEvidence(
            chunk_indices=chunk_indices,
            chunk_text_units=(4,),
            scheduled_cfg=scheduled,
            effective_cfgs=(3.0,),
            floor_reasons=(("network",),) if scheduled < 3.0 else ((),),
            chunk_candidate_ordinals=(ordinal,),
            network_conditioned=(True,),
        )

    result = run_coverage_adaptive_cascade(
        ("完整內容",),
        10,
        generator,
        lambda trajectory, chunks, seed: _joined_rejected_local(chunks),
        lambda trajectory, chunks, seed: _local_verification(chunks),
        sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
        generation_evidence_factory=network_evidence,
        max_generated_chunks=2,
        max_generated_text_units=8,
        max_sequence_paths=1,
    )

    assert result.diagnostics.attempts[1].chunk_candidate_ordinals == (1,)
    assert result.diagnostics.attempts[1].network_conditioned == (True,)
    assert result.diagnostics.attempts[1].floor_reasons == (("network",),)
    payload = json.loads(
        format_cascade_evidence_log(
            result.diagnostics,
            outcome="returned",
            generated_chunk_limit=2,
            generated_text_unit_limit=8,
            selection=result,
        ).removeprefix(CASCADE_EVIDENCE_LOG_PREFIX)
    )
    assert payload["attempts"][1]["network_conditioned"] == [True]
    assert payload["selection"]["generation"]["network_conditioned"] == [True]

    def forged_network_evidence(*args, generation_context, **kwargs):
        ordinal = generation_context.chunk_candidate_ordinals[0]
        scheduled = generation_cfg_for_candidate_offset(ordinal)
        return CandidateGenerationEvidence(
            chunk_indices=generation_context.chunk_indices,
            chunk_text_units=(4,),
            scheduled_cfg=scheduled,
            effective_cfgs=(3.0,),
            floor_reasons=((),),
            chunk_candidate_ordinals=(ordinal,),
            network_conditioned=(True,),
        )

    with pytest.raises(RuntimeError, match="generation evidence is invalid"):
        run_coverage_adaptive_cascade(
            ("完整內容",),
            10,
            generator,
            lambda trajectory, chunks, seed: _joined_rejected_local(chunks),
            lambda trajectory, chunks, seed: _local_verification(chunks),
            sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
            generation_evidence_factory=forged_network_evidence,
            max_generated_chunks=2,
            max_generated_text_units=8,
            max_sequence_paths=1,
        )


def test_zero_coverage_rows_are_refilled_before_low_coverage_rows_round_robin():
    chunks = ("第一段", "第二段", "第三段")
    generated = []

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

    def refill_verifier(trajectory, candidate_chunks, seed):
        # The first attempt for the second chunk fails.  The zero-coverage
        # third chunk must get the next seed before the second chunk retries.
        return _local_verification(candidate_chunks, passing=(seed != 101,))

    result = run_coverage_adaptive_cascade(
        chunks,
        100,
        generator,
        lambda trajectory, candidate_chunks, seed: _joined_rejected_local(
            candidate_chunks,
            passing=(True, False, False),
        ),
        refill_verifier,
        sequence_final_verifier=lambda result, candidate_chunks: _exact_final(
            candidate_chunks
        ),
        max_generated_chunks=6,
        max_generated_text_units=100,
        max_sequence_paths=1,
    )

    assert generated[1:] == [
        (("第二段",), 101),
        (("第三段",), 102),
        (("第二段",), 103),
    ]
    assert result.chunk_candidate_indices == (0, 3, 2)
    assert result.chunk_candidate_counts == (1, 1, 1)


def test_remaining_budget_prioritizes_request_endpoints_after_full_coverage():
    chunks = ("第一段", "第二段", "第三段")
    generated = []

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

    result = run_coverage_adaptive_cascade(
        chunks,
        100,
        generator,
        lambda trajectory, candidate_chunks, seed: _joined_rejected_local(
            candidate_chunks,
            passing=(False, True, False),
        ),
        lambda trajectory, candidate_chunks, seed: _local_verification(
            candidate_chunks,
            passing=(seed != 101,),
        ),
        sequence_final_verifier=lambda result, candidate_chunks: _exact_final(
            candidate_chunks
        ),
        max_generated_chunks=7,
        max_generated_text_units=100,
        max_sequence_paths=2,
    )

    assert generated[1:] == [
        (("第一段",), 101),
        (("第三段",), 102),
        (("第一段",), 103),
        (("第三段",), 104),
    ]
    assert result.chunk_candidate_counts == (1, 1, 2)


def test_exact_asr_speaker_near_miss_gets_priority_then_latest_result_reorders():
    chunks = ("第一段", "第二段")
    generated = []

    def verification(candidate_chunks, states):
        observations = []
        artifacts = []
        for chunk, state in zip(candidate_chunks, states, strict=True):
            semantic_passed = state != "semantic_failure"
            speaker_similarity = 0.05 if state == "speaker_near_miss" else 0.80
            observations.append(
                CandidateObservation(
                    target_text=chunk,
                    transcript_text=chunk if semantic_passed else "完全錯誤",
                    audio_duration_seconds=2.0,
                    speaker_similarity=speaker_similarity,
                    begin_speaker_similarity=speaker_similarity,
                    end_speaker_similarity=speaker_similarity,
                    pace_cps=2.0,
                )
            )
            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,
            max_pace_cps=4.3,
            min_speaker_similarity=0.10,
        )

    initial_local = verification(
        chunks,
        ("semantic_failure", "speaker_near_miss"),
    )
    assert initial_local.candidate_results[0].comparison.passed is False
    assert initial_local.candidate_results[1].comparison.passed is True
    assert "speaker_similarity" in initial_local.candidate_results[
        1
    ].rejection_reasons

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

    def refill_verifier(trajectory, candidate_chunks, seed):
        state = (
            "semantic_failure"
            if candidate_chunks == ("第二段",) and seed == 101
            else "passed"
        )
        return verification(candidate_chunks, (state,))

    result = run_coverage_adaptive_cascade(
        chunks,
        100,
        generator,
        lambda trajectory, candidate_chunks, seed: TrajectoryGateResult(
            passed=False,
            candidate_results=initial_local.candidate_results,
            score=math.inf,
            rejection_reasons=("joined_output:semantic_gate",),
            chunk_artifacts=initial_local.chunk_artifacts,
        ),
        refill_verifier,
        sequence_final_verifier=lambda result, candidate_chunks: _exact_final(
            candidate_chunks
        ),
        max_generated_chunks=5,
        max_generated_text_units=100,
        max_sequence_paths=1,
    )

    assert generated[1:] == [
        (("第二段",), 101),
        (("第一段",), 102),
        (("第二段",), 103),
    ]
    assert result.chunk_candidate_indices == (2, 3)
    assert result.chunk_candidate_counts == (1, 1)


def test_ragged_dp_accepts_unequal_row_widths_and_stable_cost_order():
    ranked = select_culprit_diverse_candidate_sequences(
        [[0.0], [0.3, 0.1, 0.2], [0.4, 0.0]],
        [
            [[0.0, 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 ranked] == [
        (0, 1, 1),
        (0, 2, 1),
        (0, 0, 1),
    ]
    assert [selection.total_score for selection in ranked] == pytest.approx(
        [0.1, 0.2, 0.3]
    )


def test_culprit_diversity_uses_final_checks_on_distinct_culprit_assignments():
    ranked = select_culprit_diverse_candidate_sequences(
        [[0.0, 0.001, 0.002], [0.0, 10.0, 20.0], [0.0]],
        [
            [[0.0, 0.0, 0.0]] * 3,
            [[0.0], [0.0], [0.0]],
        ],
        culprit_indices=(1,),
        max_paths=3,
    )

    assert [selection.candidate_indices for selection in ranked] == [
        (0, 0, 0),
        (0, 1, 0),
        (0, 2, 0),
    ]
    assert len({selection.candidate_indices[1] for selection in ranked}) == 3


@pytest.mark.parametrize(
    ("generator", "whole_verifier", "message"),
    [
        (
            lambda chunks, seed: np.zeros(8, dtype=np.float32),
            lambda trajectory, chunks, seed: _local_verification(chunks),
            "malformed trajectory",
        ),
        (
            lambda chunks, seed: tuple(chunks),
            lambda trajectory, chunks, seed: None,
            "invalid result",
        ),
        (
            lambda chunks, seed: tuple(chunks),
            lambda trajectory, chunks, seed: (_ for _ in ()).throw(
                RuntimeError("ASR unavailable")
            ),
            "whole trajectory verification failed",
        ),
    ],
)
def test_malformed_or_asr_initial_evidence_aborts_fail_closed(
    generator,
    whole_verifier,
    message,
):
    with pytest.raises(RuntimeError, match=message):
        run_coverage_adaptive_cascade(
            ("完整內容",),
            10,
            generator,
            whole_verifier,
            lambda *args: _local_verification(("完整內容",)),
            sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
        )


def test_passing_speaker_gate_without_ecapa_artifact_aborts_fail_closed():
    local = _local_verification(("完整內容",))
    malformed = TrajectoryGateResult(
        passed=True,
        candidate_results=local.candidate_results,
        score=local.score,
        rejection_reasons=(),
        chunk_artifacts=(ChunkCandidateArtifact(rms_db=-20.0),),
    )

    with pytest.raises(RuntimeError, match="acoustic evidence"):
        run_coverage_adaptive_cascade(
            ("完整內容",),
            10,
            lambda chunks, seed: tuple(chunks),
            lambda *args: malformed,
            lambda *args: local,
            sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
        )


def test_final_verifier_exception_invalid_evidence_and_all_reject_fail_closed():
    common = dict(
        chunks=("甲乙", "丙丁"),
        root_seed=10,
        candidate_generator=lambda chunks, seed: tuple(
            f"{seed}:{chunk}" for chunk in chunks
        ),
        whole_candidate_verifier=lambda trajectory, chunks, seed: (
            _joined_rejected_local(chunks)
        ),
        refill_candidate_verifier=lambda trajectory, chunks, seed: (
            _local_verification(chunks)
        ),
        max_generated_chunks=6,
        max_generated_text_units=100,
        max_sequence_paths=3,
    )

    with pytest.raises(RuntimeError, match="refusing unverified audio"):
        run_coverage_adaptive_cascade(
            **common,
            sequence_final_verifier=lambda *args: (_ for _ in ()).throw(
                RuntimeError("Breeze25 verifier OOM")
            ),
        )
    with pytest.raises(RuntimeError, match="invalid result"):
        run_coverage_adaptive_cascade(
            **common,
            sequence_final_verifier=lambda *args: None,
        )

    checked = []
    with pytest.raises(
        NoQualifiedCandidateError,
        match="3 assembled paths",
    ) as caught:
        run_coverage_adaptive_cascade(
            **common,
            sequence_final_verifier=lambda result, chunks: (
                checked.append(result.chunk_candidate_indices)
                or _exact_final(chunks, passed=False)
            ),
        )
    assert len(checked) == 3
    assert len(set(checked)) == 3
    search = caught.value.diagnostics.sequence_search
    assert search is not None
    assert search.ranked_path_count == 3
    assert len(search.checked_paths) == 3
    assert all(not path.final_output.passed for path in search.checked_paths)


def test_boundary_only_relaxation_is_capped_and_still_requires_exact_final_pass():
    chunks = ("第一段完整", "第二段完整")
    chunk = chunks[0]
    rejected = verify_candidate(
        CandidateObservation(
            target_text=chunk,
            transcript_text=chunk,
            audio_duration_seconds=2.0,
            speaker_similarity=0.8,
            begin_speaker_similarity=0.8,
            end_speaker_similarity=0.651,
        ),
        max_boundary_speaker_drop=0.10,
    )
    safe_second = _local_verification((chunks[1],))
    local = TrajectoryGateResult(
        passed=False,
        candidate_results=(rejected, safe_second.candidate_results[0]),
        score=math.inf,
        rejection_reasons=("chunk_0:boundary_speaker_drop",),
        chunk_artifacts=(
            ChunkCandidateArtifact(
                speaker_embedding=np.array([1.0, 0.0], dtype=np.float32),
                rms_db=-20.0,
            ),
            safe_second.chunk_artifacts[0],
        ),
    )
    callbacks = []
    result = run_coverage_adaptive_cascade(
        chunks,
        10,
        lambda chunks, seed: tuple(chunks),
        lambda *args: local,
        lambda *args: local,
        sequence_final_verifier=lambda result, chunks: (
            callbacks.append(result.chunk_candidate_indices)
            or _exact_final(chunks)
        ),
        max_generated_chunks=2,
        max_generated_text_units=100,
        max_sequence_paths=1,
        sequence_fallback_max_local_boundary_speaker_drop=(
            SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP
        ),
    )

    assert callbacks == [(0, 0)]
    assert result.selection_mode == "coverage_sequence_dp"
    assert result.verification.candidate_results[0].boundary_speaker_drop == pytest.approx(
        0.149
    )

    with pytest.raises(NoQualifiedCandidateError):
        run_coverage_adaptive_cascade(
            chunks,
            10,
            lambda chunks, seed: tuple(chunks),
            lambda *args: local,
            lambda *args: local,
            sequence_final_verifier=lambda result, chunks: _exact_final(
                chunks,
                passed=False,
            ),
            max_generated_chunks=2,
            max_generated_text_units=100,
            max_sequence_paths=1,
            sequence_fallback_max_local_boundary_speaker_drop=0.15,
        )
    with pytest.raises(ValueError, match="no greater than 0.15"):
        run_coverage_adaptive_cascade(
            chunks,
            10,
            lambda chunks, seed: tuple(chunks),
            lambda *args: local,
            lambda *args: local,
            sequence_final_verifier=lambda result, chunks: _exact_final(chunks),
            sequence_fallback_max_local_boundary_speaker_drop=0.151,
        )

    single_local = TrajectoryGateResult(
        passed=False,
        candidate_results=(rejected,),
        score=math.inf,
        rejection_reasons=("chunk_0:boundary_speaker_drop",),
        chunk_artifacts=(local.chunk_artifacts[0],),
    )
    single = run_coverage_adaptive_cascade(
        (chunk,),
        10,
        lambda candidate_chunks, seed: tuple(candidate_chunks),
        lambda *args: single_local,
        lambda *args: single_local,
        sequence_final_verifier=lambda result, candidate_chunks: _exact_final(
            candidate_chunks
        ),
        max_generated_chunks=1,
        max_generated_text_units=100,
        max_sequence_paths=1,
        sequence_fallback_max_local_boundary_speaker_drop=0.15,
    )
    assert single.selection_mode == "coverage_sequence_dp"
    assert single.verification.candidate_results[0].boundary_speaker_drop == pytest.approx(
        0.149
    )

    with pytest.raises(NoQualifiedCandidateError, match="1 assembled paths"):
        run_coverage_adaptive_cascade(
            (chunk,),
            10,
            lambda candidate_chunks, seed: tuple(candidate_chunks),
            lambda *args: single_local,
            lambda *args: single_local,
            sequence_final_verifier=lambda result, candidate_chunks: _exact_final(
                candidate_chunks,
                passed=False,
            ),
            max_generated_chunks=1,
            max_generated_text_units=100,
            max_sequence_paths=1,
            sequence_fallback_max_local_boundary_speaker_drop=0.15,
        )