import ast import hashlib from pathlib import Path from types import SimpleNamespace import numpy as np from production import ( GenerationChunkSpec, count_speech_units, fade_variable_internal_edges, join_audio_chunks_variable, match_chunk_rms, normalize_spoken_forms, pcm16_verification_waveform, plan_generation_chunks, punctuation_pause_seconds, split_quality_text_for_tts, ) from latency_timing import timed_latency_stage from quality_runtime import LocalIndependentGateEvidence ROOT = Path(__file__).resolve().parents[1] FROZEN_SPEAKER_ANCHORS = { "6f7cab914a1e27c56b504ec663c0144dc25cc0a3": { "speaker_id": "female_voice", "speaker_index": 1, "ui_label": "內建語者 B", "dtype": "float32", "shape": (192,), "sha256": ( "67cd20e7ebdde857913f2e0bd722626fdb3224ec2f3bc9d5d0635c7b5272aa3d" ), "file_sha256": ( "e9556e14723c140985a104c1659d1ff8a5078d2fa28ce2fb756f04906641a8a7" ), } } def _string_constants(path: Path) -> dict[str, str]: tree = ast.parse(path.read_text(encoding="utf-8")) values: dict[str, str] = {} for node in tree.body: if not isinstance(node, ast.Assign) or len(node.targets) != 1: continue target = node.targets[0] if isinstance(target, ast.Name) and isinstance(node.value, ast.Constant): if isinstance(node.value.value, str): values[target.id] = node.value.value return values def _literal_constants(path: Path) -> dict[str, object]: tree = ast.parse(path.read_text(encoding="utf-8")) values: dict[str, object] = {} for node in tree.body: if not isinstance(node, ast.Assign) or len(node.targets) != 1: continue target = node.targets[0] if not isinstance(target, ast.Name): continue try: values[target.id] = ast.literal_eval(node.value) except (TypeError, ValueError): continue return values def _isolated_assemble_trajectory_audio(): """Execute only the production assembly function without loading the model.""" 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 == "_assemble_trajectory_audio" ) module = ast.Module( body=[ ast.ImportFrom( module="__future__", names=[ast.alias(name="annotations")], level=0, ), function, ], type_ignores=[], ) ast.fix_missing_locations(module) constants = _literal_constants(app_path) namespace = { "np": np, "GenerationChunkSpec": GenerationChunkSpec, "SR": 1_000, "CHUNK_RMS_MATCH_DB": constants["CHUNK_RMS_MATCH_DB"], "SEMANTIC_CHUNK_MIN_SILENCE_MS": constants[ "SEMANTIC_CHUNK_MIN_SILENCE_MS" ], "NETWORK_REQUEST_SEMANTIC_CHUNK_MIN_SILENCE_MS": constants[ "NETWORK_REQUEST_SEMANTIC_CHUNK_MIN_SILENCE_MS" ], "NETWORK_INTERNAL_SILENCE_MS": constants["NETWORK_INTERNAL_SILENCE_MS"], "NETWORK_INTERNAL_FADE_MS": constants["NETWORK_INTERNAL_FADE_MS"], "CHUNK_EDGE_FADE_MS": constants["CHUNK_EDGE_FADE_MS"], "CROSSFADE_MS": constants["CROSSFADE_MS"], "FINAL_ENDPOINT_FADE_MS": constants["FINAL_ENDPOINT_FADE_MS"], "match_chunk_rms": match_chunk_rms, "punctuation_pause_seconds": punctuation_pause_seconds, "fade_variable_internal_edges": fade_variable_internal_edges, "join_audio_chunks_variable": join_audio_chunks_variable, "apply_loudness_floor": lambda waveform, **_kwargs: waveform, "_apply_speed": lambda waveform, _speed: waveform, "count_speech_units": count_speech_units, "finish_audio": lambda waveform, _sample_rate, **_kwargs: waveform, "pcm16_verification_waveform": pcm16_verification_waveform, "timed_latency_stage": timed_latency_stage, } exec(compile(module, str(app_path), "exec"), namespace) return namespace["_assemble_trajectory_audio"] def _isolated_load_speakers(*, metadata: dict, speaker_ids: tuple[str, ...]): """Execute only ``_load_speakers`` without importing the GPU application.""" 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 == "_load_speakers" ) module = ast.Module( body=[ ast.ImportFrom( module="__future__", names=[ast.alias(name="annotations")], level=0, ), function, ], type_ignores=[], ) ast.fix_missing_locations(module) load_calls = [] centroids = tuple(f"test-centroid-{index}" for index in range(len(speaker_ids))) def fake_load(path, **kwargs): load_calls.append((path, kwargs)) return {"speaker_ids": speaker_ids, "centroids": centroids} namespace = { "MODEL_DIR": "/pinned/model", "METADATA": metadata, "os": SimpleNamespace( path=SimpleNamespace( join=lambda *parts: "/".join(part.strip("/") for part in parts), exists=lambda _path: True, ) ), "torch": SimpleNamespace(load=fake_load), } exec(compile(module, app_path, "exec"), namespace) result = namespace["_load_speakers"]() return result, load_calls def test_pinned_female_voice_embedding_replaces_only_builtin_b(): constants = _string_constants(ROOT / "app.py") contract = FROZEN_SPEAKER_ANCHORS[constants["MODEL_REVISION"]] source = (ROOT / "app.py").read_text(encoding="utf-8") assert contract == { "speaker_id": "female_voice", "speaker_index": 1, "ui_label": "內建語者 B", "dtype": "float32", "shape": (192,), "sha256": ( "67cd20e7ebdde857913f2e0bd722626fdb3224ec2f3bc9d5d0635c7b5272aa3d" ), "file_sha256": ( "e9556e14723c140985a104c1659d1ff8a5078d2fa28ce2fb756f04906641a8a7" ), } assert '"speaker_b_embedding.pt"' in source assert "centroids[1] = torch.nn.functional.normalize(" in source assert "default_index = 0" in source assert contract["file_sha256"] in source def test_remote_model_and_speaker_encoder_are_revision_pinned(): app_path = ROOT / "app.py" source = app_path.read_text(encoding="utf-8") constants = _string_constants(app_path) assert constants["MODEL_REVISION"] == "6f7cab914a1e27c56b504ec663c0144dc25cc0a3" assert constants["ECAPA_REVISION"] == "0f99f2d0ebe89ac095bcc5903c4dd8f72b367286" assert "snapshot_download(REPO_ID, revision=MODEL_REVISION)" in source assert "ECAPA_REPO_ID," in source assert "revision=ECAPA_REVISION" in source assert "source=ECAPA_DIR" in source assert 'overrides={"pretrained_path": ECAPA_DIR}' in source def test_cuda_runtime_is_pinned_deterministic_before_model_import(): app_path = ROOT / "app.py" source = app_path.read_text(encoding="utf-8") constants = _literal_constants(app_path) assert constants["DETERMINISTIC_ALGORITHMS"] is True assert constants["CUBLAS_WORKSPACE_CONFIG"] == ":4096:8" assert constants["MAMBA_DETERMINISTIC"] == "1" assert constants["CUDNN_BENCHMARK"] is False assert constants["CUDNN_DETERMINISTIC"] is True assert constants["CUDA_MATMUL_ALLOW_TF32"] is False assert constants["CUDNN_ALLOW_TF32"] is False assert constants["EXPECTED_TORCH_VERSION"] == "2.11.0+cu130" assert constants["EXPECTED_TORCH_CUDA_VERSION"] == "13.0" assert constants["EXPECTED_CUDNN_VERSION"] == 91900 assert constants["EXPECTED_MAMBA_SSM_VERSION"] == ( "2.3.2.post1+bluemagpie.triton1" ) assert constants["EXPECTED_TRITON_VERSION"] == "3.6.0" assert source.index('os.environ.setdefault(\n "CUBLAS_WORKSPACE_CONFIG"') < ( source.index("import torch") ) assert source.index('os.environ.setdefault(\n "MAMBA_DETERMINISTIC"') < ( source.index("import torch") ) spaces_import = source.index(" import spaces") torch_import = source.index("import torch") mamba_import = source.index( "from mamba_ssm import __version__ as mamba_ssm_runtime_version" ) assert spaces_import < torch_import assert spaces_import < mamba_import assert "torch.use_deterministic_algorithms(DETERMINISTIC_ALGORITHMS)" in source assert "torch.backends.cudnn.benchmark = CUDNN_BENCHMARK" in source assert "torch.backends.cudnn.deterministic = CUDNN_DETERMINISTIC" in source assert ( "torch.backends.cuda.matmul.allow_tf32 = CUDA_MATMUL_ALLOW_TF32" in source ) assert "torch.backends.cudnn.allow_tf32 = CUDNN_ALLOW_TF32" in source runtime_check = source.index("if runtime_versions != expected_runtime_versions:") backend_setup = source.index( "torch.use_deterministic_algorithms(DETERMINISTIC_ALGORITHMS)" ) model_import = source.index("from bluemagpie import BlueMagpieModel") assert runtime_check < backend_setup < model_import def test_tts_runtime_is_vendored_from_the_frozen_commit(): requirements = (ROOT / "requirements.txt").read_text(encoding="utf-8").splitlines() provenance = (ROOT / "bluemagpie" / "UPSTREAM_RUNTIME.md").read_text( encoding="utf-8" ) assert not any("BlueMagpie-TTS.git" in line for line in requirements) assert "ce384c8cc54efea1aaba7b9f1d7ded6c1c99aa9a" in provenance assert (ROOT / "bluemagpie" / "LICENSE.upstream").is_file() assert (ROOT / "bluemagpie" / "_vendor" / "voxcpm" / "LICENSE").is_file() pinned_hashes = { "model.py": "91810524212b34f727880154d90653fab4ae1b75eb3471b86cafd92c75514fef", "loading.py": "e3407544e9bc888018fe5771edc01d954469e2af548b566873e0ef7f2afe6dca", "_vendor/voxcpm/model/utils.py": ( "cea16e1ab57f15129a7f5dec13c428bd14a771221abcf17dd0a90b3d65e763a2" ), } for relative_path, expected_hash in pinned_hashes.items(): payload = (ROOT / "bluemagpie" / relative_path).read_bytes() assert hashlib.sha256(payload).hexdigest() == expected_hash def test_mamba_triton_runtime_is_vendored_and_importable_without_extension(): import sys import mamba_ssm from mamba_ssm.ops.triton.layernorm_gated import RMSNorm from mamba_ssm.ops.triton.ssd_combined import mamba_chunk_scan_combined assert Path(mamba_ssm.__file__).resolve().is_relative_to(ROOT) assert mamba_ssm.__version__ == "2.3.2.post1+bluemagpie.triton1" assert RMSNorm.__name__ == "RMSNorm" assert mamba_chunk_scan_combined.__name__ == "mamba_chunk_scan_combined" assert "selective_scan_cuda" not in sys.modules pinned_hashes = { "LICENSE.upstream": ( "760939b000194d04548ede6a857bbe735d1695d8422ec85955c8e2bd7f4b95c5" ), "ops/triton/k_activations.py": ( "ede8d75600b1b01fd867df00eae4a727df9a34ef8091baf965522c36b3dfdf7b" ), "ops/triton/layernorm_gated.py": ( "eb6252e247b90f1c8a75946efbc1a221e0c4da701b6757ddae49f3495cf7a42f" ), "ops/triton/softplus.py": ( "989f7667ad7f8866dfafa2783d553555e235478cbce349905b23d8fb66b7c5ab" ), "ops/triton/ssd_bmm.py": ( "5059b16f8fa269cd84e8159ee3c6880e4ccf90e4bedf9c2d733e6bc2baff9587" ), "ops/triton/ssd_chunk_scan.py": ( "055b5ce4cb0f30c84c3e031d775f218227fa0b793359254a2465777097b224f9" ), "ops/triton/ssd_chunk_state.py": ( "6d82771b2f62a7bf84c381c8ef8e4b579a95e0352021d117507a1072e4d71dcd" ), "ops/triton/ssd_combined.py": ( "39c19c1e4c8e36982847079bc12b4f07ed3af03ac68f6a6653200ef80df5515d" ), "ops/triton/ssd_state_passing.py": ( "ae1fab6c680cb5312ef746862a2cf8e74157033c80878490cc64fa1f24f88b70" ), "utils/determinism.py": ( "cb6e1c30392c11200425c2a23ad9fa3d47f50b556d15e9b0caf79b7d483d6f1d" ), "utils/torch.py": ( "1c3132a1a747e914874e84b5f300ea80d0297faba6eb9256cb5ac4b3d6413c06" ), } for relative_path, expected_hash in pinned_hashes.items(): payload = (ROOT / "mamba_ssm" / relative_path).read_bytes() assert hashlib.sha256(payload).hexdigest() == expected_hash def test_barbet_runtime_dependency_is_commit_pinned(): requirements = (ROOT / "requirements.txt").read_text(encoding="utf-8").splitlines() barbet_lines = [line for line in requirements if "OpenFormosa/Barbet.git" in line] assert barbet_lines == [ "git+https://github.com/OpenFormosa/Barbet.git@" "6fcd7ce4aa37f2250a3242995bef0fbc3b026ba8" ] def test_breeze25_content_asr_is_revision_and_weight_pinned_once(): quality_path = ROOT / "quality_runtime.py" quality_source = quality_path.read_text(encoding="utf-8") constants = _string_constants(quality_path) app_source = (ROOT / "app.py").read_text(encoding="utf-8") assert constants["BREEZE25_MODEL_ID"] == "MediaTek-Research/Breeze-ASR-25" assert ( constants["BREEZE25_REVISION"] == "cffe7ccb404d025296a00758d0a33468bec3a9d0" ) assert ( constants["BREEZE25_WEIGHT_SHA256"] == "c5d952b3bc03ea277209aff0ef5b5c4c055d74449ff794c02d8f4e315fdef6b6" ) assert "load_pinned_breeze25_runtime" in quality_source assert "transcribe_breeze25" in quality_source assert app_source.count("snapshot_download(\n BREEZE25_MODEL_ID,") == 1 assert "snapshot_download(WHISPER_MODEL_ID" not in app_source assert "snapshot_download(\n VERIFICATION_WHISPER_MODEL_ID," not in app_source allow_patterns = _literal_constants(ROOT / "app.py")[ "BREEZE25_SNAPSHOT_ALLOW_PATTERNS" ] assert "model.safetensors" in allow_patterns assert "optimizer.bin" not in allow_patterns assert not any(pattern.startswith("whisper-github/") for pattern in allow_patterns) def test_quality_runtime_dependencies_are_version_pinned(): requirements = set( (ROOT / "requirements.txt").read_text(encoding="utf-8").splitlines() ) assert "huggingface_hub==0.36.0" in requirements assert "opencc-python-reimplemented==0.1.7" in requirements assert "pypinyin==0.55.0" in requirements assert "transformers==4.57.6" in requirements assert "accelerate==1.12.0" in requirements assert "einops==0.8.2" in requirements assert "pydantic==2.11.10" in requirements assert "packaging==26.0" in requirements assert "numpy==2.2.6" in requirements assert "scipy==1.15.3" in requirements assert "numexpr==2.10.0" in requirements assert "bottleneck==1.5.0" in requirements assert "tqdm==4.68.2" in requirements assert "safetensors==0.8.0" in requirements assert "librosa==0.11.0" in requirements assert "soundfile==0.14.0" in requirements assert "speechbrain==1.0.3" in requirements def test_readme_describes_coverage_refill_and_sequence_transition_scores(): readme = (ROOT / "README.md").read_text(encoding="utf-8") app_source = (ROOT / "app.py").read_text(encoding="utf-8") assert "exactly one same-seed whole trajectory" in readme assert "single-chunk refills for zero/low-coverage rows only" in readme assert "`(coverage, refill attempts, chunk index)`" in readme assert "32 generated TTS chunks、800 generated speech units" in readme assert "不使用 reference-style distribution score" in readme assert "median-F0 軟成本" in readme assert "5.2 CJK / 4.6 ASCII" in readme assert "4.6 CJK / 4.0 ASCII" in readme assert "URL/email-bearing chunks 8 units" in readme assert "Email 第一輪必須在 `小老鼠`" in readme assert "第一輪 DP 不得跨越這兩類" in readme assert "grammar boundary" in readme assert "所有最佳 edit alignment" in readme assert "boundary-only local rejects" in readme assert "drop 不超過 0.15" in readme assert "整段仍必須通過 similarity 0.105 與 boundary drop 0.095" in readme assert "zero/low-coverage 單 chunk refill" in app_source assert "speaker/RMS/F0 ragged DP" in app_source assert "1→5→10→15→20" not in app_source def test_app_wires_row_local_candidate_ordinal_to_generation_policy_and_logs_it(): source = (ROOT / "app.py").read_text(encoding="utf-8") assert source.count("policy: GenerationPolicy") == 2 assert "generation_context.chunk_candidate_ordinals" in source assert "policy=generation_policy_for_candidate_offset(candidate_ordinal)" in source assert "generation_context.seed != seed" in source assert 'f"name={policy.name}' in source assert "chunk_policies={selected_policies}" in source assert '"min_len": min_len' in source assert '"[BlueMagpie] generation attempt "' in source assert "scheduled_cfg={scheduled_cfg:.2f}" in source assert "effective_cfg={effective_cfg:.2f}" in source assert "network_floor_applied={network_floor_applied}" in source assert "short_floor_applied={short_floor_applied}" in source assert "short_headroom_floor_applied={short_headroom_floor_applied}" in source assert "generation_endpoint_evidence_by_seed" in source assert "chunk_stop_reasons=tuple(" in source assert "chunk_endpoint_energy_ratios=tuple(" in source assert "require_endpoint_evidence=True" in source quality_source = (ROOT / "quality_runtime.py").read_text(encoding="utf-8") assert "short_headroom_max_units=2" in quality_source assert "short_hard_stop_floor_steps=5" in quality_source assert "ENDPOINT_HARD_STOP_PENALTY = 0.05" in quality_source assert "ENDPOINT_ENERGY_WEIGHT = 0.02" in quality_source assert "ENDPOINT_PREFERRED_SQUIM_STOI_SLACK = 0.02" in quality_source assert "ENDPOINT_PREFERRED_SQUIM_PESQ_SLACK = 0.03" in quality_source def test_app_rejects_ambiguous_iri_before_frontend_normalization(): source = (ROOT / "app.py").read_text(encoding="utf-8") synthesize_start = source.index("def _synthesize(") raw_text = source.index("raw_text = str(text)", synthesize_start) iri_guard = source.index( "if network_identifier_has_ambiguous_iri(raw_text):", synthesize_start, ) normalization = source.index( 'text = normalize_spoken_forms(raw_text, locale="zh-TW")', synthesize_start, ) assert synthesize_start < raw_text < iri_guard < normalization assert "非 ASCII IRI 必須先轉成 ASCII/percent-encoded" in ( ROOT / "README.md" ).read_text(encoding="utf-8") def test_app_does_not_add_an_artificial_onset_split(): 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) } synthesize_source = ast.get_source_segment(source, functions["_synthesize"]) assert synthesize_source is not None assert "split_quality_text_for_tts(" in synthesize_source assert "QUALITY_MEDIUM_TEXT_MAX_UNITS = 48" in source assert "QUALITY_MEDIUM_CHUNK_UNITS = 24" in source assert "GLYPH_CARRIER_MEDIUM_CHUNK_UNITS = 32" in source assert "medium_chunk_units = (" in synthesize_source assert "GLYPH_CARRIER_MEDIUM_CHUNK_UNITS" in synthesize_source assert "split_leading_clause(" not in synthesize_source assert "ONSET_CLAUSE_SEARCH_CHARS = 0" in source def test_interactive_retry_budget_excludes_network_and_glyph_carriers(): app_path = ROOT / "app.py" source = app_path.read_text(encoding="utf-8") tree = ast.parse(source) helper = next( node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == "_quality_request_generated_chunk_limit" ) module = ast.Module( body=[ ast.ImportFrom( module="__future__", names=[ast.alias(name="annotations")], level=0, ), helper, ], type_ignores=[], ) ast.fix_missing_locations(module) constants = _literal_constants(app_path) namespace = { "QUALITY_MEDIUM_TEXT_MAX_UNITS": constants[ "QUALITY_MEDIUM_TEXT_MAX_UNITS" ], "QUALITY_INTERACTIVE_MAX_GENERATED_CHUNKS": constants[ "QUALITY_INTERACTIVE_MAX_GENERATED_CHUNKS" ], "QUALITY_MAX_GENERATED_CHUNKS": constants[ "QUALITY_MAX_GENERATED_CHUNKS" ], } exec(compile(module, str(app_path), "exec"), namespace) limit = namespace["_quality_request_generated_chunk_limit"] assert limit(48, network_request=False, carrier_request=False) == 10 assert limit(49, network_request=False, carrier_request=False) == 32 assert limit(10, network_request=True, carrier_request=False) == 32 assert limit(10, network_request=False, carrier_request=True) == 32 def test_glyph_carrier_keeps_legacy_single_chunk_planner_profile(): app_path = ROOT / "app.py" source = app_path.read_text(encoding="utf-8") tree = ast.parse(source) validator = next( node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == "_validate_glyph_carrier_text" ) validator_source = ast.get_source_segment(source, validator) constants = _literal_constants(app_path) text = "兩張票合計新台幣二千六百八十，租借望遠鏡另收日圓四千五百。" chunks = split_quality_text_for_tts( text, long_max_units=constants["CHUNK_CHARS"], min_chunk_units=constants["MIN_CHUNK_CHARS"], medium_max_units=constants["QUALITY_MEDIUM_TEXT_MAX_UNITS"], medium_chunk_units=constants["GLYPH_CARRIER_MEDIUM_CHUNK_UNITS"], ) assert chunks == [text] assert validator_source is not None assert ( "medium_chunk_units=GLYPH_CARRIER_MEDIUM_CHUNK_UNITS" in validator_source ) def test_app_applies_fixed_mixed_cfg_schedule_after_global_quality_floor(): 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) } synthesize_source = ast.get_source_segment(source, functions["_synthesize"]) assert synthesize_source is not None assert "DEFAULT_CFG = 3.0" in source assert "QUALITY_CFG_MIN" not in source assert "MIXED_CFG_PRIMARY = 3.0" in source assert "MIXED_CFG_ALTERNATE = 2.0" in source assert ( 'MIXED_CFG_SCHEDULE = "row_ordinal_zero_and_even_primary_odd_alternate"' in source ) assert "network_request = contains_network_identifier(raw_text)" in synthesize_source assert "cfg_value != MIXED_CFG_PRIMARY" in synthesize_source assert "request_cfg = MIXED_CFG_PRIMARY" in synthesize_source assert "cfg=candidate_cfg(candidate_ordinal)" in synthesize_source assert "candidate_ordinals = generation_context.chunk_candidate_ordinals" in ( synthesize_source ) assert "chunk_candidate_ordinals=candidate_ordinals" in synthesize_source assert "generation_cfg_for_candidate_offset(" in synthesize_source assert "chunk_cfgs={selected_cfgs}" in synthesize_source assert "attempted_schedule_cfgs={attempted_schedule_cfgs}" in synthesize_source assert "network_cfg_floor={NETWORK_TEXT_CFG_MIN:.2f}" in synthesize_source assert "mixed_cfg_primary={MIXED_CFG_PRIMARY:.2f}" in synthesize_source assert "np.isfinite(cfg_value)" in synthesize_source assert "1.0 <= cfg_value <= 4.0" in synthesize_source assert 'label="CFG (已驗證固定值)"' in source def test_app_and_quality_runtime_pin_the_same_mixed_cfg_contract(): app_constants = _literal_constants(ROOT / "app.py") quality_constants = _literal_constants(ROOT / "quality_runtime.py") assert app_constants["MIXED_CFG_SCHEDULE"] == quality_constants[ "MIXED_CFG_SCHEDULE" ] assert app_constants["MIXED_CFG_PRIMARY"] == quality_constants[ "MIXED_CFG_PRIMARY" ] assert app_constants["MIXED_CFG_ALTERNATE"] == quality_constants[ "MIXED_CFG_ALTERNATE" ] assert app_constants["SHORT_TEXT_CFG_UNITS"] == quality_constants[ "MIXED_CFG_SHORT_TEXT_MAX_UNITS" ] assert app_constants["SHORT_TEXT_CFG_MIN"] == quality_constants[ "MIXED_CFG_SHORT_TEXT_MIN" ] assert app_constants["NETWORK_TEXT_CFG_MIN"] == quality_constants[ "MIXED_CFG_NETWORK_MIN" ] def test_app_rejects_silent_text_and_coalesces_before_runtime_budgeting(): source = (ROOT / "app.py").read_text(encoding="utf-8") production_source = (ROOT / "production.py").read_text(encoding="utf-8") tree = ast.parse(source) functions = { node.name: node for node in tree.body if isinstance(node, ast.FunctionDef) } synthesize_source = ast.get_source_segment(source, functions["_synthesize"]) assemble_source = ast.get_source_segment( source, functions["_assemble_trajectory_audio"], ) assert synthesize_source is not None assert assemble_source is not None assert "if len(raw_text) > MAX_TEXT_CHARS" in synthesize_source assert "normalized_units = count_speech_units(text)" in synthesize_source assert "if not network_request:" in synthesize_source assert ( "text = strip_unspoken_cjk_quotes_for_generation(text)" in synthesize_source ) assert synthesize_source.index("if not network_request:") < ( synthesize_source.index("normalized_units = count_speech_units(text)") ) assert "if normalized_units <= 0" in synthesize_source assert ( "if normalized_units > QUALITY_MAX_GENERATED_TEXT_UNITS" in synthesize_source ) assert "split_quality_text_for_tts(" in synthesize_source assert "coalesce_text_chunks(" in synthesize_source assert "chunk_specs = plan_generation_chunks(" in synthesize_source assert "chunks = tuple(spec.text for spec in chunk_specs)" in synthesize_source assert "max_chunks=QUALITY_MAX_GENERATED_CHUNKS" in synthesize_source assert "pre_faded_edges=True" in assemble_source assert "NETWORK_GENERATION_MIN_UNITS = 8" in source assert "NETWORK_GENERATION_TARGET_UNITS = 32" in source assert "NETWORK_GENERATION_MAX_UNITS = 36" in source assert "NETWORK_SPLIT_ATOMIC_EMAIL_AT_SEPARATOR = True" in source assert "network_min_units=NETWORK_GENERATION_MIN_UNITS" in synthesize_source assert "split_atomic_email_at_separator=(" in synthesize_source assert "NETWORK_SPLIT_ATOMIC_EMAIL_AT_SEPARATOR" in synthesize_source assert "NETWORK_INTERNAL_FADE_MS = 5.0" in source assert "NETWORK_INTERNAL_SILENCE_MS = 400.0" in source assert "SEMANTIC_CHUNK_MIN_SILENCE_MS = 250.0" in source assert "NETWORK_REQUEST_SEMANTIC_CHUNK_MIN_SILENCE_MS = 350.0" in source assert 'chunk_specs[index].boundary_after == "network_internal"' in ( assemble_source ) assert "NETWORK_INTERNAL_SILENCE_MS / 1000.0" in assemble_source assert "punctuation_pause_seconds(chunk)" in assemble_source assert "semantic_min_silence_ms / 1000.0" in assemble_source assert "pauses.append(int(round(pause_seconds * SR)))" in assemble_source assert "join_audio_chunks_variable(" in assemble_source assert "network_conditioned=network_flags" in synthesize_source assert "network_conditioned=network_flag" in synthesize_source assert "mandatory_cut_offsets" in production_source assert "any(start < cut < end for cut in active_mandatory_cuts)" in ( production_source ) assert "plan = solve(mandatory_cuts - short_identifier_cuts)" in ( production_source ) assert "_protected_ranges_are_exact_in_all_optimal_alignments(" in ( production_source ) def test_network_endpoint_headroom_is_isolated_from_public_unit_contracts(): app_source = (ROOT / "app.py").read_text(encoding="utf-8") production_source = (ROOT / "production.py").read_text(encoding="utf-8") app_tree = ast.parse(app_source) production_tree = ast.parse(production_source) app_functions = { node.name: node for node in app_tree.body if isinstance(node, ast.FunctionDef) } production_functions = { node.name: node for node in production_tree.body if isinstance(node, ast.FunctionDef) } generate_source = ast.get_source_segment( app_source, app_functions["_generate_chunk"], ) public_counter_source = ast.get_source_segment( production_source, production_functions["count_speech_units"], ) network_counter_source = ast.get_source_segment( production_source, production_functions["count_network_endpoint_duration_units"], ) assert generate_source is not None assert public_counter_source is not None assert network_counter_source is not None assert "count_network_endpoint_duration_units(text)" in generate_source assert "if network_conditioned" in generate_source assert "duration_units=endpoint_duration_units" in generate_source assert "min_len = 2" in generate_source assert '"max_len": hard_stop_steps' in generate_source assert "expected_steps=expected_steps" in generate_source assert "hard_stop_steps=hard_stop_steps" in generate_source assert "duration_counter=" in generate_source assert "divisor = 2 if token.isdigit() else 4" in public_counter_source assert "math.ceil(ascii_run_length / 2)" in network_counter_source assert production_source.count( "count_network_endpoint_duration_units(" ) == 1 def test_network_local_breeze25_projection_is_range_bound_for_initial_and_refill(): app_source = (ROOT / "app.py").read_text(encoding="utf-8") quality_source = (ROOT / "quality_runtime.py").read_text(encoding="utf-8") app_tree = ast.parse(app_source) quality_tree = ast.parse(quality_source) app_functions = { node.name: node for node in app_tree.body if isinstance(node, ast.FunctionDef) } quality_functions = { node.name: node for node in quality_tree.body if isinstance(node, ast.FunctionDef) } helper_source = ast.get_source_segment( app_source, app_functions["_verify_network_local_asr_intersection"], ) initial_source = ast.get_source_segment( app_source, app_functions["_qualify_candidate_trajectory_audio"], ) refill_source = ast.get_source_segment( app_source, app_functions["_verify_refill_candidate_trajectory_audio"], ) intersection_source = ast.get_source_segment( quality_source, quality_functions["intersect_local_semantic_verification"], ) assert all( source is not None for source in ( helper_source, initial_source, refill_source, intersection_source, ) ) assert "proof_rows[index] for index in selected_indices" in helper_source assert "transcriber=transcribe_breeze25" in helper_source assert "transcriber=transcriber" in helper_source assert "semantic_only=True" in helper_source assert "network_fragment_proofs=independent_proof_rows" in helper_source assert "intersect_local_semantic_verification(" in helper_source assert "[BlueMagpie] network local Breeze25 projection " in helper_source assert "proof_count=" in helper_source assert "transcript_text" not in helper_source for caller_source in (initial_source, refill_source): turbo_index = caller_source.index( "local_verification = _verify_trajectory_audio(" ) intersection_index = caller_source.index( "_verify_network_local_asr_intersection(" ) assert turbo_index < intersection_index assert "proof_rows = _network_fragment_proof_rows(" in caller_source assert "network_fragment_proofs=proof_rows" in caller_source assert "proof_rows," in caller_source[intersection_index:] assert "candidate_seed=" in caller_source[intersection_index:] assert "transcriber=" in caller_source[intersection_index:] assert "semantic_reasons = [\"semantic_gate\"]" in intersection_source assert "\"network_protected_span_mismatch\"" in intersection_source assert "chunk_artifacts=primary_verification.chunk_artifacts" in ( intersection_source ) assert "CASCADE_EVIDENCE_SCHEMA_VERSION = 6" in quality_source assert '"chunk_text_variants"' in quality_source assert "local_candidate_has_coverage_eligibility(" in helper_source assert "independent_local_results=" in initial_source assert "independent_local_results=" in refill_source assert '"independent_local_evidence_complete"' in quality_source assert '"independent_local_results"' in quality_source def test_network_local_breeze25_projection_reuses_exact_proof_rows_and_logs_no_text( capsys, ): source = (ROOT / "app.py").read_text(encoding="utf-8") tree = ast.parse(source) function = next( node for node in tree.body if ( isinstance(node, ast.FunctionDef) and node.name == "_verify_network_local_asr_intersection" ) ) module = ast.Module( body=[ ast.ImportFrom( module="__future__", names=[ast.alias(name="annotations")], level=0, ), function, ], type_ignores=[], ) ast.fix_missing_locations(module) calls = [] proof = object() skipped_proof = object() proof_rows = ((), (proof,), (skipped_proof,)) primary_results = (object(), object(), object()) turbo = SimpleNamespace(candidate_results=primary_results) independent_result = SimpleNamespace( passed=True, rejection_reasons=(), comparison=SimpleNamespace( cer=0.0, prefix_cer=0.0, suffix_cer=0.0, extra_tail_units=0, ), ) independent = SimpleNamespace(candidate_results=(independent_result,)) verification_transcriber = object() def fake_verify(*args, **kwargs): calls.append(("verify", args, kwargs)) return independent def fake_intersect(*args): calls.append(("intersect", args)) return "combined" namespace = { "_verify_trajectory_audio": fake_verify, "QUALITY_FINAL_ASR_MAX_NEW_TOKENS": 440, "SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP": 0.15, "transcribe_breeze25": verification_transcriber, "intersect_local_semantic_verification": fake_intersect, "local_candidate_has_coverage_eligibility": ( lambda result, **_kwargs: result is primary_results[1] ), "LocalIndependentGateEvidence": LocalIndependentGateEvidence, "candidate_gate_evidence": lambda result: ("bounded", result), } exec(compile(module, "", "exec"), namespace) result = namespace["_verify_network_local_asr_intersection"]( turbo, ("ordinary-audio", "network-audio", "ordinary-audio-2"), ("PRIVATE_ORDINARY_A", "PRIVATE_NETWORK_TEXT", "PRIVATE_ORDINARY_B"), "anchor", proof_rows, candidate_seed=123, transcriber=verification_transcriber, ) assert result[0] == "combined" verify_call = calls[0] assert verify_call[0] == "verify" assert verify_call[1][:4] == ( ("network-audio",), ("PRIVATE_NETWORK_TEXT",), "anchor", 1.0, ) assert verify_call[1][4] == 440 assert verify_call[2]["transcriber"] is verification_transcriber assert verify_call[2]["semantic_only"] is True selected_proofs = verify_call[2]["network_fragment_proofs"] assert selected_proofs == ((proof,),) assert selected_proofs[0] is proof_rows[1] assert calls[1] == ( "intersect", (turbo, independent, (1,)), ) evidence = result[1] assert evidence[0] == LocalIndependentGateEvidence(False, None, 0, None) assert evidence[1].attempted is True assert evidence[1].passed is True assert evidence[1].proof_count == 1 assert evidence[1].result == ("bounded", independent_result) assert evidence[2] == LocalIndependentGateEvidence(False, None, 1, None) log = capsys.readouterr().out assert "seed=123 local_chunk_index=1 proof_count=1 passed=True" in log assert "PRIVATE_" not in log def test_app_emits_one_canonical_content_free_evidence_line_per_terminal_outcome(): 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) } synthesize_source = ast.get_source_segment(source, functions["_synthesize"]) qualifier_source = ast.get_source_segment( source, functions["_qualify_candidate_trajectory_audio"], ) assert synthesize_source is not None assert qualifier_source is not None assert synthesize_source.count("format_cascade_evidence_log(") == 3 assert synthesize_source.count( "generated_chunk_limit=request_generated_chunk_limit" ) == 3 assert synthesize_source.count( "generated_text_unit_limit=QUALITY_MAX_GENERATED_TEXT_UNITS" ) == 3 assert "generation_evidence_factory=candidate_generation_evidence" in ( synthesize_source ) assert 'outcome="no_qualified_candidate"' in synthesize_source assert 'outcome="final_output_rejected"' in synthesize_source assert 'outcome="returned"' in synthesize_source assert "error.diagnostics" in synthesize_source assert "cascade.diagnostics" in synthesize_source assert "final_output=final_evidence" in synthesize_source assert "CandidateVerification(" in qualifier_source assert "independent_local_results=independent_local_results" in ( qualifier_source ) assert "joined_evidence = trajectory_gate_evidence(joined_verification)" in ( qualifier_source ) assert "joined_output=joined_evidence" in qualifier_source assert ( "independent_output=trajectory_gate_evidence(independent_verification)" in qualifier_source ) assert "independent_final_output=independent_final_evidence" in synthesize_source def test_app_limits_boundary_relaxation_to_final_verified_sequence_fallback(): app_source = (ROOT / "app.py").read_text(encoding="utf-8") quality_source = (ROOT / "quality_runtime.py").read_text(encoding="utf-8") assert "SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP = 0.15" in quality_source assert "result.rejection_reasons != (\"boundary_speaker_drop\",)" in quality_source assert "sequence fallback boundary relaxation requires a final verifier" in quality_source assert "sequence_fallback_max_local_boundary_speaker_drop=(" in app_source assert "SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP" in app_source assert "QUALITY_RELEASE_MAX_BOUNDARY_SPEAKER_DROP = 0.095" in app_source def test_internal_synthesize_accepts_only_a_keyword_seed_while_ui_stays_unchanged(): 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, ast.AsyncFunctionDef)) } synthesize = functions["_synthesize"] assert "request_seed" not in [argument.arg for argument in synthesize.args.args] assert [argument.arg for argument in synthesize.args.kwonlyargs][-1] == "request_seed" assert isinstance(synthesize.args.kw_defaults[-1], ast.Constant) assert synthesize.args.kw_defaults[-1].value is None assert "request_seed = resolve_request_seed(request_seed, secrets.randbelow)" in source assert "run_coverage_adaptive_cascade(\n chunks,\n request_seed," in source for wrapper_name in ("tts_speaker", "tts_reference"): wrapper = functions[wrapper_name] arguments = wrapper.args.args + wrapper.args.kwonlyargs assert "request_seed" not in [argument.arg for argument in arguments] def test_app_reverifies_the_post_join_speed_adjusted_whole_waveform(): 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) } assemble_source = ast.get_source_segment( source, functions["_assemble_trajectory_audio"], ) release_verify_source = ast.get_source_segment( source, functions["_verify_release_whole_audio"], ) synthesize_source = ast.get_source_segment(source, functions["_synthesize"]) assert assemble_source is not None assert release_verify_source is not None assert synthesize_source is not None speed_index = assemble_source.index( "waveform = _apply_speed(waveform, playback_speed)" ) finish_index = assemble_source.index("waveform = finish_audio(") pcm_index = assemble_source.index( "return pcm16_verification_waveform(waveform)" ) assert speed_index < finish_index < pcm_index assert "fade_ms=FINAL_ENDPOINT_FADE_MS" in assemble_source assert "FINAL_ENDPOINT_FADE_MS = 3.0" in source assert "finish_fade_ms" not in assemble_source production_source = (ROOT / "production.py").read_text(encoding="utf-8") assert "trailing_silence_ms: float = 180.0" in production_source assemble_index = synthesize_source.index( "waveform = _assemble_trajectory_audio(" ) verify_index = synthesize_source.index( "final_verification = _verify_release_whole_audio(" ) require_index = synthesize_source.index( "require_verified_final_output(final_verification)" ) return_index = synthesize_source.index("return SR, waveform") assert assemble_index < verify_index < require_index < return_index assert "text," in synthesize_source[verify_index:require_index] assert "BREEZE25_RELEASE_VERIFICATION_PROFILE" in synthesize_source[ verify_index:require_index ] assert " 1.0," in release_verify_source assert "QUALITY_FINAL_ASR_MAX_NEW_TOKENS" in release_verify_source assert "release_speaker_gate=True" in release_verify_source assert "short_audio_seconds=(" in source assert "RELEASE_SPEAKER_TRIGGER_SECONDS" in source def test_chunk_generation_closed_loop_rerenders_final_audio_from_raw_once(): 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) } generate_source = ast.get_source_segment(source, functions["_generate_chunk"]) apply_speed_source = ast.get_source_segment(source, functions["_apply_speed"]) assert generate_source is not None assert apply_speed_source is not None total_pace_index = generate_source.index("pace_speed = target_pace_speed(") active_measure_index = generate_source.index( "active_voiced_duration_seconds(audio, SR)" ) active_pace_index = generate_source.index( "active_speed = active_pace_correction_speed(" ) combined_speed_index = generate_source.index( "combined_speed = min(pace_speed, active_speed)" ) initial_stretch_index = generate_source.index("corrected = _apply_speed(") corrected_measure_index = generate_source.index( "corrected_active_duration = active_voiced_duration_seconds(" ) rerender_speed_index = generate_source.index( "rerender_speed = active_pace_correction_speed(" ) final_speed_index = generate_source.index( "final_speed = combined_speed * rerender_speed" ) final_stretch_index = generate_source.index("final_audio = _apply_speed(") assert ( total_pace_index < active_measure_index < active_pace_index < combined_speed_index < initial_stretch_index < corrected_measure_index < rerender_speed_index < final_speed_index < final_stretch_index ) assert generate_source.count("_apply_speed(") == 4 assert "waveform = _apply_speed(audio, pace_speed)" not in generate_source assert "ACTIVE_PACE_TARGET_CPS = 4.00" in source assert "CLOSED_LOOP_ACTIVE_PACE_TARGET_CPS = 3.95" in source assert "target_cps=ACTIVE_PACE_TARGET_CPS" in generate_source assert "target_cps=CLOSED_LOOP_ACTIVE_PACE_TARGET_CPS" in generate_source assert "prior_speed=1.0" in generate_source assert "prior_speed=combined_speed" in generate_source assert "min_total_speed=MIN_PACE_SPEED" in generate_source assert "final_speed = combined_speed * rerender_speed" in generate_source assert "fallback_speed = final_speed * fallback_residual" in generate_source assert "_apply_speed(\n corrected" not in generate_source assert "_apply_speed(\n audio,\n final_speed" in generate_source assert "audio,\n fallback_speed" in generate_source assert "EXTREME_SHORT_MAX_UNITS = 2" in source assert "EXTREME_SHORT_MAX_SPEED = 0.95" in source assert "speech_units <= EXTREME_SHORT_MAX_UNITS" in generate_source assert "final_speed > EXTREME_SHORT_MAX_SPEED" in generate_source assert "final_speed = EXTREME_SHORT_MAX_SPEED" in generate_source assert "audio,\n final_speed,\n network_conditioned=False" in generate_source assert "MIN_PACE_SPEED = 0.90" in source assert "PACE_ONLY_FALLBACK_MIN_SPEED = MIN_PACE_SPEED" in source assert "from audio_wsola import wsola_time_stretch" in source assert "network_conditioned=network_conditioned" in generate_source assert "del network_conditioned" in apply_speed_source assert "speed < MIN_PACE_SPEED" in apply_speed_source assert "wsola_time_stretch(" in apply_speed_source assert "sample_rate=SR" in apply_speed_source assert "librosa.effects.time_stretch" not in apply_speed_source def test_public_tts_wrappers_serialize_the_same_pcm16_grid_used_for_verification(): 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) } synthesize_source = ast.get_source_segment(source, functions["_synthesize"]) assemble_source = ast.get_source_segment( source, functions["_assemble_trajectory_audio"], ) assert synthesize_source is not None assert assemble_source is not None assert "return pcm16_verification_waveform(waveform)" in assemble_source assert "pcm16_audio_output" not in synthesize_source assert "return SR, waveform" in synthesize_source for wrapper_name in ("tts_speaker", "tts_reference"): wrapper_source = ast.get_source_segment(source, functions[wrapper_name]) assert wrapper_source is not None assert "return pcm16_audio_output(" in wrapper_source assert "*_synthesize(" in wrapper_source def test_space_applies_pinned_squim_to_local_joined_and_final_audio(): app_source = (ROOT / "app.py").read_text(encoding="utf-8") app_tree = ast.parse(app_source) app_functions = { node.name: node for node in app_tree.body if isinstance(node, ast.FunctionDef) } verify_source = ast.get_source_segment( app_source, app_functions["_verify_trajectory_audio"], ) runtime_source = (ROOT / "quality_runtime.py").read_text(encoding="utf-8") assert verify_source is not None assert 'QUALITY_MIN_SQUIM_STOI = 0.60' in app_source assert 'QUALITY_MIN_SQUIM_PESQ = 1.12' in app_source assert 'QUALITY_PREFERRED_MIN_SQUIM_STOI = 0.72' in app_source assert 'QUALITY_PREFERRED_MIN_SQUIM_PESQ = 1.20' in app_source assert 'QUALITY_PREFERRED_MIN_SPEAKER_SIMILARITY = 0.25' in app_source assert 'QUALITY_PREFERRED_MAX_BOUNDARY_SPEAKER_DROP = 0.05' in app_source assert 'QUALITY_FAST_EXACT_MIN_SPEAKER_SIMILARITY = 0.23' in app_source assert 'QUALITY_FAST_EXACT_MAX_BOUNDARY_SPEAKER_DROP = 0.08' in app_source assert "if not semantic_only and transcript:" in verify_source score_index = verify_source.index("squim_objective_evidence_from_audio(") observation_index = verify_source.index("CandidateObservation(", score_index) gate_index = verify_source.index("squim_gate_enabled=not semantic_only") assert score_index < observation_index < gate_index artifact_gate_index = verify_source.index( "verification = _apply_echo_smearing_gate(" ) assert gate_index < artifact_gate_index assert "if not semantic_only:" in verify_source[gate_index:artifact_gate_index] assert "tuple(prepared_waveforms)" in verify_source[artifact_gate_index:] assert "from audio_artifact_gate import (" in app_source assert "evaluate_echo_smearing," in app_source assert "except ValueError:" in verify_source[score_index:observation_index] assert "except RuntimeError:" not in verify_source[score_index:observation_index] assert ( '"2c54586fea83fb5eb5394d710038ee89f55cab7011a5bf730bebed4c8777e828"' in runtime_source ) hash_index = runtime_source.index("digest = str(hasher(weight_path)).casefold()") state_index = runtime_source.index( 'state_dict = loader(weight_path, map_location="cpu", weights_only=True)' ) assert hash_index < state_index assert "device=torch.device(\"cpu\")" in runtime_source synthesize_source = ast.get_source_segment( app_source, app_functions["_synthesize"], ) assert synthesize_source is not None for argument in ( "preferred_min_speaker_similarity=", "preferred_max_boundary_speaker_drop=", "preferred_min_squim_stoi=QUALITY_PREFERRED_MIN_SQUIM_STOI", "preferred_min_squim_pesq=QUALITY_PREFERRED_MIN_SQUIM_PESQ", "QUALITY_PREFERRED_SQUIM_MIN_DURATION_SECONDS", "single_exact_min_speaker_similarity=", "single_exact_max_boundary_speaker_drop=", ): assert argument in synthesize_source assert "QUALITY_PREFERRED_SQUIM_MIN_DURATION_SECONDS = 1.50" in app_source def test_whole_candidate_qualification_uses_the_exact_return_assembler_after_local_pass(): 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) } qualify_source = ast.get_source_segment( source, functions["_qualify_candidate_trajectory_audio"], ) synthesize_source = ast.get_source_segment(source, functions["_synthesize"]) assert qualify_source is not None assert synthesize_source is not None local_index = qualify_source.index("local_verification = _verify_trajectory_audio(") local_fail_index = qualify_source.index("if not local_verification.passed:") assemble_index = qualify_source.index("waveform = _assemble_trajectory_audio(") joined_index = qualify_source.index( "joined_verification = _verify_release_whole_audio(" ) assert local_index < local_fail_index < assemble_index < joined_index assert "BREEZE25_RELEASE_VERIFICATION_PROFILE" in qualify_source[joined_index:] assert "qualify_trajectory_with_joined_output(" in qualify_source[joined_index:] assert "waveform = _assemble_trajectory_audio(" in synthesize_source assert "cascade.trajectory" in synthesize_source assert "chunk_specs" in synthesize_source assert "require_verified_final_output(final_verification)" in synthesize_source def test_space_uses_one_cached_breeze25_transcriber_for_every_semantic_gate(): 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) } verify_source = ast.get_source_segment(source, functions["_verify_trajectory_audio"]) independent_source = ast.get_source_segment( source, functions["_verify_independent_whole_audio"], ) qualify_source = ast.get_source_segment( source, functions["_qualify_candidate_trajectory_audio"], ) sequence_source = ast.get_source_segment( source, functions["_verify_sequence_trajectory_audio"], ) refill_source = ast.get_source_segment( source, functions["_verify_refill_candidate_trajectory_audio"], ) network_local_source = ast.get_source_segment( source, functions["_verify_network_local_asr_intersection"], ) synthesize_source = ast.get_source_segment(source, functions["_synthesize"]) assert all( segment is not None for segment in ( verify_source, independent_source, qualify_source, sequence_source, refill_source, network_local_source, synthesize_source, ) ) assert "transcriber=transcribe_breeze25" in verify_source assert "waveform.ndim != 1" in verify_source assert "np.asarray(audio, dtype=np.float32).reshape(-1)" not in verify_source assert "speaker_gate_enabled=not semantic_only" in verify_source assert "transcriber=transcribe_breeze25" in independent_source assert "semantic_only=True" in independent_source assert "cache.verify(" in independent_source assert "BREEZE25_SEMANTIC_VERIFICATION_PROFILE" in independent_source assert "local_verification = _verify_trajectory_audio(" in refill_source assert "_verify_independent_whole_audio" not in refill_source assert "transcriber=transcribe_breeze25" in network_local_source assert "semantic_only=True" in network_local_source assert "network_fragment_proofs=independent_proof_rows" in network_local_source assert "intersect_local_semantic_verification(" in network_local_source assert "BREEZE25_TRANSCRIPT_PROFILE" in synthesize_source assert synthesize_source.count( "cached_breeze25_transcriber = _cached_request_transcriber(" ) == 1 assert "transcribe_breeze25," in synthesize_source assert "cached_turbo_transcriber" not in synthesize_source assert "cached_large_v3_transcriber" not in synthesize_source assert "transcribe_whisper" not in synthesize_source assert "transcribe_verification_whisper" not in synthesize_source assert synthesize_source.count( "release_transcriber=cached_breeze25_transcriber" ) == 3 assert synthesize_source.count( "independent_transcriber=cached_breeze25_transcriber" ) == 3 assert "transcriber=cached_breeze25_transcriber" in synthesize_source local_index = qualify_source.index("local_verification = _verify_trajectory_audio(") local_fail_index = qualify_source.index("if not local_verification.passed:") assemble_index = qualify_source.index("waveform = _assemble_trajectory_audio(") turbo_joined_index = qualify_source.index( "joined_verification = _verify_release_whole_audio(" ) turbo_fail_index = qualify_source.index("if not qualified.passed:") independent_index = qualify_source.index("_verify_independent_whole_audio(") semantic_projection_index = qualify_source.index( "semantic_qualified = qualify_trajectory_with_joined_output(" ) assert ( local_index < local_fail_index < assemble_index < turbo_joined_index < turbo_fail_index < independent_index < semantic_projection_index ) sequence_assemble = sequence_source.index("_assemble_trajectory_audio(") sequence_large_v3 = sequence_source.index( "assembled_verification = _verify_release_whole_audio(" ) sequence_large_v3_transcriber = sequence_source.index( "transcriber=release_transcriber", sequence_large_v3, ) sequence_independent = sequence_source.index( "independent_verification = _verify_independent_whole_audio(" ) sequence_intersection = sequence_source.index( "intersected = qualify_trajectory_with_joined_output(" ) assert ( sequence_assemble < sequence_large_v3 < sequence_large_v3_transcriber < sequence_independent < sequence_intersection ) cache_create = synthesize_source.index( "independent_cache = WholeWaveformVerificationCache()" ) cascade_index = synthesize_source.index( "cascade = run_coverage_adaptive_cascade(" ) final_assemble = synthesize_source.index( "waveform = _assemble_trajectory_audio(" ) final_turbo = synthesize_source.index( "final_verification = _verify_release_whole_audio(" ) final_turbo_require = synthesize_source.index( "require_verified_final_output(final_verification)" ) final_independent = synthesize_source.index( "independent_final_verification = _verify_independent_whole_audio(" ) final_independent_require = synthesize_source.index( "require_verified_final_output(independent_final_verification)" ) return_index = synthesize_source.index("return SR, waveform") assert ( cache_create < cascade_index < final_assemble < final_turbo < final_turbo_require < final_independent < final_independent_require < return_index ) assert synthesize_source.count("independent_cache,") >= 3 assert "except (RuntimeError, ValueError) as error:" in synthesize_source def test_space_disables_experimental_gradio_ssr(): source = (ROOT / "app.py").read_text(encoding="utf-8") assert ( "demo.queue(default_concurrency_limit=1).launch(ssr_mode=False)" in source ) def test_space_automatically_selects_zero_gpu_lease_from_text_length(): source = (ROOT / "app.py").read_text(encoding="utf-8") assert "INTERACTIVE_GPU_LEASE_SECONDS = 60" in source assert "LONGFORM_GPU_LEASE_SECONDS = 120" in source assert "AUTO_LONG_TEXT_CHARS = 80" in source assert ( "speaker_gpu = spaces.GPU(duration=_speaker_gpu_duration)" in source ) assert ( "reference_gpu = spaces.GPU(duration=_reference_gpu_duration)" in source ) assert source.count("@speaker_gpu") == 1 assert source.count("@reference_gpu") == 1 assert "def tts_longform(" not in source def test_network_fragment_relaxation_is_range_bound_and_local_only(): 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) } verify_source = ast.get_source_segment(source, functions["_verify_trajectory_audio"]) proof_source = ast.get_source_segment( source, functions["_network_fragment_proof_rows"], ) qualify_source = ast.get_source_segment( source, functions["_qualify_candidate_trajectory_audio"], ) refill_source = ast.get_source_segment( source, functions["_verify_refill_candidate_trajectory_audio"], ) independent_source = ast.get_source_segment( source, functions["_verify_independent_whole_audio"], ) sequence_source = ast.get_source_segment( source, functions["_verify_sequence_trajectory_audio"], ) synthesize_source = ast.get_source_segment(source, functions["_synthesize"]) assert all( segment is not None for segment in ( verify_source, proof_source, qualify_source, refill_source, independent_source, sequence_source, synthesize_source, ) ) assert "canonicalize_asr_network_fragments(" in verify_source assert "if fragment_evidence.passed" in verify_source assert "network-conditioned chunk lacks exact fragment proof" in proof_source assert "proof.span_index for proof in proofs" in proof_source assert "proof.full_spoken_proof for proof in proofs" in proof_source local_index = qualify_source.index("local_verification = _verify_trajectory_audio(") joined_index = qualify_source.index( "joined_verification = _verify_release_whole_audio(" ) assert "network_fragment_proofs=" in qualify_source[local_index:joined_index] assert "network_fragment_proofs=" not in qualify_source[joined_index:] assert "network_fragment_proofs=" in refill_source assert "network_fragment_proofs=" not in independent_source assert "network_fragment_proofs=" not in sequence_source final_index = synthesize_source.index( "final_verification = _verify_release_whole_audio(" ) assert "network_fragment_proofs=" not in synthesize_source[final_index:] assert "generation_context_by_seed" in synthesize_source assert "generation_chunk_specs(seed, candidate_chunks)" in synthesize_source def test_local_endpoint_relaxation_is_role_bound_and_whole_gates_stay_exact(): 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) } verify_source = ast.get_source_segment(source, functions["_verify_trajectory_audio"]) role_source = ast.get_source_segment(source, functions["_local_endpoint_role_rows"]) qualify_source = ast.get_source_segment( source, functions["_qualify_candidate_trajectory_audio"], ) refill_source = ast.get_source_segment( source, functions["_verify_refill_candidate_trajectory_audio"], ) sequence_source = ast.get_source_segment( source, functions["_verify_sequence_trajectory_audio"], ) synthesize_source = ast.get_source_segment(source, functions["_synthesize"]) assert all( segment is not None for segment in ( verify_source, role_source, qualify_source, refill_source, sequence_source, synthesize_source, ) ) assert "spec.source_start == 0" in role_source assert 'spec.boundary_after == "none"' in role_source assert "spec.text != chunk" in role_source assert "local_candidate_pool = not semantic_only and not release_speaker_gate" in ( verify_source ) assert '"max_prefix_cer": (1.0 / 6.0 if local_candidate_pool else 0.0)' in ( verify_source ) assert '"max_suffix_cer": (1.0 / 6.0 if local_candidate_pool else 0.0)' in ( verify_source ) assert '"max_prefix_deletions": (0 if local_candidate_pool else None)' in ( verify_source ) assert '"max_suffix_deletions": (0 if local_candidate_pool else None)' in ( verify_source ) assert "candidate_gate_kwargs_by_index=indexed_gate_kwargs" in verify_source local_index = qualify_source.index("local_verification = _verify_trajectory_audio(") joined_index = qualify_source.index( "joined_verification = _verify_release_whole_audio(" ) assert "local_endpoint_roles=" in qualify_source[local_index:joined_index] assert "local_endpoint_roles=" not in qualify_source[joined_index:] assert "local_endpoint_roles=" in refill_source assert "local_endpoint_roles=" not in sequence_source final_index = synthesize_source.index( "final_verification = _verify_release_whole_audio(" ) assert "local_endpoint_roles=" not in synthesize_source[final_index:] assert "generation_chunk_specs(seed, candidate_chunks)" in synthesize_source def test_naturalized_url_provenance_is_revalidated_without_policy_override(): production_source = (ROOT / "production.py").read_text(encoding="utf-8") app_source = (ROOT / "app.py").read_text(encoding="utf-8") readme = (ROOT / "README.md").read_text(encoding="utf-8") production_tree = ast.parse(production_source) app_tree = ast.parse(app_source) production_functions = { node.name: node for node in production_tree.body if isinstance(node, ast.FunctionDef) } app_functions = { node.name: node for node in app_tree.body if isinstance(node, ast.FunctionDef) } inverse_source = ast.get_source_segment( production_source, production_functions["inverse_network_url_rendering"], ) planner_source = ast.get_source_segment( production_source, production_functions["plan_generation_chunks"], ) local_source = ast.get_source_segment( production_source, production_functions["canonicalize_asr_network_fragments"], ) runtime_source = ast.get_source_segment( app_source, app_functions["_network_fragment_proof_rows"], ) generate_source = ast.get_source_segment( app_source, app_functions["_generate_chunk"], ) assert all( segment is not None for segment in ( inverse_source, planner_source, local_source, runtime_source, generate_source, ) ) assert "proof != expected" in inverse_source assert "_naturalized_url_proof_from_spoken(full_proof)" in planner_source assert "_naturalized_url_proof_from_spoken(full_proof)" in local_source assert "proof.raw_identifier or proof.url_rendering_proof is not None" in ( planner_source ) assert "proof.raw_identifier or proof.url_rendering_proof is not None" in ( local_source ) assert "contains_naturalized_url_spoken_form(" in runtime_source assert "proof.url_rendering_proof != fresh_rendering" in runtime_source assert "proof.raw_identifier or proof.url_rendering_proof is not None" in ( runtime_source ) assert "generation_cps = (" in generate_source assert "policy.ascii_cps" in generate_source assert "COMPLETION_HEADROOM_GENERATION_POLICY" not in generate_source assert "不可切 component" in readme assert "Email contract" in readme def test_space_wires_bounded_k_best_paths_to_exact_assembled_whole_gate(): source = (ROOT / "app.py").read_text(encoding="utf-8") assert "sequence_final_verifier=lambda sequence_result, candidate_chunks:" in source assert "_verify_sequence_trajectory_audio(" in source assert "QUALITY_MAX_SEQUENCE_PATHS = 3" in source assert "max_sequence_paths=QUALITY_MAX_SEQUENCE_PATHS" in source assert "waveform = _assemble_trajectory_audio(" in source assert "QUALITY_FINAL_ASR_MAX_NEW_TOKENS" in source assert "sequence_rank={cascade.sequence_path_rank}" in source assert "sequence_paths_checked={cascade.sequence_paths_checked}" in source assert "cer={comparison.cer:.6f}" in source assert "prefix_cer={comparison.prefix_cer:.6f}" in source assert "suffix_cer={comparison.suffix_cer:.6f}" in source assert "tail_units={comparison.extra_tail_units}" in source def test_space_locks_validated_nfe_and_bounds_total_generation_work(): source = (ROOT / "app.py").read_text(encoding="utf-8") quality_source = (ROOT / "quality_runtime.py").read_text(encoding="utf-8") readme = (ROOT / "README.md").read_text(encoding="utf-8") assert "QUALITY_MAX_GENERATED_CHUNKS = 32" in source assert "QUALITY_INTERACTIVE_MAX_GENERATED_CHUNKS = 10" in source assert "QUALITY_MAX_GENERATED_TEXT_UNITS = 800" in source assert "EMAIL_MAIL_FALLBACK_CANDIDATE_ORDINALS = frozenset((2, 5, 6))" in source assert "NETWORK_REQUEST_ORDINARY_MAX_UNITS = 36" in source assert "ordinary_max_units=NETWORK_REQUEST_ORDINARY_MAX_UNITS" in source assert "MIN_PACE_SPEED = 0.90" in source assert "PACE_ONLY_FALLBACK_MIN_SPEED = MIN_PACE_SPEED" in source assert "PACE_ONLY_FALLBACK_MIN_UNITS = 30" in source assert "if observed_cps > QUALITY_MAX_PACE_CPS:" in source assert "final_audio = _apply_speed(" in source assert "fallback_speed = final_speed * fallback_residual" in source assert "max_generated_chunks=request_generated_chunk_limit" in source assert "normalized_units <= QUALITY_MEDIUM_TEXT_MAX_UNITS" in source assert "carrier_request=generation_audit is not None" in source assert "max_generated_text_units=QUALITY_MAX_GENERATED_TEXT_UNITS" in source assert "candidate_generation_text_transform=(" in source assert "_verify_refill_candidate_trajectory_audio(" in source assert "def run_coverage_adaptive_cascade(" in quality_source assert "proposed_generation_chunks = generation_chunks(" in quality_source assert "generated_chunks += 1" in quality_source assert "generated_units += refill_units" in quality_source assert "select_culprit_diverse_candidate_sequences(" in quality_source assert "requested_steps != DEFAULT_STEPS" in source assert "interactive=False" in source assert "NFE steps（已驗證固定值）" in source assert 'label="後處理語速（已驗證固定值）"' in source assert "float(speed) != 1.0" in source assert "最多 3 條 culprit-diverse 完整路徑" in readme assert "不再為了取得某一段替代候選而重生整篇" in readme assert "boundary-only relaxation" in readme def test_pace_only_fallback_guard_starts_at_30_ordinary_units_not_29(): app_path = ROOT / "app.py" constants = _literal_constants(app_path) tree = ast.parse(app_path.read_text(encoding="utf-8")) generate_chunk = next( node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == "_generate_chunk" ) guarded_ifs = [ node for node in ast.walk(generate_chunk) if isinstance(node, ast.If) and any( isinstance(name, ast.Name) and name.id == "PACE_ONLY_FALLBACK_MIN_UNITS" for name in ast.walk(node.test) ) ] assert constants["PACE_ONLY_FALLBACK_MIN_UNITS"] == 30 assert len(guarded_ifs) == 1 expression = ast.Expression(body=guarded_ifs[0].test) ast.fix_missing_locations(expression) guard = compile(expression, str(app_path), "eval") def eligible(units, *, network_conditioned=False): return eval( guard, { "PACE_ONLY_FALLBACK_MIN_UNITS": constants[ "PACE_ONLY_FALLBACK_MIN_UNITS" ], "count_speech_units": lambda _text: units, "network_conditioned": network_conditioned, "speech_units": units, "text": "測試", }, ) assert eligible(29) is False assert eligible(30) is True assert eligible(30, network_conditioned=True) is False def test_assembled_waveform_uses_250ms_ordinary_and_350ms_network_request_pause(): assemble = _isolated_assemble_trajectory_audio() chunks = ("第一段，", "第二段") trajectory = ( np.ones(100, dtype=np.float32), np.ones(100, dtype=np.float32), ) ordinary = assemble(trajectory, chunks, 1.0) assert ordinary.shape == (450,) np.testing.assert_array_equal(ordinary[100:350], np.zeros(250, dtype=np.float32)) first_end = len(chunks[0]) network_specs = ( GenerationChunkSpec( text=chunks[0], source_start=0, source_end=first_end, boundary_after="semantic", ), GenerationChunkSpec( text=chunks[1], source_start=first_end, source_end=first_end + len(chunks[1]), network_span_indices=(0,), boundary_after="none", ), ) network_request = assemble(trajectory, chunks, 1.0, network_specs) assert network_request.shape == (550,) np.testing.assert_array_equal( network_request[100:450], np.zeros(350, dtype=np.float32), ) def test_assembled_h06_email_separator_uses_400ms_proven_network_pause(): assemble = _isolated_assemble_trajectory_audio() raw = "若要更換導覽場次，請寄信到 tour.help@islandmuseum.tw。" specs = plan_generation_chunks( raw, normalize_spoken_forms(raw), split_atomic_email_at_separator=True, ) chunks = tuple(spec.text for spec in specs) trajectory = tuple( np.ones(100, dtype=np.float32) for _ in chunks ) output = assemble(trajectory, chunks, 1.0, specs) assert tuple(spec.boundary_after for spec in specs) == ( "network_internal", "none", ) assert output.shape == (600,) np.testing.assert_array_equal( output[100:500], np.zeros(400, dtype=np.float32), ) def test_space_locks_waveform_only_bounded_breeze25_segmentation(): source = (ROOT / "quality_runtime.py").read_text(encoding="utf-8") assert "BREEZE25_MAX_SEGMENT_SECONDS = 28.0" in source assert "BREEZE25_HARD_MAX_SEGMENT_SECONDS = 30.0" in source assert "BREEZE25_MIN_SEGMENT_SECONDS = 1.25" in source assert "BREEZE25_MIN_PAUSE_SECONDS = 0.25" in source assert "BREEZE25_MAX_VERIFICATION_SEGMENTS = 12" in source assert "GLYPH_HYBRID_MAX_VERIFICATION_SEGMENTS = 160" in source assert "BREEZE25_MAX_MICROBATCH_SEGMENTS = 6" in source