Spaces:

hugging-apps
/

ming-3-tts-tiny

Running on Zero

ming-3-tts-tiny / ming_tiny /task_protocol.py
Initial Ming-3.0-tts-tiny demo
b522cc2 verified 1 day ago
27.5 kB
"""Public task contract for the Ming-3.0-tts-tiny multi-task checkpoint.
The runtime in :mod:`ming_tiny.inference` owns tensor execution.  This
module owns the task-facing part of the contract: which conditioning path a
task uses, where its instruction is placed, and the exact structured-control
JSON learned during training.
"""

from __future__ import annotations

from dataclasses import dataclass
import json
import os
from pathlib import Path
import re
from typing import Mapping, Optional, Sequence

from .prompt_contract import (
    BASE_PROTOCOL,
    _canonical_protocol,
    normalize_target_language,
)


FORMAL_SEED = 1234
FORMAL_PODCAST_SEED = 1235
FORMAL_CFG = 2.0
FORMAL_TEMPERATURE = 0.0
FORMAL_FLOW_STEPS = 10
FORMAL_SIGMA = 0.25

CONTROL_FIELD_ORDER = (
    "序号",
    "说话人",
    "方言",
    "情感",
    "风格",
    "语速",
    "基频",
    "音量",
    "BGM",
    "IP",
)
BGM_FIELD_ORDER = ("Genre", "Mood", "Instrument", "Theme", "ENV", "SNR")

EMOTION_CONTROL_FIELD_ORDER = (
    "序号",
    "说话人",
    "方言",
    "风格",
    "语速",
    "基频",
    "音量",
    "情感",
    "BGM",
    "IP",
)

DIALECT_VALUES = frozenset({"广粤话", "川渝话"})
EMOTION_VALUES = frozenset({"中性", "悲伤", "愤怒", "高兴", "惊讶", "恐惧", "厌恶"})
SPEED_VALUES = frozenset({"慢速", "中速", "快速"})
PITCH_VALUES = frozenset({"低", "中", "高"})
VOLUME_VALUES = frozenset({"低", "中", "高"})
PARALINGUISTIC_TAGS = (
    "[Breathing]",
    "[Cough]",
    "[Sigh]",
    "[Laughter]",
    "[Surprise-ah]",
    "[Surprise-oh]",
    "[Dissatisfaction-hnn]",
    "[Uhm]",
    "[Confirmation-en]",
    "[Question-ei]",
    "[Question-ah]",
    "[Question-en]",
    "[Question-oh]",
    "[Question-huh]",
)

EXPECTED_ENGINE_CONTRACT = {
    "protocol_version": "thinking_v1",
    "prompt_format": BASE_PROTOCOL,
    "tokenizer_contract": BASE_PROTOCOL,
    "sampling_contract": "native_fp32",
    "decode_alignment": "old_shift",
    "enable_tts_target_language": True,
    "infer_tts_language_from_text": True,
    "task_contract": "lingv3_sft_multitask_v1",
    "with_rml": True,
}
# config.json may still carry the pre-rename protocol name.
_PROTOCOL_FIELDS = frozenset({"protocol_version", "prompt_format", "tokenizer_contract"})
FORMAL_RUNTIME_ENVIRONMENT = {
    "BAILING_KDA_FUSED_RECURRENT": "0",
    "BAILING_MLA_ATTN": "sdpa",
    "MING_TINY_AUDIO_VAE_ATTN": "sdpa",
    "FLA_USE_CUDA_KDA_FWD": "0",
    "FLA_USE_CUDA_KDA_BWD": "0",
}

DIALECT_ALIASES = {
    "cantonese": "广粤话",
    "canton": "广粤话",
    "粤语": "广粤话",
    "广州话": "广粤话",
    "sichuan": "川渝话",
    "sichuanese": "川渝话",
    "chuanyu": "川渝话",
    "chuanyu dialect": "川渝话",
    "四川话": "川渝话",
    "重庆话": "川渝话",
}
EMOTION_ALIASES = {
    "happy": "高兴",
    "sad": "悲伤",
    "angry": "愤怒",
    "neutral": "中性",
    "surprised": "惊讶",
    "surprise": "惊讶",
    "fearful": "恐惧",
    "fear": "恐惧",
    "disgusted": "厌恶",
    "disgust": "厌恶",
}


@dataclass(frozen=True)
class TaskSpec:
    name: str
    conditioning: str
    include_target_language: bool
    instruction_kind: str
    reference_count: str
    prompt_text_mode: str
    allowed_languages: Optional[frozenset[str]] = None


TASK_SPECS: Mapping[str, TaskSpec] = {
    "seed": TaskSpec(
        "seed",
        "prompt_latent",
        True,
        "none",
        "one",
        "one",
        frozenset({"Chinese", "English"}),
    ),
    "multilingual": TaskSpec(
        "multilingual",
        "prompt_latent",
        True,
        "none",
        "one",
        "one",
        frozenset({"German", "Spanish", "French", "Korean", "Japanese"}),
    ),
    "podcast": TaskSpec("podcast", "multi_speaker", True, "none", "many", "many"),
    "basic": TaskSpec("basic", "speaker_only", False, "basic", "one", "none"),
    "emotion": TaskSpec("emotion", "speaker_only", False, "emotion", "one", "none"),
    "dialect": TaskSpec("dialect", "speaker_only", False, "dialect", "one", "none"),
    "voice_design": TaskSpec(
        "voice_design", "masked_speaker", False, "style", "none", "none"
    ),
    "ip": TaskSpec("ip", "masked_speaker", False, "ip", "none", "none"),
    "freeform": TaskSpec(
        "freeform", "masked_speaker", False, "freeform", "optional_one", "none"
    ),
    "paralinguistic": TaskSpec(
        "paralinguistic",
        "prompt_latent",
        True,
        "none",
        "one",
        "one",
        frozenset({"Chinese", "English"}),
    ),
}
TASKS = tuple(TASK_SPECS)

# Task names used by the 0.5b API and the evaluation manifests.
TASK_ALIASES = {
    "tts": "seed",
    "clone": "seed",
    "zero_shot": "seed",
    "basic_attributes": "basic",
    "basic_attribute": "basic",
    "emotion_cv3": "emotion",
    "emotion_cv3n": "emotion",
    "style": "voice_design",
    "voice": "voice_design",
    "voice_actor": "ip",
    "nvbench": "paralinguistic",
}

_PODCAST_SPEAKER_RE = re.compile(r"^\s*speaker_(\d+)\s*:")


@dataclass(frozen=True)
class ResolvedTaskRequest:
    task: str
    text: str
    instruction: Optional[str]
    prompt_wav: Optional[str | tuple[str, ...]]
    prompt_text: Optional[str]
    target_language: Optional[str]
    include_target_language: bool
    conditioning: str
    assistant_instruction: bool = True
    allow_mixed_rml: bool = True


def configure_release_environment() -> None:
    """Install missing numerical defaults and reject conflicting ones."""

    conflicts = {}
    for name, expected in FORMAL_RUNTIME_ENVIRONMENT.items():
        actual = os.environ.get(name)
        if actual is None:
            os.environ[name] = expected
        elif actual != expected:
            conflicts[name] = (expected, actual)
    if conflicts:
        raise RuntimeError(
            "Ming-3.0-tts-tiny runtime environment mismatch: " + repr(conflicts)
        )


def validate_release_bundle(
    model_dir: str,
    *,
    tokenizer_dir: Optional[str] = None,
    campplus_path: Optional[str] = None,
) -> None:
    """Cheap preflight before allocating the model."""

    root = Path(model_dir).expanduser().resolve()
    if not root.is_dir():
        raise FileNotFoundError(root)
    tokenizer_root = Path(tokenizer_dir or root).expanduser().resolve()
    if tokenizer_root != root:
        raise RuntimeError("Ming-3.0-tts-tiny requires the tokenizer shipped in model_dir")
    expected_campplus = (root / "campplus.onnx").resolve()
    selected_campplus = Path(campplus_path or expected_campplus).expanduser().resolve()
    if selected_campplus != expected_campplus:
        raise RuntimeError("Ming-3.0-tts-tiny requires the campplus.onnx shipped in model_dir")
    required_files = (
        root / "config.json",
        root / "model.safetensors.index.json",
        root / "tokenizer.json",
        root / "tokenizer_config.json",
        expected_campplus,
    )
    missing = [str(path.name) for path in required_files if not path.is_file()]
    if missing:
        raise FileNotFoundError(
            f"incomplete Ming-3.0-tts-tiny bundle; missing files: {missing}"
        )
    try:
        config = json.loads((root / "config.json").read_text(encoding="utf-8"))
        index = json.loads(
            (root / "model.safetensors.index.json").read_text(encoding="utf-8")
        )
    except json.JSONDecodeError as error:
        raise ValueError("invalid JSON in Ming-3.0-tts-tiny bundle") from error
    tts_config = config.get("tts_config")
    if not isinstance(tts_config, Mapping):
        raise RuntimeError("Ming-3.0-tts-tiny config.json has no tts_config")
    tts_config = {
        name: (
            _canonical_protocol(value)
            if name in _PROTOCOL_FIELDS and value is not None
            else value
        )
        for name, value in tts_config.items()
    }
    expected = {
        **EXPECTED_ENGINE_CONTRACT,
        "sample_rate": 44_100,
        "speaker_embedding_dim": 192,
        "latent_dim": 64,
        "patch_size": 1,
        "history_size": 1,
        "audio_patch_token_id": 157_164,
        "audio_start_token_id": 157_165,
        "audio_end_token_id": 157_166,
        "speaker_start_token_id": 157_175,
        "speaker_end_token_id": 157_176,
        "prompt_eos_token_id": 156_895,
        "minimum_decode_steps": 13,
        "maximum_decode_steps": 250,
        "stop_threshold": 0.5,
    }
    mismatches = {
        name: (wanted, tts_config.get(name))
        for name, wanted in expected.items()
        if tts_config.get(name) != wanted
    }
    task_alignment = tts_config.get("task_decode_alignment")
    if not isinstance(task_alignment, Mapping) or (
        task_alignment.get("podcast") != "old_shift"
    ):
        mismatches["task_decode_alignment.podcast"] = (
            "old_shift",
            task_alignment,
        )
    if mismatches:
        raise RuntimeError("Ming-3.0-tts-tiny bundle contract mismatch: " + repr(mismatches))
    weight_map = index.get("weight_map")
    if not isinstance(weight_map, Mapping) or not weight_map:
        raise RuntimeError("model.safetensors.index.json has no weight_map")
    missing_shards = sorted(
        shard for shard in set(weight_map.values()) if not (root / shard).is_file()
    )
    if missing_shards:
        raise FileNotFoundError(
            f"Ming-3.0-tts-tiny bundle is missing weight shards: {missing_shards}"
        )


def validate_release_engine(engine) -> None:
    """Fail closed unless an engine matches the released checkpoint contract."""

    configure_release_environment()
    actual = {
        name: getattr(engine, name, None) for name in EXPECTED_ENGINE_CONTRACT
    }
    mismatches = {
        name: (expected, actual[name])
        for name, expected in EXPECTED_ENGINE_CONTRACT.items()
        if actual[name] != expected
    }
    if mismatches:
        raise RuntimeError(
            "Ming-3.0-tts-tiny checkpoint contract mismatch: " + repr(mismatches)
        )
    if int(getattr(engine, "sample_rate", 0)) != 44_100:
        raise RuntimeError("Ming-3.0-tts-tiny requires sample_rate=44100")
    if getattr(engine, "moe_backend", None) != "loop":
        raise RuntimeError("Ming-3.0-tts-tiny requires moe_backend=loop")
    if getattr(engine, "runtime_state_dtype", None) != "stable_fp32":
        raise RuntimeError("Ming-3.0-tts-tiny requires runtime_state_dtype=stable_fp32")
    if str(getattr(engine, "dtype", None)) != "torch.bfloat16":
        raise RuntimeError("Ming-3.0-tts-tiny requires dtype=bf16")
    if os.path.realpath(getattr(engine, "model_dir", "")) != os.path.realpath(
        getattr(engine, "tokenizer_dir", "")
    ):
        raise RuntimeError("Ming-3.0-tts-tiny requires the tokenizer shipped in model_dir")
    if getattr(engine, "campplus_path", None) is None:
        raise RuntimeError("Ming-3.0-tts-tiny bundle is missing campplus.onnx")
    expected_campplus = os.path.join(engine.model_dir, "campplus.onnx")
    if os.path.realpath(engine.campplus_path) != os.path.realpath(expected_campplus):
        raise RuntimeError("Ming-3.0-tts-tiny requires the campplus.onnx shipped in model_dir")
    if getattr(engine, "speaker_extractor", None) is None:
        raise RuntimeError("Ming-3.0-tts-tiny did not initialize CampPlus")

    tts_config = getattr(getattr(engine, "config", None), "tts_config", None)
    if not isinstance(tts_config, Mapping):
        raise RuntimeError("Ming-3.0-tts-tiny engine has no TTS config")
    expected_generation = {
        "sample_rate": 44_100,
        "speaker_embedding_dim": 192,
        "latent_dim": 64,
        "patch_size": 1,
        "history_size": 1,
        "audio_patch_token_id": 157_164,
        "audio_start_token_id": 157_165,
        "audio_end_token_id": 157_166,
        "speaker_start_token_id": 157_175,
        "speaker_end_token_id": 157_176,
        "prompt_eos_token_id": 156_895,
        "minimum_decode_steps": 13,
        "maximum_decode_steps": 250,
        "stop_threshold": 0.5,
    }
    generation_mismatches = {
        name: (expected, tts_config.get(name))
        for name, expected in expected_generation.items()
        if tts_config.get(name) != expected
    }
    if generation_mismatches:
        raise RuntimeError(
            "Ming-3.0-tts-tiny decode defaults mismatch: " + repr(generation_mismatches)
        )
    geometry = getattr(engine, "prompt_geometry", None)
    expected_geometry = {
        "sample_rate": 44_100,
        "latent_dim": 64,
        "audio_patch_size": 4,
        "flow_patch_size": 1,
        "samples_per_audio_latent": 3_528,
        "pad_align_samples": 3_528,
    }
    if not isinstance(geometry, Mapping) or any(
        geometry.get(name) != expected for name, expected in expected_geometry.items()
    ):
        raise RuntimeError("Ming-3.0-tts-tiny prompt geometry mismatch: " + repr(geometry))
    task_alignments = getattr(engine, "task_decode_alignment", None)
    if not isinstance(task_alignments, Mapping) or (
        task_alignments.get("podcast") != "old_shift"
    ):
        raise RuntimeError(
            "Ming-3.0-tts-tiny requires task_decode_alignment.podcast=old_shift"
        )

    tokenizer = getattr(engine, "tokenizer", None)
    if tokenizer is None:
        raise RuntimeError("Ming-3.0-tts-tiny engine has no tokenizer")
    special_tokens = set(getattr(tokenizer, "all_special_tokens", ()) or ())
    offenders = sorted(special_tokens.intersection(PARALINGUISTIC_TAGS))
    if offenders:
        raise RuntimeError(
            "paralinguistic tags must remain ordinary tokenizer tokens: "
            + repr(offenders)
        )
    for tag in PARALINGUISTIC_TAGS:
        if not tokenizer.encode(tag, add_special_tokens=False):
            raise RuntimeError(f"tokenizer produced no IDs for {tag!r}")


def normalize_task(task: str) -> str:
    value = str(task or "").strip().lower().replace("-", "_")
    value = TASK_ALIASES.get(value, value)
    if value not in TASK_SPECS:
        raise ValueError(
            f"unsupported task {task!r}; expected one of: "
            + ", ".join(sorted(TASK_SPECS))
        )
    return value


def resolve_generation_seed(task: str, seed: Optional[int] = None) -> int:
    if seed is not None:
        return seed
    return FORMAL_PODCAST_SEED if normalize_task(task) == "podcast" else FORMAL_SEED


def _required_text(value: object, name: str) -> str:
    if not isinstance(value, str) or not value.strip():
        raise ValueError(f"{name} must be a non-empty string")
    return value.strip()


def build_control_instruction(
    *,
    dialect: Optional[str] = None,
    emotion: Optional[str] = None,
    style: Optional[str] = None,
    speed: Optional[str] = None,
    pitch: Optional[str] = None,
    volume: Optional[str] = None,
    ip: Optional[str] = None,
) -> str:
    """Serialize the exact structured-control bytes the checkpoint learned.
    Field order is semantic.  Keep this default ordering stable; the emotion
    task applies its archived evaluation ordering separately.
    """

    for name, value in (
        ("dialect", dialect),
        ("emotion", emotion),
        ("style", style),
        ("speed", speed),
        ("pitch", pitch),
        ("volume", volume),
        ("ip", ip),
    ):
        if value is not None and not isinstance(value, str):
            raise ValueError(f"{name} must be a string")
    if dialect is not None:
        dialect = DIALECT_ALIASES.get(dialect.strip().lower(), dialect.strip())
        if dialect not in DIALECT_VALUES:
            raise ValueError(f"dialect must be one of {sorted(DIALECT_VALUES)}")
    if emotion is not None:
        emotion = EMOTION_ALIASES.get(emotion.strip().lower(), emotion.strip())
        if emotion not in EMOTION_VALUES:
            raise ValueError(f"emotion must be one of {sorted(EMOTION_VALUES)}")
    if speed is not None and speed not in SPEED_VALUES:
        raise ValueError(f"speed must be one of {sorted(SPEED_VALUES)}")
    if pitch is not None and pitch not in PITCH_VALUES:
        raise ValueError(f"pitch must be one of {sorted(PITCH_VALUES)}")
    if volume is not None and volume not in VOLUME_VALUES:
        raise ValueError(f"volume must be one of {sorted(VOLUME_VALUES)}")
    if style is not None:
        style = _required_text(style, "style")
    if ip is not None:
        ip = _required_text(ip, "ip")

    item = {
        "序号": 1,
        "说话人": "speaker_1",
        "方言": dialect,
        "情感": emotion,
        "风格": style,
        "语速": speed,
        "基频": pitch,
        "音量": volume,
        "BGM": {name: None for name in BGM_FIELD_ORDER},
        "IP": ip,
    }
    if tuple(item) != CONTROL_FIELD_ORDER:
        raise AssertionError("structured-control field order changed")
    return json.dumps({"audio_sequence": [item]}, ensure_ascii=False)


def _validate_instruction_for_task(
    spec: TaskSpec,
    *,
    instruction: Optional[str],
    dialect: Optional[str],
    emotion: Optional[str],
    style: Optional[str],
    speed: Optional[str],
    pitch: Optional[str],
    volume: Optional[str],
    ip: Optional[str],
) -> Optional[str]:
    controls = {
        "dialect": dialect,
        "emotion": emotion,
        "style": style,
        "speed": speed,
        "pitch": pitch,
        "volume": volume,
        "ip": ip,
    }
    if spec.instruction_kind == "none":
        if instruction is not None or any(
            value is not None for value in controls.values()
        ):
            raise ValueError(f"task {spec.name} does not accept an instruction")
        return None
    if spec.instruction_kind == "freeform":
        if any(value is not None for value in controls.values()):
            raise ValueError("freeform accepts --instruction, not structured controls")
        instruction = _required_text(instruction, "instruction")
        if instruction.startswith("Speaking instruction:"):
            return instruction
        return f"Speaking instruction: Speak in this style: {instruction}"
    if instruction is not None:
        raise ValueError(
            f"task {spec.name} builds canonical structured JSON; do not pass raw instruction"
        )

    if spec.instruction_kind == "basic":
        if speed is None or pitch is None or volume is None:
            raise ValueError("basic requires speed, pitch, and volume")
        allowed = {"speed", "pitch", "volume"}
    else:
        required = spec.instruction_kind
        if controls[required] is None:
            raise ValueError(f"task {spec.name} requires {required}")
        allowed = {required}
    unexpected = [
        name
        for name, value in controls.items()
        if value is not None and name not in allowed
    ]
    if unexpected:
        raise ValueError(f"task {spec.name} received unrelated controls: {unexpected}")
    serialized = build_control_instruction(**controls)
    if spec.name != "emotion":
        return serialized

    payload = json.loads(serialized)
    item = payload["audio_sequence"][0]
    payload["audio_sequence"][0] = {
        name: item[name] for name in EMOTION_CONTROL_FIELD_ORDER
    }
    return json.dumps(payload, ensure_ascii=False)


def _one_path(value: object, name: str) -> str:
    if isinstance(value, (list, tuple)):
        raise ValueError(f"{name} expects one path, not a list")
    return _required_text(value, name)


def _many_texts(value: object, name: str) -> tuple[str, ...]:
    if isinstance(value, (str, bytes)) or not isinstance(value, Sequence):
        raise ValueError(f"{name} must be an ordered list")
    result = tuple(_required_text(item, name) for item in value)
    if len(result) < 2:
        raise ValueError(f"{name} must contain at least two entries")
    return result


def build_podcast_prompt_transcript(prompt_texts: Sequence[str]) -> str:
    texts = _many_texts(prompt_texts, "prompt_text")
    return "".join(
        f" speaker_{index}:{text}\n" for index, text in enumerate(texts, start=1)
    )


def validate_podcast_script(text: str, speaker_count: int) -> None:
    seen = set()
    for line in _required_text(text, "text").splitlines():
        if not line.strip():
            continue
        match = _PODCAST_SPEAKER_RE.match(line)
        if match is None:
            raise ValueError("every Podcast line must start with speaker_N:")
        seen.add(int(match.group(1)))
    expected = set(range(1, speaker_count + 1))
    if seen != expected:
        raise ValueError(
            "Podcast speaker IDs must exactly match prompt speakers 1.."
            f"{speaker_count}; found {sorted(seen)}"
        )


def canonicalize_podcast_script(text: str, speaker_count: int) -> str:
    """Return the exact multi-speaker serialization used in training."""

    validate_podcast_script(text, speaker_count)
    turns = []
    for line in text.splitlines():
        if not line.strip():
            continue
        match = _PODCAST_SPEAKER_RE.match(line)
        if match is None:  # Kept explicit for type checkers and future edits.
            raise ValueError("every Podcast line must start with speaker_N:")
        turn_text = line[match.end() :].strip()
        if not turn_text:
            raise ValueError("Podcast turns must not be empty")
        turns.append(f" speaker_{int(match.group(1))}:{turn_text}\n")
    return "".join(turns)


def resolve_task_request(
    task: str,
    *,
    text: str,
    prompt_wav: Optional[str | Sequence[str]] = None,
    prompt_text: Optional[str | Sequence[str]] = None,
    target_language: Optional[str] = None,
    instruction: Optional[str] = None,
    dialect: Optional[str] = None,
    emotion: Optional[str] = None,
    style: Optional[str] = None,
    speed: Optional[str] = None,
    pitch: Optional[str] = None,
    volume: Optional[str] = None,
    ip: Optional[str] = None,
) -> ResolvedTaskRequest:
    """Validate and lower a public request to the HF engine contract."""

    canonical_task = normalize_task(task)
    spec = TASK_SPECS[canonical_task]
    _required_text(text, "text")
    # Target text is validated stripped but forwarded as given, which is what
    # the published evaluation did; Podcast is canonicalized below.
    target_text = text
    resolved_instruction = _validate_instruction_for_task(
        spec,
        instruction=instruction,
        dialect=dialect,
        emotion=emotion,
        style=style,
        speed=speed,
        pitch=pitch,
        volume=volume,
        ip=ip,
    )

    resolved_wav: Optional[str | tuple[str, ...]]
    resolved_prompt_text: Optional[str]
    conditioning = spec.conditioning
    if spec.reference_count == "optional_one":
        # Freeform defaults to instruction-only generation. Keep explicit
        # speaker references available for existing callers.
        if prompt_text is not None:
            raise ValueError(f"task {canonical_task} does not use prompt_text")
        resolved_wav = None if prompt_wav is None else _one_path(prompt_wav, "prompt_wav")
        resolved_prompt_text = None
        if resolved_wav is not None:
            conditioning = "speaker_only"
    elif spec.reference_count == "none":
        if prompt_wav is not None or prompt_text is not None:
            raise ValueError(f"task {canonical_task} does not accept reference audio")
        resolved_wav = None
        resolved_prompt_text = None
    elif spec.reference_count == "one":
        resolved_wav = _one_path(prompt_wav, "prompt_wav")
        if spec.prompt_text_mode == "one":
            if isinstance(prompt_text, (list, tuple)):
                raise ValueError("prompt_text expects one transcript")
            resolved_prompt_text = _required_text(prompt_text, "prompt_text")
        else:
            if prompt_text is not None:
                raise ValueError(f"task {canonical_task} does not use prompt_text")
            resolved_prompt_text = None
    else:
        wavs = _many_texts(prompt_wav, "prompt_wav")
        texts = _many_texts(prompt_text, "prompt_text")
        if len(wavs) != len(texts):
            raise ValueError("Podcast needs one prompt_text per prompt_wav")
        target_text = canonicalize_podcast_script(target_text, len(wavs))
        resolved_wav = wavs
        resolved_prompt_text = build_podcast_prompt_transcript(texts)

    language = normalize_target_language(target_language)
    if canonical_task == "podcast" and language is not None:
        raise ValueError("Podcast infers Target language from its dialogue text")
    if spec.include_target_language:
        if spec.allowed_languages is not None and language is None:
            raise ValueError(f"task {canonical_task} requires target_language")
        if spec.allowed_languages and language not in spec.allowed_languages:
            raise ValueError(
                f"task {canonical_task} does not support target_language={language!r}; "
                f"expected one of {sorted(spec.allowed_languages)}"
            )
    elif language is not None:
        raise ValueError(f"task {canonical_task} must not add a Target language line")

    return ResolvedTaskRequest(
        task=canonical_task,
        text=target_text,
        instruction=resolved_instruction,
        prompt_wav=resolved_wav,
        prompt_text=resolved_prompt_text,
        target_language=language,
        include_target_language=spec.include_target_language,
        conditioning=conditioning,
    )


def podcast_frame_bounds_to_decode_steps(
    minimum_frames: int,
    maximum_frames: int,
    flow_patch_size: int,
) -> tuple[int, int]:
    for name, value in (
        ("minimum_frames", minimum_frames),
        ("maximum_frames", maximum_frames),
        ("flow_patch_size", flow_patch_size),
    ):
        if not isinstance(value, int) or isinstance(value, bool):
            raise TypeError(f"{name} must be an integer")
    if not 8 <= minimum_frames <= maximum_frames <= 600:
        raise ValueError("Podcast frame bounds must satisfy 8 <= min <= max <= 600")
    if flow_patch_size < 1:
        raise ValueError("flow_patch_size must be positive")
    minimum_steps = minimum_frames // flow_patch_size
    maximum_steps = maximum_frames // flow_patch_size
    if maximum_steps < 1 or minimum_steps > maximum_steps:
        raise ValueError("Podcast frame bounds cannot be represented by this model")
    return minimum_steps, maximum_steps


def podcast_generation_frame_bounds(text: str) -> tuple[int, int]:
    """Derive the 12.5 Hz frame budget from the dialogue text."""

    minimum_frames = 0
    maximum_frames = 0
    for line in _required_text(text, "text").splitlines():
        if not line.strip():
            continue
        match = _PODCAST_SPEAKER_RE.match(line)
        if match is None:
            raise ValueError("every Podcast line must start with speaker_N:")
        turn_text = line[match.end() :].strip()
        if not turn_text:
            raise ValueError("Podcast turns must not be empty")
        if turn_text.isascii():
            units = len(turn_text.split())
            minimum_frames += units
            maximum_frames += units * 8
        else:
            units = len(turn_text)
            minimum_frames += units
            maximum_frames += units * 4
    maximum_frames = max(8, min(600, maximum_frames))
    minimum_frames = min(maximum_frames, max(8, minimum_frames))
    return minimum_frames, maximum_frames