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

"""The audio.cpp setup wizard: TUI screens, task lanes, CLI entry points."""

import argparse
import json
import sys
from pathlib import Path
from typing import Callable, Dict, List, Optional, Tuple

# Alias kept on this module: main()'s tty check and its tests patch it
# here.
from backends.setup import interactive as _interactive

from backends import common
from backends.common import (
    APP_DIR,
    PROMPT_TEXT_FILENAME,
    VOICES_DIR,
    find_wav_files,
    read_prompt_text,
    resolve_wav_dir_arg,
    write_prompt_text,
)
from ui import taskview, tui
from . import build as _build
from . import configsync as _configsync
from . import models as _models
from . import prebuilt as _prebuilt
from . import voices as _voices
from .catalog import (_backend_options, build_model_entry, build_server_config,
                      detect_backend, entry_model_path, hosting_task,
                      load_model_catalog, load_server_config,
                      MIOTTS_CODEC_DISPLAY_NAME, MIOTTS_CODEC_INSTALL_ID,
                      package_dir_options, server_config_selections,
                      apply_entry_session_options)
from .constants import (AUDIOCPP_DIR_NAME, AUDIOCPP_GIT_URL, BACKENDS,
                        DEFAULT_HOST, TASK_TTS, TASK_VDES)

_GO_BACK = object()


def _flag_build_mode(args: argparse.Namespace, backend: str) -> str:
    """Resolve the ``--prebuilt`` flag into a concrete build mode.

    ``auto`` (the default) downloads the prebuilt release when this
    platform/backend has one and otherwise builds from source; ``yes``
    forces the download, ``no`` forces the source build. Only consulted
    when a build/install is actually pending.
    """
    choice = getattr(args, "prebuilt", "auto")
    if choice == "no":
        return "source"
    if choice == "yes" or _prebuilt.prebuilt_supported(backend):
        return "prebuilt"
    return "source"


class _GoBack(Exception):
    """Internal signal: Esc was pressed inside one of a screen's sub-prompts.

    The wizard drives a stack of screens via ``tui.Wizard``. Helpers that ask
    several questions through callbacks (the task/id pickers inside
    ``_build_entries``, the transcription plan, the download prompt) cannot
    themselves return the wizard's ``BACK`` sentinel, so they convert the
    ``_GO_BACK`` value passed to each widget into this exception. The screen
    that invoked the helper catches it and returns ``tui.Wizard.BACK``, which
    pops back to the previous screen. Esc on the first screen aborts the
    whole wizard.
    """


class _TuiError(Exception):
    """A fatal error raised from inside the TUI wizard.

    The message is reported to stderr after the terminal is restored; the
    process exits with code 2 (matching a parser error).
    """



def _build_entries(family_keys: List[str], chosen: Dict[str, List[dict]],
                   catalog_by_family: Dict[str, dict],
                   task_picker: Callable[[str], str],
                   known_tasks: Optional[Dict[Tuple[str, str], str]] = None
                   ) -> Tuple[List[dict], List[str], List[Tuple[str, str]],
                              List[Tuple[str, str]], List[str], bool]:
    """Build server.json model entries from the selected families/packages.

    TASK_PICKER is called for each design package to choose vdes/tts.
    KNOWN_TASKS maps ``(family, target_directory)`` to a previously-stored
    task ("tts" or "vdes") so a modify run preserves how a design package
    was hosted instead of re-asking. Each entry's server id is its package
    ``target_directory`` (flattened to a token), so packages from the same
    family never collide; an id that does collide (across families) is
    auto-suffixed without prompting. Entry paths come from the catalog
    (``entry_model_path``): normally ``models/<target_directory>``, or the
    package's first GGUF file when the package ships several GGUFs into one
    directory (audio.cpp refuses multi-GGUF directories). COMPANION_GUIDANCE
    carries the companion packages hosted models require but the TTS catalog
    never offers (MioCodec for MioTTS). Returns (model_entries, entry_ids,
    install_guidance, companion_guidance, design_entry_ids, include_clone).
    """
    model_entries: List[dict] = []
    entry_ids: List[str] = []
    install_guidance: List[Tuple[str, str]] = []
    companion_guidance: List[Tuple[str, str]] = []
    design_entry_ids: List[str] = []
    include_clone = False
    for family in family_keys:
        entry = catalog_by_family[family]
        include_clone = include_clone or entry["clone_capable"]
        for opt in chosen[family]:
            if opt["design"]:
                task = known_tasks.get((family, opt["target_directory"])) \
                    if known_tasks else None
                if task is None:
                    task = task_picker(opt["install_id"])
            else:
                # Clone-only families (Chatterbox, Confucius4-TTS,
                # Echo-TTS) reject plain-TTS sessions, so they are hosted
                # with task "clon"; everything else keeps "tts".
                task = hosting_task(entry)
            base_id = opt["target_directory"].replace("/", "-")
            model_id = base_id
            if model_id in entry_ids:
                n = 2
                while f"{base_id}-{n}" in entry_ids:
                    n += 1
                model_id = f"{base_id}-{n}"
            entry_ids.append(model_id)
            model_entries.append(build_model_entry(
                family, model_id,
                entry_model_path(entry, opt["target_directory"]),
                task=task))
            install_guidance.append((entry["display_name"], opt["install_id"]))
            if task == TASK_VDES:
                design_entry_ids.append(model_id)
    if any(str(e.get("family")) == "miotts" for e in model_entries):
        companion_guidance.append(
            (MIOTTS_CODEC_DISPLAY_NAME, MIOTTS_CODEC_INSTALL_ID))
    return (model_entries, entry_ids, install_guidance, companion_guidance,
            design_entry_ids, include_clone)


def _write_and_advise(audiocpp_dir: Path, wav_dir: Optional[Path],
                      output_path: Path, model_entries: List[dict],
                      install_guidance: List[Tuple[str, str]], host: str,
                      port: int, backend: str, lazy_load: bool,
                      transcripts: Dict[str, str], write_prompt: bool) -> None:
    """Console phase shared by both UI modes: write files, print summary.

    After a successful run the console output is the path of the written
    server.json. The model install commands (and optional automatic
    download) are handled separately by _install_models, called by both
    UI modes once the user has decided whether to download. Before the
    document is written, apply_entry_session_options bakes in the
    per-entry session options heavy families need (MioTTS's codec path;
    a VoxCPM AudioVAE encoder capacity sized to the voice directory's
    longest reference), reported as one summary line.
    """
    voice_dir: Optional[str] = None
    if transcripts:
        if write_prompt:
            prompt_path = wav_dir / PROMPT_TEXT_FILENAME
            write_prompt_text(wav_dir, transcripts)
            print(f"[OK] Wrote {prompt_path}")
        voice_dir = str(wav_dir.resolve())

    configured = apply_entry_session_options(model_entries, wav_dir,
                                             audiocpp_dir)
    if configured:
        print(f"[OK] Added family session options to: {', '.join(configured)}")

    server_config = build_server_config(
        host=host, port=port, backend=backend, lazy_load=lazy_load,
        model_entries=model_entries, voice_dir=voice_dir)

    with output_path.open("w", encoding="utf-8") as handle:
        json.dump(server_config, handle, indent=2, ensure_ascii=False)
        handle.write("\n")

    count = len(model_entries)
    print(f"Wrote {output_path.resolve()} with {count} "
          f"{'entry' if count == 1 else 'entries'}.")


def _plan_from_mode(mode: str, wav_files: list,
                    existing: Dict[str, str]) -> dict:
    """Build the transcription PLAN for the chosen form MODE.

    The plan dict is what ``voices._transcribe`` consumes: "missing"
    carries the .wavs lacking a transcript plus the existing mapping,
    "all" re-transcribes everything; both reuse the mapping read while
    applying the form.
    """
    if mode == "missing":
        missing = [wav for wav in wav_files
                   if not existing.get(wav.stem, "").strip()]
        return {"mode": mode, "missing": missing, "existing": dict(existing)}
    return {"mode": mode, "missing": [], "existing": dict(existing)}


def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser
            ) -> Optional[dict]:
    """Run every TUI screen; return the collected settings, or None to abort.

    The wizard has two screens: the model tree ("Select TTS Models to
    Host") and one combined configuration form (backend choice when it is
    ambiguous, build offer when needed, clone-voice directory,
    transcription plan, model download/defaults/cleanup), laid out like
    the Generate Audiobooks screen — every option appears on one screen,
    and options that do not apply are hidden instead of asked separately.
    The bind host is always 127.0.0.1 and the port comes from
    AUDIOCPP_API_URL in app/converter/config.py (the Settings screen),
    so neither is ever asked. Esc on the first screen aborts the whole
    wizard; Esc on the form pops back to the model tree.
    """

    s: dict = {}

    def resolve_checkout(audiocpp_dir: Path) -> None:
        """Validate the audio.cpp checkout and populate the wizard state ``s``."""
        audiocpp_dir = Path(audiocpp_dir).resolve()
        try:
            catalog = load_model_catalog(audiocpp_dir)
        except NotADirectoryError as exc:
            raise _TuiError(str(exc)) from exc
        if not catalog:
            raise _TuiError(f"No TTS model families found in "
                            f"{audiocpp_dir}/model_specs; check the "
                            "checkout is up to date")
        catalog_by_family = {entry["family"]: entry for entry in catalog}
        output_path = args.output if args.output is not None \
            else audiocpp_dir / "server.json"
        # Modify flow: an existing server.json seeds the wizard's screens
        # instead of being overwritten from scratch (an explicit --force
        # still starts fresh).
        existing_config = load_server_config(output_path) \
            if not args.force else None
        if existing_config is not None:
            existing_selected, existing_tasks = \
                server_config_selections(existing_config, catalog)
        else:
            existing_selected, existing_tasks = {}, {}
        s.update({
            "audiocpp_dir": audiocpp_dir,
            "catalog": catalog,
            "catalog_by_family": catalog_by_family,
            "output_path": output_path,
            "existing_config": existing_config,
            "existing_selected": existing_selected,
            "existing_tasks": existing_tasks,
            "existing_backend": existing_config.get("backend")
            if existing_config else None,
            "existing_voice_dir": existing_config.get("voice_dir")
            if existing_config else None,
            "detected_backend": detect_backend(audiocpp_dir),
        })

    def _families_from_flag() -> None:
        requested = [f.strip() for f in args.families.split(",") if f.strip()]
        unknown = [f for f in requested if f not in s["catalog_by_family"]]
        if unknown:
            raise _TuiError(
                f"Unknown family in --families: {', '.join(unknown)}. "
                f"Available: {', '.join(s['catalog_by_family'])}")
        chosen: Dict[str, List[dict]] = {}
        family_keys: List[str] = []
        for family in requested:
            if family not in family_keys:
                family_keys.append(family)
                chosen[family] = [opt for opt in package_dir_options(
                    s["catalog_by_family"][family]) if opt["recommended"]]
        s["chosen"] = chosen
        s["family_keys"] = family_keys

    def _compute_entries() -> None:
        # Design task menu. Esc raises _GoBack, which the caller turns into
        # Wizard.BACK (the design prompts are grouped: Esc returns to the
        # families tree).
        def task_picker(install_id: str) -> str:
            result = tui.menu(
                stdscr,
                f"How should the '{install_id}' package be hosted?",
                [
                    ("design (vdes) - describe the voice with "
                     "--instructions", TASK_VDES),
                    ("tts - normal synthesis", TASK_TTS),
                ], default_index=0, back_value=_GO_BACK)
            if result is _GO_BACK:
                raise _GoBack()
            return result

        model_entries, entry_ids, install_guidance, companion_guidance, \
            design_entry_ids, include_clone = _build_entries(
                s["family_keys"], s["chosen"], s["catalog_by_family"],
                task_picker, known_tasks=s["existing_tasks"])
        s.update({
            "model_entries": model_entries,
            "entry_ids": entry_ids,
            "install_guidance": install_guidance,
            "companion_guidance": companion_guidance,
            "design_entry_ids": design_entry_ids,
            "include_clone": include_clone,
        })

    def _finalize() -> dict:
        host = DEFAULT_HOST
        port = _configsync.config_port()
        backend = s["backend"]
        # Build decision: --build-backend builds when no single-backend
        # binary was detected; a plain --backend or a detected build never
        # rebuilds; the interactive answer comes from the form.
        build = s["build"]
        return {
            "audiocpp_dir": s["audiocpp_dir"],
            "catalog": s["catalog"],
            "catalog_by_family": s["catalog_by_family"],
            "output_path": s["output_path"],
            "family_keys": s["family_keys"],
            "chosen": s["chosen"],
            "model_entries": s["model_entries"],
            "entry_ids": s["entry_ids"],
            "install_guidance": s["install_guidance"],
            "companion_guidance": s["companion_guidance"],
            "design_entry_ids": s["design_entry_ids"],
            "include_clone": s["include_clone"],
            "host": host,
            "port": port,
            "backend": backend,
            "build": build,
            "build_mode": s.get("build_mode"),
            "prebuilt_forced": False,
            "lazy_load": True,
            "wav_dir": s["wav_dir"],
            "plan": s["plan"],
            "download": s["download"],
            "delete_unused": s["delete_unused"],
            "unused_entries": s["unused_entries"],
        }

    def screen_families():
        """Pick TTS model families and packages (the modify tree)."""
        tree_families = _models._build_tree_families(s["catalog"])
        # Modify flow: pre-check the models an existing server.json hosts,
        # so the tree opens as a "modify" list rather than a fresh one.
        checked_set = set()
        for family, dirs in s["existing_selected"].items():
            if family not in s["catalog_by_family"]:
                continue
            family_index = s["catalog"].index(s["catalog_by_family"][family])
            valid_dirs = {opt["target_directory"]
                          for opt in package_dir_options(
                              s["catalog_by_family"][family])}
            for target in dirs:
                if target in valid_dirs:
                    checked_set.add((family_index, target))
        picked = tui.checkbox_tree(
            stdscr, "Select TTS Models to Host",
            tree_families, expand_all=args.all_packages,
            back_value=_GO_BACK, checked=checked_set,
            start_on_buttons=True)
        if picked is _GO_BACK:
            return tui.Wizard.BACK
        chosen: Dict[str, List[dict]] = {}
        family_keys: List[str] = []
        for family_index, option_key in picked:
            family = s["catalog"][family_index]["family"]
            if family not in chosen:
                chosen[family] = []
                family_keys.append(family)
            chosen[family].append(option_key)
        for family in list(chosen):
            keyed = {opt["target_directory"]: opt
                     for opt in package_dir_options(
                         s["catalog_by_family"][family])}
            chosen[family] = [keyed[key] for key in chosen[family]]
        s["chosen"] = chosen
        s["family_keys"] = family_keys
        return screen_config

    def _after_families():
        if args.families is not None:
            _families_from_flag()
            return screen_config
        return screen_families

    def _field_val(fields_list, key, default=None):
        return next((f["value"] for f in fields_list
                     if f.get("key") == key), default)

    def _apply_form(result: dict) -> dict:
        """Fold the form's answers into the settings and finalize."""
        # Backend/build: the interactive combination. A backend whose
        # binary already exists (switching to an already-built one) hides
        # the build row — honor that by re-checking at apply time.
        if s["backend"] is None:
            s["backend"] = result["backend"]
            if _build.built_server_binary(s["audiocpp_dir"],
                                          s["backend"]) is not None:
                s["build"] = False
                s["build_mode"] = None
            else:
                # The form reports every field's value, including hidden
                # ones, so the mode is chosen by whether this backend has
                # a prebuilt asset at all — not by which field was shown.
                # Only the visible field's answer is meaningful: the
                # choice field when a prebuilt asset exists, otherwise the
                # plain build question.
                if _prebuilt.prebuilt_supported(s["backend"]):
                    mode = result.get("build_mode")
                    if mode not in ("prebuilt", "source", "skip"):
                        mode = "prebuilt"
                else:
                    mode = "source" if bool(result.get("build")) else "skip"
                s["build_mode"] = mode
                s["build"] = mode in ("prebuilt", "source")

        # Clone-voice directory: only meaningful for clone-capable picks.
        if args.input_dir is not None:
            s["wav_dir"] = args.input_dir
        elif s["include_clone"]:
            raw = result.get("wav_dir")
            s["wav_dir"] = Path(raw) if raw else None
        else:
            s["wav_dir"] = None

        # Transcription plan (transcription itself runs in the tail).
        s["plan"] = None
        if s["include_clone"]:
            wav_files = find_wav_files(Path(s["wav_dir"])) \
                if s["wav_dir"] is not None else []
            prompt_path = Path(s["wav_dir"]) / PROMPT_TEXT_FILENAME \
                if s["wav_dir"] is not None else None
            existing = {}
            if prompt_path is not None and prompt_path.exists() \
                    and not args.force:
                existing = read_prompt_text(prompt_path)
            mode = result.get("transcription")
            if mode not in ("missing", "all"):
                mode = "missing"
            s["plan"] = _plan_from_mode(mode, wav_files, existing)

        s["download"] = bool(result.get("download")) and (
            _models.download_applicable(
                s["audiocpp_dir"], s["model_entries"],
                companions=s.get("companion_guidance")))
        s["delete_unused"] = bool(result.get("delete_unused")) \
            and bool(s["unused_entries"])
        return _finalize()

    def screen_config():
        """One combined configuration screen for everything else.

        The Generate-audiobooks-style form replaces the old one-question-
        per-screen chain (host, port, port sync, backend, build offer,
        wav directory, transcription plan, delete unused, download). Rows
        whose question does not apply are hidden rather than skipped
        silently. Esc or Cancel pops back to the model tree.
        """
        try:
            _compute_entries()
        except _GoBack:
            return tui.Wizard.BACK

        # Backend: pinned by a flag or an existing build when possible;
        # only otherwise does it become a form question. Not built for any
        # pinned backend yet still asks — even on a modify run, so a user
        # who declined the build the first time is never stranded without
        # a way to build from the TUI.
        if args.build_backend is not None:
            s["backend"] = args.build_backend
            s["build"] = s["detected_backend"] is None
            s["build_mode"] = _flag_build_mode(args, s["backend"]) \
                if s["build"] else None
        elif args.backend is not None:
            s["backend"] = args.backend
            s["build"] = False
            s["build_mode"] = None
        elif s["detected_backend"] is not None:
            # Already built: use the detected backend, no menu, no build.
            s["backend"] = s["detected_backend"]
            s["build"] = False
            s["build_mode"] = None
        else:
            s["backend"] = None  # decided by the form
            s["build"] = None
            s["build_mode"] = None

        # Clone-voice directory seed: the voice_dir recorded by the
        # server.json being modified, else the project voices/ dir (the
        # same default the --wavs flag documents). No auto-detection: the
        # field must never start blank.
        wav_start = None
        if s["include_clone"]:
            if isinstance(s["existing_voice_dir"], str) \
                    and s["existing_voice_dir"]:
                wav_start = Path(s["existing_voice_dir"])
            else:
                wav_start = VOICES_DIR
        s["wav_dir"] = wav_start

        fields: List[dict] = []
        if s["backend"] is None:
            options, default_index = _backend_options(None)
            default_backend = options[default_index][1]
            if s["existing_backend"] in BACKENDS:
                default_backend = next(
                    (value for _label, value in options
                     if value == s["existing_backend"]), default_backend)

            def needs_build(fs) -> bool:
                chosen = _field_val(fs, "backend", default_backend)
                return _build.built_server_binary(
                    s["audiocpp_dir"], chosen) is None

            fields.append({
                "key": "backend", "label": "Inference backend",
                "kind": "choice", "value": default_backend,
                "choices": options,
            })
            # How to get the binary: macOS and Windows have upstream
            # release binaries (no toolchain needed — see
            # backends.audiocpp.prebuilt), everything else builds from
            # source. HIP on Windows has no release asset, so it keeps
            # the plain build question.
            fields.append({
                "key": "build_mode",
                "label": "Get audiocpp_server",
                "kind": "choice", "value": "prebuilt",
                "choices": [
                    ("Download prebuilt server (recommended)", "prebuilt"),
                    ("Build from source", "source"),
                    ("Skip for now", "skip"),
                ],
                "visible": lambda fs: (
                    needs_build(fs) and _prebuilt.prebuilt_supported(
                        _field_val(fs, "backend", default_backend))),
            })
            fields.append({
                "key": "build", "label": "Build audiocpp_server now?",
                "kind": "bool", "value": True,
                "visible": lambda fs: (
                    needs_build(fs) and not _prebuilt.prebuilt_supported(
                        _field_val(fs, "backend", default_backend))),
            })

        wav_field = {
            "key": "wav_dir", "label": "Voice clone .wav directory",
            "kind": "dir", "value": Path(wav_start) if wav_start else None,
            "info": common.wav_dir_info, "preview": common.wav_dir_preview,
            "visible": lambda fs: bool(s["include_clone"]),
        }
        fields.append(wav_field)

        # Voice transcript handling: a plain in-place toggle, always
        # offered whenever any clone-capable model is hosted (no
        # dependency on what the picked directory currently holds).
        fields.append({
            "key": "transcription", "label": "Voice transcripts",
            "kind": "toggle", "value": "missing",
            "choices": [("Transcribe new voices", "missing"),
                        ("Re-transcribe all voices", "all")],
            "visible": lambda fs: bool(s["include_clone"]),
        })

        if _models.download_applicable(
                s["audiocpp_dir"], s["model_entries"],
                companions=s.get("companion_guidance")):
            fields.append({
                "key": "download",
                "label": "Download the selected models automatically?",
                "kind": "bool", "value": True,
            })

        new_paths = {entry["path"] for entry in s["model_entries"]}
        s["unused_entries"] = _models.unused_installed_entries(
            s["output_path"], new_paths) \
            if s["existing_config"] is not None else []
        s["delete_unused"] = False
        if s["unused_entries"]:
            count = len(s["unused_entries"])
            fields.append({
                "key": "delete_unused",
                "label": f"Delete {count} unused downloaded model "
                         f"{'entry' if count == 1 else 'entries'} from disk?",
                "kind": "bool", "value": False,
            })

        result = tui.form(
            stdscr, "Configure audio.cpp", fields,
            buttons=("Continue", "Cancel"),
            start_on_buttons=True, back_value=tui.Wizard.BACK)
        if result is tui.Wizard.BACK:
            return tui.Wizard.BACK
        return _apply_form(result)

    # First screen: resolve the checkout directly when it already exists
    # (the modify flow), so the wizard starts on a real screen. When no
    # checkout exists, clone it into ./app/audio.cpp (streaming inside the
    # TUI task view, not by dropping to the console) without asking, then
    # continue the same way.
    audiocpp_dir = _build.find_local_checkout()
    if audiocpp_dir is None:
        target = APP_DIR / AUDIOCPP_DIR_NAME
        # The ggml build patches are deliberately NOT applied here: they
        # belong to the source-build path (build_audiocpp applies them
        # right before building), and a prebuilt install checks out the
        # release tag, which the patches may not fit.
        rc = taskview.run_steps(stdscr, "Clone audio.cpp", [
            taskview.TaskStep(
                f"Cloning audio.cpp into {target}",
                lambda emit, cancel: common.git_clone(
                    AUDIOCPP_GIT_URL, target, emit=emit, cancel=cancel)),
        ])
        if rc == 130:
            # Cancelled from the task view: abort the wizard quietly.
            return None
        if rc != 0:
            raise _TuiError(
                f"audio.cpp setup step failed (exit {rc}). Clone "
                f"audio.cpp manually: git clone "
                f"{AUDIOCPP_GIT_URL} {target}, then re-run")
        audiocpp_dir = target
    resolve_checkout(audiocpp_dir)
    first = _after_families()
    return tui.Wizard().run(first)


def _execute_lanes(settings: dict,
                   args: argparse.Namespace) -> List[taskview.TaskLane]:
    """Build the ordered setup steps for the in-TUI task view, per lane.

    The same work ``_execute`` runs on the console, split into two lanes so
    the view can run the build in one pane while configuring and downloading
    models in the other (both progress bars visible at once). The install
    lane exists only when ``settings["build"]`` is set: it downloads the
    prebuilt release (``settings["build_mode"] == "prebuilt"``) or builds
    from source, per the user's choice. The models lane always exists
    (transcribe → write server.json → download/print commands). Shared
    results (the transcription mapping) travel through a small closure
    dict scoped to the models lane. Each step's ``work(emit, cancel)``
    returns its exit code; subprocess steps stream through EMIT and abort on
    CANCEL, while print()-based steps are captured by the view's stdout
    routing.
    """
    audiocpp_dir = settings["audiocpp_dir"]
    state: dict = {}
    build = settings.get("build")
    lanes: List[taskview.TaskLane] = []

    if build:
        mode = settings.get("build_mode") or "source"
        forced = bool(settings.get("prebuilt_forced"))

        def source_build_step(emit, cancel) -> int:
            """Build from source, warning-and-continue like the setup."""
            rc = _build.build_audiocpp(audiocpp_dir, settings["backend"],
                                       emit=emit, cancel=cancel)
            if rc == 124:
                print("[WARNING] build went silent and was stopped; the "
                      "server.json was still written — build "
                      "audiocpp_server manually before starting it")
            elif rc != 0:
                print(f"[WARNING] build exited with code {rc}; the "
                      "server.json was still written — build "
                      "audiocpp_server manually before starting it")
            else:
                print("[OK] build complete")
            return rc

        if mode == "prebuilt":
            def build_step(emit, cancel):
                rc = _prebuilt.install_prebuilt(
                    audiocpp_dir, settings["backend"],
                    emit=emit, cancel=cancel)
                if rc == 0:
                    print("[OK] prebuilt audiocpp_server installed")
                    return 0
                if rc == 130 or forced:
                    # A cancelled download, or one the user explicitly
                    # forced with --prebuilt yes: fail fast.
                    return rc
                # The usual failure is GitHub's API rate limit — recover
                # in place instead of leaving the setup half-installed.
                print(f"[WARNING] prebuilt download failed (exit {rc}); "
                      "falling back to a source build...")
                return source_build_step(emit, cancel)
            build_title = f"Install audiocpp_server ({settings['backend']})"
        else:
            build_step = source_build_step
            build_title = f"Build audiocpp_server ({settings['backend']})"
        lanes.append(taskview.TaskLane(
            "Build",
            [taskview.TaskStep(build_title, build_step)]))

    def transcribe(emit, cancel):
        args.input_dir = settings["wav_dir"]
        if settings["include_clone"] and args.input_dir is not None:
            transcripts, write_prompt = _voices._transcribe(
                args, plan=settings["plan"], cancel=cancel)
        elif args.input_dir is not None:
            print(f"[WARNING] Ignoring {args.input_dir}: no clone-capable "
                  "family selected, so voice presets are not used")
            transcripts, write_prompt = {}, False
        else:
            transcripts, write_prompt = {}, False
        state["transcripts"] = transcripts
        state["write_prompt"] = write_prompt
        # A blank transcript means the affected clone voices cannot work:
        # transcript-conditioned families (Qwen3-TTS Base ICL) reject every
        # request without it. Report the step as failed instead of a silent
        # [OK]; the setup continues (warn-and-continue) and still writes
        # server.json and prompt_text.
        blank = sorted(name for name, text in transcripts.items()
                       if not text.strip())
        if blank:
            print("[ERROR] No transcript for: " + ", ".join(blank))
            print("[ERROR] Those clone voices will NOT work until prompt_text "
                  "carries an accurate transcript for each one; see the "
                  "summary below for how to fix them by hand.")
            return 1
        return 0

    def write(emit, cancel):
        _write_and_advise(
            audiocpp_dir, settings["wav_dir"], settings["output_path"],
            settings["model_entries"], settings["install_guidance"],
            settings["host"], settings["port"], settings["backend"],
            settings["lazy_load"], state["transcripts"], state["write_prompt"])

        # Delete-unused cleanup (modify flow): remove the already-downloaded
        # models the new selection dropped. The regenerated server.json
        # already only lists the kept entries.
        if settings.get("delete_unused"):
            removed = _models.delete_model_files(settings["output_path"],
                                         settings["unused_entries"])
            print(f"[OK] Deleted {removed} unused model "
                  f"{'entry' if removed == 1 else 'entries'} from disk.")

        _voices.print_empty_transcript_warning(state["transcripts"])
        return 0

    def install(emit, cancel):
        _models._install_models(audiocpp_dir, settings["install_guidance"],
                        settings["download"], emit=emit, cancel=cancel,
                        model_entries=settings["model_entries"],
                        companions=settings.get("companion_guidance"))
        _build._print_launch_hint(audiocpp_dir, settings["output_path"])
        return 0
    # Everything already on disk: the install step just reports it, so the
    # step title says so instead of promising a download.
    everything_installed = bool(settings["model_entries"]) \
        and _models._all_models_present(audiocpp_dir,
                                        settings["model_entries"])
    if everything_installed:
        install_title = "Verify models"
    elif settings.get("download"):
        install_title = "Download models"
    else:
        install_title = "Print model install commands"

    lanes.append(taskview.TaskLane(
        "Configure & download",
        [taskview.TaskStep("Transcribe reference voices", transcribe),
         taskview.TaskStep("Write server.json & sync config", write),
         taskview.TaskStep(install_title, install)]))

    return lanes


def _execute_steps(settings: dict,
                   args: argparse.Namespace) -> List[taskview.TaskStep]:
    """The ordered setup steps for the sequential console path.

    The lanes ``_execute_lanes`` builds, flattened into one ordered list
    (build first, then transcribe → write → download), so the console tail
    is byte-identical to the pre-lanes behavior.
    """
    steps: List[taskview.TaskStep] = []
    for lane in _execute_lanes(settings, args):
        steps.extend(lane.steps)
    return steps


def _execute(settings: dict, args: argparse.Namespace) -> int:
    """Shared console tail: build, sync, transcribe, write, install, advise.

    Runs after the TUI wizard returns (or after _collect_from_flags for a
    non-interactive run): the terminal is plain, so subprocess output and
    transcription progress appear normally. The same work as
    ``_execute_steps``, run with no emit (console streaming).
    """
    return taskview.run_steps_inline(_execute_steps(settings, args))


def setup_screen(stdscr) -> int:
    """Run the setup wizard on an existing curses screen (the hub's).

    The hub drives this as one screen of its own ``tui.Wizard`` stack, so
    Esc on the wizard's first screen simply returns here and the hub pops
    back to the menu that launched it. The setup tail (build, transcribe,
    write, download) runs inside the TUI task view on this same screen, so
    the hub's curses session stays intact and the user sees per-step status
    and progress instead of being dropped to the console. On a fresh install
    the build and the model setup run as two parallel lanes (a split view),
    so cloning → configuring → building+downloading is one continuous,
    one-click flow; the individual "Build" and "Download Missing Models" hub
    actions remain only as fallbacks when something fails or is interrupted.
    Returns 0 on completion, 1 when the user aborted.
    """
    parser = build_parser()
    args = parser.parse_args([])
    settings = _wizard(stdscr, args, parser)
    if settings is None:
        return 1
    return taskview.run_lanes(stdscr, "Setting up audio.cpp",
                              _execute_lanes(settings, args))


def build_screen(stdscr) -> int:
    """Install audiocpp_server from the hub when the checkout has none.

    Asks which backend to install for (pre-selecting the backend an
    existing server.json records, else cuda), then — where a prebuilt
    release asset exists (macOS and Windows, see
    ``backends.audiocpp.prebuilt``) — whether to download it or build
    from source. The chosen action runs inside the TUI task view —
    alongside a download of any missing models when server.json is
    already configured and those models map to an install command (the
    split view), or alone otherwise — then updates server.json's
    ``backend`` field to match. Returns 0 on success, non-zero when the
    user backed out, cancelled, or the install failed. This is the hub's
    "Build audio.cpp Server" action, so a checkout that was cloned but
    never built is always installable from the TUI; the standalone
    "Download Missing Models" action stays as the fallback when a model
    download fails or is interrupted.
    """
    checkout = _build.find_local_checkout()
    if checkout is None:
        tui.flash(stdscr, "No audio.cpp checkout found — install audio.cpp "
                  "first.", "err")
        return 1
    if _build.find_audiocpp_server_bin(checkout) is not None:
        tui.flash(stdscr, "audiocpp_server is already built.", "ok")
        return 0
    server_config = load_server_config(checkout / "server.json") or {}
    recorded = server_config.get("backend")
    options, default = _backend_options(None)
    if recorded in BACKENDS:
        default = next((i for i, (_label, value) in enumerate(options)
                        if value == recorded), default)
    backend = tui.menu(
        stdscr, "Which inference backend should audiocpp_server be built "
        "for?", options, default_index=default, back_value=_GO_BACK)
    if backend is _GO_BACK:
        return 1

    mode = "source"
    if _prebuilt.prebuilt_supported(backend):
        mode = tui.menu(
            stdscr, "Install audiocpp_server:",
            [("Download prebuilt server from GitHub releases "
              "(recommended)", "prebuilt"),
             ("Build from source", "source")],
            default_index=0, back_value=_GO_BACK)
        if mode is _GO_BACK:
            return 1

    outcome: dict = {"prebuilt": False}

    def build_step(emit, cancel):
        if mode == "prebuilt":
            rc = _prebuilt.install_prebuilt(checkout, backend,
                                            emit=emit, cancel=cancel)
            if rc == 0:
                outcome["prebuilt"] = True
                return 0
            if rc == 130:
                return rc
            # The usual failure is GitHub's API rate limit — recover in
            # place instead of sending the user back to the menus.
            print(f"[WARNING] prebuilt download failed (exit {rc}); "
                  "falling back to a source build...")
        return _build.build_audiocpp(checkout, backend, emit=emit,
                                     cancel=cancel)

    action = f"Install audiocpp_server ({backend})" if mode == "prebuilt" \
        else f"Build audiocpp_server ({backend})"
    lanes = [taskview.TaskLane("Build", [taskview.TaskStep(action,
                                                           build_step)])]

    # Missing models this build can also fetch, so a configured backend that
    # lost its binary is restored to "installed" in one step.
    server_json = checkout / "server.json"
    missing = _models.missing_model_entries(server_json) if server_json.exists() else []
    guidance = _models.missing_model_install_guidance(checkout, missing) \
        if missing else []

    if guidance:
        def download_step(emit, cancel):
            _models.install_models(checkout, guidance, emit=emit, cancel=cancel)
            return 0
        lanes.append(taskview.TaskLane(
            "Download models",
            [taskview.TaskStep("Download missing models", download_step)]))

    install_title = ("Install audiocpp_server" if mode == "prebuilt"
                     else "Build audiocpp_server")
    title = f"{install_title} & download models" if len(lanes) == 2 \
        else install_title
    rc = taskview.run_lanes(stdscr, title, lanes)
    if rc != 0:
        return rc
    if not _configsync.update_server_backend(backend):
        verb = "installed" if outcome["prebuilt"] else "built"
        tui.flash(stdscr, f"audiocpp_server {verb} for {backend}. (Could not "
                  "update server.json's backend field — reconfigure audio.cpp "
                  "if it was already configured.)", "warn")
    # Models that can't be mapped to an install command still need hand
    # installation; say so now rather than leaving the user in the dark.
    if missing and not guidance:
        tui.flash(stdscr, _models.hand_install_guidance(checkout, missing), "err")
    return 0


def run_tui(args: Optional[argparse.Namespace] = None,
            parser: Optional[argparse.ArgumentParser] = None) -> int:
    """Run the audio.cpp setup wizard end-to-end.

    With no ARGS (the hub's call) a default namespace is built so the full
    wizard runs. Called from ``main`` after argparse when the terminal is
    interactive. Returns the process exit code.
    """
    import curses
    if args is None:
        parser = build_parser()
        args = parser.parse_args([])
        if args.input_dir is not None and not args.input_dir.is_dir():
            print(f"[ERROR] --wavs not found: {args.input_dir}",
                  file=sys.stderr)
            return 2
    try:
        settings = curses.wrapper(_wizard, args, parser)
    except _TuiError as exc:
        print(f"[ERROR] {exc}", file=sys.stderr)
        return 2
    except tui.WizardCancelled:
        print("\n[INFO] Cancelled; nothing was written")
        return 1
    try:
        curses.curs_set(1)  # restore the text cursor hidden by the TUI
    except curses.error:
        pass
    if settings is None:
        print("[INFO] Aborted; existing server.json kept")
        return 1
    return _execute(settings, args)


def _collect_from_flags(args: argparse.Namespace,
                        parser: argparse.ArgumentParser) -> Optional[dict]:
    """Build the settings dict from flags for a non-interactive run.

    Every required value must come from a flag (there are no prompts in a
    non-interactive run); a missing one is a hard ``parser.error``. Returns
    the settings dict, or None when the user declined an overwrite (the
    default-location fallback then also exists).
    """
    # Checkout: ./app/audio.cpp, else --clone clones one there.
    # (The ggml build patches are applied by build_audiocpp itself, so a
    # prebuilt install never needs them.)
    audiocpp_dir = _build.find_local_checkout()
    if audiocpp_dir is None and args.clone:
        target = APP_DIR / AUDIOCPP_DIR_NAME
        rc = common.git_clone(AUDIOCPP_GIT_URL, target)
        if rc != 0:
            parser.error(f"git clone failed (exit {rc}); clone audio.cpp "
                         f"manually: git clone {AUDIOCPP_GIT_URL} {target}")
        audiocpp_dir = target
    if audiocpp_dir is None:
        parser.error(
            "An audio.cpp checkout is required. Pass --clone to clone "
            "app/audio.cpp, or run without flags for the TUI wizard.")
    try:
        catalog = load_model_catalog(audiocpp_dir)
    except NotADirectoryError as exc:
        parser.error(str(exc))
    if not catalog:
        parser.error(
            f"No TTS model families found in {audiocpp_dir}/model_specs; "
            "check the checkout is up to date")
    catalog_by_family = {entry["family"]: entry for entry in catalog}

    # Families: required from --families in a non-interactive run.
    if args.families is None:
        parser.error("--families is required in a non-interactive run (or run "
                     "without flags for the TUI wizard)")
    requested = [f.strip() for f in args.families.split(",") if f.strip()]
    unknown = [f for f in requested if f not in catalog_by_family]
    if unknown:
        parser.error(
            f"Unknown family in --families: {', '.join(unknown)}. "
            f"Available: {', '.join(catalog_by_family)}")
    family_keys: List[str] = []
    for fam in requested:
        if fam not in family_keys:
            family_keys.append(fam)

    chosen: Dict[str, List[dict]] = {}
    for family in family_keys:
        opts = package_dir_options(catalog_by_family[family])
        if args.all_packages:
            chosen[family] = opts
        else:
            chosen[family] = [opt for opt in opts if opt["recommended"]]

    # Non-interactive picker: design packages default to vdes.
    def task_picker(install_id: str) -> str:
        return TASK_VDES

    model_entries, entry_ids, install_guidance, companion_guidance, \
        design_entry_ids, include_clone = _build_entries(
            family_keys, chosen, catalog_by_family, task_picker)

    # Server settings. Host is always 127.0.0.1 and the port comes from
    # AUDIOCPP_API_URL in app/converter/config.py (the Settings screen) —
    # neither is a CLI option.
    host = DEFAULT_HOST
    detected_backend = detect_backend(audiocpp_dir)
    if args.build_backend:
        backend = args.build_backend
        build = detected_backend is None
    elif args.backend:
        backend = args.backend
        build = False
    elif detected_backend is not None:
        backend = detected_backend
        build = False
    else:
        backend = "cuda"
        build = False
    # How a pending install happens: the prebuilt release by default on
    # macOS/Windows (``--prebuilt no`` forces the source build), always
    # the source build elsewhere. A ``--prebuilt yes`` download that
    # fails stays a failure (no source-build fallback) — the flag's
    # whole point is "do not build".
    build_mode = _flag_build_mode(args, backend) if build else None
    prebuilt_forced = bool(build) and getattr(args, "prebuilt", "auto") \
        == "yes"
    port = _configsync.config_port()
    lazy_load = True

    # Output path / overwrite (decline falls back to cwd, then aborts).
    output_path = args.output if args.output is not None \
        else audiocpp_dir / "server.json"
    if output_path.exists() and not args.force:
        if args.output is None:
            output_path = Path.cwd() / "server.json"
            if output_path.exists() and not args.force:
                print("[INFO] Aborted; existing server.json kept")
                return None
        else:
            print("[INFO] Aborted; existing server.json kept")
            return None

    # Wav dir + transcription plan (defaults to the project's voices/ dir).
    wav_dir = args.input_dir if args.input_dir is not None else VOICES_DIR
    plan: Optional[dict] = None
    if include_clone and wav_dir is not None:
        wav_files = find_wav_files(wav_dir)
        if wav_files:
            prompt_path = wav_dir / PROMPT_TEXT_FILENAME
            plan = _voices._flag_plan(wav_files, prompt_path, args.force)

    return {
        "audiocpp_dir": audiocpp_dir,
        "catalog": catalog,
        "catalog_by_family": catalog_by_family,
        "output_path": output_path,
        "family_keys": family_keys,
        "chosen": chosen,
        "model_entries": model_entries,
        "entry_ids": entry_ids,
        "install_guidance": install_guidance,
        "companion_guidance": companion_guidance,
        "design_entry_ids": design_entry_ids,
        "include_clone": include_clone,
        "host": host,
        "port": port,
        "backend": backend,
        "build": build,
        "build_mode": build_mode,
        "prebuilt_forced": prebuilt_forced,
        "lazy_load": lazy_load,
        "wav_dir": wav_dir,
        "plan": plan,
        "download": args.download,
    }


def build_parser() -> argparse.ArgumentParser:
    """The audio.cpp setup CLI (also used to build a default namespace)."""
    parser = argparse.ArgumentParser(
        description="Set up the audio.cpp TTS backend: clone/build, pick "
                    "models, write server.json, and sync the configured port.")
    parser.add_argument("--wavs", type=resolve_wav_dir_arg, default=None,
                        dest="input_dir", metavar="WAV_DIR",
                        help="Directory with .wav reference files to publish as "
                             "a server-level voice_dir cloning library "
                             f"(default: {VOICES_DIR}; asked for when omitted "
                             "in the TUI)")
    parser.add_argument("--output", type=Path, default=None,
                        help="Output path for server.json (default: "
                             "server.json inside the audio.cpp checkout; an "
                             "existing file is overwritten only with --force "
                             "or a TUI confirm)")
    parser.add_argument("--clone", action="store_true",
                        help="Non-interactive: clone audio.cpp into "
                             "./app/audio.cpp when no checkout is found")
    parser.add_argument("--families", type=str, default=None,
                        help="Comma-separated model families to host, as named "
                             "in the audio.cpp catalog (e.g. "
                             "qwen3_tts,higgs_audio_tts). Required in a "
                             "non-interactive run; skips the family tree in "
                             "the TUI")
    parser.add_argument("--all-packages", action="store_true",
                        help="Host every installable package of each selected "
                             "family (distinct target_directory) instead of "
                             "only the recommended one. Voice-design packages "
                             "are hosted with task 'vdes'")
    parser.add_argument("--backend", choices=BACKENDS, default=None,
                        help="Inference backend recorded in server.json "
                             "(default: auto-detected from the checkout's "
                             "build/ directory, else cuda)")
    parser.add_argument("--build-backend", choices=BACKENDS, default=None,
                        help="Build audiocpp_server for this backend when it "
                             "is not built yet, and use it in server.json")
    parser.add_argument("--prebuilt", choices=("auto", "yes", "no"),
                        default="auto",
                        help="How to install audiocpp_server when it is "
                             "missing: auto downloads the prebuilt release "
                             "on macOS/Windows (falling back to a source "
                             "build if the download fails) and builds from "
                             "source elsewhere; yes forces the prebuilt "
                             "download and fails if it cannot; no forces a "
                             "source build")
    parser.add_argument("--whisper-model", type=str, default="base",
                        help="Whisper model size for transcription "
                             "(default: base)")
    parser.add_argument("--force", action="store_true",
                        help="Overwrite the output file (and prompt_text) "
                             "without prompting; in the TUI, start the "
                             "wizard fresh instead of loading the existing "
                             "server.json")
    parser.add_argument("--download", action="store_true",
                        help="Run model_manager_v2.py install for each hosted "
                             "model automatically (default: print the commands "
                             "only)")
    return parser


def main() -> int:
    parser = build_parser()
    args = parser.parse_args()

    if args.input_dir is not None and not args.input_dir.is_dir():
        parser.error(
            f"WAV directory not found: {args.input_dir}\n"
            f"      (resolved from the current working directory: "
            f"{Path.cwd()})\n"
            "      --wavs must be a directory containing the .wav "
            "reference files to use as voice cloning presets")

    if _interactive():
        return run_tui(args, parser)

    # Non-interactive (no terminal, or all flags supplied): flag-only path.
    settings = _collect_from_flags(args, parser)
    if settings is None:
        return 1
    return _execute(settings, args)