import ast from dataclasses import replace from pathlib import Path from types import SimpleNamespace import numpy as np import pytest import torch from latency_timing import latency_stage, timed_latency_stage from production import count_speech_units from quality_runtime import ( CandidateObservation, ChunkCandidateArtifact, verify_candidate, verify_trajectory, ) ROOT = Path(__file__).resolve().parents[1] def _isolated_app_function(name, namespace): namespace.setdefault("latency_stage", latency_stage) namespace.setdefault("timed_latency_stage", timed_latency_stage) app_path = ROOT / "app.py" tree = ast.parse(app_path.read_text(encoding="utf-8")) function = next( node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == name ) module = ast.Module( body=[ ast.ImportFrom( module="__future__", names=[ast.alias(name="annotations")], level=0, ), function, ], type_ignores=[], ) ast.fix_missing_locations(module) exec(compile(module, app_path, "exec"), namespace) return namespace[name] def test_app_speed_wrapper_uses_wsola_and_rejects_below_release_floor(): calls = [] def stretch(audio, *, rate, sample_rate): calls.append((np.asarray(audio).copy(), rate, sample_rate)) return np.full(12, rate, dtype=np.float32) apply_speed = _isolated_app_function( "_apply_speed", { "np": np, "wsola_time_stretch": stretch, "MIN_PACE_SPEED": 0.90, "SR": 48_000, }, ) waveform = np.linspace(-0.5, 0.5, 16, dtype=np.float32) output = apply_speed(waveform, 0.90, network_conditioned=True) assert output.dtype == np.float32 assert np.array_equal(output, np.full(12, 0.90, dtype=np.float32)) assert len(calls) == 1 assert np.array_equal(calls[0][0], waveform) assert calls[0][1:] == (0.90, 48_000) with pytest.raises(ValueError, match="below the production WSOLA floor"): apply_speed(waveform, 0.899) with pytest.raises(ValueError, match="finite and positive"): apply_speed(waveform, float("nan")) def test_echo_smearing_gate_projects_acoustic_failure_into_trajectory(): base = verify_trajectory( ( CandidateObservation( target_text="測試內容", transcript_text="測試內容", audio_duration_seconds=2.0, ), ), speaker_gate_enabled=False, ) assert base.passed diagnostics = SimpleNamespace( passed=False, rejection_reasons=("delayed_copy",), delayed_copy_score=0.6, smearing_score=0.1, dominant_delay_ms=42.0, cepstral_peak_ratio=11.0, analysis_window_count=5, ) gate = _isolated_app_function( "_apply_echo_smearing_gate", { "np": np, "replace": replace, "SR": 48_000, "DEFAULT_ECHO_SMEARING_GATE_CONFIG": SimpleNamespace( min_duration_seconds=1.0 ), "evaluate_echo_smearing": lambda waveform, sample_rate: diagnostics, }, ) rejected = gate( base, (np.ones(2 * 48_000, dtype=np.float32),), ) assert not rejected.passed assert np.isinf(rejected.score) assert rejected.rejection_reasons == ( "chunk_0:audio_artifact_delayed_copy", ) assert rejected.candidate_results[0].rejection_reasons == ( "audio_artifact_delayed_copy", ) def test_echo_smearing_gate_does_not_apply_estimator_below_one_second(): base = verify_trajectory( ( CandidateObservation( target_text="短句", transcript_text="短句", audio_duration_seconds=0.8, ), ), speaker_gate_enabled=False, ) calls = [] gate = _isolated_app_function( "_apply_echo_smearing_gate", { "np": np, "replace": replace, "SR": 48_000, "DEFAULT_ECHO_SMEARING_GATE_CONFIG": SimpleNamespace( min_duration_seconds=1.0 ), "evaluate_echo_smearing": lambda *args: calls.append(args), }, ) unchanged = gate( base, (np.ones(int(0.8 * 48_000), dtype=np.float32),), ) assert unchanged.passed assert unchanged.rejection_reasons == () assert calls == [] def test_echo_smearing_gate_skips_candidates_already_rejected_by_hard_gate(): rejected = verify_trajectory( ( CandidateObservation( target_text="正確內容", transcript_text="錯誤內容", audio_duration_seconds=2.0, ), ), speaker_gate_enabled=False, ) calls = [] gate = _isolated_app_function( "_apply_echo_smearing_gate", { "np": np, "replace": replace, "SR": 48_000, "DEFAULT_ECHO_SMEARING_GATE_CONFIG": SimpleNamespace( min_duration_seconds=1.0 ), "evaluate_echo_smearing": lambda *args: calls.append(args), }, ) unchanged = gate( rejected, (np.ones(2 * 48_000, dtype=np.float32),), ) assert not unchanged.passed assert unchanged.rejection_reasons == rejected.rejection_reasons assert calls == [] def test_trajectory_verifier_skips_acoustic_models_after_semantic_preflight_failure(): acoustic_calls = [] echo_waveforms = [] waveform = np.ones(2 * 48_000, dtype=np.float32) def forbidden(*args, **kwargs): acoustic_calls.append((args, kwargs)) raise AssertionError("acoustic model should not run") def echo_gate(verification, waveforms): echo_waveforms.append(waveforms) return verification verify_audio = _isolated_app_function( "_verify_trajectory_audio", { "np": np, "CandidateObservation": CandidateObservation, "ChunkCandidateArtifact": ChunkCandidateArtifact, "transcribe_breeze25": lambda *_args, **_kwargs: "", "prepare_candidate_audio": lambda *_args, **_kwargs: SimpleNamespace( waveform=waveform, transcript_text="錯誤內容", ), "active_voiced_duration_seconds": lambda *_args, **_kwargs: 2.0, "count_speech_units": count_speech_units, "verify_candidate": verify_candidate, "verify_trajectory": verify_trajectory, "_get_ecapa_encoder": forbidden, "speaker_evidence_from_audio": forbidden, "release_speaker_evidence_from_audio": forbidden, "squim_objective_evidence_from_audio": forbidden, "active_audio_rms_db": forbidden, "_apply_echo_smearing_gate": echo_gate, "_attach_transition_f0": lambda verification, *_args, **_kwargs: verification, "SR": 48_000, "QUALITY_MAX_CER": 0.20, "QUALITY_PREFIX_SUFFIX_UNITS": 6, "QUALITY_MAX_PACE_CPS": 4.85, "QUALITY_PREFLIGHT_PACE_MARGIN_CPS": 0.05, "SHORT_AUDIO_SPEAKER_GATE_SECONDS": 1.50, "RELEASE_SPEAKER_TRIGGER_SECONDS": 1.48, "QUALITY_RELEASE_MIN_SPEAKER_SIMILARITY": 0.105, "QUALITY_MIN_SPEAKER_SIMILARITY": 0.10, "QUALITY_RELEASE_MAX_BOUNDARY_SPEAKER_DROP": 0.095, "QUALITY_MAX_BOUNDARY_SPEAKER_DROP": 0.10, "QUALITY_MIN_SQUIM_STOI": 0.60, "QUALITY_MIN_SQUIM_PESQ": 1.12, }, ) verification = verify_audio( (waveform,), ("正確內容",), np.ones(4, dtype=np.float32), 1.0, ) assert not verification.passed assert verification.candidate_results[0].rejection_reasons == ( "semantic_gate", ) assert acoustic_calls == [] assert echo_waveforms == [(None,)] def test_trajectory_verifier_skips_acoustics_above_conservative_pace_preflight(): acoustic_calls = [] asr_calls = [] waveform = np.ones(2 * 48_000, dtype=np.float32) target = "甲乙丙丁戊己庚辛壬癸" def forbidden(*args, **kwargs): acoustic_calls.append((args, kwargs)) raise AssertionError("acoustic model should not run") def forbidden_asr(*args, **kwargs): asr_calls.append((args, kwargs)) raise AssertionError("Breeze25 should not run after hard pace rejection") verify_audio = _isolated_app_function( "_verify_trajectory_audio", { "np": np, "CandidateObservation": CandidateObservation, "ChunkCandidateArtifact": ChunkCandidateArtifact, "transcribe_breeze25": lambda *_args, **_kwargs: "", "prepare_candidate_audio": forbidden_asr, "active_voiced_duration_seconds": lambda *_args, **_kwargs: 2.0, "count_speech_units": count_speech_units, "verify_candidate": verify_candidate, "verify_trajectory": verify_trajectory, "_get_ecapa_encoder": forbidden, "speaker_evidence_from_audio": forbidden, "release_speaker_evidence_from_audio": forbidden, "squim_objective_evidence_from_audio": forbidden, "active_audio_rms_db": forbidden, "_apply_echo_smearing_gate": lambda verification, _waveforms: verification, "_attach_transition_f0": lambda verification, *_args, **_kwargs: verification, "SR": 48_000, "QUALITY_MAX_CER": 0.20, "QUALITY_PREFIX_SUFFIX_UNITS": 6, "QUALITY_MAX_PACE_CPS": 4.85, "QUALITY_PREFLIGHT_PACE_MARGIN_CPS": 0.05, "SHORT_AUDIO_SPEAKER_GATE_SECONDS": 1.50, "RELEASE_SPEAKER_TRIGGER_SECONDS": 1.48, "QUALITY_RELEASE_MIN_SPEAKER_SIMILARITY": 0.105, "QUALITY_MIN_SPEAKER_SIMILARITY": 0.10, "QUALITY_RELEASE_MAX_BOUNDARY_SPEAKER_DROP": 0.095, "QUALITY_MAX_BOUNDARY_SPEAKER_DROP": 0.10, "QUALITY_MIN_SQUIM_STOI": 0.60, "QUALITY_MIN_SQUIM_PESQ": 1.12, }, ) verification = verify_audio( (waveform,), (target,), np.ones(4, dtype=np.float32), 1.0, ) assert not verification.passed assert verification.candidate_results[0].rejection_reasons == ( "semantic_gate", "pace_too_fast", ) assert asr_calls == [] assert acoustic_calls == [] def test_borderline_preflight_still_runs_acoustics_then_enforces_final_pace_gate(): calls = [] waveform = np.ones(2 * 48_000, dtype=np.float32) target = "甲乙丙丁戊己庚辛壬癸" active_duration = count_speech_units(target) / 4.875 def speaker_measurement(*_args, **_kwargs): calls.append("speaker") return SimpleNamespace( active_duration_seconds=active_duration, similarity=0.80, begin_similarity=0.80, end_similarity=0.79, speaker_embedding=np.ones(4, dtype=np.float32), active_rms_db=-20.0, ) def squim_measurement(*_args, **_kwargs): calls.append("squim") return SimpleNamespace(stoi=0.90, pesq=2.0, si_sdr=10.0) verify_audio = _isolated_app_function( "_verify_trajectory_audio", { "np": np, "CandidateObservation": CandidateObservation, "ChunkCandidateArtifact": ChunkCandidateArtifact, "transcribe_breeze25": lambda *_args, **_kwargs: "", "prepare_candidate_audio": lambda *_args, **_kwargs: SimpleNamespace( waveform=waveform, transcript_text=target, ), "active_voiced_duration_seconds": lambda *_args, **_kwargs: active_duration, "count_speech_units": count_speech_units, "verify_candidate": verify_candidate, "verify_trajectory": verify_trajectory, "_get_ecapa_encoder": lambda: object(), "speaker_evidence_from_audio": speaker_measurement, "release_speaker_evidence_from_audio": speaker_measurement, "release_speaker_measurement_required": lambda duration: duration >= 1.48, "squim_objective_evidence_from_audio": squim_measurement, "active_audio_rms_db": lambda _waveform: -20.0, "_apply_echo_smearing_gate": lambda verification, _waveforms: verification, "_attach_transition_f0": lambda verification, *_args, **_kwargs: verification, "SR": 48_000, "QUALITY_MAX_CER": 0.20, "QUALITY_PREFIX_SUFFIX_UNITS": 6, "QUALITY_MAX_PACE_CPS": 4.85, "QUALITY_PREFLIGHT_PACE_MARGIN_CPS": 0.05, "SHORT_AUDIO_SPEAKER_GATE_SECONDS": 1.50, "RELEASE_SPEAKER_TRIGGER_SECONDS": 1.48, "QUALITY_RELEASE_MIN_SPEAKER_SIMILARITY": 0.105, "QUALITY_MIN_SPEAKER_SIMILARITY": 0.10, "QUALITY_RELEASE_MAX_BOUNDARY_SPEAKER_DROP": 0.095, "QUALITY_MAX_BOUNDARY_SPEAKER_DROP": 0.10, "QUALITY_MIN_SQUIM_STOI": 0.60, "QUALITY_MIN_SQUIM_PESQ": 1.12, }, ) verification = verify_audio( (waveform,), (target,), np.ones(4, dtype=np.float32), 1.0, ) assert not verification.passed assert verification.candidate_results[0].rejection_reasons == ( "pace_too_fast", ) assert calls == ["speaker", "squim"] def test_closed_loop_pace_rerenders_once_from_untouched_model_waveform(): apply_calls = [] active_calls = [] correction_calls = [] def apply_speed(audio, speed, *, network_conditioned=False): source = np.asarray(audio, dtype=np.float32).copy() apply_calls.append((source, speed, network_conditioned)) return source * np.float32(speed) def active_duration(audio, sample_rate): active_calls.append((np.asarray(audio).copy(), sample_rate)) return 2.0 def active_correction(active_seconds, text, **kwargs): correction_calls.append((active_seconds, text, kwargs)) return 0.95 namespace = { "np": np, "torch": torch, "model": SimpleNamespace(generate=lambda **kwargs: torch.ones(16)), "select_generation_cps": lambda *args, **kwargs: 5.0, "count_network_endpoint_duration_units": lambda text: 8, "count_speech_units": lambda text: 8, "endpoint_generation_plan": lambda *args, **kwargs: ("測試內容。", 10, 12), "effective_generation_cfg": lambda text, cfg, **kwargs: cfg, "set_generation_seed": lambda seed: None, "target_pace_speed": lambda *args, **kwargs: 0.95, "active_voiced_duration_seconds": active_duration, "active_pace_correction_speed": active_correction, "endpoint_tail_energy_ratio": lambda audio, sample_rate: 0.5, "_ChunkGenerationOutcome": SimpleNamespace, "_apply_speed": apply_speed, "_NATIVE_STOP_POLICY": False, "_GENERATE_PARAMETERS": set(), "_STOP_CONTROLLER": None, "SR": 48_000, "STEP_SECONDS": 0.16, "MIN_ENDPOINT_CUE_UNITS": 6, "SHORT_TEXT_CFG_UNITS": 6, "SHORT_TEXT_CFG_MIN": 3.0, "TARGET_CPS": 4.0, "ACTIVE_PACE_TARGET_CPS": 4.0, "CLOSED_LOOP_ACTIVE_PACE_TARGET_CPS": 3.95, "MIN_PACE_SPEED": 0.90, "STOP_THRESHOLD": 0.50, "STOP_CONSECUTIVE": 1, } generate = _isolated_app_function("_generate_chunk", namespace) policy = SimpleNamespace( name="test", cjk_cps=5.0, ascii_cps=4.6, hard_stop_margin_steps=1, ) output = generate( "測試內容。", torch.ones(4), cfg=3.0, steps=10, request_seed=123, policy=policy, network_conditioned=True, ) assert len(apply_calls) == 2 assert np.array_equal(apply_calls[0][0], np.ones(16, dtype=np.float32)) assert np.array_equal(apply_calls[1][0], np.ones(16, dtype=np.float32)) assert apply_calls[0][1] == pytest.approx(0.95) assert apply_calls[1][1] == pytest.approx(0.95 * 0.95) assert all(call[2] is True for call in apply_calls) assert np.array_equal( output.audio, np.ones(16, dtype=np.float32) * np.float32(0.9025), ) assert output.stop_reason == "native_stop" assert output.endpoint_energy_ratio == 0.5 assert output.generated_steps == 12 assert output.hard_stop_steps == 12 assert len(active_calls) == 2 assert np.array_equal(active_calls[0][0], np.ones(16, dtype=np.float32)) assert np.array_equal(active_calls[1][0], np.ones(16, dtype=np.float32) * 0.95) assert correction_calls[1][2] == { "target_cps": 3.95, "prior_speed": 0.95, "min_total_speed": 0.90, } def test_sparse_completion_policy_adds_only_bounded_one_two_unit_headroom(): def run(units, policy): generation_calls = [] namespace = { "np": np, "torch": torch, "model": SimpleNamespace( generate=lambda **kwargs: ( generation_calls.append(kwargs) or torch.ones(64) ) ), "select_generation_cps": lambda *args, **kwargs: 5.0, "count_network_endpoint_duration_units": lambda text: units, "count_speech_units": lambda text: units, "endpoint_generation_plan": ( lambda *args, **kwargs: ("短句。", 3, 4) ), "effective_generation_cfg": lambda text, cfg, **kwargs: cfg, "set_generation_seed": lambda seed: None, "target_pace_speed": lambda *args, **kwargs: 1.0, "active_voiced_duration_seconds": ( lambda *args, **kwargs: (_ for _ in ()).throw(ValueError()) ), "active_pace_correction_speed": lambda *args, **kwargs: 1.0, "endpoint_tail_energy_ratio": lambda audio, sample_rate: 0.25, "_apply_speed": lambda audio, speed, **kwargs: np.asarray(audio), "_ChunkGenerationOutcome": SimpleNamespace, "_NATIVE_STOP_POLICY": False, "_GENERATE_PARAMETERS": set(), "_STOP_CONTROLLER": None, "SR": 48_000, "STEP_SECONDS": 0.16, "MIN_ENDPOINT_CUE_UNITS": 6, "SHORT_TEXT_CFG_UNITS": 6, "SHORT_TEXT_CFG_MIN": 3.0, "TARGET_CPS": 4.0, "ACTIVE_PACE_TARGET_CPS": 4.0, "CLOSED_LOOP_ACTIVE_PACE_TARGET_CPS": 3.95, "MIN_PACE_SPEED": 0.90, "STOP_THRESHOLD": 0.50, "STOP_CONSECUTIVE": 1, } generate = _isolated_app_function("_generate_chunk", namespace) outcome = generate( "短句", torch.ones(4), cfg=3.0, steps=10, request_seed=123, policy=policy, network_conditioned=True, ) return generation_calls[0]["max_len"], outcome base = SimpleNamespace( name="base", cjk_cps=5.2, ascii_cps=4.6, hard_stop_margin_steps=1, short_headroom_max_units=0, short_hard_stop_floor_steps=0, ) completion = SimpleNamespace( name="completion_headroom", cjk_cps=4.2, ascii_cps=3.6, hard_stop_margin_steps=1, short_headroom_max_units=2, short_hard_stop_floor_steps=5, ) assert run(2, base)[0] == 4 headroom_cap, headroom = run(2, completion) assert headroom_cap == 5 assert headroom.hard_stop_steps == 5 assert run(3, completion)[0] == 4 def test_lazy_transition_f0_enricher_preserves_existing_evidence(): waveform = np.ones(100, dtype=np.float32) measured = [] namespace = { "np": np, "replace": replace, "ChunkCandidateArtifact": ChunkCandidateArtifact, "active_audio_median_f0_hz": ( lambda waveform, sample_rate: measured.append( (waveform.copy(), sample_rate) ) or 210.0 ), "SR": 48_000, } enrich = _isolated_app_function( "_enrich_transition_artifact_f0", namespace, ) artifact = ChunkCandidateArtifact(rms_db=-20.0) updated = enrich(waveform, artifact) assert len(measured) == 1 assert np.array_equal(measured[0][0], waveform) assert measured[0][1] == 48_000 assert updated.rms_db == artifact.rms_db assert updated.median_f0_hz == 210.0 existing = ChunkCandidateArtifact( rms_db=-21.0, median_f0_hz=180.0, ) assert enrich(waveform, existing) is existing assert len(measured) == 1 with pytest.raises(TypeError, match="artifact is invalid"): enrich(waveform, object()) def test_space_defers_transition_f0_to_actual_multi_path_ranking(): source = (ROOT / "app.py").read_text(encoding="utf-8") tree = ast.parse(source) functions = { node.name: ast.get_source_segment(source, node) for node in tree.body if isinstance(node, ast.FunctionDef) } assert "active_audio_median_f0_hz" not in functions[ "_verify_trajectory_audio" ] for name in ( "_qualify_candidate_trajectory_audio", "_verify_refill_candidate_trajectory_audio", "_verify_independent_whole_audio", "_verify_sequence_trajectory_audio", ): assert "collect_transition_f0=" not in functions[name] assert "transition_artifact_enricher=(" in functions["_synthesize"] assert "_enrich_transition_artifact_f0" in functions["_synthesize"] generate = functions["_generate_chunk"] assert "corrected_active_duration = active_voiced_duration_seconds(" in generate assert "final_speed = combined_speed * rerender_speed" in generate assert "_apply_speed(\n corrected" not in generate assert "_apply_speed(\n audio,\n final_speed" in generate def test_space_preloads_release_cpu_runtime_before_serving_requests(): source = (ROOT / "app.py").read_text(encoding="utf-8") tree = ast.parse(source) functions = { node.name: node for node in tree.body if isinstance(node, ast.FunctionDef) } preload_calls = [ node for node in tree.body if isinstance(node, ast.Expr) and isinstance(node.value, ast.Call) and isinstance(node.value.func, ast.Name) and node.value.func.id == "_preload_release_cpu_runtime" ] assert len(preload_calls) == 1 assert preload_calls[0].lineno > functions["_speaker_anchor_array"].end_lineno preload_source = ast.get_source_segment( source, functions["_preload_release_cpu_runtime"], ) assert "speaker_evidence_from_audio(" in preload_source assert "release_speaker_evidence_from_audio(" in preload_source assert "squim_objective_evidence_from_audio(" in preload_source assert "active_audio_median_f0_hz(" in preload_source assert "np.arange(sample_count" in preload_source