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