tests/test_coverage_adaptive.py

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,
):
    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,
        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 = []

    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),
    )

    assert generated == [(chunks, 80)]
    assert refill_calls == []
    assert final_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_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_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 = []

    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),
        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 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_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,
        )


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"] == 5
    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"],
    }


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("large-v3 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,
        )