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