app.py

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

from __future__ import annotations

from dataclasses import replace
import inspect
import json
import os
import secrets
import threading

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

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,
    plan_generation_chunks,
    select_generation_cps,
    set_generation_seed,
    split_text_for_tts,
    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,
    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
CROSSFADE_MS = 80.0
CHUNK_EDGE_FADE_MS = 80.0
CHUNK_RMS_MATCH_DB = 4.0
NETWORK_GENERATION_MIN_UNITS = 8
NETWORK_GENERATION_TARGET_UNITS = 32
NETWORK_GENERATION_MAX_UNITS = 36
NETWORK_REQUEST_ORDINARY_MAX_UNITS = 36
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
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,
    )


def _generate_chunk(
    text: str,
    centroid: torch.Tensor,
    *,
    cfg: float,
    steps: int,
    request_seed: int,
    policy: GenerationPolicy,
    network_conditioned: bool = False,
) -> np.ndarray:
    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,
    )
    # 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()
    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}"
    )
    audio = audio.detach().float().cpu().numpy().reshape(-1)
    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 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,
        )

    # 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 count_speech_units(text) >= PACE_ONLY_FALLBACK_MIN_UNITS
    ):
        try:
            final_active_duration = active_voiced_duration_seconds(final_audio, SR)
        except (TypeError, ValueError, RuntimeError, OverflowError):
            return final_audio
        observed_cps = count_speech_units(text) / 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 _apply_speed(
                    audio,
                    fallback_speed,
                    network_conditioned=False,
                )
    return 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[np.ndarray, ...]:
    scheduled_cfg = float(cfg)
    trajectory: list[np.ndarray] = []
    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}"
        )
        trajectory.append(
            _generate_chunk(
                chunk,
                centroid,
                cfg=network_adjusted_cfg,
                steps=steps,
                request_seed=request_seed,
                policy=policy,
                network_conditioned=network_chunk,
            )
        )
    return tuple(trajectory)


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)
    finish_fade_ms = 0.0 if count_speech_units("".join(chunks)) <= 6 else 5.0
    return finish_audio(waveform, SR, fade_ms=finish_fade_ms)


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 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 count_speech_units(text) <= 0:
        raise gr.Error("輸入必須包含可發音的文字或數字。")
    if len(text) > MAX_TEXT_CHARS:
        raise gr.Error(f"單次最多 {MAX_TEXT_CHARS} 個字元，請分段合成。")
    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,
            )
            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_text_for_tts(
                        text,
                        max_chars=CHUNK_CHARS,
                        min_chunk_chars=MIN_CHUNK_CHARS,
                    ),
                    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] = {}

    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,
        )
        return _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,
        )

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

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