app.py

"""BlueMagpie-TTS Gradio demo using a latency-first inference profile."""

from __future__ import annotations

from dataclasses import dataclass, replace
import hashlib
import inspect
import json
import os
from pathlib import Path
import secrets
import tempfile
import threading
from types import MappingProxyType

# The public demo intentionally favors throughput and first-audio latency over
# cross-process bit identity.  Let cuDNN select the fastest kernels and allow
# TensorFloat-32 on hardware that supports it.
DETERMINISTIC_ALGORITHMS = False
CUDNN_BENCHMARK = True
CUDNN_DETERMINISTIC = False
CUDA_MATMUL_ALLOW_TF32 = True
CUDNN_ALLOW_TF32 = True
os.environ["MAMBA_DETERMINISTIC"] = "0"

INTERACTIVE_GPU_LEASE_SECONDS = 60
LONGFORM_GPU_LEASE_SECONDS = 120
AUTO_LONG_TEXT_CHARS = 80


def _gpu_lease_seconds(text: object) -> int:
    """Reserve the long-form lease only when automatic chunking needs it."""

    return (
        LONGFORM_GPU_LEASE_SECONDS
        if len(str(text)) > AUTO_LONG_TEXT_CHARS
        else INTERACTIVE_GPU_LEASE_SECONDS
    )


def _speaker_gpu_duration(text, speaker, cfg, steps, speed, seed) -> int:
    del speaker, cfg, steps, speed, seed
    return _gpu_lease_seconds(text)


def _reference_gpu_duration(text, reference_wav, cfg, steps, speed, seed) -> int:
    del reference_wav, cfg, steps, speed, seed
    return _gpu_lease_seconds(text)

try:
    import spaces

    speaker_gpu = spaces.GPU(duration=_speaker_gpu_duration)
    reference_gpu = spaces.GPU(duration=_reference_gpu_duration)
    USING_ZERO_GPU = True
except ImportError:
    def speaker_gpu(function):
        return function

    reference_gpu = speaker_gpu
    USING_ZERO_GPU = False

import gradio as gr
import numpy as np
import soundfile as sf
import torch
from huggingface_hub import snapshot_download
from transformers import PreTrainedTokenizerFast

torch.use_deterministic_algorithms(DETERMINISTIC_ALGORITHMS)
torch.backends.cudnn.benchmark = CUDNN_BENCHMARK
torch.backends.cudnn.deterministic = CUDNN_DETERMINISTIC
torch.backends.cuda.matmul.allow_tf32 = CUDA_MATMUL_ALLOW_TF32
torch.backends.cudnn.allow_tf32 = CUDNN_ALLOW_TF32
torch.set_float32_matmul_precision("high")

from audio_wsola import wsola_time_stretch
from audio_artifact_gate import (
    DEFAULT_ECHO_SMEARING_GATE_CONFIG,
    evaluate_echo_smearing,
)
from bluemagpie import BlueMagpieModel
from glyph_ascii_v1 import GLYPH_ASCII_V1_SYMBOLS
from glyph_assets import (
    MAX_GENERATED_CHUNK_BUDGET as GLYPH_MAX_GENERATED_CHUNK_BUDGET,
    MAX_GENERATED_TEXT_UNIT_BUDGET as GLYPH_MAX_GENERATED_TEXT_UNIT_BUDGET,
    AssetSegment,
    CarrierSegment,
    GenerationAttemptRecord,
    GlyphAssetBundle,
    GlyphAssetError,
    SemanticPlanCommitment,
    assemble_hybrid_pcm16,
    compute_hybrid_plan_sha256,
    compute_ordered_attempt_ledger_sha256,
    load_glyph_asset_bundle,
    verify_hybrid_assembly,
)
from glyph_hybrid_plan import (
    GlyphHybridPlanError,
    HybridRenderPlan,
    build_glyph_hybrid_plan,
    glyph_hybrid_plan_sha256,
    inverse_glyph_hybrid_plan,
)
from glyph_hybrid_adapter import (
    GlyphAdapterRuntime,
    HardenedGlyphHybridAdapter,
)
from glyph_hybrid_evidence import GatePolicy as GlyphEvidenceGatePolicy
from latency_timing import (
    latency_stage,
    timed_latency_request,
    timed_latency_stage,
)
from production import (
    StopHysteresisController,
    GenerationChunkSpec,
    NetworkFragmentProof,
    active_pace_correction_speed,
    apply_loudness_floor,
    canonicalize_asr_network_fragments,
    coalesce_text_chunks,
    count_network_endpoint_duration_units,
    contains_naturalized_url_spoken_form,
    contains_network_identifier,
    count_speech_units,
    effective_generation_cfg,
    email_domain_mail_generation_variant,
    endpoint_generation_plan,
    estimate_step_seconds,
    fade_variable_internal_edges,
    finish_audio,
    join_audio_chunks_variable,
    inverse_network_url_rendering,
    match_chunk_rms,
    network_identifier_has_ambiguous_iri,
    network_protected_spoken_spans,
    naturalized_url_rendering_proof,
    normalize_spoken_forms,
    punctuation_pause_seconds,
    pcm16_audio_output,
    pcm16_verification_waveform,
    plan_generation_chunks,
    select_generation_cps,
    set_generation_seed,
    split_quality_text_for_tts,
    strip_unspoken_cjk_quotes_for_generation,
    target_pace_speed,
)
from quality_runtime import (
    BASE_GENERATION_POLICY,
    BREEZE25_MAX_VERIFICATION_SEGMENTS,
    BREEZE25_MODEL_ID,
    BREEZE25_REVISION,
    BREEZE25_WEIGHT_SHA256,
    GLYPH_HYBRID_MAX_VERIFICATION_SEGMENTS as QUALITY_GLYPH_HYBRID_MAX_VERIFICATION_SEGMENTS,
    CandidateGenerationContext,
    CandidateGenerationEvidence,
    CandidateVerification,
    CandidateObservation,
    ChunkCandidateArtifact,
    FinalOutputRejectedError,
    GenerationPolicy,
    LocalIndependentGateEvidence,
    NoQualifiedCandidateError,
    RELEASE_SPEAKER_TRIGGER_SECONDS,
    SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP,
    WholeWaveformTranscriptCache,
    WholeWaveformVerificationCache,
    active_voiced_duration_seconds,
    candidate_gate_evidence,
    active_audio_rms_db,
    active_audio_median_f0_hz,
    endpoint_tail_energy_ratio,
    format_cascade_evidence_log,
    generation_cfg_for_candidate_offset,
    generation_policy_for_candidate_offset,
    intersect_local_semantic_verification,
    local_candidate_has_coverage_eligibility,
    prepare_candidate_audio,
    qualify_trajectory_with_joined_output,
    require_verified_final_output,
    release_speaker_evidence_from_audio,
    release_speaker_measurement_required,
    resolve_request_seed,
    run_coverage_adaptive_cascade,
    speaker_evidence_from_audio,
    squim_objective_evidence_from_audio,
    transcribe_breeze25,
    trajectory_gate_evidence,
    verify_candidate,
    verify_trajectory,
)

DEVICE = "cuda" if USING_ZERO_GPU or torch.cuda.is_available() else "cpu"


REPO_ID = "OpenFormosa/BlueMagpie-TTS"
MODEL_REVISION = "6f7cab914a1e27c56b504ec663c0144dc25cc0a3"
ECAPA_REPO_ID = "speechbrain/spkrec-ecapa-voxceleb"
ECAPA_REVISION = "0f99f2d0ebe89ac095bcc5903c4dd8f72b367286"
DEFAULT_CFG = 3.0
REFERENCE_CFG = 2.0
DEFAULT_STEPS = 10
TARGET_CPS = 4.0
ACTIVE_PACE_TARGET_CPS = 4.00
CLOSED_LOOP_ACTIVE_PACE_TARGET_CPS = 3.95
MIXED_CFG_PRIMARY = 3.0
MIXED_CFG_ALTERNATE = 2.0
MIXED_CFG_SCHEDULE = "row_ordinal_zero_and_even_primary_odd_alternate"
MIN_ENDPOINT_CUE_UNITS = 6
SHORT_TEXT_CFG_MIN = 3.0
SHORT_TEXT_CFG_UNITS = 6
NETWORK_TEXT_CFG_MIN = 3.0
CHUNK_CHARS = AUTO_LONG_TEXT_CHARS
ONSET_CLAUSE_SEARCH_CHARS = 0
MIN_CHUNK_CHARS = 12
QUALITY_MEDIUM_TEXT_MAX_UNITS = 48
QUALITY_MEDIUM_CHUNK_UNITS = 24
GLYPH_CARRIER_MEDIUM_CHUNK_UNITS = 32
CROSSFADE_MS = 80.0
CHUNK_EDGE_FADE_MS = 80.0
CHUNK_RMS_MATCH_DB = 4.0
FINAL_ENDPOINT_FADE_MS = 3.0
NETWORK_GENERATION_MIN_UNITS = 8
NETWORK_GENERATION_TARGET_UNITS = 32
NETWORK_GENERATION_MAX_UNITS = 36
NETWORK_REQUEST_ORDINARY_MAX_UNITS = 36
NETWORK_SPLIT_ATOMIC_EMAIL_AT_SEPARATOR = True
NETWORK_INTERNAL_FADE_MS = 5.0
NETWORK_INTERNAL_SILENCE_MS = 400.0
SEMANTIC_CHUNK_MIN_SILENCE_MS = 250.0
NETWORK_REQUEST_SEMANTIC_CHUNK_MIN_SILENCE_MS = 350.0
STOP_THRESHOLD = 0.50
STOP_LATE_THRESHOLD = 0.05
STOP_LATE_START_RATIO = 0.75
STOP_LATE_FULL_RATIO = 0.95
STOP_CONSECUTIVE = 1
MIN_PACE_SPEED = 0.90
EXTREME_SHORT_MAX_UNITS = 2
EXTREME_SHORT_MAX_SPEED = 0.95
PACE_ONLY_FALLBACK_MIN_SPEED = MIN_PACE_SPEED
PACE_ONLY_FALLBACK_MIN_UNITS = 30
MAX_TEXT_CHARS = 360
QUALITY_MAX_GENERATED_CHUNKS = 32
QUALITY_INTERACTIVE_MAX_GENERATED_CHUNKS = 10
QUALITY_MAX_GENERATED_TEXT_UNITS = 800
EMAIL_MAIL_FALLBACK_CANDIDATE_ORDINALS = frozenset((2, 5, 6))
BASE_CHUNK_TEXT_VARIANT = "base"
EMAIL_DOMAIN_MAIL_CHUNK_TEXT_VARIANT = "email_domain_mail_v1"
QUALITY_FINAL_ASR_MAX_NEW_TOKENS = 440
QUALITY_MAX_CER = 0.20
QUALITY_MAX_PACE_CPS = 4.85
QUALITY_PREFLIGHT_PACE_MARGIN_CPS = 0.05
QUALITY_PREFIX_SUFFIX_UNITS = 6
QUALITY_MIN_SPEAKER_SIMILARITY = 0.10
QUALITY_MAX_BOUNDARY_SPEAKER_DROP = 0.10
QUALITY_RELEASE_MIN_SPEAKER_SIMILARITY = 0.105
QUALITY_RELEASE_MAX_BOUNDARY_SPEAKER_DROP = 0.095
QUALITY_MIN_SQUIM_STOI = 0.60
QUALITY_MIN_SQUIM_PESQ = 1.12
QUALITY_PREFERRED_MIN_SQUIM_STOI = 0.72
QUALITY_PREFERRED_MIN_SQUIM_PESQ = 1.20
QUALITY_PREFERRED_SQUIM_MIN_DURATION_SECONDS = 1.50
QUALITY_PREFERRED_MIN_SPEAKER_SIMILARITY = 0.25
QUALITY_PREFERRED_MAX_BOUNDARY_SPEAKER_DROP = 0.05
# A single-chunk waveform that has passed both Breeze25 semantic projections
# and every hard release gate may stop the cascade at this exact-only tier.
# Local chunk proxies and multi-chunk eager paths remain on the stricter tier.
QUALITY_FAST_EXACT_MIN_SPEAKER_SIMILARITY = 0.23
QUALITY_FAST_EXACT_MAX_BOUNDARY_SPEAKER_DROP = 0.08
QUALITY_MAX_SEQUENCE_PATHS = 3
SHORT_AUDIO_SPEAKER_GATE_SECONDS = 1.50
SPEAKER_GENERATION_SEED = 56_789
SPEAKER_B_LABEL = "內建語者 B"
SPEAKER_B_PROJECTOR_SCALE = 1.125
SPEAKER_B_EMBEDDING_SHA256 = (
    "e9556e14723c140985a104c1659d1ff8a5078d2fa28ce2fb756f04906641a8a7"
)
REFERENCE_GENERATION_SEED = 1_337
REFERENCE_AUDIO_SECONDS = 6.0
GLYPH_ASSET_MANIFEST_RELATIVE_PATH = "glyph_assets_v1/manifest.json"
# The asset builder replaces this unavailable sentinel with the exact frozen
# manifest SHA-256 before a candidate can expose the V5 profile.
GLYPH_ASSET_MANIFEST_SHA256 = ""
GLYPH_SPEAKER_ANCHOR_SHA256 = (
    "e33e4cb6a741d4d1237aa4ff557f1e663d0a6427dccda1f51e83bc149d4188ca"
)
GLYPH_NETWORK_MARKERS = ("http://", "https://", "ftp://", "www.", "@", "://")
GLYPH_FINAL_MAX_CER = 0.08
GLYPH_EVIDENCE_LOG_PREFIX = "[BlueMagpie][glyph-hybrid-evidence] "
GLYPH_HYBRID_MAX_VERIFICATION_SEGMENTS = 160
if (
    GLYPH_HYBRID_MAX_VERIFICATION_SEGMENTS
    != QUALITY_GLYPH_HYBRID_MAX_VERIFICATION_SEGMENTS
):
    raise RuntimeError("glyph-hybrid Breeze25 segment caps disagree")
BREEZE25_SEMANTIC_VERIFICATION_PROFILE = (
    f"{BREEZE25_MODEL_ID}@{BREEZE25_REVISION}:"
    f"weights{BREEZE25_WEIGHT_SHA256}:"
    f"greedy:zh:short{QUALITY_PREFIX_SUFFIX_UNITS}:cer{QUALITY_MAX_CER}:"
    f"prefix{QUALITY_PREFIX_SUFFIX_UNITS}=0:"
    f"suffix{QUALITY_PREFIX_SUFFIX_UNITS}=0:tail=0:"
    f"max_tokens{QUALITY_FINAL_ASR_MAX_NEW_TOKENS}"
)
BREEZE25_TRANSCRIPT_PROFILE = (
    f"{BREEZE25_MODEL_ID}@{BREEZE25_REVISION}:"
    f"weights{BREEZE25_WEIGHT_SHA256}:greedy:zh:transcribe:"
    f"max_tokens{QUALITY_FINAL_ASR_MAX_NEW_TOKENS}:"
    f"segments{BREEZE25_MAX_VERIFICATION_SEGMENTS}"
)
BREEZE25_RELEASE_VERIFICATION_PROFILE = (
    f"{BREEZE25_TRANSCRIPT_PROFILE}:release-speaker-squim-echo-wsola-v2:"
    f"pace{QUALITY_MAX_PACE_CPS}"
)
BREEZE25_SNAPSHOT_ALLOW_PATTERNS = (
    "added_tokens.json",
    "config.json",
    "generation_config.json",
    "merges.txt",
    "model.safetensors",
    "normalizer.json",
    "preprocessor_config.json",
    "special_tokens_map.json",
    "tokenizer.json",
    "tokenizer_config.json",
    "vocab.json",
)


print(f"[BlueMagpie] downloading model from {REPO_ID}@{MODEL_REVISION} ...")
MODEL_DIR = snapshot_download(REPO_ID, revision=MODEL_REVISION)
# Optional reference-speaker conditioning is loaded on first use.  The former
# Breeze/SQUIM verification stack is deliberately absent from the hot path.
ECAPA_DIR: str | None = None
BREEZE25_ASR_DIR: str | None = None
tokenizer = PreTrainedTokenizerFast(tokenizer_file=os.path.join(MODEL_DIR, "tokenizer.json"))
print(f"[BlueMagpie] loading model on device={DEVICE} ...")
model = BlueMagpieModel.from_local(MODEL_DIR, tokenizer=tokenizer, training=False, device=DEVICE)
SR = int(model.sample_rate)
STEP_SECONDS = estimate_step_seconds(model, SR)


METADATA: dict = {}
try:
    with open(os.path.join(MODEL_DIR, "release_metadata.json"), encoding="utf-8") as handle:
        METADATA = json.load(handle)
except (OSError, ValueError) as error:
    print(f"[BlueMagpie] release metadata unavailable: {error}")
CHECKPOINT = str(METADATA.get("checkpoint", "release"))


def _load_speakers() -> tuple[dict[str, torch.Tensor], str]:
    path = os.path.join(MODEL_DIR, "checkpoints", "speaker_centroids.pt")
    if not os.path.exists(path):
        raise RuntimeError("speaker_centroids.pt is missing from the model release")
    table = torch.load(path, map_location="cpu", weights_only=True)
    centroids = table["centroids"].detach().float().cpu().clone()
    embedding_path = os.path.join(
        MODEL_DIR,
        "checkpoints",
        "speaker_b_embedding.pt",
    )
    if not os.path.exists(embedding_path):
        raise RuntimeError("speaker_b_embedding.pt is missing from the model release")
    with open(embedding_path, "rb") as handle:
        embedding_sha256 = hashlib.sha256(handle.read()).hexdigest()
    if embedding_sha256 != SPEAKER_B_EMBEDDING_SHA256:
        raise RuntimeError("speaker_b_embedding.pt has an unexpected SHA-256")
    embedding_payload = torch.load(
        embedding_path,
        map_location="cpu",
        weights_only=True,
    )
    embedding = embedding_payload.get("embedding")
    if (
        embedding_payload.get("speaker_id") != "female_voice"
        or not isinstance(embedding, torch.Tensor)
        or centroids.ndim != 2
        or centroids.shape[0] < 2
        or embedding.shape != centroids[1].shape
        or not torch.isfinite(embedding).all()
        or float(torch.linalg.vector_norm(embedding.float())) <= 0.0
        or embedding_payload.get("generation_seed") != SPEAKER_GENERATION_SEED
        or embedding_payload.get("speaker_projector_scale")
        != SPEAKER_B_PROJECTOR_SCALE
    ):
        raise RuntimeError("speaker_b_embedding.pt has an invalid contract")
    centroids[1] = torch.nn.functional.normalize(embedding.float(), dim=0)
    labels = {f"內建語者 {chr(65 + index)}": centroid for index, centroid in enumerate(centroids)}

    default_index = 0
    return labels, f"內建語者 {chr(65 + default_index)}"


SPEAKERS, DEFAULT_SPEAKER = _load_speakers()
DEFAULT_CENTROID = SPEAKERS[DEFAULT_SPEAKER]


@dataclass(frozen=True)
class _GlyphProfileState:
    bundle: GlyphAssetBundle | None
    asset_entry_sha256_by_asset_id: object
    unavailable_reason: str | None

    @property
    def available(self) -> bool:
        return (
            self.bundle is not None
            and self.unavailable_reason is None
            and isinstance(
                self.asset_entry_sha256_by_asset_id,
                MappingProxyType,
            )
        )


def _canonical_json_sha256(value: object) -> str:
    encoded = json.dumps(
        value,
        ensure_ascii=True,
        separators=(",", ":"),
        sort_keys=True,
    ).encode("ascii")
    return hashlib.sha256(encoded).hexdigest()


def _glyph_asset_entry_sha256(asset) -> str:
    """Bind the exact schema-v7 trusted-asset fields used by the evaluator."""

    return _canonical_json_sha256(
        {
            "asset_id": asset.asset_id,
            "raw_text": asset.raw_byte,
            "canonical_text": asset.canonical_token,
            "pcm_sha256": hashlib.sha256(asset.pcm16_le).hexdigest(),
            "pcm_sample_count": len(asset.pcm16_le) // 2,
        }
    )


def _load_glyph_profile_state() -> _GlyphProfileState:
    def unavailable(reason: str) -> _GlyphProfileState:
        return _GlyphProfileState(
            bundle=None,
            asset_entry_sha256_by_asset_id=MappingProxyType({}),
            unavailable_reason=reason,
        )

    expected_manifest = GLYPH_ASSET_MANIFEST_SHA256
    if (
        type(expected_manifest) is not str
        or len(expected_manifest) != 64
        or any(character not in "0123456789abcdef" for character in expected_manifest)
    ):
        return unavailable("manifest_sha256_not_frozen")
    manifest_path = (
        Path(__file__).resolve().parent / GLYPH_ASSET_MANIFEST_RELATIVE_PATH
    )
    try:
        bundle = load_glyph_asset_bundle(
            manifest_path,
            expected_manifest_sha256=expected_manifest,
        )
        centroid = (
            torch.as_tensor(DEFAULT_CENTROID)
            .detach()
            .cpu()
            .to(dtype=torch.float32)
            .contiguous()
            .numpy()
        )
        centroid_sha256 = hashlib.sha256(
            np.asarray(centroid, dtype="<f4").tobytes(order="C")
        ).hexdigest()
        if (
            DEFAULT_SPEAKER != "內建語者 B"
            or SPEAKERS.get(DEFAULT_SPEAKER) is not DEFAULT_CENTROID
            or centroid_sha256 != GLYPH_SPEAKER_ANCHOR_SHA256
            or bundle.speaker_anchor_sha256 != GLYPH_SPEAKER_ANCHOR_SHA256
            or bundle.manifest_sha256 != expected_manifest
            or len(bundle.assets) != len(GLYPH_ASCII_V1_SYMBOLS)
        ):
            return unavailable("speaker_or_bundle_profile_mismatch")
        entries: dict[str, str] = {}
        for symbol in GLYPH_ASCII_V1_SYMBOLS:
            raw_text = chr(symbol.raw_byte)
            asset = bundle.asset_by_raw_byte(raw_text)
            if (
                asset.asset_id != symbol.asset_id
                or asset.canonical_token != symbol.canonical_token
            ):
                return unavailable("renderer_asset_grammar_mismatch")
            entries[symbol.asset_id] = _glyph_asset_entry_sha256(asset)
        if len(entries) != len(GLYPH_ASCII_V1_SYMBOLS):
            return unavailable("renderer_asset_set_mismatch")
        return _GlyphProfileState(
            bundle=bundle,
            asset_entry_sha256_by_asset_id=MappingProxyType(entries),
            unavailable_reason=None,
        )
    except (GlyphAssetError, OSError, RuntimeError, TypeError, ValueError):
        return unavailable("manifest_or_asset_integrity_failure")


_GLYPH_PROFILE_STATE = _GlyphProfileState(
    bundle=None,
    asset_entry_sha256_by_asset_id=MappingProxyType({}),
    unavailable_reason="disabled_in_latency_first_mode",
)
_GLYPH_HYBRID_EVIDENCE_ADAPTER = None
print(
    "[BlueMagpie] glyph hybrid profile "
    + (
        "available"
        if _GLYPH_PROFILE_STATE.available
        else f"unavailable reason={_GLYPH_PROFILE_STATE.unavailable_reason}"
    )
)


# The pinned public package predates native stop hysteresis. Wrap only that
# version; newer packages receive the same policy through native arguments.
_GENERATE_PARAMETERS = set(inspect.signature(model._generate).parameters)
_NATIVE_STOP_POLICY = {"stop_threshold", "stop_consecutive"}.issubset(_GENERATE_PARAMETERS)
_STOP_CONTROLLER: StopHysteresisController | None = None
if not _NATIVE_STOP_POLICY:
    _STOP_CONTROLLER = StopHysteresisController(
        model.stop_head,
        threshold=STOP_THRESHOLD,
        late_threshold=STOP_LATE_THRESHOLD,
        consecutive=STOP_CONSECUTIVE,
        late_start_ratio=STOP_LATE_START_RATIO,
        late_full_ratio=STOP_LATE_FULL_RATIO,
    )
    model.stop_head = _STOP_CONTROLLER

_GENERATION_LOCK = threading.Lock()
_ECAPA_ENCODER = None
_ECAPA_LOCK = threading.Lock()
print(
    f"[BlueMagpie] ready checkpoint={CHECKPOINT} sample_rate={SR} "
    f"step_seconds={STEP_SECONDS} native_stop_policy={_NATIVE_STOP_POLICY}"
)


def _get_ecapa_encoder():
    global _ECAPA_ENCODER, ECAPA_DIR
    with _ECAPA_LOCK:
        if _ECAPA_ENCODER is None:
            with latency_stage("ecapa"):
                if ECAPA_DIR is None:
                    ECAPA_DIR = snapshot_download(
                        ECAPA_REPO_ID,
                        revision=ECAPA_REVISION,
                    )
                import torchaudio

                # SpeechBrain 1.0.3 still probes this API during import, while the
                # ZeroGPU torchaudio build has removed it. Reference loading is
                # handled outside torchaudio, so an empty compatibility result is correct.
                if not hasattr(torchaudio, "list_audio_backends"):
                    torchaudio.list_audio_backends = lambda: []
                from speechbrain.inference.speaker import EncoderClassifier

                cache_dir = os.path.join(
                    os.environ.get("HF_HOME", "/tmp"),
                    "speechbrain",
                    "ecapa",
                )
                _ECAPA_ENCODER = EncoderClassifier.from_hparams(
                    source=ECAPA_DIR,
                    # The upstream hyperparams otherwise points back to the repo
                    # and SpeechBrain 1.0.3 uses a removed Hub keyword. Keep every
                    # weight fetch inside the already pinned local snapshot.
                    overrides={"pretrained_path": ECAPA_DIR},
                    savedir=cache_dir,
                    run_opts={"device": "cpu"},
                )
    return _ECAPA_ENCODER


def _preload_release_cpu_runtime() -> None:
    """Move deterministic CPU verifier loads and first forwards out of requests."""

    print("[BlueMagpie] preloading release CPU verifiers ...")
    encoder = _get_ecapa_encoder()
    if encoder is None:
        raise RuntimeError("release speaker verifier preload failed")
    sample_count = int(round(6.0 * SR))
    timeline = np.arange(sample_count, dtype=np.float64) / float(SR)
    warmup_waveform = np.asarray(
        0.08 * np.sin(2.0 * np.pi * 180.0 * timeline)
        + 0.04 * np.sin(2.0 * np.pi * 360.0 * timeline)
        + 0.02 * np.sin(2.0 * np.pi * 540.0 * timeline),
        dtype=np.float32,
    )
    anchor = _speaker_anchor_array(DEFAULT_CENTROID)
    warmup_evidence = (
        speaker_evidence_from_audio(
            warmup_waveform,
            SR,
            encoder,
            anchor,
            device="cpu",
        ),
        release_speaker_evidence_from_audio(
            warmup_waveform,
            SR,
            encoder,
            anchor,
            device="cpu",
        ),
    )
    if any(
        not np.isfinite(evidence.similarity)
        or not np.isfinite(evidence.speaker_embedding).all()
        for evidence in warmup_evidence
    ):
        raise RuntimeError("release speaker verifier warmup failed")
    squim_evidence = squim_objective_evidence_from_audio(
        warmup_waveform,
        SR,
    )
    if not all(
        np.isfinite(value)
        for value in (
            squim_evidence.stoi,
            squim_evidence.pesq,
            squim_evidence.si_sdr,
        )
    ):
        raise RuntimeError("release SQUIM verifier warmup failed")
    median_f0_hz = active_audio_median_f0_hz(
        warmup_waveform,
        SR,
    )
    if (
        median_f0_hz is None
        or not np.isfinite(median_f0_hz)
        or median_f0_hz < 1.0
    ):
        raise RuntimeError("release F0 transition runtime warmup failed")
    print("[BlueMagpie] release CPU verifiers ready")


def _apply_speed(
    audio: np.ndarray,
    speed: float,
    *,
    network_conditioned: bool = False,
) -> np.ndarray:
    del network_conditioned
    try:
        speed = float(speed)
    except (TypeError, ValueError, OverflowError) as error:
        raise ValueError("speed must be finite and positive") from error
    if not np.isfinite(speed) or speed <= 0.0:
        raise ValueError("speed must be finite and positive")
    if speed < MIN_PACE_SPEED:
        raise ValueError(
            f"speed is below the production WSOLA floor {MIN_PACE_SPEED:.2f}"
        )
    if abs(speed - 1.0) < 1.0e-3:
        return np.asarray(audio, dtype=np.float32)
    return wsola_time_stretch(
        audio,
        rate=speed,
        sample_rate=SR,
    )


def _apply_echo_smearing_gate(
    verification,
    waveforms: tuple[np.ndarray | None, ...],
):
    """Reject analyzable candidates with fixed echo or smearing evidence.

    The no-reference metric needs at least one second of waveform evidence.
    Shorter utterances remain governed by the existing semantic, endpoint and
    SQUIM contracts instead of being rejected by an inapplicable estimator.
    Every analyzable waveform is fail-closed: missing/non-finite/insufficient
    active-window evidence becomes an acoustic rejection.
    """

    if len(waveforms) != len(verification.candidate_results):
        raise ValueError("audio artifact evidence does not align")
    updated_results = []
    for index, (waveform, result) in enumerate(
        zip(waveforms, verification.candidate_results, strict=True)
    ):
        if waveform is None or not result.passed:
            updated_results.append(result)
            continue
        duration_seconds = waveform.size / float(SR)
        if (
            duration_seconds
            < DEFAULT_ECHO_SMEARING_GATE_CONFIG.min_duration_seconds
        ):
            print(
                "[BlueMagpie] audio artifact gate "
                f"chunk={index} applicable=false "
                f"duration={duration_seconds:.6f}"
            )
            updated_results.append(result)
            continue

        diagnostics = evaluate_echo_smearing(waveform, SR)
        print(
            "[BlueMagpie] audio artifact gate "
            f"chunk={index} applicable=true passed={diagnostics.passed} "
            f"delayed_score={diagnostics.delayed_copy_score:.6f} "
            f"smearing_score={diagnostics.smearing_score:.6f} "
            f"delay_ms={diagnostics.dominant_delay_ms:.3f} "
            f"peak_ratio={diagnostics.cepstral_peak_ratio:.6f} "
            f"windows={diagnostics.analysis_window_count} "
            f"reasons={diagnostics.rejection_reasons}"
        )
        if diagnostics.passed:
            updated_results.append(result)
            continue
        artifact_reasons = tuple(
            f"audio_artifact_{reason}"
            for reason in diagnostics.rejection_reasons
        ) or ("audio_artifact_unavailable",)
        updated_results.append(
            replace(
                result,
                passed=False,
                score=float("inf"),
                rejection_reasons=tuple(
                    dict.fromkeys(
                        (*result.rejection_reasons, *artifact_reasons)
                    )
                ),
            )
        )

    candidate_results = tuple(updated_results)
    passed = bool(candidate_results) and all(
        result.passed for result in candidate_results
    )
    score = (
        sum(result.score for result in candidate_results)
        if passed
        else float("inf")
    )
    rejection_reasons = tuple(
        f"chunk_{index}:{reason}"
        for index, result in enumerate(candidate_results)
        for reason in result.rejection_reasons
    )
    return replace(
        verification,
        passed=passed,
        candidate_results=candidate_results,
        score=score,
        rejection_reasons=rejection_reasons,
    )


@dataclass(frozen=True)
class _ChunkGenerationOutcome:
    audio: np.ndarray
    stop_reason: str
    endpoint_energy_ratio: float
    generated_steps: int
    hard_stop_steps: int


def _generate_chunk(
    text: str,
    centroid: torch.Tensor,
    *,
    cfg: float,
    steps: int,
    request_seed: int,
    policy: GenerationPolicy,
    network_conditioned: bool = False,
) -> _ChunkGenerationOutcome:
    generation_cps = (
        policy.ascii_cps
        if network_conditioned
        else select_generation_cps(
            text,
            cjk_cps=policy.cjk_cps,
            ascii_cps=policy.ascii_cps,
        )
    )
    endpoint_duration_units = (
        count_network_endpoint_duration_units(text)
        if network_conditioned
        else count_speech_units(text)
    )
    model_text, expected_steps, hard_stop_steps = endpoint_generation_plan(
        text,
        generation_cps=generation_cps,
        step_seconds=STEP_SECONDS,
        margin_steps=policy.hard_stop_margin_steps,
        add_terminal_punctuation=count_speech_units(text) >= MIN_ENDPOINT_CUE_UNITS,
        duration_units=endpoint_duration_units,
    )
    public_text_units = count_speech_units(text)
    short_headroom_max_units = int(
        getattr(policy, "short_headroom_max_units", 0)
    )
    short_hard_stop_floor_steps = int(
        getattr(policy, "short_hard_stop_floor_steps", 0)
    )
    short_headroom_floor_applied = bool(
        0 < public_text_units <= short_headroom_max_units
        and hard_stop_steps < short_hard_stop_floor_steps
    )
    if short_headroom_floor_applied:
        hard_stop_steps = short_hard_stop_floor_steps
    # Do not hold generation open to enforce pace. The model can finish the
    # requested text early; extending its latent sequence creates tail speech.
    min_len = 2
    generation_cfg = effective_generation_cfg(
        text,
        cfg,
        short_text_unit_threshold=SHORT_TEXT_CFG_UNITS,
        short_text_min_cfg=SHORT_TEXT_CFG_MIN,
    )
    set_generation_seed(request_seed)
    kwargs = {
        "target_text": model_text,
        "speaker_centroid": centroid,
        "cfg_value": generation_cfg,
        "inference_timesteps": int(steps),
        "min_len": min_len,
        "max_len": hard_stop_steps,
        "retry_badcase": False,
        "retry_badcase_max_times": 1,
        "retry_badcase_ratio_threshold": 6.0,
    }
    if _NATIVE_STOP_POLICY:
        kwargs["stop_threshold"] = STOP_THRESHOLD
        kwargs["stop_consecutive"] = STOP_CONSECUTIVE
    if "generation_seed" in _GENERATE_PARAMETERS:
        kwargs["generation_seed"] = request_seed

    if _STOP_CONTROLLER is not None:
        _STOP_CONTROLLER.begin(
            min_len,
            expected_steps=expected_steps,
            hard_stop_steps=hard_stop_steps,
        )
    with latency_stage("generation"):
        try:
            audio = model.generate(**kwargs)
        finally:
            if _STOP_CONTROLLER is not None:
                _STOP_CONTROLLER.end()
        # Copying to CPU closes the asynchronous CUDA work inside this stage,
        # so the reported wall time is not shifted into later verification.
        audio = audio.detach().float().cpu().numpy().reshape(-1)
    stop_reason = (
        _STOP_CONTROLLER.last_stop_reason
        if _STOP_CONTROLLER is not None
        else "native_stop"
    )
    generated_steps = (
        _STOP_CONTROLLER.last_generated_steps
        if _STOP_CONTROLLER is not None
        else hard_stop_steps
    )
    if _STOP_CONTROLLER is not None:
        print(
            "[BlueMagpie] endpoint "
            f"expected_steps={expected_steps} hard_stop_steps={hard_stop_steps} "
            f"generated_steps={_STOP_CONTROLLER.last_generated_steps} "
            f"reason={_STOP_CONTROLLER.last_stop_reason} cfg={generation_cfg:.2f}"
        )
    print(
        "[BlueMagpie] generation policy "
        f"name={policy.name} seed={request_seed} generation_cps={generation_cps:.2f} "
        f"duration_units={endpoint_duration_units} "
        f"duration_counter={'network_conservative' if network_conditioned else 'public'} "
        f"expected_steps={expected_steps} hard_stop_steps={hard_stop_steps} "
        f"min_len={min_len} "
        f"short_headroom_floor_applied={short_headroom_floor_applied}"
    )
    def outcome(waveform: np.ndarray) -> _ChunkGenerationOutcome:
        normalized = np.asarray(waveform, dtype=np.float32).reshape(-1)
        return _ChunkGenerationOutcome(
            audio=normalized,
            stop_reason=stop_reason,
            endpoint_energy_ratio=endpoint_tail_energy_ratio(normalized, SR),
            generated_steps=generated_steps,
            hard_stop_steps=hard_stop_steps,
        )

    pace_speed = target_pace_speed(
        audio.size,
        SR,
        text,
        target_cps=TARGET_CPS,
        min_speed=MIN_PACE_SPEED,
    )
    combined_speed = pace_speed
    try:
        active_duration = active_voiced_duration_seconds(audio, SR)
    except (TypeError, ValueError, RuntimeError, OverflowError):
        # Invalid pace evidence must never turn into an unbounded correction.
        # Total-duration pace remains bounded and faces the ordinary local gate.
        pass
    else:
        active_speed = active_pace_correction_speed(
            active_duration,
            text,
            target_cps=ACTIVE_PACE_TARGET_CPS,
            prior_speed=1.0,
            min_total_speed=MIN_PACE_SPEED,
        )
        combined_speed = min(pace_speed, active_speed)
        if combined_speed < 1.0:
            print(
                "[BlueMagpie] single-pass pace correction "
                f"active_duration={active_duration:.3f} total_rate={pace_speed:.6f} "
                f"active_rate={active_speed:.6f} combined_rate={combined_speed:.6f}"
            )
    corrected = _apply_speed(
        audio,
        combined_speed,
        network_conditioned=network_conditioned,
    )
    try:
        corrected_active_duration = active_voiced_duration_seconds(corrected, SR)
    except (TypeError, ValueError, RuntimeError, OverflowError):
        return outcome(corrected)
    rerender_speed = active_pace_correction_speed(
        corrected_active_duration,
        text,
        target_cps=CLOSED_LOOP_ACTIVE_PACE_TARGET_CPS,
        prior_speed=combined_speed,
        min_total_speed=MIN_PACE_SPEED,
    )
    final_speed = combined_speed
    final_audio = corrected
    if rerender_speed < 1.0:
        final_speed = combined_speed * rerender_speed
        print(
            "[BlueMagpie] closed-loop pace rerender "
            f"active_duration={corrected_active_duration:.3f} "
            f"prior_rate={combined_speed:.6f} residual_rate={rerender_speed:.6f} "
            f"total_rate={final_speed:.6f}"
        )
        # Re-render from the untouched model waveform so the returned samples
        # have exactly one pitch-preserving WSOLA pass.
        final_audio = _apply_speed(
            audio,
            final_speed,
            network_conditioned=network_conditioned,
        )

    speech_units = count_speech_units(text)
    if (
        not network_conditioned
        and speech_units <= EXTREME_SHORT_MAX_UNITS
        and final_speed > EXTREME_SHORT_MAX_SPEED
    ):
        final_speed = EXTREME_SHORT_MAX_SPEED
        print(
            "[BlueMagpie] extreme-short robustness rerender "
            f"units={speech_units} total_rate={final_speed:.6f}"
        )
        final_audio = _apply_speed(
            audio,
            final_speed,
            network_conditioned=False,
        )

    # Preserve every already-compliant waveform bit-for-bit. A lower stretch
    # floor is available only to long, ordinary chunks whose exact base output
    # would otherwise fail the frozen local pace gate. Re-rendering still uses
    # the untouched model waveform, so no candidate receives two vocoder passes.
    if (
        not network_conditioned
        and speech_units >= PACE_ONLY_FALLBACK_MIN_UNITS
    ):
        try:
            final_active_duration = active_voiced_duration_seconds(final_audio, SR)
        except (TypeError, ValueError, RuntimeError, OverflowError):
            return outcome(final_audio)
        observed_cps = speech_units / final_active_duration
        if observed_cps > QUALITY_MAX_PACE_CPS:
            fallback_residual = active_pace_correction_speed(
                final_active_duration,
                text,
                target_cps=CLOSED_LOOP_ACTIVE_PACE_TARGET_CPS,
                prior_speed=final_speed,
                min_total_speed=PACE_ONLY_FALLBACK_MIN_SPEED,
            )
            if fallback_residual < 1.0:
                fallback_speed = final_speed * fallback_residual
                print(
                    "[BlueMagpie] pace-only fallback rerender "
                    f"active_duration={final_active_duration:.3f} "
                    f"observed_cps={observed_cps:.6f} "
                    f"prior_rate={final_speed:.6f} "
                    f"residual_rate={fallback_residual:.6f} "
                    f"total_rate={fallback_speed:.6f}"
                )
                return outcome(
                    _apply_speed(
                        audio,
                        fallback_speed,
                        network_conditioned=False,
                    )
                )
    return outcome(final_audio)


def _speaker_anchor_array(centroid: torch.Tensor) -> np.ndarray:
    anchor = torch.as_tensor(centroid).detach().float().cpu().numpy().reshape(-1)
    if anchor.size == 0 or not np.isfinite(anchor).all():
        raise ValueError("speaker anchor is invalid")
    norm = float(np.linalg.norm(anchor))
    if not np.isfinite(norm) or norm <= 1.0e-8:
        raise ValueError("speaker anchor has zero norm")
    return np.asarray(anchor / norm, dtype=np.float32)


# Verifiers are intentionally not preloaded in latency-first mode.


def _generate_trajectory(
    chunks: tuple[str, ...],
    centroid: torch.Tensor,
    *,
    cfg: float,
    steps: int,
    request_seed: int,
    policy: GenerationPolicy,
    network_cfg_min: float = NETWORK_TEXT_CFG_MIN,
    network_conditioned: tuple[bool, ...] | None = None,
) -> tuple[tuple[np.ndarray, ...], tuple[_ChunkGenerationOutcome, ...]]:
    scheduled_cfg = float(cfg)
    trajectory: list[np.ndarray] = []
    endpoint_outcomes: list[_ChunkGenerationOutcome] = []
    if network_conditioned is not None and len(network_conditioned) != len(chunks):
        raise ValueError("network provenance must align with generation chunks")
    for local_chunk_index, chunk in enumerate(chunks):
        network_chunk = (
            bool(network_conditioned[local_chunk_index])
            if network_conditioned is not None
            else bool(network_protected_spoken_spans(chunk))
        )
        network_floor_applied = bool(
            network_chunk and scheduled_cfg < float(network_cfg_min)
        )
        network_adjusted_cfg = (
            max(scheduled_cfg, float(network_cfg_min))
            if network_chunk
            else scheduled_cfg
        )
        effective_cfg = effective_generation_cfg(
            chunk,
            network_adjusted_cfg,
            short_text_unit_threshold=SHORT_TEXT_CFG_UNITS,
            short_text_min_cfg=SHORT_TEXT_CFG_MIN,
        )
        short_floor_applied = effective_cfg > network_adjusted_cfg
        print(
            "[BlueMagpie] generation attempt "
            f"seed={request_seed} local_chunk_index={local_chunk_index} "
            f"policy={policy.name} scheduled_cfg={scheduled_cfg:.2f} "
            f"effective_cfg={effective_cfg:.2f} "
            f"network_floor_applied={network_floor_applied} "
            f"short_floor_applied={short_floor_applied}"
        )
        outcome = _generate_chunk(
            chunk,
            centroid,
            cfg=network_adjusted_cfg,
            steps=steps,
            request_seed=request_seed,
            policy=policy,
            network_conditioned=network_chunk,
        )
        trajectory.append(outcome.audio)
        endpoint_outcomes.append(outcome)
    return tuple(trajectory), tuple(endpoint_outcomes)


def _enrich_transition_artifact_f0(
    waveform: np.ndarray,
    artifact: ChunkCandidateArtifact,
) -> ChunkCandidateArtifact:
    """Attach the unchanged pYIN statistic only when DP must compare paths."""

    if not isinstance(artifact, ChunkCandidateArtifact):
        raise TypeError("transition artifact is invalid")
    if artifact.median_f0_hz is not None:
        return artifact
    try:
        median_f0_hz = active_audio_median_f0_hz(waveform, SR)
    except (RuntimeError, TypeError, ValueError):
        median_f0_hz = None
    return replace(artifact, median_f0_hz=median_f0_hz)


def _verify_trajectory_audio(
    trajectory: tuple[np.ndarray, ...],
    chunks: tuple[str, ...],
    anchor: np.ndarray,
    playback_speed: float,
    asr_max_new_tokens: int = 128,
    *,
    release_speaker_gate: bool = False,
    transcriber=transcribe_breeze25,
    semantic_only: bool = False,
    network_fragment_proofs: (
        tuple[tuple[NetworkFragmentProof, ...], ...] | None
    ) = None,
    local_endpoint_roles: tuple[tuple[bool, bool], ...] | None = None,
):
    if len(trajectory) != len(chunks):
        return verify_trajectory(())
    if network_fragment_proofs is not None and len(network_fragment_proofs) != len(
        chunks
    ):
        raise ValueError("network fragment proofs must align with local chunks")
    if local_endpoint_roles is not None and len(local_endpoint_roles) != len(chunks):
        raise ValueError("local endpoint roles must align with local chunks")
    local_candidate_pool = not semantic_only and not release_speaker_gate
    indexed_gate_kwargs = None
    if local_candidate_pool and local_endpoint_roles is not None:
        indexed_gate_kwargs = tuple(
            {
                **({"max_prefix_cer": 0.0} if request_first else {}),
                **({"max_suffix_cer": 0.0} if request_last else {}),
            }
            for request_first, request_last in local_endpoint_roles
        )
    semantic_gate_kwargs = {
        "short_text_units": 6,
        "short_text_max_cer": 0.0,
        "max_cer": QUALITY_MAX_CER,
        "prefix_units": QUALITY_PREFIX_SUFFIX_UNITS,
        "suffix_units": QUALITY_PREFIX_SUFFIX_UNITS,
        "max_prefix_cer": (1.0 / 6.0 if local_candidate_pool else 0.0),
        "max_suffix_cer": (1.0 / 6.0 if local_candidate_pool else 0.0),
        "max_prefix_deletions": (0 if local_candidate_pool else None),
        "max_suffix_deletions": (0 if local_candidate_pool else None),
        "max_extra_tail_units": 0,
        "max_pace_cps": None if semantic_only else QUALITY_MAX_PACE_CPS,
    }
    encoder = None
    observations: list[CandidateObservation] = []
    artifacts: list[ChunkCandidateArtifact] = []
    prepared_waveforms: list[np.ndarray | None] = []
    proof_rows = network_fragment_proofs or ((),) * len(chunks)
    for index, (chunk, audio, fragment_proofs) in enumerate(
        zip(
            chunks,
            trajectory,
            proof_rows,
            strict=True,
        )
    ):
        try:
            waveform = np.asarray(audio, dtype=np.float32)
            if (
                waveform.ndim != 1
                or waveform.size == 0
                or not np.isfinite(waveform).all()
            ):
                raise ValueError("candidate waveform is invalid")
            duration = active_voiced_duration_seconds(waveform, SR)
        except (TypeError, ValueError, OverflowError):
            observations.append(
                CandidateObservation(
                    target_text=chunk,
                    transcript_text="",
                    audio_duration_seconds=0.0,
                    pace_cps=None,
                )
            )
            artifacts.append(ChunkCandidateArtifact())
            prepared_waveforms.append(None)
            continue
        speech_units = count_speech_units(chunk)
        pace_cps = (
            speech_units / duration * float(playback_speed)
            if duration > 0.0
            else None
        )
        if (
            not semantic_only
            and (
                pace_cps is None
                or pace_cps
                > QUALITY_MAX_PACE_CPS + QUALITY_PREFLIGHT_PACE_MARGIN_CPS
            )
        ):
            # A candidate that cannot pass the immutable pace gate must never
            # be returned. Reject it before Breeze25, ECAPA, SQUIM and echo
            # work; later cascade candidates still face every unchanged gate.
            observations.append(
                CandidateObservation(
                    target_text=chunk,
                    transcript_text="",
                    audio_duration_seconds=duration,
                    pace_cps=pace_cps,
                )
            )
            artifacts.append(ChunkCandidateArtifact())
            prepared_waveforms.append(None)
            continue
        prepared = prepare_candidate_audio(
            waveform,
            SR,
            transcriber=lambda waveform, sample_rate: transcriber(
                waveform,
                sample_rate,
                max_new_tokens=asr_max_new_tokens,
            ),
        )
        if prepared is None:
            observations.append(
                CandidateObservation(
                    target_text=chunk,
                    transcript_text="",
                    audio_duration_seconds=0.0,
                    pace_cps=None,
                )
            )
            artifacts.append(ChunkCandidateArtifact())
            prepared_waveforms.append(None)
            continue
        waveform = prepared.waveform
        transcript = prepared.transcript_text
        if fragment_proofs:
            fragment_evidence = canonicalize_asr_network_fragments(
                transcript,
                chunk,
                fragment_proofs,
            )
            # A range-bound identifier mismatch is a hard local semantic
            # failure even when its contribution to whole-chunk CER is small.
            transcript = (
                fragment_evidence.transcript_text
                if fragment_evidence.passed
                else ""
            )
        semantic_observation = CandidateObservation(
            target_text=chunk,
            transcript_text=transcript,
            audio_duration_seconds=duration,
            pace_cps=pace_cps,
        )
        if not semantic_only:
            row_gate_kwargs = (
                indexed_gate_kwargs[index]
                if indexed_gate_kwargs is not None
                else {}
            )
            semantic_preflight = verify_candidate(
                semantic_observation,
                **{
                    **semantic_gate_kwargs,
                    **row_gate_kwargs,
                    "max_pace_cps": (
                        QUALITY_MAX_PACE_CPS
                        + QUALITY_PREFLIGHT_PACE_MARGIN_CPS
                    ),
                },
                speaker_gate_enabled=False,
                squim_gate_enabled=False,
            )
            if not semantic_preflight.passed:
                # Preserve the exact semantic/pace evidence while avoiding
                # ECAPA, SQUIM, echo and F0 work for an already rejected row.
                observations.append(semantic_observation)
                artifacts.append(ChunkCandidateArtifact())
                prepared_waveforms.append(None)
                continue
        prepared_waveforms.append(waveform)
        speaker_similarity = None
        begin_similarity = None
        end_similarity = None
        speaker_embedding = None
        rms_db = None
        squim_stoi = None
        squim_pesq = None
        squim_si_sdr = None
        measure_speaker = (
            not semantic_only
            and (
                release_speaker_measurement_required(duration)
                if release_speaker_gate
                else duration >= SHORT_AUDIO_SPEAKER_GATE_SECONDS
            )
        )
        if measure_speaker:
            try:
                if encoder is None:
                    encoder = _get_ecapa_encoder()
                speaker_measurement = (
                    release_speaker_evidence_from_audio
                    if release_speaker_gate
                    else speaker_evidence_from_audio
                )
                evidence = speaker_measurement(
                    waveform,
                    SR,
                    encoder,
                    anchor,
                    device="cpu",
                )
                duration = evidence.active_duration_seconds
                speaker_similarity = evidence.similarity
                begin_similarity = evidence.begin_similarity
                end_similarity = evidence.end_similarity
                speaker_embedding = evidence.speaker_embedding
                rms_db = evidence.active_rms_db
            except ValueError:
                # A malformed/empty speaker measurement remains missing and is
                # rejected by the fail-closed gate for non-short candidates.
                pass
        if rms_db is None:
            try:
                rms_db = active_audio_rms_db(waveform)
            except ValueError:
                pass
        if not semantic_only and transcript:
            try:
                squim_evidence = squim_objective_evidence_from_audio(
                    waveform,
                    SR,
                )
                squim_stoi = squim_evidence.stoi
                squim_pesq = squim_evidence.pesq
                squim_si_sdr = squim_evidence.si_sdr
            except ValueError:
                # Candidate-local invalid/non-finite acoustic evidence remains
                # missing and is rejected; runtime/hash failures propagate.
                pass
        observations.append(
            CandidateObservation(
                target_text=chunk,
                transcript_text=transcript,
                audio_duration_seconds=duration,
                speaker_similarity=speaker_similarity,
                begin_speaker_similarity=begin_similarity,
                end_speaker_similarity=end_similarity,
                pace_cps=(
                    speech_units / duration * float(playback_speed)
                    if duration > 0.0
                    else None
                ),
                squim_stoi=squim_stoi,
                squim_pesq=squim_pesq,
                squim_si_sdr=squim_si_sdr,
            )
        )
        artifacts.append(
            ChunkCandidateArtifact(
                speaker_embedding=speaker_embedding,
                rms_db=rms_db,
            )
        )
    verification = verify_trajectory(
        observations,
        chunk_artifacts=artifacts,
        candidate_gate_kwargs_by_index=indexed_gate_kwargs,
        **semantic_gate_kwargs,
        speaker_gate_enabled=not semantic_only,
        short_audio_seconds=(
            RELEASE_SPEAKER_TRIGGER_SECONDS
            if release_speaker_gate
            else SHORT_AUDIO_SPEAKER_GATE_SECONDS
        ),
        min_speaker_similarity=(
            QUALITY_RELEASE_MIN_SPEAKER_SIMILARITY
            if release_speaker_gate
            else QUALITY_MIN_SPEAKER_SIMILARITY
        ),
        max_boundary_speaker_drop=(
            QUALITY_RELEASE_MAX_BOUNDARY_SPEAKER_DROP
            if release_speaker_gate
            else QUALITY_MAX_BOUNDARY_SPEAKER_DROP
        ),
        squim_gate_enabled=not semantic_only,
        min_squim_stoi=QUALITY_MIN_SQUIM_STOI,
        min_squim_pesq=QUALITY_MIN_SQUIM_PESQ,
        skip_acoustic_on_hard_failure=True,
    )
    if not semantic_only:
        verification = _apply_echo_smearing_gate(
            verification,
            tuple(prepared_waveforms),
        )
    return verification


def _verify_independent_whole_audio(
    waveform: np.ndarray,
    target_text: str,
    anchor: np.ndarray,
    cache: WholeWaveformVerificationCache,
    *,
    transcriber=transcribe_breeze25,
):
    """Apply the semantic-only projection using the cached Breeze25 transcript."""

    def verifier(exact_waveform, sample_rate, exact_target):
        if sample_rate != SR:
            raise ValueError("independent verifier sample rate mismatch")
        return _verify_trajectory_audio(
            (exact_waveform,),
            (exact_target,),
            anchor,
            1.0,
            QUALITY_FINAL_ASR_MAX_NEW_TOKENS,
            transcriber=transcriber,
            semantic_only=True,
        )

    return cache.verify(
        waveform,
        SR,
        target_text,
        BREEZE25_SEMANTIC_VERIFICATION_PROFILE,
        verifier,
    )


def _cached_request_transcriber(
    cache: WholeWaveformTranscriptCache,
    decoder_profile: str,
    transcriber,
):
    """Share one pinned ASR decode for an exact waveform within a request."""

    def cached_transcriber(
        waveform: np.ndarray,
        sample_rate: int,
        *,
        max_new_tokens: int = QUALITY_FINAL_ASR_MAX_NEW_TOKENS,
    ) -> str:
        if (
            type(max_new_tokens) is not int
            or max_new_tokens != QUALITY_FINAL_ASR_MAX_NEW_TOKENS
        ):
            raise ValueError("cached whole-waveform ASR token cap mismatch")
        return cache.transcribe(
            waveform,
            sample_rate,
            decoder_profile,
            lambda exact_waveform, exact_rate: transcriber(
                exact_waveform,
                exact_rate,
                max_new_tokens=QUALITY_FINAL_ASR_MAX_NEW_TOKENS,
            ),
        )

    return cached_transcriber


def _verify_release_whole_audio(
    waveform: np.ndarray,
    target_text: str,
    anchor: np.ndarray,
    cache: WholeWaveformVerificationCache,
    *,
    transcriber,
    verifier_profile: str,
):
    """Cache the full semantic, speaker, SQUIM and echo release gate."""

    def verifier(exact_waveform, sample_rate, exact_target):
        if sample_rate != SR:
            raise ValueError("release verifier sample rate mismatch")
        return _verify_trajectory_audio(
            (exact_waveform,),
            (exact_target,),
            anchor,
            1.0,
            QUALITY_FINAL_ASR_MAX_NEW_TOKENS,
            release_speaker_gate=True,
            transcriber=transcriber,
        )

    return cache.verify(
        waveform,
        SR,
        target_text,
        verifier_profile,
        verifier,
    )


def _transcribe_glyph_hybrid_breeze_primary(
    audio: np.ndarray,
    sample_rate: int,
    *,
    language: str = "zh",
    task: str = "transcribe",
    max_new_tokens: int = 128,
    max_verification_segments: int = GLYPH_HYBRID_MAX_VERIFICATION_SEGMENTS,
) -> str:
    """Run the fixed-zh Breeze25 profile with the V5 segment cap."""

    return transcribe_breeze25(
        audio,
        sample_rate,
        language=language,
        task=task,
        max_new_tokens=max_new_tokens,
        max_verification_segments=max_verification_segments,
    )


def _transcribe_glyph_hybrid_breeze_confirmation(
    audio: np.ndarray,
    sample_rate: int,
    *,
    language: str | None = None,
    task: str = "transcribe",
    max_new_tokens: int = 128,
    max_verification_segments: int = GLYPH_HYBRID_MAX_VERIFICATION_SEGMENTS,
) -> str:
    """Run the auto-language Breeze25 confirmation profile for V5."""

    return transcribe_breeze25(
        audio,
        sample_rate,
        language=language,
        task=task,
        max_new_tokens=max_new_tokens,
        max_verification_segments=max_verification_segments,
    )


_GLYPH_HYBRID_EVIDENCE_ADAPTER = HardenedGlyphHybridAdapter(
    GlyphAdapterRuntime(
        sample_rate=SR,
        inference_steps=DEFAULT_STEPS,
        model_repo_id=REPO_ID,
        model_revision=MODEL_REVISION,
        ecapa_repo_id=ECAPA_REPO_ID,
        ecapa_revision=ECAPA_REVISION,
        speaker_anchor_sha256=GLYPH_SPEAKER_ANCHOR_SHA256,
        speaker_centroid=DEFAULT_CENTROID,
        generation_lock=_GENERATION_LOCK,
        generate_chunk=_generate_chunk,
        generation_policy_resolver=(
            generation_policy_for_candidate_offset
        ),
        generation_cfg_resolver=lambda ordinal: (
            generation_cfg_for_candidate_offset(
                ordinal,
                primary_cfg=MIXED_CFG_PRIMARY,
                alternate_cfg=MIXED_CFG_ALTERNATE,
            )
        ),
        speaker_encoder_factory=_get_ecapa_encoder,
        speaker_anchor_factory=lambda: _speaker_anchor_array(
            DEFAULT_CENTROID
        ),
        gate_policy=GlyphEvidenceGatePolicy(
            sample_rate=SR,
            max_cer=GLYPH_FINAL_MAX_CER,
            max_prefix_cer=0.0,
            max_suffix_cer=0.0,
            max_prefix_deletions=0,
            max_suffix_deletions=0,
            max_extra_tail_units=0,
            max_pace_cps=QUALITY_MAX_PACE_CPS,
            min_speaker_similarity=(
                QUALITY_RELEASE_MIN_SPEAKER_SIMILARITY
            ),
            max_boundary_speaker_drop=(
                QUALITY_RELEASE_MAX_BOUNDARY_SPEAKER_DROP
            ),
            min_squim_stoi=QUALITY_PREFERRED_MIN_SQUIM_STOI,
            min_squim_pesq=QUALITY_PREFERRED_MIN_SQUIM_PESQ,
            min_squim_si_sdr=-20.0,
            max_endpoint_tail_energy_ratio=0.02,
        ),
        asr_max_new_tokens=QUALITY_FINAL_ASR_MAX_NEW_TOKENS,
        asr_max_verification_segments=(
            GLYPH_HYBRID_MAX_VERIFICATION_SEGMENTS
        ),
        breeze_primary_transcriber=(
            _transcribe_glyph_hybrid_breeze_primary
        ),
        breeze_confirmation_transcriber=(
            _transcribe_glyph_hybrid_breeze_confirmation
        ),
    )
)


def _network_fragment_proof_rows(
    chunks: tuple[str, ...],
    chunk_specs: tuple[GenerationChunkSpec, ...] | None,
) -> tuple[tuple[NetworkFragmentProof, ...], ...] | None:
    """Validate explicit planner provenance before any local ASR relaxation."""

    if chunk_specs is None:
        return None
    if len(chunk_specs) != len(chunks):
        raise ValueError("network fragment provenance must align with chunks")
    rows: list[tuple[NetworkFragmentProof, ...]] = []
    for chunk, spec in zip(chunks, chunk_specs, strict=True):
        proofs = spec.network_fragment_proofs
        if spec.text != chunk:
            raise ValueError("network fragment provenance text does not match")
        if spec.network_conditioned:
            if (
                not proofs
                or len(proofs) != len(spec.network_span_indices)
                or tuple(proof.span_index for proof in proofs)
                != spec.network_span_indices
                or tuple(proof.full_spoken_proof for proof in proofs)
                != spec.network_full_spoken_proofs
            ):
                raise ValueError("network-conditioned chunk lacks exact fragment proof")
        elif proofs or spec.network_full_spoken_proofs:
            raise ValueError("ordinary chunk carries network fragment proof")
        for proof in proofs:
            naturalized_spoken = contains_naturalized_url_spoken_form(
                proof.full_spoken_proof
            )
            if naturalized_spoken:
                fresh_rendering = naturalized_url_rendering_proof(
                    proof.raw_identifier
                )
                if (
                    not proof.raw_identifier
                    or fresh_rendering is None
                    or proof.url_rendering_proof != fresh_rendering
                    or inverse_network_url_rendering(proof.url_rendering_proof)
                    != proof.raw_identifier
                    or proof.url_rendering_proof.spoken_text
                    != proof.full_spoken_proof
                    or proof.parent_start != 0
                    or proof.parent_end != len(proof.full_spoken_proof)
                    or chunk[proof.chunk_start : proof.chunk_end]
                    != proof.full_spoken_proof
                ):
                    raise ValueError(
                        "naturalized URL runtime proof is inconsistent"
                    )
            elif proof.raw_identifier or proof.url_rendering_proof is not None:
                raise ValueError(
                    "non-naturalized runtime proof carries URL rendering provenance"
                )
        rows.append(proofs)
    return tuple(rows)


def _local_endpoint_role_rows(
    chunks: tuple[str, ...],
    chunk_specs: tuple[GenerationChunkSpec, ...] | None,
) -> tuple[tuple[bool, bool], ...] | None:
    """Bind exact local onset/ending gates to immutable planner positions."""

    if chunk_specs is None:
        return None
    if len(chunk_specs) != len(chunks) or any(
        spec.text != chunk
        for spec, chunk in zip(chunk_specs, chunks, strict=True)
    ):
        raise ValueError("generation chunk endpoint provenance does not align")
    return tuple(
        (spec.source_start == 0, spec.boundary_after == "none")
        for spec in chunk_specs
    )


@timed_latency_stage("assemble")
def _assemble_trajectory_audio(
    trajectory: tuple[np.ndarray, ...],
    chunks: tuple[str, ...],
    playback_speed: float,
    chunk_specs: tuple[GenerationChunkSpec, ...] | None = None,
) -> np.ndarray:
    """Assemble chunks exactly as they will be returned to the listener."""

    if not trajectory or len(trajectory) != len(chunks):
        raise ValueError("trajectory and text chunks must be non-empty and aligned")
    if chunk_specs is not None and (
        len(chunk_specs) != len(chunks)
        or any(spec.text != chunk for spec, chunk in zip(chunk_specs, chunks, strict=True))
    ):
        raise ValueError("generation chunk provenance does not align with text")
    if chunk_specs is not None:
        for index, spec in enumerate(chunk_specs):
            if (
                spec.source_start < 0
                or spec.source_end <= spec.source_start
                or spec.boundary_after
                not in {"semantic", "network_internal", "none"}
            ):
                raise ValueError("generation chunk provenance is invalid")
            is_last = index + 1 == len(chunk_specs)
            if is_last:
                if spec.boundary_after != "none":
                    raise ValueError("final generation boundary must be none")
                continue
            following = chunk_specs[index + 1]
            if (
                spec.source_end != following.source_start
                or spec.boundary_after == "none"
            ):
                raise ValueError("generation chunk boundaries are incomplete")
            if spec.boundary_after == "network_internal" and (
                not spec.network_conditioned
                or not following.network_conditioned
                or not set(spec.network_span_indices).intersection(
                    following.network_span_indices
                )
            ):
                raise ValueError("network boundary lacks shared identifier proof")
    audio_chunks = [np.asarray(audio, dtype=np.float32).copy() for audio in trajectory]
    semantic_min_silence_ms = (
        NETWORK_REQUEST_SEMANTIC_CHUNK_MIN_SILENCE_MS
        if chunk_specs is not None and any(spec.network_conditioned for spec in chunk_specs)
        else SEMANTIC_CHUNK_MIN_SILENCE_MS
    )
    pauses: list[int] = []
    fades_ms: list[float] = []
    crossfades_ms: list[float] = []
    for index, chunk in enumerate(chunks):
        if index > 0:
            audio_chunks[index] = match_chunk_rms(
                audio_chunks[0],
                audio_chunks[index],
                max_adjust_db=CHUNK_RMS_MATCH_DB,
            )
        if index + 1 < len(chunks):
            network_internal = bool(
                chunk_specs is not None
                and chunk_specs[index].boundary_after == "network_internal"
            )
            pause_seconds = (
                NETWORK_INTERNAL_SILENCE_MS / 1000.0
                if network_internal
                else max(
                    punctuation_pause_seconds(chunk),
                    semantic_min_silence_ms / 1000.0,
                )
            )
            pauses.append(int(round(pause_seconds * SR)))
            fades_ms.append(
                NETWORK_INTERNAL_FADE_MS
                if network_internal
                else CHUNK_EDGE_FADE_MS
            )
            crossfades_ms.append(
                NETWORK_INTERNAL_FADE_MS
                if network_internal
                else CROSSFADE_MS
            )

    audio_chunks = fade_variable_internal_edges(audio_chunks, SR, fades_ms)
    waveform = join_audio_chunks_variable(
        audio_chunks,
        pauses,
        crossfade_samples_by_boundary=[
            int(round(crossfade_ms * SR / 1000.0))
            for crossfade_ms in crossfades_ms
        ],
        pre_faded_edges=True,
    )
    waveform = apply_loudness_floor(
        waveform,
        min_rms=0.07,
        peak_limit=0.95,
        max_gain=3.0,
    )
    waveform = _apply_speed(waveform, playback_speed)
    waveform = finish_audio(
        waveform,
        SR,
        fade_ms=FINAL_ENDPOINT_FADE_MS,
    )
    # Joined/final ASR, SQUIM, and speaker gates must inspect the exact
    # decoder-equivalent waveform represented by the public PCM16 response.
    return pcm16_verification_waveform(waveform)


def _verification_metric_log_fields(verification) -> str:
    """Format normalized semantic metrics without logging transcript content."""

    if len(verification.candidate_results) != 1:
        return (
            "cer=nan prefix_cer=nan suffix_cer=nan tail_units=nan "
            f"reasons={verification.rejection_reasons}"
        )
    comparison = verification.candidate_results[0].comparison
    candidate = verification.candidate_results[0]
    return (
        f"cer={comparison.cer:.6f} prefix_cer={comparison.prefix_cer:.6f} "
        f"suffix_cer={comparison.suffix_cer:.6f} "
        f"tail_units={comparison.extra_tail_units} "
        f"squim_stoi={candidate.squim_stoi} "
        f"squim_pesq={candidate.squim_pesq} "
        f"squim_si_sdr={candidate.squim_si_sdr} "
        f"squim_cost={candidate.squim_quality_cost} "
        f"reasons={verification.rejection_reasons}"
    )


def _verify_network_local_asr_intersection(
    primary_verification,
    trajectory: tuple[np.ndarray, ...],
    chunks: tuple[str, ...],
    anchor: np.ndarray,
    proof_rows: tuple[tuple[NetworkFragmentProof, ...], ...] | None,
    *,
    candidate_seed: int,
    transcriber=transcribe_breeze25,
):
    """Project cached Breeze25 semantics before retaining network locals.

    Only proof-bearing rows receive the duplicate semantic projection. Their
    exact ``NetworkFragmentProof`` tuples are reused unchanged, so neither
    semantic projection can borrow a whole URL/email proof for another local
    range. The
    intersection helper keeps primary speaker/SQUIM/pace/artifact evidence
    while projecting every semantic rejection into a hard local failure. The
    request-local transcriber cache prevents a second decode of exact PCM.
    """

    if proof_rows is None:
        return primary_verification, tuple(
            LocalIndependentGateEvidence(False, None, 0, None)
            for _ in chunks
        )
    if (
        len(proof_rows) != len(chunks)
        or len(trajectory) != len(chunks)
        or len(primary_verification.candidate_results) != len(chunks)
    ):
        raise ValueError("network local verification provenance does not align")
    independent_evidence = [
        LocalIndependentGateEvidence(
            attempted=False,
            passed=None,
            proof_count=len(proofs),
            result=None,
        )
        for proofs in proof_rows
    ]
    selected_indices = tuple(
        index
        for index, proofs in enumerate(proof_rows)
        if (
            proofs
            and local_candidate_has_coverage_eligibility(
                primary_verification.candidate_results[index],
                max_local_boundary_speaker_drop=(
                    SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP
                ),
            )
        )
    )
    if not selected_indices:
        return primary_verification, tuple(independent_evidence)
    independent_proof_rows = tuple(
        proof_rows[index] for index in selected_indices
    )
    independent_verification = _verify_trajectory_audio(
        tuple(trajectory[index] for index in selected_indices),
        tuple(chunks[index] for index in selected_indices),
        anchor,
        1.0,
        QUALITY_FINAL_ASR_MAX_NEW_TOKENS,
        transcriber=transcriber,
        semantic_only=True,
        network_fragment_proofs=independent_proof_rows,
    )
    for local_index, primary_index in enumerate(selected_indices):
        result = independent_verification.candidate_results[local_index]
        independent_evidence[primary_index] = LocalIndependentGateEvidence(
            attempted=True,
            passed=result.passed is True,
            proof_count=len(proof_rows[primary_index]),
            result=candidate_gate_evidence(result),
        )
        comparison = result.comparison
        print(
            "[BlueMagpie] network local Breeze25 projection "
            f"seed={candidate_seed} local_chunk_index={primary_index} "
            f"proof_count={len(proof_rows[primary_index])} "
            f"passed={result.passed} cer={comparison.cer:.6f} "
            f"prefix_cer={comparison.prefix_cer:.6f} "
            f"suffix_cer={comparison.suffix_cer:.6f} "
            f"tail_units={comparison.extra_tail_units} "
            f"reasons={result.rejection_reasons}"
        )
    return (
        intersect_local_semantic_verification(
            primary_verification,
            independent_verification,
            selected_indices,
        ),
        tuple(independent_evidence),
    )


def _qualify_candidate_trajectory_audio(
    trajectory: tuple[np.ndarray, ...],
    chunks: tuple[str, ...],
    whole_target_text: str,
    anchor: np.ndarray,
    playback_speed: float,
    independent_cache: WholeWaveformVerificationCache,
    *,
    candidate_seed: int,
    chunk_specs: tuple[GenerationChunkSpec, ...] | None = None,
    release_transcriber=transcribe_breeze25,
    independent_transcriber=transcribe_breeze25,
):
    """Run whole-output qualification only after every local chunk passes."""

    proof_rows = _network_fragment_proof_rows(chunks, chunk_specs)
    local_verification = _verify_trajectory_audio(
        trajectory,
        chunks,
        anchor,
        playback_speed,
        QUALITY_FINAL_ASR_MAX_NEW_TOKENS,
        transcriber=release_transcriber,
        network_fragment_proofs=proof_rows,
        local_endpoint_roles=_local_endpoint_role_rows(chunks, chunk_specs),
    )
    local_verification, independent_local_results = (
        _verify_network_local_asr_intersection(
            local_verification,
            trajectory,
            chunks,
            anchor,
            proof_rows,
            candidate_seed=candidate_seed,
            transcriber=independent_transcriber,
        )
    )
    if not local_verification.passed:
        return CandidateVerification(
            local_verification,
            independent_local_results=independent_local_results,
        )

    waveform = _assemble_trajectory_audio(
        trajectory,
        chunks,
        playback_speed,
        chunk_specs,
    )
    joined_verification = _verify_release_whole_audio(
        waveform,
        whole_target_text,
        anchor,
        independent_cache,
        transcriber=release_transcriber,
        verifier_profile=BREEZE25_RELEASE_VERIFICATION_PROFILE,
    )
    qualified = qualify_trajectory_with_joined_output(
        local_verification,
        joined_verification,
    )
    joined_evidence = trajectory_gate_evidence(joined_verification)
    if not qualified.passed:
        print(
            "[BlueMagpie] candidate joined output rejected "
            f"seed={candidate_seed} "
            f"{_verification_metric_log_fields(joined_verification)}"
        )
        return CandidateVerification(
            qualified,
            independent_local_results=independent_local_results,
            joined_output=joined_evidence,
        )

    independent_verification = _verify_independent_whole_audio(
        waveform,
        whole_target_text,
        anchor,
        independent_cache,
        transcriber=independent_transcriber,
    )
    semantic_qualified = qualify_trajectory_with_joined_output(
        qualified,
        independent_verification,
    )
    if not semantic_qualified.passed:
        print(
            "[BlueMagpie] candidate Breeze25 semantic projection rejected "
            f"seed={candidate_seed} "
            f"{_verification_metric_log_fields(independent_verification)}"
        )
    return CandidateVerification(
        semantic_qualified,
        independent_local_results=independent_local_results,
        joined_output=joined_evidence,
        independent_output=trajectory_gate_evidence(independent_verification),
    )


def _verify_refill_candidate_trajectory_audio(
    trajectory: tuple[np.ndarray, ...],
    chunks: tuple[str, ...],
    anchor: np.ndarray,
    playback_speed: float,
    chunk_specs: tuple[GenerationChunkSpec, ...] | None = None,
    *,
    candidate_seed: int,
    transcriber=transcribe_breeze25,
):
    """Apply strict cached-Breeze25 local gates to one safe-duration refill."""

    if len(trajectory) != 1 or len(chunks) != 1:
        return verify_trajectory(())
    proof_rows = _network_fragment_proof_rows(chunks, chunk_specs)
    local_verification = _verify_trajectory_audio(
        trajectory,
        chunks,
        anchor,
        playback_speed,
        QUALITY_FINAL_ASR_MAX_NEW_TOKENS,
        transcriber=transcriber,
        network_fragment_proofs=proof_rows,
        local_endpoint_roles=_local_endpoint_role_rows(chunks, chunk_specs),
    )
    local_verification, independent_local_results = (
        _verify_network_local_asr_intersection(
            local_verification,
            trajectory,
            chunks,
            anchor,
            proof_rows,
            candidate_seed=candidate_seed,
            transcriber=transcriber,
        )
    )
    return CandidateVerification(
        local_verification,
        independent_local_results=independent_local_results,
    )


def _verify_sequence_trajectory_audio(
    sequence_result,
    chunks: tuple[str, ...],
    whole_target_text: str,
    anchor: np.ndarray,
    playback_speed: float,
    independent_cache: WholeWaveformVerificationCache,
    chunk_specs: tuple[GenerationChunkSpec, ...] | None = None,
    *,
    release_transcriber=transcribe_breeze25,
    independent_transcriber=transcribe_breeze25,
):
    """Verify one ranked DP path after exact production assembly."""

    waveform = _assemble_trajectory_audio(
        sequence_result.trajectory,
        chunks,
        playback_speed,
        chunk_specs,
    )
    # Local and assembled checks share one request-local Breeze25 transcript
    # cache. The semantic-only projection remains a separate hard gate over
    # the same exact transcript, never a second decoder invocation.
    assembled_verification = _verify_release_whole_audio(
        waveform,
        whole_target_text,
        anchor,
        independent_cache,
        transcriber=release_transcriber,
        verifier_profile=BREEZE25_RELEASE_VERIFICATION_PROFILE,
    )
    status = "verified" if assembled_verification.passed else "rejected"
    print(
        f"[BlueMagpie] sequence path Breeze25 {status} "
        f"rank={sequence_result.sequence_path_rank} "
        f"chunk_candidates={sequence_result.chunk_candidate_indices} "
        f"{_verification_metric_log_fields(assembled_verification)}"
    )
    if not assembled_verification.passed:
        return assembled_verification
    independent_verification = _verify_independent_whole_audio(
        waveform,
        whole_target_text,
        anchor,
        independent_cache,
        transcriber=independent_transcriber,
    )
    intersected = qualify_trajectory_with_joined_output(
        assembled_verification,
        independent_verification,
    )
    independent_status = "verified" if independent_verification.passed else "rejected"
    print(
        f"[BlueMagpie] sequence path Breeze25 projection {independent_status} "
        f"rank={sequence_result.sequence_path_rank} "
        f"chunk_candidates={sequence_result.chunk_candidate_indices} "
        f"{_verification_metric_log_fields(independent_verification)}"
    )
    return intersected


@dataclass(frozen=True)
class _GlyphAttemptReservation:
    carrier_segment_id: str
    candidate_index: int
    candidate_ordinal: int
    network_conditioned: bool
    attempt_id: str
    seed: int
    text: str


@dataclass(frozen=True)
class _GlyphCompletedAttempt:
    reservation: _GlyphAttemptReservation
    endpoint_outcome: _ChunkGenerationOutcome
    record: GenerationAttemptRecord


class _GlyphRequestAttemptLedger:
    """Transient exact accounting for every model invocation in one request."""

    def __init__(self) -> None:
        self._records: list[GenerationAttemptRecord] = []
        self._record_by_carrier_candidate: dict[
            tuple[str, int],
            GenerationAttemptRecord,
        ] = {}
        self._selected_by_carrier: dict[str, GenerationAttemptRecord] = {}
        self._completed: list[_GlyphCompletedAttempt] = []
        self._selected_waveform_by_carrier: dict[str, np.ndarray] = {}

    @property
    def records(self) -> tuple[GenerationAttemptRecord, ...]:
        return tuple(self._records)

    @property
    def completed_attempts(self) -> tuple[_GlyphCompletedAttempt, ...]:
        return tuple(self._completed)

    def begin(
        self,
        carrier_segment_id: str,
        candidate_chunks: tuple[str, ...],
        seed: int,
        generation_context: CandidateGenerationContext,
        candidate_ordinal: int,
        network_conditioned: bool,
    ) -> _GlyphAttemptReservation:
        if (
            type(carrier_segment_id) is not str
            or not carrier_segment_id
            or type(candidate_chunks) is not tuple
            or len(candidate_chunks) != 1
            or len(generation_context.chunk_indices) != 1
            or generation_context.chunk_candidate_ordinals
            != (candidate_ordinal,)
            or type(network_conditioned) is not bool
        ):
            raise ValueError(
                "one hybrid carrier candidate must be one exact model invocation"
            )
        text = candidate_chunks[0]
        units = count_speech_units(text)
        generated_chunks = sum(
            record.generated_chunk_count for record in self._records
        )
        generated_units = sum(
            record.generated_text_units for record in self._records
        )
        if (
            units <= 0
            or generated_chunks + 1 > GLYPH_MAX_GENERATED_CHUNK_BUDGET
            or generated_units + units > GLYPH_MAX_GENERATED_TEXT_UNIT_BUDGET
        ):
            raise ValueError("hybrid request exceeds its frozen generation budget")
        key = (carrier_segment_id, generation_context.candidate_index)
        if key in self._record_by_carrier_candidate:
            raise ValueError("hybrid carrier candidate invocation is duplicated")
        return _GlyphAttemptReservation(
            carrier_segment_id=carrier_segment_id,
            candidate_index=generation_context.candidate_index,
            candidate_ordinal=candidate_ordinal,
            network_conditioned=network_conditioned,
            attempt_id=(
                f"glyph-carrier:{carrier_segment_id}:attempt:{len(self._records)}"
            ),
            seed=int(seed),
            text=text,
        )

    def complete(
        self,
        reservation: _GlyphAttemptReservation,
        endpoint_outcomes: tuple[_ChunkGenerationOutcome, ...],
    ) -> None:
        if (
            type(reservation) is not _GlyphAttemptReservation
            or type(endpoint_outcomes) is not tuple
            or len(endpoint_outcomes) != 1
        ):
            raise ValueError("hybrid generation endpoint evidence is incomplete")
        outcome = endpoint_outcomes[0]
        endpoint_sha256 = _canonical_json_sha256(
            {
                "schema": "bluemagpie.glyph-carrier.endpoint.v1",
                "stop_reason": outcome.stop_reason,
                "tail_energy_ratio": float(outcome.endpoint_energy_ratio),
                "generated_steps": int(outcome.generated_steps),
                "hard_cap_steps": int(outcome.hard_stop_steps),
            }
        )
        record = GenerationAttemptRecord(
            ordinal=len(self._records),
            attempt_id=reservation.attempt_id,
            seed=reservation.seed,
            text_sha256=hashlib.sha256(
                reservation.text.encode("utf-8")
            ).hexdigest(),
            endpoint_evidence_sha256=endpoint_sha256,
            generated_chunk_count=1,
            generated_text_units=count_speech_units(reservation.text),
        )
        key = (
            reservation.carrier_segment_id,
            reservation.candidate_index,
        )
        if key in self._record_by_carrier_candidate:
            raise ValueError("hybrid generation completion is duplicated")
        self._records.append(record)
        self._record_by_carrier_candidate[key] = record
        self._completed.append(
            _GlyphCompletedAttempt(
                reservation=reservation,
                endpoint_outcome=outcome,
                record=record,
            )
        )

    def select(
        self,
        carrier_segment_id: str,
        cascade,
        waveform: np.ndarray,
    ) -> None:
        if (
            carrier_segment_id in self._selected_by_carrier
            or len(cascade.chunk_candidate_indices) != 1
        ):
            raise ValueError("hybrid carrier selection is inconsistent")
        key = (carrier_segment_id, cascade.chunk_candidate_indices[0])
        record = self._record_by_carrier_candidate.get(key)
        if record is None:
            raise ValueError("hybrid selected carrier lacks generation provenance")
        self._selected_by_carrier[carrier_segment_id] = record
        selected_waveform = np.asarray(waveform, dtype=np.float32).reshape(-1).copy()
        if selected_waveform.size <= 0 or not np.isfinite(selected_waveform).all():
            raise ValueError("hybrid selected carrier waveform is invalid")
        self._selected_waveform_by_carrier[carrier_segment_id] = selected_waveform

    def selected(self, carrier_segment_id: str) -> GenerationAttemptRecord:
        record = self._selected_by_carrier.get(carrier_segment_id)
        if record is None:
            raise ValueError("hybrid selected carrier provenance is unavailable")
        return record

    def selected_waveform(self, carrier_segment_id: str) -> np.ndarray:
        waveform = self._selected_waveform_by_carrier.get(carrier_segment_id)
        if waveform is None:
            raise ValueError("hybrid selected carrier waveform is unavailable")
        return waveform.copy()


class _GlyphCarrierGenerationAudit:
    def __init__(
        self,
        request_ledger: _GlyphRequestAttemptLedger,
        carrier_segment_id: str,
    ) -> None:
        self._request_ledger = request_ledger
        self._carrier_segment_id = carrier_segment_id

    def begin(
        self,
        candidate_chunks: tuple[str, ...],
        seed: int,
        generation_context: CandidateGenerationContext,
        candidate_ordinal: int,
        network_conditioned: bool,
    ) -> _GlyphAttemptReservation:
        return self._request_ledger.begin(
            self._carrier_segment_id,
            candidate_chunks,
            seed,
            generation_context,
            candidate_ordinal,
            network_conditioned,
        )

    def complete(
        self,
        reservation: _GlyphAttemptReservation,
        endpoint_outcomes: tuple[_ChunkGenerationOutcome, ...],
    ) -> None:
        self._request_ledger.complete(reservation, endpoint_outcomes)

    def select(self, cascade, waveform: np.ndarray) -> None:
        self._request_ledger.select(
            self._carrier_segment_id,
            cascade,
            waveform,
        )

    def selected(self) -> GenerationAttemptRecord:
        return self._request_ledger.selected(self._carrier_segment_id)

    def selected_waveform(self) -> np.ndarray:
        return self._request_ledger.selected_waveform(self._carrier_segment_id)


def _quality_request_generated_chunk_limit(
    normalized_units: int,
    *,
    network_request: bool,
    carrier_request: bool,
) -> int:
    """Bound interactive retries without changing network/carrier contracts."""

    if (
        not network_request
        and not carrier_request
        and normalized_units <= QUALITY_MEDIUM_TEXT_MAX_UNITS
    ):
        return QUALITY_INTERACTIVE_MAX_GENERATED_CHUNKS
    return QUALITY_MAX_GENERATED_CHUNKS


def _fast_text_chunks(text: str) -> tuple[str, ...]:
    """Split text cheaply without invoking any release-gate planner."""

    text = str(text).strip()
    if not text:
        return ()
    chunks: list[str] = []
    break_characters = frozenset("。！？!?；;，,、：:\n")
    while len(text) > CHUNK_CHARS:
        cut = CHUNK_CHARS
        for index in range(CHUNK_CHARS - 1, CHUNK_CHARS // 2 - 1, -1):
            if text[index] in break_characters:
                cut = index + 1
                break
        chunk = text[:cut].strip()
        if chunk:
            chunks.append(chunk)
        text = text[cut:].strip()
    if text:
        chunks.append(text)
    return tuple(chunks)


def _prepare_reference_audio(reference_wav: str) -> tuple[str, str | None]:
    """Deterministically center-crop long references to the trained 6s cap."""

    info = sf.info(reference_wav)
    max_frames = int(round(REFERENCE_AUDIO_SECONDS * int(info.samplerate)))
    if int(info.frames) <= max_frames:
        return reference_wav, None
    start = max(0, (int(info.frames) - max_frames) // 2)
    waveform, sample_rate = sf.read(
        reference_wav,
        start=start,
        frames=max_frames,
        dtype="float32",
        always_2d=True,
    )
    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as handle:
        temporary_path = handle.name
    try:
        sf.write(
            temporary_path,
            waveform,
            sample_rate,
            subtype="PCM_16",
        )
    except Exception:
        os.unlink(temporary_path)
        raise
    return temporary_path, temporary_path


def _generate_fast_chunk(
    text: str,
    centroid: torch.Tensor | None,
    *,
    cfg: float,
    steps: int,
    request_seed: int,
    reference_wav_path: str = "",
    short_text_min_cfg: float = SHORT_TEXT_CFG_MIN,
) -> np.ndarray:
    """Run one model call and return it without candidate qualification."""

    generation_cps = select_generation_cps(
        text,
        cjk_cps=BASE_GENERATION_POLICY.cjk_cps,
        ascii_cps=BASE_GENERATION_POLICY.ascii_cps,
    )
    model_text, expected_steps, hard_stop_steps = endpoint_generation_plan(
        text,
        generation_cps=generation_cps,
        step_seconds=STEP_SECONDS,
        margin_steps=BASE_GENERATION_POLICY.hard_stop_margin_steps,
        add_terminal_punctuation=count_speech_units(text) >= MIN_ENDPOINT_CUE_UNITS,
    )
    generation_cfg = effective_generation_cfg(
        text,
        float(cfg),
        short_text_unit_threshold=SHORT_TEXT_CFG_UNITS,
        short_text_min_cfg=float(short_text_min_cfg),
    )
    set_generation_seed(request_seed)
    kwargs = {
        "target_text": model_text,
        "speaker_centroid": centroid,
        "cfg_value": generation_cfg,
        "inference_timesteps": int(steps),
        "min_len": 2,
        "max_len": hard_stop_steps,
        "retry_badcase": False,
    }
    if reference_wav_path:
        kwargs["reference_wav_path"] = reference_wav_path
    if _NATIVE_STOP_POLICY:
        kwargs["stop_threshold"] = STOP_THRESHOLD
        kwargs["stop_consecutive"] = STOP_CONSECUTIVE
    if "generation_seed" in _GENERATE_PARAMETERS:
        kwargs["generation_seed"] = request_seed
    if _STOP_CONTROLLER is not None:
        _STOP_CONTROLLER.begin(
            2,
            expected_steps=expected_steps,
            hard_stop_steps=hard_stop_steps,
        )
    try:
        with latency_stage("generation"):
            audio = model.generate(**kwargs)
            return audio.detach().float().cpu().numpy().reshape(-1)
    finally:
        if _STOP_CONTROLLER is not None:
            _STOP_CONTROLLER.end()


def _assemble_fast_audio(trajectory: tuple[np.ndarray, ...]) -> np.ndarray:
    """Concatenate first-pass chunks with a short seam and no analysis."""

    if not trajectory:
        return np.zeros(1, dtype=np.float32)
    if len(trajectory) == 1:
        waveform = np.asarray(trajectory[0], dtype=np.float32).reshape(-1)
    else:
        seam = np.zeros(int(round(0.12 * SR)), dtype=np.float32)
        parts: list[np.ndarray] = []
        for index, audio in enumerate(trajectory):
            if index:
                parts.append(seam)
            parts.append(np.asarray(audio, dtype=np.float32).reshape(-1))
        waveform = np.concatenate(parts)
    return finish_audio(waveform, SR, fade_ms=FINAL_ENDPOINT_FADE_MS)


def _synthesize(
    text: str,
    centroid: torch.Tensor | None,
    *,
    cfg: float,
    steps: int,
    speed: float,
    generation_audit: _GlyphCarrierGenerationAudit | None = None,
    request_seed: int | None = None,
    speaker_projector_scale: float = 1.0,
    reference_wav_path: str = "",
    short_text_min_cfg: float = SHORT_TEXT_CFG_MIN,
) -> tuple[int, np.ndarray]:
    """Latency-first single-pass synthesis with no output qualification."""

    del speed, generation_audit
    raw_text = str(text)
    try:
        normalized = normalize_spoken_forms(raw_text, locale="zh-TW")
    except (TypeError, ValueError):
        normalized = raw_text
    normalized = strip_unspoken_cjk_quotes_for_generation(normalized)
    chunks = _fast_text_chunks(normalized)
    if not chunks:
        return SR, np.zeros(1, dtype=np.float32)
    seed = (
        secrets.randbelow(2_147_483_648)
        if request_seed is None
        else int(request_seed)
    )
    with _GENERATION_LOCK:
        projector_hook = None
        if speaker_projector_scale != 1.0:
            projector_hook = model.speaker_projector.register_forward_hook(
                lambda _module, _inputs, output: output * speaker_projector_scale
            )
        try:
            trajectory = tuple(
                _generate_fast_chunk(
                    chunk,
                    centroid,
                    cfg=float(cfg),
                    steps=int(steps),
                    request_seed=seed,
                    reference_wav_path=reference_wav_path,
                    short_text_min_cfg=short_text_min_cfg,
                )
                for chunk in chunks
            )
        finally:
            if projector_hook is not None:
                projector_hook.remove()
    waveform = _assemble_fast_audio(trajectory)
    print(
        "[BlueMagpie] latency-first synthesis "
        f"chunks={len(chunks)} model_calls={len(chunks)} verification=disabled"
    )
    return SR, waveform


def _synthesize_verified(
    text: str,
    centroid: torch.Tensor,
    *,
    cfg: float,
    steps: int,
    speed: float,
    generation_audit: _GlyphCarrierGenerationAudit | None = None,
    request_seed: int | None = None,
) -> tuple[int, np.ndarray]:
    def emit_public_cascade_evidence(line: str) -> None:
        # Carrier-local cascades are implementation details of one hybrid
        # request.  Publishing schema 6 here would make an external evaluator
        # mistake a carrier for the returned request.  Ordinary requests keep
        # their byte-for-byte existing schema-6 emission.
        if generation_audit is None:
            print(line)

    if isinstance(steps, (bool, np.bool_)):
        raise gr.Error(f"目前只支援已驗證的 NFE {DEFAULT_STEPS}。")
    try:
        requested_steps = int(steps)
        exact_steps = float(steps) == float(requested_steps)
    except (TypeError, ValueError, OverflowError):
        requested_steps = -1
        exact_steps = False
    if not exact_steps or requested_steps != DEFAULT_STEPS:
        raise gr.Error(f"目前只支援已驗證的 NFE {DEFAULT_STEPS}。")
    steps = requested_steps
    try:
        cfg_value = float(cfg)
    except (TypeError, ValueError, OverflowError):
        raise gr.Error("CFG 必須是有限數值。") from None
    if not np.isfinite(cfg_value) or not 1.0 <= cfg_value <= 4.0:
        raise gr.Error("CFG 必須介於 1.0 與 4.0。")
    if cfg_value != MIXED_CFG_PRIMARY:
        raise gr.Error(f"目前只支援已驗證的主 CFG {MIXED_CFG_PRIMARY:.1f}。")
    raw_text = str(text)
    if len(raw_text) > MAX_TEXT_CHARS:
        raise gr.Error(f"單次最多 {MAX_TEXT_CHARS} 個字元，請分段合成。")
    if network_identifier_has_ambiguous_iri(raw_text):
        raise gr.Error("網址目前只支援 ASCII 字元；非 ASCII IRI 會與字母讀音混淆。")
    network_request = contains_network_identifier(raw_text)
    request_cfg = MIXED_CFG_PRIMARY
    try:
        text = normalize_spoken_forms(raw_text, locale="zh-TW")
    except ValueError as error:
        raise gr.Error(str(error)) from None
    if not text:
        raise gr.Error("請先輸入要合成的文字。")
    if not network_request:
        text = strip_unspoken_cjk_quotes_for_generation(text)
    normalized_units = count_speech_units(text)
    if normalized_units <= 0:
        raise gr.Error("輸入必須包含可發音的文字或數字。")
    if normalized_units > QUALITY_MAX_GENERATED_TEXT_UNITS:
        raise gr.Error("正規化後的語音單位超過安全生成上限，請分段合成。")
    if not np.isfinite(float(speed)) or float(speed) != 1.0:
        raise gr.Error("後處理語速固定為 1.0；減速已由品質管線控制。")

    chunk_specs: tuple[GenerationChunkSpec, ...] | None = None
    try:
        if network_request:
            chunk_specs = plan_generation_chunks(
                raw_text,
                text,
                min_units=MIN_CHUNK_CHARS,
                network_min_units=NETWORK_GENERATION_MIN_UNITS,
                target_units=NETWORK_GENERATION_TARGET_UNITS,
                network_max_units=NETWORK_GENERATION_MAX_UNITS,
                ordinary_max_units=NETWORK_REQUEST_ORDINARY_MAX_UNITS,
                split_atomic_email_at_separator=(
                    NETWORK_SPLIT_ATOMIC_EMAIL_AT_SEPARATOR
                ),
            )
            chunks = tuple(spec.text for spec in chunk_specs)
            if len(chunks) > QUALITY_MAX_GENERATED_CHUNKS:
                raise ValueError(
                    "network request exceeds the generated-chunk budget"
                )
        else:
            medium_chunk_units = (
                GLYPH_CARRIER_MEDIUM_CHUNK_UNITS
                if generation_audit is not None
                else QUALITY_MEDIUM_CHUNK_UNITS
            )
            chunks = tuple(
                coalesce_text_chunks(
                    split_quality_text_for_tts(
                        text,
                        long_max_units=CHUNK_CHARS,
                        min_chunk_units=MIN_CHUNK_CHARS,
                        medium_max_units=QUALITY_MEDIUM_TEXT_MAX_UNITS,
                        medium_chunk_units=medium_chunk_units,
                    ),
                    max_chunks=QUALITY_MAX_GENERATED_CHUNKS,
                    max_units=CHUNK_CHARS,
                )
            )
    except ValueError as error:
        raise gr.Error("文字無法在已驗證的生成限制內安全分段。") from error
    request_generated_chunk_limit = _quality_request_generated_chunk_limit(
        normalized_units,
        network_request=network_request,
        carrier_request=generation_audit is not None,
    )
    request_seed = resolve_request_seed(request_seed, secrets.randbelow)
    anchor = _speaker_anchor_array(centroid)
    independent_cache = WholeWaveformVerificationCache()
    transcript_cache = WholeWaveformTranscriptCache()
    cached_breeze25_transcriber = _cached_request_transcriber(
        transcript_cache,
        BREEZE25_TRANSCRIPT_PROFILE,
        transcribe_breeze25,
    )
    generation_context_by_seed: dict[int, CandidateGenerationContext] = {}
    generation_endpoint_evidence_by_seed: dict[
        int,
        tuple[_ChunkGenerationOutcome, ...],
    ] = {}

    def canonical_chunks_for_context(
        generation_context: CandidateGenerationContext,
    ) -> tuple[str, ...]:
        canonical: list[str] = []
        for chunk_index in generation_context.chunk_indices:
            if not 0 <= chunk_index < len(chunks):
                raise ValueError("candidate generation provenance is out of range")
            canonical.append(chunks[chunk_index])
        return tuple(canonical)

    def candidate_generation_text_transform(
        canonical_chunks: tuple[str, ...],
        generation_context: CandidateGenerationContext,
    ) -> tuple[str, ...]:
        expected_canonical = canonical_chunks_for_context(generation_context)
        if canonical_chunks != expected_canonical:
            raise ValueError("candidate generation text lacks canonical provenance")
        if chunk_specs is None:
            return canonical_chunks
        transformed: list[str] = []
        for chunk_index, chunk, candidate_ordinal in zip(
            generation_context.chunk_indices,
            canonical_chunks,
            generation_context.chunk_candidate_ordinals,
            strict=True,
        ):
            spec = chunk_specs[chunk_index]
            if spec.text != chunk:
                raise ValueError("candidate generation text does not match its proof")
            transformed.append(
                email_domain_mail_generation_variant(
                    chunk,
                    spec.network_fragment_proofs,
                )
                if candidate_ordinal in EMAIL_MAIL_FALLBACK_CANDIDATE_ORDINALS
                else chunk
            )
        return tuple(transformed)

    def candidate_generation_text_variants(
        canonical_chunks: tuple[str, ...],
        generation_context: CandidateGenerationContext,
    ) -> tuple[str, ...]:
        expected_canonical = canonical_chunks_for_context(generation_context)
        if canonical_chunks != expected_canonical:
            raise ValueError("candidate text variant lacks canonical provenance")
        generated_chunks = candidate_generation_text_transform(
            canonical_chunks,
            generation_context,
        )
        variants: list[str] = []
        for chunk_index, canonical, generated, candidate_ordinal in zip(
            generation_context.chunk_indices,
            canonical_chunks,
            generated_chunks,
            generation_context.chunk_candidate_ordinals,
            strict=True,
        ):
            if generated == canonical:
                variants.append(BASE_CHUNK_TEXT_VARIANT)
                continue
            if chunk_specs is None:
                raise ValueError("candidate text variant lacks network proof")
            spec = chunk_specs[chunk_index]
            expected_variant = email_domain_mail_generation_variant(
                canonical,
                spec.network_fragment_proofs,
            )
            if (
                candidate_ordinal not in EMAIL_MAIL_FALLBACK_CANDIDATE_ORDINALS
                or generated != expected_variant
                or generated == canonical
                or not any(
                    proof.identifier_kind == "email"
                    for proof in spec.network_fragment_proofs
                )
            ):
                raise ValueError("candidate text variant is not proof-bound")
            variants.append(EMAIL_DOMAIN_MAIL_CHUNK_TEXT_VARIANT)
        return tuple(variants)

    def generation_chunk_specs(
        seed: int,
        candidate_chunks: tuple[str, ...],
    ) -> tuple[GenerationChunkSpec, ...] | None:
        if chunk_specs is None:
            return None
        context = generation_context_by_seed.get(seed)
        if context is None or len(context.chunk_indices) != len(candidate_chunks):
            raise ValueError("candidate verification lacks generation provenance")
        selected: list[GenerationChunkSpec] = []
        for chunk_index, chunk in zip(
            context.chunk_indices,
            candidate_chunks,
            strict=True,
        ):
            if not 0 <= chunk_index < len(chunk_specs):
                raise ValueError("candidate verification provenance is out of range")
            spec = chunk_specs[chunk_index]
            if spec.text != chunk:
                raise ValueError("candidate verification provenance text does not match")
            selected.append(spec)
        return tuple(selected)

    def candidate_cfg(candidate_ordinal: int) -> float:
        return generation_cfg_for_candidate_offset(
            candidate_ordinal,
            primary_cfg=request_cfg,
            alternate_cfg=MIXED_CFG_ALTERNATE,
        )

    def chunk_cfg_evidence(
        chunk: str,
        candidate_ordinal: int,
        *,
        network_conditioned: bool | None = None,
    ) -> tuple[float, tuple[str, ...]]:
        scheduled = generation_cfg_for_candidate_offset(
            candidate_ordinal,
            primary_cfg=request_cfg,
            alternate_cfg=MIXED_CFG_ALTERNATE,
        )
        reasons: list[str] = []
        network_adjusted = scheduled
        is_network = (
            bool(network_protected_spoken_spans(chunk))
            if network_conditioned is None
            else bool(network_conditioned)
        )
        if is_network and scheduled < NETWORK_TEXT_CFG_MIN:
            network_adjusted = NETWORK_TEXT_CFG_MIN
            reasons.append("network")
        effective = effective_generation_cfg(
            chunk,
            network_adjusted,
            short_text_unit_threshold=SHORT_TEXT_CFG_UNITS,
            short_text_min_cfg=SHORT_TEXT_CFG_MIN,
        )
        if effective > network_adjusted:
            reasons.append("short_text")
        return effective, tuple(reasons)

    def generation_network_flags(
        chunk_indices: tuple[int, ...],
        candidate_chunks: tuple[str, ...],
        *,
        generation_context: CandidateGenerationContext,
    ) -> tuple[bool, ...]:
        if len(chunk_indices) != len(candidate_chunks):
            raise ValueError("generation provenance does not align with chunks")
        canonical_chunks = canonical_chunks_for_context(generation_context)
        expected_generation_chunks = candidate_generation_text_transform(
            canonical_chunks,
            generation_context,
        )
        if candidate_chunks != expected_generation_chunks:
            raise ValueError("candidate generation text disagrees with its schedule")
        if chunk_specs is None:
            return tuple(
                bool(network_protected_spoken_spans(chunk))
                for chunk in canonical_chunks
            )
        flags: list[bool] = []
        for chunk_index, chunk in zip(
            chunk_indices,
            canonical_chunks,
            strict=True,
        ):
            if not 0 <= chunk_index < len(chunk_specs):
                raise ValueError("generation provenance index is out of range")
            spec = chunk_specs[chunk_index]
            if spec.text != chunk:
                raise ValueError("generation provenance text does not match")
            flags.append(spec.network_conditioned)
        return tuple(flags)

    def effective_chunk_cfg(
        chunk: str,
        candidate_ordinal: int,
        *,
        network_conditioned: bool | None = None,
    ) -> float:
        return chunk_cfg_evidence(
            chunk,
            candidate_ordinal,
            network_conditioned=network_conditioned,
        )[0]

    def candidate_generation(
        candidate_chunks: tuple[str, ...],
        seed: int,
        *,
        generation_context: CandidateGenerationContext,
    ) -> tuple[np.ndarray, ...]:
        if generation_context.seed != seed:
            raise ValueError("generation context seed does not match the request seed")
        previous_context = generation_context_by_seed.get(seed)
        if previous_context is not None and previous_context != generation_context:
            raise ValueError("one candidate seed cannot carry two generation contexts")
        generation_context_by_seed[seed] = generation_context
        ordinals = generation_context.chunk_candidate_ordinals
        if not ordinals or len(set(ordinals)) != 1:
            raise ValueError("one generation call must use one candidate ordinal")
        candidate_ordinal = ordinals[0]
        network_flags = generation_network_flags(
            generation_context.chunk_indices,
            candidate_chunks,
            generation_context=generation_context,
        )
        audit_reservation = (
            generation_audit.begin(
                candidate_chunks,
                seed,
                generation_context,
                candidate_ordinal,
                network_flags[0],
            )
            if generation_audit is not None
            else None
        )
        trajectory, endpoint_outcomes = _generate_trajectory(
            candidate_chunks,
            centroid,
            cfg=candidate_cfg(candidate_ordinal),
            steps=steps,
            request_seed=seed,
            policy=generation_policy_for_candidate_offset(candidate_ordinal),
            network_conditioned=network_flags,
        )
        if len(endpoint_outcomes) != len(candidate_chunks):
            raise ValueError("generation endpoint evidence is incomplete")
        if generation_audit is not None:
            generation_audit.complete(
                audit_reservation,
                endpoint_outcomes,
            )
        generation_endpoint_evidence_by_seed[seed] = endpoint_outcomes
        return trajectory

    def candidate_generation_evidence(
        candidate_index: int,
        seed: int,
        chunk_indices: tuple[int, ...],
        candidate_chunks: tuple[str, ...],
        *,
        generation_context: CandidateGenerationContext,
    ) -> CandidateGenerationEvidence:
        if candidate_index != seed - request_seed:
            raise ValueError("candidate index does not match request seed offset")
        if (
            generation_context.candidate_index != candidate_index
            or generation_context.seed != seed
            or generation_context.chunk_indices != chunk_indices
        ):
            raise ValueError("candidate generation context does not match the attempt")
        candidate_ordinals = generation_context.chunk_candidate_ordinals
        if not candidate_ordinals or len(set(candidate_ordinals)) != 1:
            raise ValueError("one generation call must use one candidate ordinal")
        candidate_ordinal = candidate_ordinals[0]
        scheduled = candidate_cfg(candidate_ordinal)
        network_flags = generation_network_flags(
            chunk_indices,
            candidate_chunks,
            generation_context=generation_context,
        )
        canonical_chunks = canonical_chunks_for_context(generation_context)
        expected_generation_chunks = candidate_generation_text_transform(
            canonical_chunks,
            generation_context,
        )
        if candidate_chunks != expected_generation_chunks:
            raise ValueError("generation evidence text disagrees with its schedule")
        text_variants = candidate_generation_text_variants(
            canonical_chunks,
            generation_context,
        )
        rows = tuple(
            chunk_cfg_evidence(
                chunk,
                candidate_ordinal,
                network_conditioned=network_flag,
            )
            for chunk, network_flag in zip(
                candidate_chunks,
                network_flags,
                strict=True,
            )
        )
        endpoint_outcomes = generation_endpoint_evidence_by_seed.get(seed)
        if endpoint_outcomes is None or len(endpoint_outcomes) != len(
            candidate_chunks
        ):
            raise ValueError("generation endpoint evidence is missing")
        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(row[0] for row in rows),
            floor_reasons=tuple(row[1] for row in rows),
            chunk_candidate_ordinals=candidate_ordinals,
            network_conditioned=network_flags,
            chunk_text_variants=text_variants,
            chunk_stop_reasons=tuple(
                outcome.stop_reason for outcome in endpoint_outcomes
            ),
            chunk_endpoint_energy_ratios=tuple(
                outcome.endpoint_energy_ratio for outcome in endpoint_outcomes
            ),
            chunk_generated_steps=tuple(
                outcome.generated_steps for outcome in endpoint_outcomes
            ),
            chunk_hard_stop_steps=tuple(
                outcome.hard_stop_steps for outcome in endpoint_outcomes
            ),
        )

    try:
        with _GENERATION_LOCK:
            cascade = run_coverage_adaptive_cascade(
                chunks,
                request_seed,
                candidate_generation,
                lambda trajectory, candidate_chunks, seed: _qualify_candidate_trajectory_audio(
                    trajectory,
                    candidate_chunks,
                    text,
                    anchor,
                    speed,
                    independent_cache,
                    candidate_seed=seed,
                    chunk_specs=generation_chunk_specs(seed, candidate_chunks),
                    release_transcriber=cached_breeze25_transcriber,
                    independent_transcriber=cached_breeze25_transcriber,
                ),
                lambda trajectory, candidate_chunks, seed: (
                    _qualify_candidate_trajectory_audio(
                        trajectory,
                        candidate_chunks,
                        text,
                        anchor,
                        speed,
                        independent_cache,
                        candidate_seed=seed,
                        chunk_specs=generation_chunk_specs(seed, candidate_chunks),
                        release_transcriber=cached_breeze25_transcriber,
                        independent_transcriber=cached_breeze25_transcriber,
                    )
                    if len(chunks) == 1
                    else _verify_refill_candidate_trajectory_audio(
                        trajectory,
                        candidate_chunks,
                        anchor,
                        speed,
                        generation_chunk_specs(seed, candidate_chunks),
                        candidate_seed=seed,
                        transcriber=cached_breeze25_transcriber,
                    )
                ),
                sequence_final_verifier=lambda sequence_result, candidate_chunks: (
                    _verify_sequence_trajectory_audio(
                        sequence_result,
                        candidate_chunks,
                        text,
                        anchor,
                        speed,
                        independent_cache,
                        chunk_specs,
                        release_transcriber=cached_breeze25_transcriber,
                        independent_transcriber=cached_breeze25_transcriber,
                    )
                ),
                generation_evidence_factory=candidate_generation_evidence,
                candidate_generation_text_transform=(
                    candidate_generation_text_transform
                ),
                transition_artifact_enricher=(
                    _enrich_transition_artifact_f0
                ),
                max_generated_chunks=request_generated_chunk_limit,
                max_generated_text_units=QUALITY_MAX_GENERATED_TEXT_UNITS,
                max_sequence_paths=QUALITY_MAX_SEQUENCE_PATHS,
                sequence_fallback_max_local_boundary_speaker_drop=(
                    SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP
                ),
                preferred_min_speaker_similarity=(
                    QUALITY_PREFERRED_MIN_SPEAKER_SIMILARITY
                ),
                preferred_max_boundary_speaker_drop=(
                    QUALITY_PREFERRED_MAX_BOUNDARY_SPEAKER_DROP
                ),
                preferred_min_squim_stoi=QUALITY_PREFERRED_MIN_SQUIM_STOI,
                preferred_min_squim_pesq=QUALITY_PREFERRED_MIN_SQUIM_PESQ,
                preferred_min_squim_audio_duration_seconds=(
                    QUALITY_PREFERRED_SQUIM_MIN_DURATION_SECONDS
                ),
                single_exact_min_speaker_similarity=(
                    QUALITY_FAST_EXACT_MIN_SPEAKER_SIMILARITY
                ),
                single_exact_max_boundary_speaker_drop=(
                    QUALITY_FAST_EXACT_MAX_BOUNDARY_SPEAKER_DROP
                ),
                require_endpoint_evidence=True,
            )
    except NoQualifiedCandidateError as error:
        emit_public_cascade_evidence(
            format_cascade_evidence_log(
                error.diagnostics,
                outcome="no_qualified_candidate",
                generated_chunk_limit=request_generated_chunk_limit,
                generated_text_unit_limit=QUALITY_MAX_GENERATED_TEXT_UNITS,
            )
        )
        raise gr.Error("目前沒有候選通過內容與音色驗證，請稍後重試或調整文字。") from error
    except (RuntimeError, ValueError) as error:
        raise gr.Error("品質驗證暫時無法完成，未回傳未驗證的語音。") from error

    attempts_by_index = {
        attempt.candidate_index: attempt for attempt in cascade.diagnostics.attempts
    }
    selected_ordinals: list[int] = []
    for chunk_index, candidate_index in enumerate(cascade.chunk_candidate_indices):
        attempt = attempts_by_index.get(candidate_index)
        if attempt is None or chunk_index not in attempt.chunk_indices:
            raise gr.Error("品質驗證紀錄不完整，未回傳未驗證的語音。")
        local_index = attempt.chunk_indices.index(chunk_index)
        try:
            selected_ordinals.append(attempt.chunk_candidate_ordinals[local_index])
        except IndexError as error:
            raise gr.Error("品質驗證紀錄不完整，未回傳未驗證的語音。") from error
    selected_policies = tuple(
        generation_policy_for_candidate_offset(ordinal).name
        for ordinal in selected_ordinals
    )
    attempted_policies = tuple(
        generation_policy_for_candidate_offset(
            attempt.chunk_candidate_ordinals[0]
        ).name
        for attempt in cascade.diagnostics.attempts
    )
    selected_network_flags = (
        tuple(spec.network_conditioned for spec in chunk_specs)
        if chunk_specs is not None
        else tuple(
            bool(network_protected_spoken_spans(chunk)) for chunk in chunks
        )
    )
    selected_cfgs = tuple(
        effective_chunk_cfg(
            chunk,
            candidate_ordinal,
            network_conditioned=network_flag,
        )
        for chunk, candidate_ordinal, network_flag in zip(
            chunks,
            selected_ordinals,
            selected_network_flags,
            strict=True,
        )
    )
    attempted_schedule_cfgs = tuple(
        candidate_cfg(attempt.chunk_candidate_ordinals[0])
        for attempt in cascade.diagnostics.attempts
    )
    print(
        "[BlueMagpie] quality cascade "
        f"candidate_index={cascade.candidate_index} attempts={len(cascade.attempted_seeds)} "
        f"selection={cascade.selection_mode} "
        f"chunk_candidates={cascade.chunk_candidate_indices} "
        f"chunk_seeds={cascade.chunk_seeds} score={cascade.verification.score:.6f}"
        f" chunk_policies={selected_policies} attempted_policies={attempted_policies}"
        f" chunk_cfgs={selected_cfgs}"
        f" attempted_schedule_cfgs={attempted_schedule_cfgs}"
        f" sequence_rank={cascade.sequence_path_rank}"
        f" sequence_paths_checked={cascade.sequence_paths_checked}"
        f" generated_chunks={cascade.generated_chunk_count}"
        f" generated_text_units={cascade.generated_text_units}"
        f" chunk_candidate_counts={cascade.chunk_candidate_counts}"
        f" request_cfg={request_cfg:.2f} mixed_cfg_primary={MIXED_CFG_PRIMARY:.2f}"
        f" network_cfg_floor={NETWORK_TEXT_CFG_MIN:.2f}"
        f" cfg_schedule={MIXED_CFG_SCHEDULE}"
        f" network_request={network_request}"
    )
    waveform = _assemble_trajectory_audio(
        cascade.trajectory,
        chunks,
        speed,
        chunk_specs,
    )
    final_verification = _verify_release_whole_audio(
        waveform,
        text,
        anchor,
        independent_cache,
        transcriber=cached_breeze25_transcriber,
        verifier_profile=BREEZE25_RELEASE_VERIFICATION_PROFILE,
    )
    final_evidence = trajectory_gate_evidence(final_verification)
    independent_final_evidence = None
    independent_final_verification = None
    try:
        require_verified_final_output(final_verification)
        independent_final_verification = _verify_independent_whole_audio(
            waveform,
            text,
            anchor,
            independent_cache,
            transcriber=cached_breeze25_transcriber,
        )
        independent_final_evidence = trajectory_gate_evidence(
            independent_final_verification
        )
        require_verified_final_output(independent_final_verification)
    except FinalOutputRejectedError as error:
        independent_fields = (
            "not_run"
            if independent_final_verification is None
            else _verification_metric_log_fields(independent_final_verification)
        )
        emit_public_cascade_evidence(
            format_cascade_evidence_log(
                cascade.diagnostics,
                outcome="final_output_rejected",
                generated_chunk_limit=request_generated_chunk_limit,
                generated_text_unit_limit=QUALITY_MAX_GENERATED_TEXT_UNITS,
                selection=cascade,
                final_output=final_evidence,
                independent_final_output=independent_final_evidence,
            )
        )
        print(
            "[BlueMagpie] final output rejected "
            f"breeze25={_verification_metric_log_fields(final_verification)} "
            f"semantic_projection={independent_fields}"
        )
        raise gr.Error("最終合成結果未通過整段內容、語速與音色驗證，未回傳音訊。") from error
    except (RuntimeError, ValueError) as error:
        raise gr.Error("品質驗證暫時無法完成，未回傳未驗證的語音。") from error
    emit_public_cascade_evidence(
        format_cascade_evidence_log(
            cascade.diagnostics,
            outcome="returned",
            generated_chunk_limit=request_generated_chunk_limit,
            generated_text_unit_limit=QUALITY_MAX_GENERATED_TEXT_UNITS,
            selection=cascade,
            final_output=final_evidence,
            independent_final_output=independent_final_evidence,
        )
    )
    print(
        "[BlueMagpie] final output verified "
        f"breeze25_score={final_verification.score:.6f} "
        f"semantic_projection_score={independent_final_verification.score:.6f} "
        f"independent_cache_entries={independent_cache.entry_count} "
        f"transcript_cache_entries={transcript_cache.entry_count}"
    )
    if generation_audit is not None:
        generation_audit.select(cascade, waveform)
    return SR, waveform


def _glyph_request_network_looking(text: object) -> bool:
    folded = str(text).lower()
    return any(marker in folded for marker in GLYPH_NETWORK_MARKERS)


def _glyph_profile_selected(speaker: object) -> bool:
    return (
        type(speaker) is str
        and speaker == DEFAULT_SPEAKER
        and DEFAULT_SPEAKER == "內建語者 B"
        and SPEAKERS.get(speaker) is DEFAULT_CENTROID
    )


def _validate_glyph_runtime_controls(
    *,
    cfg: object,
    steps: object,
    speed: object,
) -> None:
    if isinstance(steps, (bool, np.bool_)):
        raise gr.Error(f"目前只支援已驗證的 NFE {DEFAULT_STEPS}。")
    try:
        exact_steps = float(steps) == float(DEFAULT_STEPS)
        cfg_value = float(cfg)
        speed_value = float(speed)
    except (TypeError, ValueError, OverflowError):
        raise gr.Error("V5 字元語音只支援固定的已驗證參數。") from None
    if (
        not exact_steps
        or not np.isfinite(cfg_value)
        or cfg_value != MIXED_CFG_PRIMARY
        or not np.isfinite(speed_value)
        or speed_value != 1.0
    ):
        raise gr.Error("V5 字元語音只支援 CFG 3.0、NFE 10 與原始語速 1.0。")


def _validate_glyph_carrier_text(carrier_text: str) -> None:
    try:
        normalized = normalize_spoken_forms(carrier_text, locale="zh-TW")
    except ValueError as error:
        raise GlyphHybridPlanError("hybrid carrier normalization failed") from error
    generated_text = strip_unspoken_cjk_quotes_for_generation(normalized)
    if normalized != carrier_text or generated_text != carrier_text:
        raise GlyphHybridPlanError(
            "hybrid carrier text is not stable under the legacy frontend"
        )
    chunks = tuple(
        coalesce_text_chunks(
            split_quality_text_for_tts(
                generated_text,
                long_max_units=CHUNK_CHARS,
                min_chunk_units=MIN_CHUNK_CHARS,
                medium_max_units=QUALITY_MEDIUM_TEXT_MAX_UNITS,
                medium_chunk_units=GLYPH_CARRIER_MEDIUM_CHUNK_UNITS,
            ),
            max_chunks=QUALITY_MAX_GENERATED_CHUNKS,
            max_units=CHUNK_CHARS,
        )
    )
    if chunks != (carrier_text,):
        raise GlyphHybridPlanError(
            "one renderer carrier must remain one exact generation chunk"
        )


def _require_glyph_final_verification(verification) -> None:
    require_verified_final_output(verification)
    if len(verification.candidate_results) != 1:
        raise FinalOutputRejectedError(
            "glyph final verifier did not return one whole-waveform result"
        )
    result = verification.candidate_results[0]
    comparison = result.comparison
    if (
        not np.isfinite(float(comparison.cer))
        or float(comparison.cer) > GLYPH_FINAL_MAX_CER
        or comparison.prefix_deletions != 0
        or comparison.suffix_deletions != 0
        or comparison.extra_tail_units != 0
    ):
        raise FinalOutputRejectedError(
            "glyph final semantic gate exceeded the frozen release limits"
        )


def _glyph_inverse_proof_sha256(
    plan: HybridRenderPlan,
    reconstructed_raw: str,
) -> str:
    return _canonical_json_sha256(
        {
            "schema": "bluemagpie.glyph-hybrid.inverse.v1",
            "plan_sha256": plan.plan_sha256,
            "raw_input_sha256": hashlib.sha256(
                reconstructed_raw.encode("utf-8")
            ).hexdigest(),
            "canonical_target_sha256": plan.canonical_target_sha256,
        }
    )


def _emit_glyph_hybrid_evidence(
    *,
    raw_text: str,
    plan: HybridRenderPlan,
    profile_state: _GlyphProfileState,
    segments: tuple[AssetSegment | CarrierSegment, ...],
    assembly,
    semantic_commitment: SemanticPlanCommitment,
    request_ledger: _GlyphRequestAttemptLedger,
    final_verification,
    independent_final_verification,
) -> None:
    adapter = _GLYPH_HYBRID_EVIDENCE_ADAPTER
    if not callable(adapter):
        raise RuntimeError("schema-7 hybrid evidence adapter is unavailable")
    serialized = adapter(
        raw_text=raw_text,
        plan=plan,
        profile_state=profile_state,
        segments=segments,
        assembly=assembly,
        semantic_commitment=semantic_commitment,
        request_ledger=request_ledger,
        final_verification=final_verification,
        independent_final_verification=independent_final_verification,
    )
    if type(serialized) is not bytes or not serialized or b"\n" in serialized:
        raise RuntimeError("schema-7 hybrid evidence is not one canonical record")
    try:
        payload = json.loads(serialized.decode("ascii"))
    except (UnicodeDecodeError, json.JSONDecodeError) as error:
        raise RuntimeError("schema-7 hybrid evidence is invalid") from error
    try:
        canonical_serialized = json.dumps(
            payload,
            ensure_ascii=True,
            allow_nan=False,
            separators=(",", ":"),
            sort_keys=True,
        ).encode("ascii")
    except (TypeError, ValueError) as error:
        raise RuntimeError("schema-7 hybrid evidence is not finite JSON") from error
    if canonical_serialized != serialized:
        raise RuntimeError("schema-7 hybrid evidence is not canonical JSON")

    forbidden_content_keys = {
        "raw_input",
        "raw_text",
        "canonical_target",
        "canonical_text",
        "target_text",
        "transcript_text",
        "text",
    }

    def contains_content(value: object) -> bool:
        if isinstance(value, dict):
            return bool(forbidden_content_keys.intersection(value)) or any(
                contains_content(item) for item in value.values()
            )
        if isinstance(value, list):
            return any(contains_content(item) for item in value)
        return False

    if not isinstance(payload, dict):
        raise RuntimeError("schema-7 hybrid evidence root is invalid")
    request_proof = payload.get("request_proof")
    strict_gates = (
        request_proof.get("additional_strict_gates")
        if isinstance(request_proof, dict)
        else None
    )
    final_gate = (
        request_proof.get("final_gate")
        if isinstance(request_proof, dict)
        else None
    )
    independent_gate = (
        request_proof.get("independent_gate")
        if isinstance(request_proof, dict)
        else None
    )
    exact_pcm_sha256 = hashlib.sha256(assembly.pcm16_le).hexdigest()
    budget_rows = (
        request_proof.get("budgets")
        if isinstance(request_proof, dict)
        else None
    )
    expected_budget_names = (
        "model_invocations",
        "generated_chunks",
        "generated_speech_units",
        "asset_count",
        "asset_atoms",
        "asset_canonical_units",
        "asset_audio_samples",
        "asr_calls",
        "ecapa_calls",
        "squim_calls",
    )
    valid_budget_rows = (
        isinstance(budget_rows, list)
        and len(budget_rows) == len(expected_budget_names)
        and all(
            isinstance(row, dict)
            and set(row) == {"name", "actual", "cap"}
            and type(row["name"]) is str
            and type(row["actual"]) is int
            and type(row["cap"]) is int
            for row in budget_rows
        )
        and tuple(row["name"] for row in budget_rows) == expected_budget_names
    )
    budgets = (
        {
            row["name"]: (row["actual"], row["cap"])
            for row in budget_rows
        }
        if valid_budget_rows
        else {}
    )
    records = request_ledger.records
    renderer_contract = payload.get("renderer_contract")
    renderer_contract_sha256 = payload.get("renderer_contract_sha256")
    digest_payload = dict(payload)
    evidence_sha256 = digest_payload.pop("evidence_sha256", None)
    recomputed_evidence_sha256 = _canonical_json_sha256(digest_payload)
    expected_strict_gate_names = {
        "strict_inverse_reconstruction",
        "asset_reconstruction",
        "public_pcm_equality",
    }

    def has_exact_internal_bindings() -> bool:
        if not isinstance(request_proof, dict):
            return False
        segment_rows = request_proof.get("segments")
        call_rows = payload.get("runtime_call_ledger")
        selection_rows = payload.get("carrier_selections")
        if not all(
            isinstance(rows, list)
            for rows in (segment_rows, call_rows, selection_rows)
        ):
            return False
        if not all(
            isinstance(row, dict)
            for rows in (segment_rows, call_rows, selection_rows)
            for row in rows
        ):
            return False
        segment_by_id = {
            row.get("segment_id"): row
            for row in segment_rows
            if type(row.get("segment_id")) is str
        }
        call_by_id = {
            row.get("call_id"): row
            for row in call_rows
            if type(row.get("call_id")) is str
        }
        if (
            len(segment_by_id) != len(segment_rows)
            or len(call_by_id) != len(call_rows)
        ):
            return False

        canonical_commitment = request_proof.get(
            "canonical_target_sha256"
        )
        for gate in (final_gate, independent_gate):
            if not isinstance(gate, dict):
                return False
            asr_call = call_by_id.get(gate.get("asr_call_id"))
            if (
                not isinstance(asr_call, dict)
                or asr_call.get("kind") != "asr"
                or "subject_model_call_id" not in asr_call
                or asr_call.get("subject_model_call_id") is not None
                or asr_call.get("scope") != gate.get("scope")
                or asr_call.get("target_sha256")
                != canonical_commitment
            ):
                return False

        asset_segments = {
            segment_id: row
            for segment_id, row in segment_by_id.items()
            if row.get("kind") == "asset"
        }
        asset_loads: dict[str, list[dict]] = {}
        for row in call_rows:
            if row.get("kind") == "asset_load":
                asset_loads.setdefault(row.get("segment_id"), []).append(row)
        if set(asset_loads) != set(asset_segments):
            return False
        for segment_id, segment in asset_segments.items():
            loads = asset_loads[segment_id]
            if len(loads) != 1:
                return False
            loaded = loads[0]
            if (
                loaded.get("asset_id") != segment.get("asset_id")
                or loaded.get("manifest_entry_sha256")
                != segment.get("asset_manifest_entry_sha256")
                or loaded.get("loaded_pcm_sha256")
                != segment.get("realized_pcm_sha256")
                or loaded.get("loaded_sample_count")
                != segment.get("output_sample_count")
            ):
                return False

        carrier_segments = {
            segment_id: row
            for segment_id, row in segment_by_id.items()
            if row.get("kind") == "carrier"
        }
        selections_by_segment: dict[str, list[dict]] = {}
        for row in selection_rows:
            selections_by_segment.setdefault(
                row.get("segment_id"),
                [],
            ).append(row)
        if set(selections_by_segment) != set(carrier_segments):
            return False
        for segment_id, segment in carrier_segments.items():
            selections = selections_by_segment[segment_id]
            if len(selections) != 1:
                return False
            selected = call_by_id.get(selections[0].get("selected_call_id"))
            if (
                not isinstance(selected, dict)
                or selected.get("kind") != "model_generation"
                or selected.get("segment_id") != segment_id
                or selected.get("text_sha256")
                != segment.get("carrier_text_sha256")
                or selected.get("output_pcm_sha256")
                != segment.get("realized_pcm_sha256")
                or selected.get("output_sample_count")
                != segment.get("output_sample_count")
            ):
                return False
        return True

    if (
        payload.get("schema") != "bluemagpie.glyph-hybrid.evidence.v7"
        or payload.get("version") != 7
        or contains_content(payload)
        or evidence_sha256 != recomputed_evidence_sha256
        or not isinstance(renderer_contract, dict)
        or renderer_contract_sha256
        != _canonical_json_sha256(renderer_contract)
        or renderer_contract.get("speaker_anchor_sha256")
        != GLYPH_SPEAKER_ANCHOR_SHA256
        or not isinstance(request_proof, dict)
        or request_proof.get("content_commitment_scheme")
        != "hmac-sha256-v1"
        or not has_exact_internal_bindings()
        or request_proof.get("assembled_pcm_sha256") != exact_pcm_sha256
        or request_proof.get("public_pcm_sha256") != exact_pcm_sha256
        or budgets.get("model_invocations")
        != (len(records), GLYPH_MAX_GENERATED_CHUNK_BUDGET)
        or budgets.get("generated_chunks")
        != (
            sum(record.generated_chunk_count for record in records),
            GLYPH_MAX_GENERATED_CHUNK_BUDGET,
        )
        or budgets.get("generated_speech_units")
        != (
            sum(record.generated_text_units for record in records),
            GLYPH_MAX_GENERATED_TEXT_UNIT_BUDGET,
        )
        or not valid_budget_rows
        or not isinstance(strict_gates, dict)
        or set(strict_gates) != expected_strict_gate_names
        or not all(value is True for value in strict_gates.values())
        or not isinstance(final_gate, dict)
        or final_gate.get("passed") is not True
        or not isinstance(independent_gate, dict)
        or independent_gate.get("passed") is not True
    ):
        raise RuntimeError("schema-7 hybrid evidence is not bound to this output")
    print(GLYPH_EVIDENCE_LOG_PREFIX + serialized.decode("ascii"))


def _synthesize_glyph_hybrid(
    raw_text: str,
    plan: HybridRenderPlan,
    profile_state: _GlyphProfileState,
    *,
    cfg: float,
    steps: int,
    speed: float,
) -> tuple[int, np.ndarray]:
    if (
        not profile_state.available
        or profile_state.bundle is None
        or glyph_hybrid_plan_sha256(plan) != plan.plan_sha256
    ):
        raise GlyphHybridPlanError("hybrid profile or renderer plan is unavailable")
    manifest_entries = profile_state.asset_entry_sha256_by_asset_id
    reconstructed_raw = inverse_glyph_hybrid_plan(
        plan,
        asset_entry_sha256_by_asset_id=manifest_entries,
        expected_raw_request=raw_text,
    )
    if reconstructed_raw != raw_text:
        raise GlyphHybridPlanError("hybrid inverse reconstruction failed")
    inverse_proof_sha256 = _glyph_inverse_proof_sha256(
        plan,
        reconstructed_raw,
    )
    request_seed = resolve_request_seed(None, secrets.randbelow)
    request_ledger = _GlyphRequestAttemptLedger()
    assembled_segments: list[AssetSegment | CarrierSegment] = []
    carrier_count = 0
    for segment in plan.segments:
        segment_id = f"glyph-plan-segment-{segment.ordinal}"
        if segment.kind == "asset_atom":
            if segment.asset_id is None:
                raise GlyphHybridPlanError("asset segment has no immutable asset id")
            assembled_segments.append(
                AssetSegment(
                    segment_id=segment_id,
                    asset_id=segment.asset_id,
                    speed=1.0,
                )
            )
        elif segment.kind == "carrier":
            _validate_glyph_carrier_text(segment.canonical_text)
            generation_audit = _GlyphCarrierGenerationAudit(
                request_ledger,
                segment_id,
            )
            carrier_rate, carrier_waveform = _synthesize(
                segment.canonical_text,
                DEFAULT_CENTROID,
                cfg=cfg,
                steps=steps,
                speed=1.0,
                generation_audit=generation_audit,
                request_seed=request_seed,
            )
            if (
                carrier_rate != SR
                or not np.array_equal(
                    carrier_waveform,
                    generation_audit.selected_waveform(),
                )
            ):
                raise RuntimeError("hybrid carrier finalization is inconsistent")
            selected = generation_audit.selected()
            assembled_segments.append(
                CarrierSegment(
                    segment_id=segment_id,
                    audio=carrier_waveform,
                    selected_attempt_id=selected.attempt_id,
                    selected_seed=selected.seed,
                    selected_text_sha256=selected.text_sha256,
                    selected_endpoint_evidence_sha256=(
                        selected.endpoint_evidence_sha256
                    ),
                    sample_rate=SR,
                    speed=1.0,
                )
            )
            carrier_count += 1
        elif segment.kind not in {"control", "separator"}:
            raise GlyphHybridPlanError("renderer plan has an unknown segment kind")

    immutable_segments = tuple(assembled_segments)
    if not immutable_segments:
        raise GlyphHybridPlanError("hybrid renderer produced no audio segments")
    records = request_ledger.records
    if (carrier_count == 0) != (len(records) == 0):
        raise RuntimeError("hybrid request model accounting is inconsistent")
    commitment = SemanticPlanCommitment(
        external_semantic_plan_sha256=plan.plan_sha256,
        inverse_proof_sha256=inverse_proof_sha256,
        ordered_attempt_ledger_sha256=(
            compute_ordered_attempt_ledger_sha256(records)
        ),
        attempt_record_count=len(records),
        assembled_carrier_count=carrier_count,
        request_model_generator_invocations=len(records),
        request_generated_chunk_count=sum(
            record.generated_chunk_count for record in records
        ),
        request_generated_text_units=sum(
            record.generated_text_units for record in records
        ),
        request_generated_chunk_budget=GLYPH_MAX_GENERATED_CHUNK_BUDGET,
        request_generated_text_unit_budget=(
            GLYPH_MAX_GENERATED_TEXT_UNIT_BUDGET
        ),
        ordered_attempt_records=records,
    )
    assembly_plan_sha256 = compute_hybrid_plan_sha256(
        profile_state.bundle,
        immutable_segments,
        semantic_commitment=commitment,
    )
    with latency_stage("assemble"):
        assembly = assemble_hybrid_pcm16(
            profile_state.bundle,
            immutable_segments,
            semantic_commitment=commitment,
            expected_plan_sha256=assembly_plan_sha256,
            speed=1.0,
            post_gain=1.0,
        )
    verify_hybrid_assembly(
        assembly,
        trusted_bundle=profile_state.bundle,
        original_segments=immutable_segments,
        expected_semantic_commitment=commitment,
    )
    if (
        inverse_glyph_hybrid_plan(
            plan,
            asset_entry_sha256_by_asset_id=manifest_entries,
            expected_raw_request=raw_text,
        )
        != raw_text
    ):
        raise GlyphHybridPlanError("post-assembly inverse reconstruction failed")
    public_pcm = assembly.public_pcm16()
    waveform = public_pcm.astype(np.float32) / np.float32(32768.0)
    if (
        assembly.public_pcm16_le is not assembly.verification_pcm16_le
        or not np.array_equal(
            pcm16_audio_output(assembly.sample_rate, waveform)[1],
            public_pcm,
        )
    ):
        raise RuntimeError("hybrid public and verification PCM differ")
    anchor = _speaker_anchor_array(DEFAULT_CENTROID)
    independent_cache = WholeWaveformVerificationCache()
    final_verification = _verify_trajectory_audio(
        (waveform,),
        (plan.canonical_target,),
        anchor,
        1.0,
        QUALITY_FINAL_ASR_MAX_NEW_TOKENS,
        release_speaker_gate=True,
        transcriber=_transcribe_glyph_hybrid_breeze_primary,
    )
    independent_final_verification = None
    try:
        _require_glyph_final_verification(final_verification)
        independent_final_verification = _verify_independent_whole_audio(
            waveform,
            plan.canonical_target,
            anchor,
            independent_cache,
            transcriber=_transcribe_glyph_hybrid_breeze_confirmation,
        )
        _require_glyph_final_verification(independent_final_verification)
    except FinalOutputRejectedError as error:
        raise gr.Error(
            "V5 字元語音未通過整段內容、音色、音質與語速驗證，未回傳音訊。"
        ) from error
    _emit_glyph_hybrid_evidence(
        raw_text=raw_text,
        plan=plan,
        profile_state=profile_state,
        segments=immutable_segments,
        assembly=assembly,
        semantic_commitment=commitment,
        request_ledger=request_ledger,
        final_verification=final_verification,
        independent_final_verification=independent_final_verification,
    )
    return assembly.sample_rate, waveform


def _maybe_synthesize_glyph_speaker_request(
    text: object,
    speaker: object,
    *,
    cfg: object,
    steps: object,
    speed: object,
) -> tuple[int, np.ndarray] | None:
    if not _glyph_profile_selected(speaker):
        return None
    raw_text = str(text)
    if not _glyph_request_network_looking(raw_text):
        return None
    _validate_glyph_runtime_controls(cfg=cfg, steps=steps, speed=speed)
    profile_state = _GLYPH_PROFILE_STATE
    if not profile_state.available:
        raise gr.Error(
            "V5 字元語音資產目前不可用或完整性驗證失敗，未改用未驗證路徑。"
        )
    try:
        plan = build_glyph_hybrid_plan(
            raw_text,
            asset_entry_sha256_by_asset_id=(
                profile_state.asset_entry_sha256_by_asset_id
            ),
        )
    except (GlyphHybridPlanError, GlyphAssetError) as error:
        raise gr.Error("V5 字元語音規劃失敗，未改用未驗證路徑。") from error
    if plan is None:
        raise gr.Error(
            "此網路字串不符合 V5 嚴格小寫 ASCII 規格，未進行部分合成或舊路徑 fallback。"
        )
    try:
        return _synthesize_glyph_hybrid(
            raw_text,
            plan,
            profile_state,
            cfg=float(cfg),
            steps=int(steps),
            speed=float(speed),
        )
    except gr.Error:
        raise
    except (GlyphHybridPlanError, GlyphAssetError, RuntimeError, ValueError) as error:
        raise gr.Error(
            "V5 字元語音完整性或品質驗證失敗，未回傳未驗證音訊。"
        ) from error


@speaker_gpu
@timed_latency_request
def tts_speaker(
    text: str,
    speaker: str = DEFAULT_SPEAKER,
    cfg: float = DEFAULT_CFG,
    steps: int = DEFAULT_STEPS,
    speed: float = 1.0,
    seed: int = SPEAKER_GENERATION_SEED,
):
    return pcm16_audio_output(
        *_synthesize(
            text,
            SPEAKERS.get(speaker, DEFAULT_CENTROID),
            cfg=cfg,
            steps=steps,
            speed=speed,
            request_seed=int(seed),
            speaker_projector_scale=(
                SPEAKER_B_PROJECTOR_SCALE
                if speaker == SPEAKER_B_LABEL
                else 1.0
            ),
        )
    )


@reference_gpu
@timed_latency_request
def tts_reference(
    text: str,
    reference_wav: str,
    cfg: float = REFERENCE_CFG,
    steps: int = DEFAULT_STEPS,
    speed: float = 1.0,
    seed: int = REFERENCE_GENERATION_SEED,
):
    if not reference_wav:
        raise gr.Error("請先錄音或上傳參考音檔。")
    prepared_reference, temporary_reference = _prepare_reference_audio(reference_wav)
    try:
        return pcm16_audio_output(
            *_synthesize(
                text,
                None,
                cfg=cfg,
                steps=steps,
                speed=speed,
                request_seed=int(seed),
                reference_wav_path=prepared_reference,
                short_text_min_cfg=REFERENCE_CFG,
            )
        )
    finally:
        if temporary_reference is not None:
            os.unlink(temporary_reference)

EXAMPLE_TEXTS = [
    "今天天氣真好，我們一起去散步吧。",
    "我要吃蚵仔煎，然後去丟垃圾。",
    "這學期的成績包括研究報告和期末考。",
    "這是 AI TTS code switching 測試，混合中英文也沒問題。",
    "注音符號測試：ㄅ、ㄆ、ㄇ、ㄈ。",
    "今天的會議會先整理目前進度，再確認下一階段的工作。遇到需要討論的項目時，"
    "請先記下問題，等報告結束後再一起處理。最後，我們會確認負責人和預計完成時間。",
]


HEADER = """
# BlueMagpie-TTS Demo

台灣華語與中英混合文字轉語音。模型版本：`BlueMagpie-TTS`。
"""


with gr.Blocks(title="BlueMagpie-TTS Demo", theme=gr.themes.Soft()) as demo:
    gr.Markdown(HEADER)

    with gr.Accordion("進階生成參數", open=False):
        with gr.Row():
            cfg_input = gr.Number(
                value=DEFAULT_CFG,
                label="內建語者 CFG",
                interactive=True,
            )
            reference_cfg_input = gr.Number(
                value=REFERENCE_CFG,
                label="參考音色 CFG",
                interactive=True,
            )
            steps_input = gr.Number(
                value=DEFAULT_STEPS,
                precision=0,
                interactive=True,
                label="NFE steps",
            )
            speed_input = gr.Number(
                value=1.0,
                interactive=False,
                label="後處理語速（效率模式停用）",
            )
        with gr.Row():
            speaker_seed_input = gr.Number(
                value=SPEAKER_GENERATION_SEED,
                minimum=0,
                maximum=2_147_483_647,
                precision=0,
                interactive=True,
                label="內建語者 Seed",
            )
            reference_seed_input = gr.Number(
                value=REFERENCE_GENERATION_SEED,
                minimum=0,
                maximum=2_147_483_647,
                precision=0,
                interactive=True,
                label="參考音色 Seed",
            )

    with gr.Tab("內建語者"):
        with gr.Row():
            with gr.Column():
                speaker_input = gr.Dropdown(list(SPEAKERS), value=DEFAULT_SPEAKER, label="語者")
                speaker_text = gr.Textbox(label="文字", lines=4, max_lines=8)
                speaker_button = gr.Button("合成", variant="primary")
            with gr.Column():
                speaker_output = gr.Audio(label="合成結果", type="numpy")
        gr.Examples(EXAMPLE_TEXTS, inputs=speaker_text, label="範例")
        speaker_button.click(
            tts_speaker,
            [
                speaker_text,
                speaker_input,
                cfg_input,
                steps_input,
                speed_input,
                speaker_seed_input,
            ],
            speaker_output,
        )

    with gr.Tab("參考音色"):
        gr.Markdown(
            "建議錄製至少 6 秒、單一語者且背景乾淨的授權參考音檔；"
            "超過 6 秒會固定取中央 6 秒。錄音會直接作為模型的 "
            "reference audio，不轉成 ECAPA 音色。"
        )
        with gr.Row():
            with gr.Column():
                reference_text = gr.Textbox(label="文字", lines=4, max_lines=8)
                reference_audio = gr.Audio(
                    label="參考音檔",
                    type="filepath",
                    sources=["microphone", "upload"],
                )
                reference_button = gr.Button("合成", variant="primary")
            with gr.Column():
                reference_output = gr.Audio(label="合成結果", type="numpy")
        reference_button.click(
            tts_reference,
            [
                reference_text,
                reference_audio,
                reference_cfg_input,
                steps_input,
                speed_input,
                reference_seed_input,
            ],
            reference_output,
        )

    gr.Markdown(
        "合成語音僅供研究與評估展示；正式使用前請人工檢視。 "
        "[模型](https://huggingface.co/OpenFormosa/BlueMagpie-TTS) · "
        "[程式碼](https://github.com/OpenFormosa/BlueMagpie-TTS)"
    )


if __name__ == "__main__":
    demo.queue(default_concurrency_limit=1).launch(ssr_mode=False)