"""BlueMagpie-TTS Gradio demo using the validated production inference profile."""

from __future__ import annotations

from dataclasses import dataclass, replace
import hashlib
from importlib import metadata as importlib_metadata
import inspect
import json
import os
from pathlib import Path
import secrets
import threading
from types import MappingProxyType

# Cross-process release reproducibility is a hard contract.  Configure CUDA
# before importing Spaces/Torch so cuBLAS and Mamba choose their deterministic
# paths.  ZeroGPU must then be imported before every CUDA-related package.
DETERMINISTIC_ALGORITHMS = True
CUBLAS_WORKSPACE_CONFIG = ":4096:8"
MAMBA_DETERMINISTIC = "1"
CUDNN_BENCHMARK = False
CUDNN_DETERMINISTIC = True
CUDA_MATMUL_ALLOW_TF32 = False
CUDNN_ALLOW_TF32 = False
EXPECTED_TORCH_VERSION = "2.11.0+cu130"
EXPECTED_TORCH_CUDA_VERSION = "13.0"
EXPECTED_CUDNN_VERSION = 91900
EXPECTED_MAMBA_SSM_VERSION = "2.3.2.post1+bluemagpie.triton1"
EXPECTED_TRITON_VERSION = "3.6.0"
if (
    os.environ.setdefault(
        "CUBLAS_WORKSPACE_CONFIG",
        CUBLAS_WORKSPACE_CONFIG,
    )
    != CUBLAS_WORKSPACE_CONFIG
):
    raise RuntimeError("CUBLAS_WORKSPACE_CONFIG conflicts with the release contract")
if (
    os.environ.setdefault(
        "MAMBA_DETERMINISTIC",
        MAMBA_DETERMINISTIC,
    )
    != MAMBA_DETERMINISTIC
):
    raise RuntimeError("MAMBA_DETERMINISTIC conflicts with the release contract")

GPU_LEASE_SECONDS = 120

try:
    import spaces

    gpu = spaces.GPU(duration=GPU_LEASE_SECONDS)
    USING_ZERO_GPU = True
except ImportError:
    def gpu(function):
        return function

    USING_ZERO_GPU = False

import gradio as gr
import librosa
import numpy as np
import torch
from huggingface_hub import snapshot_download
from mamba_ssm import __version__ as mamba_ssm_runtime_version
from transformers import PreTrainedTokenizerFast

runtime_versions = {
    "torch": str(torch.__version__),
    "cuda": str(torch.version.cuda),
    "cudnn": torch.backends.cudnn.version(),
    "mamba-ssm": mamba_ssm_runtime_version,
    "triton": importlib_metadata.version("triton"),
}
expected_runtime_versions = {
    "torch": EXPECTED_TORCH_VERSION,
    "cuda": EXPECTED_TORCH_CUDA_VERSION,
    "cudnn": EXPECTED_CUDNN_VERSION,
    "mamba-ssm": EXPECTED_MAMBA_SSM_VERSION,
    "triton": EXPECTED_TRITON_VERSION,
}
if runtime_versions != expected_runtime_versions:
    raise RuntimeError(
        "ZeroGPU runtime conflicts with the deterministic release contract: "
        f"expected={expected_runtime_versions!r} actual={runtime_versions!r}"
    )

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

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 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,
    extract_windowed_speaker_embedding,
    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,
    GLYPH_HYBRID_MAX_VERIFICATION_SEGMENTS as QUALITY_GLYPH_HYBRID_MAX_VERIFICATION_SEGMENTS,
    SAFE_DURATION_GENERATION_POLICY,
    VERIFICATION_WHISPER_MODEL_ID,
    VERIFICATION_WHISPER_REVISION,
    WHISPER_MAX_VERIFICATION_SEGMENTS,
    WHISPER_MODEL_ID,
    WHISPER_REVISION,
    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_verification_whisper,
    transcribe_whisper,
    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 = "aaf1a0878e37875382bb0e5c8a3a2ba43be67297"
ECAPA_REPO_ID = "speechbrain/spkrec-ecapa-voxceleb"
ECAPA_REVISION = "0f99f2d0ebe89ac095bcc5903c4dd8f72b367286"
DEFAULT_CFG = 3.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 = 80
ONSET_CLAUSE_SEARCH_CHARS = 0
MIN_CHUNK_CHARS = 12
QUALITY_MEDIUM_TEXT_MAX_UNITS = 48
QUALITY_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_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 already passed both ASR checks 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 preferred 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
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 Whisper segment caps disagree")
VERIFICATION_ASR_PROFILE = (
    f"{VERIFICATION_WHISPER_MODEL_ID}@{VERIFICATION_WHISPER_REVISION}:"
    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}"
)
TURBO_TRANSCRIPT_PROFILE = (
    f"{WHISPER_MODEL_ID}@{WHISPER_REVISION}:greedy:zh:transcribe:"
    f"max_tokens{QUALITY_FINAL_ASR_MAX_NEW_TOKENS}:"
    f"segments{WHISPER_MAX_VERIFICATION_SEGMENTS}"
)
LARGE_V3_TRANSCRIPT_PROFILE = (
    f"{VERIFICATION_WHISPER_MODEL_ID}@{VERIFICATION_WHISPER_REVISION}:"
    f"greedy:zh:transcribe:max_tokens{QUALITY_FINAL_ASR_MAX_NEW_TOKENS}:"
    f"segments{WHISPER_MAX_VERIFICATION_SEGMENTS}"
)
RELEASE_TURBO_VERIFICATION_PROFILE = (
    f"{TURBO_TRANSCRIPT_PROFILE}:release-speaker-squim-echo-wsola-v2:"
    f"pace{QUALITY_MAX_PACE_CPS}"
)
RELEASE_LARGE_V3_VERIFICATION_PROFILE = (
    f"{LARGE_V3_TRANSCRIPT_PROFILE}:release-speaker-squim-echo-wsola-v2:"
    f"pace{QUALITY_MAX_PACE_CPS}"
)


print(f"[BlueMagpie] downloading model from {REPO_ID}@{MODEL_REVISION} ...")
MODEL_DIR = snapshot_download(REPO_ID, revision=MODEL_REVISION)
print(f"[BlueMagpie] caching speaker encoder from {ECAPA_REPO_ID}@{ECAPA_REVISION} ...")
ECAPA_DIR = snapshot_download(ECAPA_REPO_ID, revision=ECAPA_REVISION)
print(f"[BlueMagpie] caching quality ASR from {WHISPER_MODEL_ID}@{WHISPER_REVISION} ...")
ASR_DIR = snapshot_download(WHISPER_MODEL_ID, revision=WHISPER_REVISION)
print(
    "[BlueMagpie] caching independent final ASR from "
    f"{VERIFICATION_WHISPER_MODEL_ID}@{VERIFICATION_WHISPER_REVISION} ..."
)
VERIFICATION_ASR_DIR = snapshot_download(
    VERIFICATION_WHISPER_MODEL_ID,
    revision=VERIFICATION_WHISPER_REVISION,
)
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)
    speaker_ids = [str(value) for value in table["speaker_ids"]]
    centroids = table["centroids"]
    labels = {f"內建語者 {chr(65 + index)}": centroid for index, centroid in enumerate(centroids)}

    requested_id = (
        METADATA.get("recommended_generation_defaults", {}).get("speaker_id")
        if isinstance(METADATA.get("recommended_generation_defaults"), dict)
        else None
    )
    if requested_id not in speaker_ids and "female_voice" in speaker_ids:
        requested_id = "female_voice"
    default_index = speaker_ids.index(requested_id) if requested_id in speaker_ids else 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 = _load_glyph_profile_state()
_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
    with _ECAPA_LOCK:
        if _ECAPA_ENCODER is None:
            import torchaudio

            # SpeechBrain 1.0.3 still probes this API during import, while the
            # ZeroGPU torchaudio build has removed it. Audio loading below is
            # handled by librosa, 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 _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,
        )
    try:
        audio = model.generate(**kwargs)
    finally:
        if _STOP_CONTROLLER is not None:
            _STOP_CONTROLLER.end()
    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}"
    )
    audio = audio.detach().float().cpu().numpy().reshape(-1)

    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)


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 _attach_transition_f0(
    verification,
    prepared_waveforms: tuple[np.ndarray | None, ...],
    *,
    collect_transition_f0: bool,
    local_candidate_pool: bool,
):
    """Measure F0 only for rows that can enter multi-chunk sequence DP."""

    if not collect_transition_f0:
        return verification
    if not local_candidate_pool:
        raise ValueError("transition F0 is only valid for local candidate pools")
    if (
        len(prepared_waveforms) != len(verification.candidate_results)
        or len(prepared_waveforms) != len(verification.chunk_artifacts)
    ):
        raise RuntimeError("transition F0 evidence does not align")

    updated_artifacts: list[ChunkCandidateArtifact] = []
    for waveform, result, artifact in zip(
        prepared_waveforms,
        verification.candidate_results,
        verification.chunk_artifacts,
        strict=True,
    ):
        median_f0_hz = None
        if waveform is not None and local_candidate_has_coverage_eligibility(
            result,
            max_local_boundary_speaker_drop=(
                SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP
            ),
        ):
            try:
                median_f0_hz = active_audio_median_f0_hz(waveform, SR)
            except (RuntimeError, TypeError, ValueError):
                median_f0_hz = None
        updated_artifacts.append(
            replace(artifact, median_f0_hz=median_f0_hz)
        )
    return replace(
        verification,
        chunk_artifacts=tuple(updated_artifacts),
    )


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_whisper,
    semantic_only: bool = False,
    network_fragment_proofs: (
        tuple[tuple[NetworkFragmentProof, ...], ...] | None
    ) = None,
    local_endpoint_roles: tuple[tuple[bool, bool], ...] | None = None,
    collect_transition_f0: bool = False,
):
    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,
        )
    ):
        prepared = prepare_candidate_audio(
            audio,
            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
        try:
            duration = active_voiced_duration_seconds(waveform, SR)
        except ValueError:
            duration = 0.0
        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 ""
            )
        pace_cps = (
            count_speech_units(chunk) / duration * float(playback_speed)
            if duration > 0.0
            else None
        )
        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=(
                    count_speech_units(chunk) / 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 _attach_transition_f0(
        verification,
        tuple(prepared_waveforms),
        collect_transition_f0=collect_transition_f0,
        local_candidate_pool=local_candidate_pool,
    )


def _verify_independent_whole_audio(
    waveform: np.ndarray,
    target_text: str,
    anchor: np.ndarray,
    cache: WholeWaveformVerificationCache,
    *,
    transcriber=transcribe_verification_whisper,
):
    """Verify exact assembled audio with the independent full large-v3 ASR."""

    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,
        VERIFICATION_ASR_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_turbo(
    audio: np.ndarray,
    sample_rate: int,
    *,
    language: str = "zh",
    task: str = "transcribe",
    max_new_tokens: int = 128,
) -> str:
    """Run Turbo with the bounded 128-asset + 32-carrier V5 cap."""

    return transcribe_whisper(
        audio,
        sample_rate,
        language=language,
        task=task,
        max_new_tokens=max_new_tokens,
        max_verification_segments=GLYPH_HYBRID_MAX_VERIFICATION_SEGMENTS,
    )


def _transcribe_glyph_hybrid_independent(
    audio: np.ndarray,
    sample_rate: int,
    *,
    language: str = "zh",
    task: str = "transcribe",
    max_new_tokens: int = 128,
) -> str:
    """Run independent large-v3 with the same V5-only segment cap."""

    return transcribe_verification_whisper(
        audio,
        sample_rate,
        language=language,
        task=task,
        max_new_tokens=max_new_tokens,
        max_verification_segments=GLYPH_HYBRID_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
        ),
        turbo_transcriber=_transcribe_glyph_hybrid_turbo,
        verification_transcriber=(
            _transcribe_glyph_hybrid_independent
        ),
    )
)


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
    )


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(
    turbo_verification,
    trajectory: tuple[np.ndarray, ...],
    chunks: tuple[str, ...],
    anchor: np.ndarray,
    proof_rows: tuple[tuple[NetworkFragmentProof, ...], ...] | None,
    *,
    candidate_seed: int,
):
    """Require pinned large-v3 semantics before retaining network locals.

    Only proof-bearing rows are sent through the independent model.  Their
    exact ``NetworkFragmentProof`` tuples are reused unchanged, so neither ASR
    can borrow a whole URL/email proof for a different local range.  The
    intersection helper keeps turbo speaker/SQUIM/pace/artifact evidence while
    projecting every independent rejection into a hard semantic local failure.
    """

    if proof_rows is None:
        return turbo_verification, tuple(
            LocalIndependentGateEvidence(False, None, 0, None)
            for _ in chunks
        )
    if (
        len(proof_rows) != len(chunks)
        or len(trajectory) != len(chunks)
        or len(turbo_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(
                turbo_verification.candidate_results[index],
                max_local_boundary_speaker_drop=(
                    SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP
                ),
            )
        )
    )
    if not selected_indices:
        return turbo_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=transcribe_verification_whisper,
        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 independent "
            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(
            turbo_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_whisper,
    independent_transcriber=transcribe_verification_whisper,
):
    """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,
        network_fragment_proofs=proof_rows,
        local_endpoint_roles=_local_endpoint_role_rows(chunks, chunk_specs),
        collect_transition_f0=len(chunks) > 1,
    )
    local_verification, independent_local_results = (
        _verify_network_local_asr_intersection(
            local_verification,
            trajectory,
            chunks,
            anchor,
            proof_rows,
            candidate_seed=candidate_seed,
        )
    )
    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=RELEASE_TURBO_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,
    )
    dual_qualified = qualify_trajectory_with_joined_output(
        qualified,
        independent_verification,
    )
    if not dual_qualified.passed:
        print(
            "[BlueMagpie] candidate independent whole output rejected "
            f"seed={candidate_seed} "
            f"{_verification_metric_log_fields(independent_verification)}"
        )
    return CandidateVerification(
        dual_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,
):
    """Apply strict dual-ASR 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,
        network_fragment_proofs=proof_rows,
        local_endpoint_roles=_local_endpoint_role_rows(chunks, chunk_specs),
        collect_transition_f0=True,
    )
    local_verification, independent_local_results = (
        _verify_network_local_asr_intersection(
            local_verification,
            trajectory,
            chunks,
            anchor,
            proof_rows,
            candidate_seed=candidate_seed,
        )
    )
    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_verification_whisper,
    independent_transcriber=transcribe_verification_whisper,
):
    """Verify one ranked DP path after exact production assembly."""

    waveform = _assemble_trajectory_audio(
        sequence_result.trajectory,
        chunks,
        playback_speed,
        chunk_specs,
    )
    # The release contract assigns local candidate coverage (including exact
    # range-bound network fragments) to turbo and the exact assembled
    # waveform to the separately pinned full large-v3 verifier.  Do not let a
    # redundant long-form turbo decode veto a path that passes that stronger
    # assembled-waveform gate.
    assembled_verification = _verify_release_whole_audio(
        waveform,
        whole_target_text,
        anchor,
        independent_cache,
        transcriber=release_transcriber,
        verifier_profile=RELEASE_LARGE_V3_VERIFICATION_PROFILE,
    )
    status = "verified" if assembled_verification.passed else "rejected"
    print(
        f"[BlueMagpie] sequence path large-v3 {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 independent large-v3 {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 _synthesize(
    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:
            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=QUALITY_MEDIUM_CHUNK_UNITS,
                    ),
                    max_chunks=QUALITY_MAX_GENERATED_CHUNKS,
                    max_units=CHUNK_CHARS,
                )
            )
    except ValueError as error:
        raise gr.Error("文字無法在已驗證的生成限制內安全分段。") from error
    request_seed = resolve_request_seed(request_seed, secrets.randbelow)
    anchor = _speaker_anchor_array(centroid)
    independent_cache = WholeWaveformVerificationCache()
    transcript_cache = WholeWaveformTranscriptCache()
    cached_turbo_transcriber = _cached_request_transcriber(
        transcript_cache,
        TURBO_TRANSCRIPT_PROFILE,
        transcribe_whisper,
    )
    cached_large_v3_transcriber = _cached_request_transcriber(
        transcript_cache,
        LARGE_V3_TRANSCRIPT_PROFILE,
        transcribe_verification_whisper,
    )
    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_turbo_transcriber,
                    independent_transcriber=cached_large_v3_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_turbo_transcriber,
                        independent_transcriber=cached_large_v3_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,
                    )
                ),
                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_large_v3_transcriber,
                        independent_transcriber=cached_large_v3_transcriber,
                    )
                ),
                generation_evidence_factory=candidate_generation_evidence,
                candidate_generation_text_transform=(
                    candidate_generation_text_transform
                ),
                max_generated_chunks=QUALITY_MAX_GENERATED_CHUNKS,
                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=QUALITY_MAX_GENERATED_CHUNKS,
                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_large_v3_transcriber
            if len(chunks) > 1
            else cached_turbo_transcriber
        ),
        verifier_profile=(
            RELEASE_LARGE_V3_VERIFICATION_PROFILE
            if len(chunks) > 1
            else RELEASE_TURBO_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_large_v3_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=QUALITY_MAX_GENERATED_CHUNKS,
                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"turbo={_verification_metric_log_fields(final_verification)} "
            f"independent={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=QUALITY_MAX_GENERATED_CHUNKS,
            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"turbo_score={final_verification.score:.6f} "
        f"independent_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=QUALITY_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,
    )
    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_turbo,
    )
    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_independent,
        )
        _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


@gpu
def tts_speaker(
    text: str,
    speaker: str = DEFAULT_SPEAKER,
    cfg: float = DEFAULT_CFG,
    steps: int = DEFAULT_STEPS,
    speed: float = 1.0,
):
    glyph_result = _maybe_synthesize_glyph_speaker_request(
        text,
        speaker,
        cfg=cfg,
        steps=steps,
        speed=speed,
    )
    if glyph_result is not None:
        return pcm16_audio_output(*glyph_result)
    return pcm16_audio_output(
        *_synthesize(
            text,
            SPEAKERS.get(speaker, DEFAULT_CENTROID),
            cfg=cfg,
            steps=steps,
            speed=speed,
        )
    )


@gpu
def tts_reference(
    text: str,
    reference_wav: str,
    cfg: float = DEFAULT_CFG,
    steps: int = DEFAULT_STEPS,
    speed: float = 1.0,
):
    if not reference_wav:
        raise gr.Error("請先錄音或上傳參考音檔。")
    try:
        centroid = extract_windowed_speaker_embedding(
            reference_wav,
            _get_ecapa_encoder(),
            device="cpu",
            min_duration_seconds=3.0,
            window_seconds=3.0,
            hop_seconds=1.5,
            max_windows=12,
            full_clip_max_seconds=12.0,
        )
    except ValueError as error:
        raise gr.Error(str(error)) from error
    return pcm16_audio_output(
        *_synthesize(text, centroid, cfg=cfg, steps=steps, speed=speed)
    )


@gpu
def tts_longform(
    text: str,
    speaker: str = DEFAULT_SPEAKER,
    cfg: float = DEFAULT_CFG,
    steps: int = DEFAULT_STEPS,
    speed: float = 1.0,
):
    glyph_result = _maybe_synthesize_glyph_speaker_request(
        text,
        speaker,
        cfg=cfg,
        steps=steps,
        speed=speed,
    )
    if glyph_result is not None:
        return pcm16_audio_output(*glyph_result)
    return pcm16_audio_output(
        *_synthesize(
            text,
            SPEAKERS.get(speaker, DEFAULT_CENTROID),
            cfg=cfg,
            steps=steps,
            speed=speed,
        )
    )


EXAMPLE_TEXTS = [
    "今天天氣真好，我們一起去散步吧。",
    "我要吃蚵仔煎，然後去丟垃圾。",
    "這學期的成績包括研究報告和期末考。",
    "這是 AI TTS code switching 測試，混合中英文也沒問題。",
    "注音符號測試：ㄅ、ㄆ、ㄇ、ㄈ。",
]

LONGFORM_EXAMPLES = [
    "今天的會議會先整理目前進度，再確認下一階段的工作。遇到需要討論的項目時，"
    "請先記下問題，等報告結束後再一起處理。最後，我們會確認負責人和預計完成時間。",
    "歡迎收聽今天的內容。第一段會介紹背景，第二段整理實際案例，最後一段則說明後續安排。"
    "如果中途聽到英文術語，不必擔心，我們會用中文補充它的意思。",
]


HEADER = f"""
# BlueMagpie-TTS Demo

台灣華語與中英混合文字轉語音。模型版本：`{CHECKPOINT}`。

執行環境固定於已驗證的 model revision `{MODEL_REVISION[:8]}`、ECAPA revision
`{ECAPA_REVISION[:8]}` 與 Whisper revision `{WHISPER_REVISION[:8]}`，避免服務重啟時
無聲變更權重、speaker embedding 或語意驗證空間。

目前預設採用穩定推論設定：候選交錯 `CFG 3.0 / 2.0`（網路內容與極短句最低 3.0）、
`NFE 10`、目標語速 `4.0 字/秒`、
初始完整 trajectory 使用 base duration estimate（{BASE_GENERATION_POLICY.cjk_cps:.1f} CJK /
{BASE_GENERATION_POLICY.ascii_cps:.1f} ASCII），後續 zero/low-coverage 單 chunk refill 使用 safe duration estimate
（{SAFE_DURATION_GENERATION_POLICY.cjk_cps:.1f} CJK /
{SAFE_DURATION_GENERATION_POLICY.ascii_cps:.1f} ASCII）；兩者只調整生成上限，生成完成後才校正至
目標語速。另補齊句末提示、套用尾端 weak-stop 保護、
一般 32 speech units 內維持單一 chunk，較長內容才依自然標點與 80-unit 上限切段。每個 request 只生成一次
same-seed 完整 trajectory；失敗後只補低覆蓋 chunks，再以 speaker/RMS/F0 ragged DP 選出最多三條
culprit-diverse exact paths；所有 chunk 首次具備 coverage 時先驗證 rank-1 exact path。單一 chunk 若 exact joined
波形已通過雙 ASR、preferred SQUIM、release speaker、語速與 echo/smearing gate，會立即返回而不再填滿候選池。
NFE 固定為已驗證的 10，且每個 request 最多生成 32 個 TTS chunks
與 800 speech units。語速校正使用 deterministic WSOLA，總 rate 不低於 0.90；不少於 1 秒的
候選與最終波形另須通過 echo/smearing gate。
"""


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=False,
            )
            steps_input = gr.Number(
                value=DEFAULT_STEPS,
                precision=0,
                interactive=False,
                label="NFE steps（已驗證固定值）",
            )
            speed_input = gr.Number(
                value=1.0,
                interactive=False,
                label="後處理語速（已驗證固定值）",
            )

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

    with gr.Tab("參考音色"):
        gr.Markdown("參考音檔至少 3 秒；只使用已取得授權的聲音。參考內容不需要逐字稿。")
        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, cfg_input, steps_input, speed_input],
            reference_output,
        )

    with gr.Tab("穩定長文"):
        with gr.Row():
            with gr.Column():
                longform_speaker = gr.Dropdown(list(SPEAKERS), value=DEFAULT_SPEAKER, label="語者")
                longform_text = gr.Textbox(
                    label=f"長文（最多 {MAX_TEXT_CHARS} 字元）",
                    lines=8,
                    max_lines=12,
                )
                longform_button = gr.Button("合成完整長文", variant="primary")
            with gr.Column():
                longform_output = gr.Audio(label="合成結果", type="numpy")
        gr.Examples(LONGFORM_EXAMPLES, inputs=longform_text, label="長文範例")
        longform_button.click(
            tts_longform,
            [longform_text, longform_speaker, cfg_input, steps_input, speed_input],
            longform_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()