app.py

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

from __future__ import annotations

from dataclasses import dataclass, replace
from importlib import metadata as importlib_metadata
import inspect
import json
import os
import secrets
import threading

# Cross-process release reproducibility is a hard contract.  Configure CUDA
# before importing torch so cuBLAS and Mamba choose their deterministic paths.
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"
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")

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

runtime_versions = {
    "torch": str(torch.__version__),
    "cuda": str(torch.version.cuda),
    "cudnn": torch.backends.cudnn.version(),
    "mamba-ssm": importlib_metadata.version("mamba-ssm"),
    "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 bluemagpie import BlueMagpieModel
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,
    SAFE_DURATION_GENERATION_POLICY,
    VERIFICATION_WHISPER_MODEL_ID,
    VERIFICATION_WHISPER_REVISION,
    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,
    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_trajectory,
)

GPU_LEASE_SECONDS = 120

try:
    import spaces

    gpu = spaces.GPU(duration=GPU_LEASE_SECONDS)
    DEVICE = "cuda"
except ImportError:
    def gpu(function):
        return function

    DEVICE = "cuda" if 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 = 24
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.80
EXTREME_SHORT_MAX_UNITS = 2
EXTREME_SHORT_MAX_SPEED = 0.95
PACE_ONLY_FALLBACK_MIN_SPEED = 0.76
PACE_ONLY_FALLBACK_MIN_UNITS = 30
PACE_STRETCH_N_FFT = 1536
PACE_STRETCH_HOP_LENGTH = 384
NETWORK_PACE_STRETCH_N_FFT = 2048
NETWORK_PACE_STRETCH_HOP_LENGTH = 512
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.30
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
QUALITY_MAX_SEQUENCE_PATHS = 3
SHORT_AUDIO_SPEAKER_GATE_SECONDS = 1.50
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}"
)


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]

# 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:
    speed = float(speed or 1.0)
    if abs(speed - 1.0) < 1.0e-3:
        return audio
    n_fft = (
        NETWORK_PACE_STRETCH_N_FFT
        if network_conditioned
        else PACE_STRETCH_N_FFT
    )
    hop_length = (
        NETWORK_PACE_STRETCH_HOP_LENGTH
        if network_conditioned
        else PACE_STRETCH_HOP_LENGTH
    )
    return librosa.effects.time_stretch(
        np.asarray(audio, dtype=np.float32),
        rate=speed,
        n_fft=n_fft,
        hop_length=hop_length,
    )


@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 phase-vocoder 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")
    encoder = None
    observations: list[CandidateObservation] = []
    artifacts: list[ChunkCandidateArtifact] = []
    prepared_waveforms: list[np.ndarray | None] = []
    proof_rows = network_fragment_proofs or ((),) * len(chunks)
    for chunk, audio, fragment_proofs in 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
        prepared_waveforms.append(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 ""
            )
        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,
            )
        )
    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
        )
    verification = verify_trajectory(
        observations,
        chunk_artifacts=artifacts,
        candidate_gate_kwargs_by_index=indexed_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,
        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
        ),
        max_pace_cps=None if semantic_only else QUALITY_MAX_PACE_CPS,
        squim_gate_enabled=not semantic_only,
        min_squim_stoi=QUALITY_MIN_SQUIM_STOI,
        min_squim_pesq=QUALITY_MIN_SQUIM_PESQ,
    )
    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,
):
    """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=transcribe_verification_whisper,
            semantic_only=True,
        )

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


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,
):
    """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_trajectory_audio(
        (waveform,),
        (whole_target_text,),
        anchor,
        1.0,
        QUALITY_FINAL_ASR_MAX_NEW_TOKENS,
        release_speaker_gate=True,
    )
    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,
    )
    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,
):
    """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_trajectory_audio(
        (waveform,),
        (whole_target_text,),
        anchor,
        1.0,
        QUALITY_FINAL_ASR_MAX_NEW_TOKENS,
        release_speaker_gate=True,
        transcriber=transcribe_verification_whisper,
    )
    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,
    )
    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


def _synthesize(
    text: str,
    centroid: torch.Tensor,
    *,
    cfg: float,
    steps: int,
    speed: float,
    request_seed: int | None = None,
) -> tuple[int, np.ndarray]:
    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 not 0.85 <= float(speed) <= 1.05:
        raise gr.Error("後處理語速必須介於 0.85 與 1.05。")

    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()
    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,
        )
        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")
        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),
                ),
                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),
                    )
                    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,
                    )
                ),
                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
                ),
                require_endpoint_evidence=True,
            )
    except NoQualifiedCandidateError as error:
        print(
            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_trajectory_audio(
        (waveform,),
        (text,),
        anchor,
        1.0,
        QUALITY_FINAL_ASR_MAX_NEW_TOKENS,
        release_speaker_gate=True,
        transcriber=(
            transcribe_verification_whisper
            if len(chunks) > 1
            else transcribe_whisper
        ),
    )
    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,
        )
        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)
        )
        print(
            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
    print(
        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}"
    )
    return SR, waveform


@gpu
def tts_speaker(
    text: str,
    speaker: str = DEFAULT_SPEAKER,
    cfg: float = DEFAULT_CFG,
    steps: int = DEFAULT_STEPS,
    speed: float = 1.0,
):
    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,
):
    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 保護、
不額外切開 80 字內的首段，長文才依自然標點與 80 字上限切段。每個 request 只生成一次
same-seed 完整 trajectory；失敗後只補低覆蓋 chunks，再以 speaker/RMS/F0 ragged DP 選出最多三條
culprit-diverse exact paths。NFE 固定為已驗證的 10，且每個 request 最多生成 32 個 TTS chunks
與 800 speech units。
"""


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.Slider(0.85, 1.05, value=1.0, step=0.05, 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()