path: root/app/backends/audiocpp/catalog.py

"""The model_specs catalog, server.json building and selection views."""

import json
import re
import sys
import wave
from pathlib import Path
from typing import Dict, List, Optional, Set, Tuple

from backends.common import find_wav_files
from converter.clients import AUDIOCPP_CLONE_ONLY_FAMILIES, audiocpp_family_spec_tasks

from .constants import TASK_CLON, TASK_TTS, TASK_VDES

DESIGN_PACKAGE_RE = re.compile(r"voice[\s_\-]?design", re.IGNORECASE)

# The spec task vocabulary for "can this family turn text into audio":
# plain synthesis ("tts"), reference-voice synthesis ("clone"; specs spell
# it out, unlike the hosted "clon" task) and described-voice synthesis
# ("vdes"). Families whose tasks name none of these only transform audio
# (speech-to-speech, voice conversion, ...) and are kept out of the
# install catalog.
NARRATION_TASKS = frozenset({TASK_TTS, "clone", TASK_VDES})


def request_options_families(audiocpp_dir: Path) -> Dict[str, dict]:
    """Map the families whose spec defines per-request options.

    Reads every ``model_specs/<family>.json`` in AUDIOCPP_DIR once and
    returns ``{family_key: {"display_name": ...}}`` for the specs that
    list request options (a non-empty ``options.request`` array) — the
    families a "Request options" field makes sense for. The key is the
    spec's ``family`` field (falling back to the file stem), matching
    what ``GET /v1/models`` reports, so callers can look an entry up by
    its family id. A missing or unreadable specs directory yields {}
    (every family then counts as unknown rather than unsupported).
    """
    specs_dir = audiocpp_dir / "model_specs"
    if not specs_dir.is_dir():
        return {}
    families: Dict[str, dict] = {}
    for spec_path in sorted(specs_dir.glob("*.json")):
        try:
            spec = json.loads(spec_path.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            continue
        options = spec.get("options")
        request = options.get("request") if isinstance(options, dict) else None
        if not isinstance(request, list) or not request:
            continue
        family = str(spec.get("family") or spec_path.stem)
        families[family] = {
            "display_name": str(spec.get("display_name") or family),
        }
    return families


def supports_request_options(families: Dict[str, dict],
                             family: str) -> Optional[bool]:
    """Whether FAMILY accepts per-request options — None when unknown.

    True only when FAMILIES (from request_options_families) lists the
    family; False when it was read but does not define request options;
    None when support cannot be determined from the local specs (no
    checkout, or a family the specs do not describe).
    """
    if not families:
        return None
    return family in families


_BACKEND_DESCRIPTIONS = (
    ("cuda", "NVIDIA GPUs"),
    ("vulkan", "cross-vendor GPU"),
    ("hip", "AMD GPUs"),
    ("cpu", "no GPU required"),
)


def _backend_options(detected: Optional[str] = None
                     ) -> Tuple[List[Tuple[str, str]], int]:
    """Build the aligned backend menu options and the default index.

    The backend names are padded to a common width so the ``-`` dashes
    before the descriptions line up. When DETECTED matches one of the
    options, that option gets ``[auto-detected]`` appended and is the
    default (cursor/start) selection; otherwise the first option is the
    default as before. Returns (options, default_index).

    On macOS only ``cpu`` is offered: Metal (via ``build_metal.sh``) is
    the only buildable inference backend there, and it is recorded as
    ``cpu`` (see the ``-metal-`` -> ``cpu`` mapping in ``detect_backend``
    and ``built_server_binary``), so cuda/vulkan/hip — which cannot build
    on macOS — never reach the build step.
    """
    if sys.platform == "darwin":
        label = "cpu - Apple Metal"
        if detected == "cpu":
            label += " [auto-detected]"
        return [(label, "cpu")], 0
    width = max(len(name) for name, _ in _BACKEND_DESCRIPTIONS)
    options = []
    default_index = 0
    for index, (name, desc) in enumerate(_BACKEND_DESCRIPTIONS):
        label = f"{name.ljust(width)} - {desc}"
        if detected == name:
            label += " [auto-detected]"
            default_index = index
        options.append((label, name))
    return options, default_index


_BACKEND_TOKEN_RE = re.compile(r"-(cuda|vulkan|hip|cpu|metal)(?:-|$)")


def detect_backend(audiocpp_dir: Path) -> Optional[str]:
    """Best-effort detection of the backend audiocpp_server was built for.

    Scans ``audiocpp_dir/build/*`` for build directories that contain a
    built ``bin/audiocpp_server`` (``.exe`` allowed on Windows) and reads
    the backend token out of the directory name (``-cuda-``, ``-vulkan-``,
    ``-hip-`` or ``-cpu-``; ``-metal-`` is mapped to ``cpu``). Returns the
    backend only when exactly one distinct backend was built, so a checkout
    with builds for several backends does not silently pick one. Returns
    None when there is no ``build/`` directory, no built server, or more
    than one distinct backend.
    """
    build_root = audiocpp_dir / "build"
    if not build_root.is_dir():
        return None
    backends: Set[str] = set()
    try:
        build_dirs = sorted(build_root.iterdir(),
                            key=lambda p: p.name.lower())
    except OSError:
        return None
    for build_dir in build_dirs:
        if not build_dir.is_dir():
            continue
        server = build_dir / "bin" / "audiocpp_server"
        if not server.exists():
            server_exe = build_dir / "bin" / "audiocpp_server.exe"
            if not server_exe.exists():
                continue
        match = _BACKEND_TOKEN_RE.search(build_dir.name.lower())
        if not match:
            continue
        token = match.group(1)
        backends.add("cpu" if token == "metal" else token)
    if len(backends) == 1:
        return next(iter(backends))
    return None


def _default_package(packages: List[dict]) -> Optional[dict]:
    """Pick the default package from a list of packages.

    Prefers the package flagged ``default: true``, then the first GGUF
    package, then the first package overall. Returns None for an empty list.
    """
    if not packages:
        return None
    for package in packages:
        if package.get("default"):
            return package
    for package in packages:
        if package.get("format") == "gguf":
            return package
    return packages[0]


def spec_gguf_rooted(spec: dict) -> bool:
    """True when SPEC's gguf source resolves its weights from ``$gguf``.

    The ``$gguf`` root is audio.cpp's single-GGUF convention: the model
    directory must hold exactly one top-level GGUF (or ``model.gguf``),
    which then carries the weights. Packages that feed such a source must
    therefore install their GGUF at the top of the target directory, not
    nested under a repository subdirectory.
    """
    for source in spec.get("sources") or []:
        if not isinstance(source, dict) or source.get("format") != "gguf":
            continue
        roots = source.get("roots")
        if isinstance(roots, dict) and any(
                value == "$gguf" for value in roots.values()
                if isinstance(value, str)):
            return True
    return False


def package_common_prefix(files: list) -> Optional[str]:
    """The one directory prefix every file path in FILES shares, or None.

    Only exact single-component matches count: every file must contain a
    ``/`` and start with the same first component. A flat package (no ``/``
    at all) or a mixed one (some files under ``config/``, some at the root,
    like minimax_music3's intentional multi-file layout) yields None.
    """
    prefixes: List[str] = []
    for item in files:
        if not isinstance(item, str) or "/" not in item:
            return None
        prefix = item.split("/", 1)[0]
        if not prefix:
            return None
        prefixes.append(prefix)
    if not prefixes:
        return None
    first = prefixes[0]
    if first in (".", ".."):
        return None
    return first if all(prefix == first for prefix in prefixes) else None


def sanitize_model_spec(spec: dict) -> bool:
    """Repair broken package ``strip_prefix`` entries in SPEC, in place.

    Two upstream spec bug classes have shipped in the audio.cpp checkout,
    and both make the model manager install files where the server cannot
    find them:

    - A dot ``strip_prefix`` ("." or "./") is meant for files written
      ``./<file>``; when the package instead lists repo-root files bare
      (``model.gguf``), the manager rejects the whole package ("file path
      does not start with strip_prefix '.': ...") and nothing can be
      downloaded. Root-level files need no prefix at all (upstream specs
      like minimax_music3.json store ""), so dropping the dot prefix is
      the safe repair. Prefixes naming a real directory are left alone —
      the correct remote paths cannot be guessed.

    - A package with NO strip_prefix whose files all nest under one
      repository directory (``Text to audio (TTS)/GLM-TTS_Q8.gguf`` —
      glm_tts and outetts shipped like this) installs the GGUF under that
      subdirectory. The server then finds no top-level GGUF and falls back
      to the safetensors source, failing on a companion file the GGUF
      package never ships ("missing model package file 'tokenizer_merges'").
      When the spec's gguf source resolves weights from ``$gguf`` (the
      single-GGUF convention) and every file shares one directory prefix,
      that prefix is what the strip_prefix should have been, so set it.
      Packages whose gguf source names tensors by explicit file paths
      (minimax_music3's nested multi-GGUF layout) are left untouched.

    Returns True when SPEC changed.
    """
    changed = False
    for package in spec.get("packages") or []:
        if not isinstance(package, dict):
            continue
        prefix = str(package.get("strip_prefix") or "").rstrip("/")
        if prefix in (".", ".."):
            files = package.get("files")
            if isinstance(files, list) and files \
                    and all(isinstance(item, str)
                            and item.startswith(prefix + "/")
                            for item in files):
                continue
            package["strip_prefix"] = ""
            changed = True
            continue
        if prefix or package.get("format") != "gguf" \
                or not spec_gguf_rooted(spec):
            continue
        common = package_common_prefix(package.get("files") or [])
        if common is None:
            continue
        package["strip_prefix"] = common
        changed = True
    return changed


def load_model_catalog(audiocpp_dir: Path) -> List[dict]:
    """Read model_specs/*.json and return the TTS-capable families.

    Each returned entry has: family, display_name, description, languages,
    clone_capable, packages (the full list from the spec, with broken
    ``strip_prefix`` entries repaired in memory via sanitize_model_spec —
    the checkout's files are never written back; the download path
    materializes the same repair through its sanitized specs copy),
    install_id (recommended package id), and default_path
    (``models/<target_directory>``). All families are treated equally and
    listed in alphabetical order by display name.
    """
    specs_dir = audiocpp_dir / "model_specs"
    if not specs_dir.is_dir():
        raise NotADirectoryError(
            f"{audiocpp_dir} has no model_specs/ directory; re-run setup "
            "to refresh the audio.cpp checkout")
    entries: List[dict] = []
    for spec_path in sorted(specs_dir.glob("*.json")):
        try:
            spec = json.loads(spec_path.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            continue
        sanitize_model_spec(spec)
        tasks = spec.get("tasks") or []
        if tasks:
            # A task list that names no text-synthesis capability means the
            # family cannot narrate text at all (e.g. PersonaPlex:
            # categorized "tts" but speech-to-speech only) — hosting one
            # fails every request, so it is never offered for install.
            if not (NARRATION_TASKS & set(tasks)):
                continue
        elif spec.get("category") != "tts":
            # No task list: fall back to the category as before.
            continue
        family = spec.get("family") or spec_path.stem
        packages = spec.get("packages") or []
        package = _default_package(packages)
        if package is None:
            # No installable package: skip (cannot be hosted from a path).
            continue
        target_directory = package.get("target_directory") or family
        languages = spec.get("languages") or []
        display_name = spec.get("display_name") or family
        description = spec.get("description") or ""
        entries.append({
            "family": family,
            "display_name": display_name,
            "description": description,
            "languages": languages,
            "tasks": list(tasks),
            "clone_capable": "clone" in tasks,
            "packages": packages,
            "install_id": package.get("id") or family,
            "default_path": f"models/{target_directory}",
        })

    # All families are treated equally: alphabetical by display name.
    entries.sort(key=lambda entry: entry["display_name"].lower())
    return entries


def is_design_package(package: dict) -> bool:
    """Return True when a package's name marks it a voice-design model.

    audio.cpp voice-design packages (whose id, display name, or target
    directory mentions "voice design") are the only packages that must be
    hosted with task "vdes"; their role is not in the schema, only in those
    strings, so it is detected from them.
    """
    text = " ".join(str(package.get(key, ""))
                    for key in ("id", "display_name", "target_directory"))
    return bool(DESIGN_PACKAGE_RE.search(text))


def is_clone_only_family(family: str, tasks: Optional[Set[str]] = None
                         ) -> bool:
    """True when FAMILY's audio.cpp implementation rejects plain TTS.

    Such families can only synthesize by cloning a reference voice, so
    their server entries must be hosted with task "clon" — hosting them
    with "tts" fails every speech request at session-creation time.
    Families are classified from the explicit known-clone-only set
    (AUDIOCPP_CLONE_ONLY_FAMILIES, which also covers specs that wrongly
    claim "tts" — Chatterbox) or from a spec task list that names only
    "clone" (TASKS, when the caller has it; without one the specs are
    read best-effort).
    """
    if family in AUDIOCPP_CLONE_ONLY_FAMILIES:
        return True
    if tasks is None:
        tasks = audiocpp_family_spec_tasks(family)
    return bool(tasks) and set(tasks) == {"clone"}


def hosting_task(entry: dict) -> str:
    """The server.json task a family's non-design packages are hosted with.

    Clone-only families (see ``is_clone_only_family``) get "clon" so their
    cloning sessions can be created at all; every other family keeps
    "tts", which serves plain TTS and — where the family supports it —
    cloning through the request's voice field alike.
    """
    if is_clone_only_family(str(entry.get("family") or ""),
                            tasks=set(entry.get("tasks") or []) or None):
        return TASK_CLON
    return TASK_TTS


def rehost_clone_only_entries(server_json: Path, data: dict) -> List[str]:
    """Re-host "tts"-tasked clone-only entries in DATA as "clon", in place.

    Server.json files written before clone-only hosting existed carry
    task "tts" for families whose audio.cpp implementation rejects plain
    TTS sessions (e.g. Chatterbox), so every speech request fails with
    HTTP 500. Each such entry's task is rewritten to "clon"; when anything
    changed, the document is written back to SERVER_JSON (same layout the
    wizard writes). Returns the repaired entries' ids, in order — empty
    when nothing needed changing (or the document is unusable).
    """
    models = data.get("models")
    if not isinstance(models, list):
        return []
    repaired: List[str] = []
    for entry in models:
        if not isinstance(entry, dict):
            continue
        if str(entry.get("task") or "") != TASK_TTS:
            continue
        family = str(entry.get("family") or "")
        if not family or not is_clone_only_family(family):
            continue
        entry["task"] = TASK_CLON
        repaired.append(str(entry.get("id") or family))
    if repaired:
        try:
            with server_json.open("w", encoding="utf-8") as handle:
                json.dump(data, handle, indent=2, ensure_ascii=False)
                handle.write("\n")
        except OSError:
            # The in-memory document is fixed either way; a failed write
            # only means the fix does not survive the process.
            pass
    return repaired


def package_dir_options(entry: dict) -> List[dict]:
    """Return one option per distinct target_directory of a family's packages.

    Each option is a dict with: target_directory, install_id (the recommended
    package id inside that directory), design (voice-design package flag), and
    recommended (whether it holds the family's default package). Precisions
    that share a directory (q8_0/bf16/...) collapse to a single option.
    """
    packages = entry.get("packages") or []
    default_pkg = _default_package(packages)
    default_dir = (default_pkg or {}).get("target_directory") or entry["family"]
    by_dir: Dict[str, List[dict]] = {}
    order: List[str] = []
    for package in packages:
        directory = package.get("target_directory") or entry["family"]
        if directory not in by_dir:
            by_dir[directory] = []
            order.append(directory)
        by_dir[directory].append(package)
    options: List[dict] = []
    for directory in order:
        package = _default_package(by_dir[directory])
        options.append({
            "target_directory": directory,
            "install_id": (package or {}).get("id") or directory,
            "design": is_design_package(package or {}),
            "recommended": directory == default_dir,
        })
    # Put the recommended package first for a friendlier checklist.
    options.sort(key=lambda opt: not opt["recommended"])
    return options


def entry_model_path(entry: dict, target_directory: Optional[str] = None) -> str:
    """The server.json model path that hosts ENTRY's package for TARGET_DIRECTORY.

    ``models/<target_directory>`` normally (the recommended package's
    directory when TARGET_DIRECTORY is None). audio.cpp only loads a model
    directory that holds exactly one top-level GGUF, so a package that
    ships several GGUFs into one directory (MiniMax-H3's text-encoder /
    DiT / audio-VAE / video-VAE bundle) is hosted from its first GGUF
    file instead: the explicit file path makes the server pick the gguf
    source, and the directory stays the model root that the spec's named
    tensors resolve from.
    """
    directory = target_directory
    if directory is None:
        default_path = str(entry.get("default_path") or "")
        directory = default_path[len("models/"):] \
            if default_path.startswith("models/") else default_path
    if not directory:
        directory = str(entry.get("family") or "")
    package = next(
        (item for item in entry.get("packages") or []
         if isinstance(item, dict)
         and str(item.get("target_directory") or entry.get("family")) == directory
         and item.get("format") == "gguf"),
        None)
    if package is None:
        return f"models/{directory}"
    prefix = str(package.get("strip_prefix") or "").rstrip("/")
    ggufs: List[str] = []
    for item in package.get("files") or []:
        if not isinstance(item, str) or not item.lower().endswith(".gguf"):
            continue
        local = item
        if prefix and local.startswith(prefix + "/"):
            local = local[len(prefix) + 1:]
        elif prefix:
            continue
        ggufs.append(local)
    if len(ggufs) > 1:
        return f"models/{directory}/{ggufs[0]}"
    return f"models/{directory}"


def build_model_entry(family: str, model_id: str, model_path: str,
                      task: str = TASK_TTS,
                      session_options: Optional[Dict[str, str]] = None) -> dict:
    """Assemble one server.json model entry.

    ``task`` defaults to "tts"; voice design packages are hosted with
    "vdes" so the server runs its design session for speech requests
    (audiobook.py then requires --instructions with that entry).
    SESSION_OPTIONS carries per-entry session-level options the server
    applies at session creation (e.g. MioTTS's codec model path, or a
    VoxCPM AudioVAE encoder capacity sized for long voice references);
    an empty mapping is omitted so the entry keeps its minimal shape.
    """
    entry = {
        "id": model_id,
        "family": family,
        "path": model_path,
        "task": task,
        "mode": "offline",
    }
    if session_options:
        entry["session_options"] = dict(session_options)
    return entry


def build_server_config(host: str, port: int, backend: str, lazy_load: bool,
                        model_entries: List[dict],
                        voice_dir: Optional[str] = None) -> dict:
    """Assemble the server.json document.

    ``voice_dir`` is a server-level cloning voice library; when set, every
    hosted clone-capable family can use its voices with ``--voice``.
    """
    config_doc = {
        "host": host,
        "port": port,
        "backend": backend,
        "lazy_load": lazy_load,
        "models": model_entries,
    }
    if voice_dir:
        config_doc["voice_dir"] = voice_dir
    return config_doc


def load_server_config(server_json: Path) -> Optional[dict]:
    """Read server.json into a dict, or None when it cannot be used.

    Returns None for a missing file, unreadable content, or a non-dict
    document. Used by the wizard's modify flow to pre-fill its screens
    from an existing config instead of prompting to overwrite it.
    """
    if not server_json.exists():
        return None
    try:
        data = json.loads(server_json.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return None
    if not isinstance(data, dict):
        return None
    return data


def server_config_selections(server_config: dict,
                             catalog: List[dict]
                             ) -> Tuple[Dict[str, List[str]],
                                        Dict[Tuple[str, str], str]]:
    """Map an existing server.json's models back to catalog selections.

    Returns ``(selected_dirs, tasks)``: ``selected_dirs`` maps a catalog
    family to the target directories it hosts (``models/<target>`` paths
    with the ``models/`` prefix stripped, in server.json order), and
    ``tasks`` maps ``(family, target_directory)`` to the entry's task
    (``"tts"`` or ``"vdes"``) so the wizard can preserve how design
    packages were hosted. Entries whose family is not in the CATALOG are
    ignored — the wizard cannot offer them again.
    """
    families = {entry["family"] for entry in catalog}
    selected_dirs: Dict[str, List[str]] = {}
    tasks: Dict[Tuple[str, str], str] = {}
    for entry in server_config.get("models") or []:
        if not isinstance(entry, dict):
            continue
        family = entry.get("family")
        if not isinstance(family, str) or family not in families:
            continue
        path = entry.get("path")
        if not isinstance(path, str):
            continue
        rel = path[len("models/"):] if path.startswith("models/") else path
        # Entries hosted from a specific model file (e.g. MiniMax-H3's
        # multi-GGUF directory, "models/<dir>/<gguf>") belong to the
        # directory's catalog option, not to a made-up nested one. Unprefixed
        # and absolute paths keep their raw target as before (the wizard's
        # valid-directory check filters those it cannot offer again).
        target = rel.split("/", 1)[0] \
            if path.startswith("models/") and "/" in rel else rel
        if family not in selected_dirs:
            selected_dirs[family] = []
        if target not in selected_dirs[family]:
            selected_dirs[family].append(target)
        tasks[(family, target)] = str(entry.get("task") or TASK_TTS)
    return selected_dirs, tasks




# ---------------------------------------------------------------------------
# Session options the setup bakes into server.json entries
# ---------------------------------------------------------------------------

# MioTTS loads its MioCodec companion through the ``miotts.codec_model_path``
# session option: the server's built-in default looks for the codec as a
# sibling of the GGUF's materialized sidecar root (/tmp/audiocpp-gguf/...),
# which is never where the model manager installs it. Pointing the option at
# the installed codec package directory is the supported path, so the wizard
# writes it (and downloads the codec alongside the model — it is an
# audio_tools family, not a TTS one, so the model catalog never offers it).
MIOTTS_CODEC_INSTALL_ID = "miocodec_q8_0"
MIOTTS_CODEC_DIRECTORY = "MioCodec-25Hz-44.1kHz-v2-GGUF"
MIOTTS_CODEC_MODEL_PATH = f"models/{MIOTTS_CODEC_DIRECTORY}"
MIOTTS_CODEC_DISPLAY_NAME = "MioCodec 25Hz 44.1kHz v2 (required by MioTTS)"

# VoxCPM-style AudioVAE encoders cap the reference audio they encode at a
# fixed sample count (240000 samples ≈ 15 s at the VAE's 16 kHz rate, per
# the audio.cpp runtime). A voice reference longer than that fails every
# cloning request ("sample capacity exceeded"). The session option below
# raises the cap; the wizard sizes it to the longest wav in the voice
# directory so the configured voices all work without hand-trimming.
VOXCPM_ENCODER_CAPACITY_DEFAULT_SAMPLES = 240_000
VOXCPM_ENCODER_CAPACITY_SAMPLE_RATE = 16_000
VOXCPM_ENCODER_CAPACITY_MAX_SAMPLES = 240_000 * 20
# Fallback key when a family's spec does not name the option (older
# checkouts may lack voxcpm1.json; the spec is preferred when present).
VOXCPM_ENCODER_CAPACITY_FALLBACK_KEYS = {
    "voxcpm2": "voxcpm2.audiovae_encoder_sample_capacity",
    "voxcpm1": "voxcpm1.audiovae_encoder_sample_capacity",
}


def wav_seconds(path: Path) -> Optional[float]:
    """A PCM WAV file's duration in seconds, or None when unreadable.

    The stdlib wave module handles the PCM variants voice references use;
    float-format or non-WAV files (mp3 renamed, exotic headers) yield None
    so callers skip them instead of guessing a duration.
    """
    try:
        with wave.open(str(path), "rb") as handle:
            frames = handle.getnframes()
            rate = handle.getframerate()
    except (OSError, EOFError, wave.Error):
        return None
    if rate <= 0 or frames <= 0:
        return None
    return frames / rate


def voxcpm_encoder_capacity_option(family: str, audiocpp_dir: Path) -> Optional[str]:
    """FAMILY's AudioVAE encoder-capacity session option key, or None.

    Discovered from the family's model spec (the option whose name carries
    ``encoder_sample_capacity``), falling back to the known VoxCPM keys so
    a spec-less family still gets the right option name rather than a
    broken server.json entry.
    """
    try:
        spec = json.loads(
            (audiocpp_dir / "model_specs" / f"{family}.json")
            .read_text(encoding="utf-8"))
    except (OSError, ValueError):
        spec = None
    if isinstance(spec, dict):
        options = spec.get("options")
        session = options.get("session") if isinstance(options, dict) else None
        if isinstance(session, list):
            for item in session:
                name = item.get("name") if isinstance(item, dict) else None
                if isinstance(name, str) and "encoder_sample_capacity" in name:
                    return name
    return VOXCPM_ENCODER_CAPACITY_FALLBACK_KEYS.get(family)


def voxcpm_encoder_capacity_samples(max_seconds: Optional[float]) -> Optional[int]:
    """The encoder-sample capacity that fits MAX_SECONDS, or None.

    Rounds up to a multiple of the model's default 240000-sample capacity
    and clamps to a generous ceiling (8 minutes) so an absurdly long
    reference cannot push VRAM through the roof; anything at or below the
    default needs no override at all.
    """
    if max_seconds is None or max_seconds <= 0:
        return None
    default = VOXCPM_ENCODER_CAPACITY_DEFAULT_SAMPLES
    needed = max_seconds * VOXCPM_ENCODER_CAPACITY_SAMPLE_RATE
    samples = ((int(needed) + default - 1) // default) * default
    if samples <= default:
        return None
    return min(samples, VOXCPM_ENCODER_CAPACITY_MAX_SAMPLES)


def apply_entry_session_options(model_entries: List[dict],
                                wav_dir: Optional[Path],
                                audiocpp_dir: Path) -> List[str]:
    """Add the per-entry session options heavy families need, in place.

    MioTTS entries get ``miotts.codec_model_path`` (see MIOTTS_CODEC_*:
    the server's default codec path cannot be satisfied by a normal
    install). VoxCPM-family entries get an AudioVAE encoder-sample
    capacity sized to the longest wav in WAV_DIR, so voice references
    longer than the model's built-in 15 s ceiling still clone (returned
    as None when the voice directory is unknown, unreadable, or only
    holds short wavs — the default then applies as before). Existing
    session options are preserved; a hand-set codec path is never
    overridden. Returns the entry ids that gained options (for the
    wizard's summary line), in order.
    """
    max_seconds: Optional[float] = None
    if wav_dir is not None:
        for path in find_wav_files(Path(wav_dir)):
            seconds = wav_seconds(path)
            if seconds is not None and (max_seconds is None
                                        or seconds > max_seconds):
                max_seconds = seconds
    applied: List[str] = []
    for entry in model_entries:
        family = str(entry.get("family") or "")
        current = dict(entry.get("session_options") or {})
        options = dict(current)
        if family == "miotts" \
                and "miotts.codec_model_path" not in options:
            options["miotts.codec_model_path"] = MIOTTS_CODEC_MODEL_PATH
        if family.startswith("voxcpm") and max_seconds is not None:
            key = voxcpm_encoder_capacity_option(family, audiocpp_dir)
            samples = voxcpm_encoder_capacity_samples(max_seconds)
            if key and samples:
                options[key] = str(samples)
        if options != current:
            entry["session_options"] = options
            applied.append(str(entry.get("id") or family))
    return applied