Blame for src/nida/cli.py

"""Nida command-line interface.



Every subcommand accepts the common run flags (``--runs-root``, ``--run-id``,

``--git-commit``) so the Colab runner can call any command the same way.

"""

from __future__ import annotations

import argparse
import json
import logging
import sys
from collections.abc import Sequence
from pathlib import Path

from nida.envreport import collect_env, summarize_env
from nida.runs import RunContext
from nida.utils.io import atomic_write_json, atomic_write_text

__all__ = ["main"]

# src/nida/cli.py → repository root (editable installs, as on Colab).
_REPO_DIR = Path(__file__).resolve().parents[2]


def _common_parser() -> argparse.ArgumentParser:
    common = argparse.ArgumentParser(add_help=False)
    common.add_argument(
        "--runs-root",
        type=Path,
        default=None,
        help="Directory holding run folders. If omitted, nothing is written to disk.",
    )
    common.add_argument("--run-id", default=None, help="Explicit run id (default: generated).")
    common.add_argument("--git-commit", default=None, help="Commit SHA of the code being run.")
    return common


def _positive_int(text: str) -> int:
    """argparse type: an integer >= 1 (clear error otherwise)."""
    try:
        value = int(text)
    except ValueError:
        raise argparse.ArgumentTypeError(f"expected an integer >= 1, got {text!r}") from None
    if value < 1:
        raise argparse.ArgumentTypeError(f"expected an integer >= 1, got {value}")
    return value


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="nida", description="Nida expressive Turkish TTS.")
    sub = parser.add_subparsers(dest="command", required=True)
    sub.add_parser(
        "env-report",
        parents=[_common_parser()],
        help="Print (and optionally record) Python/torch/GPU facts.",
    )
    bench = sub.add_parser(
        "bench-phase0",
        parents=[_common_parser()],
        help="Phase 0 benchmarks B1 (ASR) and B2 (VAE); see docs/benchmarks/B1-B2-tasarim.md.",
    )
    bench.add_argument("--stage", default="all", help="A stage name or 'all'.")
    bench.add_argument(
        "--workspace",
        type=Path,
        default=None,
        help="Durable benchmark directory (default: <runs-root>/../bench/phase0).",
    )
    bench.add_argument("--cache", type=Path, default=Path("/content/nida_cache"))
    bench.add_argument("--envs", type=Path, default=Path("/content/nida_envs"))
    bench.add_argument(
        "--repo", type=Path, default=_REPO_DIR, help="Repository root (for configs and envs)."
    )
    bench.add_argument("--c-results", type=Path, default=None, help="B1 layer-C panel results CSV.")
    bench.add_argument(
        "--b2-panel-results", type=Path, default=None, help="B2 listening panel results CSV."
    )
    bench.add_argument(
        "--retry-failed", action="store_true", help="Retry worker items that failed before."
    )
    pilot = sub.add_parser(
        "data-pilot",
        parents=[_common_parser()],
        help="Phase 1 data pipeline pilot on YODAS (docs/plans/faz-1.md).",
    )
    pilot.add_argument("--stage", default="all", help="A stage name or 'all'.")
    pilot.add_argument(
        "--root",
        type=Path,
        default=None,
        help="Durable pilot directory (default: <runs-root>/../data/pilot).",
    )
    pilot.add_argument("--cache", type=Path, default=Path("/content/nida_cache/pilot"))
    pilot.add_argument("--envs", type=Path, default=Path("/content/nida_envs"))
    pilot.add_argument("--repo", type=Path, default=_REPO_DIR)
    pilot.add_argument(
        "--subset",
        choices=("tr000", "tr100"),  # keep in sync with nida.data.pilot.YODAS_SUBSETS
        default="tr000",
        help=(
            "YODAS Turkish subset: tr000 (manual subtitles, 64 shards) or tr100 (automatic "
            "subtitles, 430 shards; videos that occur in tr000 are dropped)."
        ),
    )
    pilot.add_argument(
        "--shards", default="00000002", help="Comma-separated shards of the chosen subset."
    )
    pilot.add_argument("--max-hours", type=float, default=10.0)
    pilot.add_argument("--asr-primary", default="omniasr_llm_7b")
    pilot.add_argument("--asr-secondary", default="qwen3_asr_1_7b")
    pilot.add_argument("--retry-failed", action="store_true")
    pilot.add_argument(
        "--asr-batch-size",
        type=int,
        default=None,
        help="Override the ASR workers' batch size (bench sizes fit a 22 GB L4).",
    )
    pilot.add_argument(
        "--parallel-workers",
        type=_positive_int,
        default=1,
        help=(
            "Run up to N of the independent GPU worker stages (sed, diar, both ASR systems, "
            "ctc, emotion) at the same time in 'all' (default 1 = one after another)."
        ),
    )
    ears = sub.add_parser(
        "data-ears",
        parents=[_common_parser()],
        help="EARS -> latent dataset (ADR-0007; docs/03-data-pipeline.md).",
    )
    ears.add_argument(
        "--root",
        type=Path,
        default=None,
        help="Durable directory (default: <runs-root>/../data/ears); dataset is <root>/dataset.",
    )
    ears.add_argument(
        "--speakers", default="all", help="Comma-separated speakers (p001,p002) or 'all'."
    )
    ears.add_argument("--cache", type=Path, default=Path("/content/nida_cache/ears"))
    ears.add_argument("--envs", type=Path, default=Path("/content/nida_envs"))
    ears.add_argument("--repo", type=Path, default=_REPO_DIR)
    ears.add_argument("--seed", type=int, default=20261009, help="Speaker split seed.")
    ears.add_argument("--config-dir", type=Path, default=None, help="Default: <repo>/configs/data.")
    imp = sub.add_parser(
        "data-import",
        parents=[_common_parser()],
        help="External transcribed corpus -> latent dataset (nida.data.importer).",
    )
    imp.add_argument(
        "--source",
        # keep in sync with nida.data.importer.sources.SOURCES
        choices=("alania", "tr_full", "wavdir", "issai_tsc", "mediaspeech_tr", "common_voice_tr"),
        required=True,
        help=(
            "alania (HF parquet), tr_full (HF data.jsonl + wavs), wavdir (--source-dir), "
            "issai_tsc (HF tar.gz), mediaspeech_tr (OpenSLR TR.tgz) or common_voice_tr "
            "(--source-dir: a Common Voice release folder or tarball you downloaded)."
        ),
    )
    imp.add_argument(
        "--stage",
        default="all",
        help="select | audio | latents | ctc | timings | dataset | report | all.",
    )
    imp.add_argument(
        "--root",
        type=Path,
        default=None,
        help="Durable directory (default: <runs-root>/../data/import-<source>).",
    )
    imp.add_argument(
        "--cache",
        type=Path,
        default=None,
        help="Local scratch (default: /content/nida_cache/import/<root name>).",
    )
    imp.add_argument("--envs", type=Path, default=Path("/content/nida_envs"))
    imp.add_argument("--repo", type=Path, default=_REPO_DIR)
    imp.add_argument("--config-dir", type=Path, default=None, help="Default: <repo>/configs/data.")
    imp.add_argument("--max-hours", type=float, default=10.0, help="Stop selecting at this many h.")
    imp.add_argument(
        "--shard-offset",
        type=int,
        default=0,
        help="First work unit (parquet shard / row chunk); roots take disjoint ranges.",
    )
    imp.add_argument(
        "--shards", type=_positive_int, default=None, help="Number of work units (default: rest)."
    )
    imp.add_argument(
        "--source-dir",
        type=Path,
        default=None,
        help="wavdir: folder with manifest.jsonl; common_voice_tr: release folder or tarball.",
    )
    imp.add_argument(
        "--revision",
        default=None,
        help=(
            "Hub commit sha to pin (default: current head, recorded); mediaspeech_tr: sha256 of "
            "TR.tgz (default: the one in configs/data/import_corpora.yaml)."
        ),
    )
    imp.add_argument("--retry-failed", action="store_true")
    imp.add_argument(
        "--audio-threads",
        type=_positive_int,
        default=4,
        help="Threads converting raw audio to 16 kHz wavs (does not change the outputs).",
    )
    train = sub.add_parser(
        "train",
        parents=[_common_parser()],
        help="Phase 2 flow-matching training on a latent dataset (docs/04-architecture.md).",
    )
    train.add_argument("--config", type=Path, required=True, help="YAML under configs/train/.")
    train.add_argument(
        "--data",
        type=Path,
        required=True,
        action="append",
        help="Latent dataset directory; repeat the flag to train on the union of several.",
    )
    train.add_argument(
        "--out", type=Path, required=True, help="Training workspace (checkpoints, metrics.jsonl)."
    )
    train.add_argument(
        "--resume", action="store_true", help="Continue from the newest checkpoint in --out."
    )
    train.add_argument("--device", default=None, help="cpu | cuda | auto (default: auto).")
    train.add_argument(
        "--init-from",
        type=Path,
        default=None,
        help="Fine-tune: start from this checkpoint's weights and latent stats (same model "
        "config, except dur_cond none -> emotion) when --out has no checkpoint yet; "
        "optimizer and schedule start fresh.",
    )
    samples = sub.add_parser(
        "synth-samples",
        parents=[_common_parser()],
        help="Listening page: validation items re-spoken by a trained checkpoint.",
    )
    samples.add_argument(
        "--checkpoint", type=Path, required=True, help="ckpt-*.pt or a training workspace."
    )
    samples.add_argument("--data", type=Path, required=True, help="Latent dataset directory.")
    samples.add_argument("--out", type=Path, required=True, help="Output directory.")
    samples.add_argument("--n", type=int, default=12, help="Validation items to generate.")
    samples.add_argument("--steps", type=int, default=32, help="ODE steps.")
    samples.add_argument("--w-text", type=float, default=2.0)
    # Keeps 1.0 on purpose (sample CER stays comparable with earlier runs); `say` uses 2.0.
    samples.add_argument("--w-emo", type=float, default=1.0)
    samples.add_argument("--seed", type=int, default=0)
    samples.add_argument("--method", choices=["euler", "midpoint"], default="euler")
    samples.add_argument(
        "--sway", type=float, default=-1.0, help="Sway time-step warp (0 = uniform steps)."
    )
    samples.add_argument("--envs", type=Path, default=Path("/content/nida_envs"))
    samples.add_argument("--repo", type=Path, default=_REPO_DIR)
    samples.add_argument("--device", default=None, help="cpu | cuda | auto (default: auto).")
    samples.add_argument(
        "--no-asr", action="store_true", help="Skip the Qwen3-ASR read-back (CER) check."
    )
    say = sub.add_parser(
        "say",
        parents=[_common_parser()],
        help="Speak free text after a validation speaker prompt (aligned checkpoints).",
    )
    say.add_argument("--checkpoint", type=Path, required=True, help="ckpt-*.pt or a workspace.")
    say.add_argument(
        "--data",
        type=Path,
        default=None,
        help="Latent dataset (prompt source); not needed with only --prompt-list prompts.",
    )
    say.add_argument("--out", type=Path, required=True, help="Output directory.")
    text = say.add_mutually_exclusive_group(required=True)
    text.add_argument(
        "--text",
        help=(
            "Text to speak (sentences split on . ! ? …; long ones also at , ; :). Numbers, "
            "%%, TL, clock times, abbreviations and acronyms need not be written out: they are "
            "normalised to spoken Turkish first (--no-normalize disables this). "
            "'@mutlu:0.8' sets the emotion of the text after it; '[duraksama:0.3]' inserts "
            "exactly 0.3 s of silence (it ends the sentence); '[duraksama]' lets the model pause."
        ),
    )
    text.add_argument("--text-file", type=Path, help="UTF-8 file with the text to speak.")
    say.add_argument(
        "--prompt-id", action="append", default=None, help="Validation segment (repeatable)."
    )
    say.add_argument(
        "--n-prompts", type=int, default=3, help="Prompts to pick when no --prompt-id is given."
    )
    say.add_argument(
        "--prompt-list",
        type=Path,
        default=None,
        help="TSV '<wav>\t<transcript>' of prompts from outside the datasets.",
    )
    say.add_argument(
        "--rank-prompts",
        type=int,
        default=0,
        help="Decode this many prompt candidates and keep the cleanest --n-prompts (0 = off).",
    )
    say.add_argument("--steps", type=int, default=32, help="ODE steps.")
    say.add_argument("--w-text", type=float, default=3.0)
    say.add_argument("--w-emo", type=float, default=2.0)
    say.add_argument("--seed", type=int, default=0)
    say.add_argument("--method", choices=["euler", "midpoint"], default="euler")
    say.add_argument("--sway", type=float, default=-1.0)
    say.add_argument("--speed", type=float, default=1.0, help=">1 faster, <1 slower.")
    say.add_argument("--envs", type=Path, default=Path("/content/nida_envs"))
    say.add_argument("--repo", type=Path, default=_REPO_DIR)
    say.add_argument("--device", default=None, help="cpu | cuda | auto (default: auto).")
    say.add_argument("--no-asr", action="store_true", help="Skip the ASR read-back check.")
    say.add_argument(
        "--no-normalize",
        action="store_true",
        help=(
            "Use the text as written: no expansion of digits, symbols, abbreviations or "
            "acronyms (they must then be spelled out, or the text is rejected)."
        ),
    )
    say.add_argument(
        "--strict-normalize",
        action="store_true",
        help=(
            "Stop on every text-normalisation issue (a guessed acronym, an ambiguous time or "
            "ordinal, ...); by default only issues that would be misread stop the run."
        ),
    )
    say.add_argument(
        "--single-env",
        action="store_true",
        help=(
            "Run every worker (VAE, CTC, ASR) with this interpreter instead of building one "
            "venv per worker (a local install with all packages, e.g. a release user)."
        ),
    )
    queue = sub.add_parser(
        "queue",
        parents=[_common_parser()],
        help="Run nida.cli commands listed in a file, one after another (nida.jobqueue).",
    )
    queue.add_argument("--file", type=Path, required=True, help="Queue file (one job a line).")
    queue.add_argument(
        "--wait-minutes",
        type=float,
        default=0.0,
        help="Keep polling the file this long after the last job before returning.",
    )
    queue.add_argument(
        "--bundle-dir", type=Path, default=None, help="Default: <runs-root>/../code."
    )
    queue.add_argument("--log-dir", type=Path, default=Path("/content/nida_logs"))
    queue.add_argument("--repo", type=Path, default=_REPO_DIR)
    return parser


def _env_report_md(metrics: dict[str, object]) -> str:
    def show(value: object) -> str:
        return "—" if value is None else str(value)

    lines = ["# env-report", ""]
    if metrics["gpu_name"] is None:
        lines.append("- device: CPU only")
    else:
        lines.append(f"- device: {metrics['gpu_name']} (count {metrics['gpu_count']})")
    for key in (
        "gpu_memory_gb",
        "compute_capability",
        "bf16_supported",
        "torch_version",
        "cuda_version",
        "torchaudio_version",
        "python",
        "cpu_count",
        "ram_total_gb",
        "disk_free_gb",
    ):
        lines.append(f"- {key}: {show(metrics[key])}")
    return "\n".join(lines) + "\n"


def _resolve_device(choice: str | None, cuda_available: bool) -> str:
    """``--device``: ``None`` or ``auto`` → cuda when available, else cpu; other values as is."""
    if choice in (None, "auto"):
        return "cuda" if cuda_available else "cpu"
    return str(choice)


def _cmd_env_report(args: argparse.Namespace) -> int:
    env = collect_env()
    print(json.dumps(env, ensure_ascii=False, indent=2, sort_keys=True))
    if args.runs_root is None:
        return 0
    with RunContext(
        args.runs_root, "env-report", run_id=args.run_id, git_commit=args.git_commit
    ) as run:
        atomic_write_json(run.dir / "env.json", env)
        metrics = summarize_env(env)
        atomic_write_json(run.dir / "metrics.json", metrics)
        atomic_write_text(run.dir / "report.md", _env_report_md(metrics))
        print(f"run_id={run.run_id}")
    return 0


def _cmd_bench_phase0(args: argparse.Namespace) -> int:
    from nida.bench.phase0 import Phase0
    from nida.bench.prepare import Workspace

    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    workspace_root = args.workspace
    if workspace_root is None:
        if args.runs_root is None:
            raise SystemExit("bench-phase0 needs --workspace or --runs-root")
        workspace_root = args.runs_root.parent / "bench" / "phase0"
    phase0 = Phase0(
        workspace=Workspace(root=workspace_root, cache=args.cache),
        envs_root=args.envs,
        repo_dir=args.repo,
        sources_path=args.repo / "configs" / "sources.yaml",
        retry_failed=args.retry_failed,
    )
    if args.runs_root is None:
        result = phase0.run(
            args.stage, c_results=args.c_results, b2_panel_results=args.b2_panel_results
        )
        print(json.dumps(result, ensure_ascii=False, indent=2, default=str))
        return 0
    with RunContext(
        args.runs_root, f"bench-phase0-{args.stage}", run_id=args.run_id, git_commit=args.git_commit
    ) as run:
        result = phase0.run(
            args.stage, c_results=args.c_results, b2_panel_results=args.b2_panel_results
        )
        atomic_write_json(run.dir / "metrics.json", json.loads(json.dumps(result, default=str)))
        reports = [
            (workspace_root / "reports" / name)
            for name in ("b1_report.md", "b2_report.md", "b2_panel_report.md")
            if (workspace_root / "reports" / name).exists()
        ]
        body = "\n\n".join(r.read_text(encoding="utf-8") for r in reports)
        atomic_write_text(
            run.dir / "report.md",
            f"# bench-phase0 ({args.stage})\n\nworkspace: {workspace_root}\n\n{body}\n",
        )
        print(f"run_id={run.run_id}")
    return 0


def _cmd_data_pilot(args: argparse.Namespace) -> int:
    from nida.data.pilot import Pilot, PilotPlan

    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    root = args.root
    if root is None:
        if args.runs_root is None:
            raise SystemExit("data-pilot needs --root or --runs-root")
        root = args.runs_root.parent / "data" / "pilot"
    pilot = Pilot(
        root=root,
        cache=args.cache,
        envs_root=args.envs,
        repo_dir=args.repo,
        sources_path=args.repo / "configs" / "sources.yaml",
        config_dir=args.repo / "configs" / "data",
        plan=PilotPlan(
            subset=args.subset,
            shards=args.shards,
            max_hours=args.max_hours,
            asr_primary=args.asr_primary,
            asr_secondary=args.asr_secondary,
        ),
        retry_failed=args.retry_failed,
        asr_batch_size=args.asr_batch_size,
        parallel_workers=args.parallel_workers,
    )
    if args.runs_root is None:
        print(json.dumps(pilot.run(args.stage), ensure_ascii=False, indent=2, default=str))
        return 0
    with RunContext(
        args.runs_root, f"data-pilot-{args.stage}", run_id=args.run_id, git_commit=args.git_commit
    ) as run:
        result = pilot.run(args.stage)
        atomic_write_json(run.dir / "metrics.json", json.loads(json.dumps(result, default=str)))
        report = root / "reports" / "pilot.md"
        body = report.read_text(encoding="utf-8") if report.exists() else ""
        result_json = json.dumps(result, ensure_ascii=False, indent=2, default=str)
        lines = [
            f"# data-pilot ({args.stage})",
            "",
            f"root: {root}",
            "",
            "```json",
            result_json,
            "```",
            "",
            body,
        ]
        atomic_write_text(run.dir / "report.md", "\n".join(lines) + "\n")
        print(f"run_id={run.run_id}")
    return 0


def _cmd_data_ears(args: argparse.Namespace) -> int:
    from nida.bench import fetch
    from nida.bench.b1 import _stream_runner
    from nida.data.ears import Ears, load_settings, parse_speakers

    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    root = args.root
    if root is None:
        if args.runs_root is None:
            raise SystemExit("data-ears needs --root or --runs-root")
        root = args.runs_root.parent / "data" / "ears"
    config_dir = args.config_dir or args.repo / "configs" / "data"
    speakers = parse_speakers(args.speakers, load_settings(config_dir).ears.n_speakers)
    ears = Ears(
        root=root,
        cache=args.cache,
        envs_root=args.envs,
        repo_dir=args.repo,
        sources_path=args.repo / "configs" / "sources.yaml",
        config_dir=config_dir,
        seed=args.seed,
        runner=_stream_runner,
        opener=fetch.default_opener,
    )
    if args.runs_root is None:
        print(json.dumps(ears.run(speakers), ensure_ascii=False, indent=2, default=str))
        return 0
    with RunContext(
        args.runs_root, "data-ears", run_id=args.run_id, git_commit=args.git_commit
    ) as run:
        result = ears.run(speakers)
        atomic_write_json(run.dir / "metrics.json", json.loads(json.dumps(result, default=str)))
        report = root / "report.md"
        body = report.read_text(encoding="utf-8") if report.exists() else ""
        result_json = json.dumps(result, ensure_ascii=False, indent=2, default=str)
        lines = ["# data-ears", "", f"root: {root}", "", "```json", result_json, "```", "", body]
        atomic_write_text(run.dir / "report.md", "\n".join(lines) + "\n")
        print(f"run_id={run.run_id}")
    return 0


def _cmd_data_import(args: argparse.Namespace) -> int:
    from nida.data.importer import Importer, ImportRequest

    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    root = args.root
    if root is None:
        if args.runs_root is None:
            raise SystemExit("data-import needs --root or --runs-root")
        root = args.runs_root.parent / "data" / f"import-{args.source}"
    cache = args.cache or Path("/content/nida_cache/import") / root.name
    importer = Importer(
        root=root,
        cache=cache,
        envs_root=args.envs,
        repo_dir=args.repo,
        request=ImportRequest(
            source=args.source,
            max_hours=args.max_hours,
            unit_offset=args.shard_offset,
            unit_count=args.shards,
            revision=args.revision,
        ),
        config_dir=args.config_dir,
        wavdir=args.source_dir,
        retry_failed=args.retry_failed,
        audio_threads=args.audio_threads,
    )
    if args.runs_root is None:
        print(json.dumps(importer.run(args.stage), ensure_ascii=False, indent=2, default=str))
        return 0
    with RunContext(
        args.runs_root, f"data-import-{args.stage}", run_id=args.run_id, git_commit=args.git_commit
    ) as run:
        result = importer.run(args.stage)
        atomic_write_json(run.dir / "metrics.json", json.loads(json.dumps(result, default=str)))
        report = root / "report.md"
        body = report.read_text(encoding="utf-8") if report.exists() else ""
        result_json = json.dumps(result, ensure_ascii=False, indent=2, default=str)
        lines = [
            f"# data-import {args.source} ({args.stage})",
            "",
            f"root: {root}",
            "",
            "```json",
            result_json,
            "```",
            "",
            body,
        ]
        atomic_write_text(run.dir / "report.md", "\n".join(lines) + "\n")
        print(f"run_id={run.run_id}")
    return 0


#: Exit code of ``nida train`` when a SIGTERM stopped it after saving a checkpoint.
EXIT_TRAIN_INTERRUPTED = 3


def _cmd_train(args: argparse.Namespace) -> int:
    from nida.train.runner import TrainOutcome, run_training

    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")

    def execute() -> TrainOutcome:
        return run_training(
            args.config,
            args.data,
            args.out,
            resume=args.resume,
            device=args.device,
            git_commit=args.git_commit,
            repo_dir=_REPO_DIR,
            init_from=args.init_from,
        )

    if args.runs_root is None:
        outcome = execute()
        atomic_write_json(args.out / "metrics.json", outcome.metrics)
        atomic_write_text(args.out / "report.md", outcome.report_md)
    else:
        with RunContext(
            args.runs_root, "train", run_id=args.run_id, git_commit=args.git_commit
        ) as run:
            outcome = execute()
            if outcome.result.stopped_by_signal:
                run.status.extra["interrupted"] = "sigterm"
            atomic_write_json(run.dir / "metrics.json", outcome.metrics)
            atomic_write_text(run.dir / "report.md", outcome.report_md)
            print(f"run_id={run.run_id}")
    return EXIT_TRAIN_INTERRUPTED if outcome.result.stopped_by_signal else 0


def _cmd_synth_samples(args: argparse.Namespace) -> int:
    import torch

    from nida.bench import envs as bench_envs
    from nida.bench.b1 import _stream_runner
    from nida.data.envs import data_env_for_worker
    from nida.infer.samples import SampleSettings, make_listening_page

    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    settings = SampleSettings(
        n_items=args.n,
        steps=args.steps,
        w_text=args.w_text,
        w_emo=args.w_emo,
        seed=args.seed,
        method=args.method,
        sway=args.sway,
    )
    device = _resolve_device(args.device, torch.cuda.is_available())

    def execute() -> Path:
        python = bench_envs.ensure_env(
            data_env_for_worker("vae_decode"),
            args.envs,
            args.repo,
            uv=bench_envs.find_uv(bench_envs.subprocess_runner),
        )
        asr_python = None
        if not args.no_asr:
            asr_python = bench_envs.ensure_env(
                bench_envs.env_for_worker("asr_qwen"),
                args.envs,
                args.repo,
                uv=bench_envs.find_uv(bench_envs.subprocess_runner),
            )
        return make_listening_page(
            args.checkpoint,
            args.data,
            args.out,
            settings,
            decoder_python=python,
            runner=_stream_runner,
            device=torch.device(device),
            asr_python=asr_python,
        )

    if args.runs_root is None:
        print(execute())
        return 0
    with RunContext(
        args.runs_root, "synth-samples", run_id=args.run_id, git_commit=args.git_commit
    ) as run:
        page = execute()
        settings_json = (args.out / "settings.json").read_text(encoding="utf-8")
        atomic_write_json(run.dir / "metrics.json", {"page": str(page)})
        atomic_write_text(
            run.dir / "report.md",
            f"# synth-samples\n\npage: {page}\n\n```json\n{settings_json}\n```\n",
        )
        print(f"run_id={run.run_id}")
    return 0


#: Package pins of the isolated worker envs that ``say --single-env`` replaces
#: (``nida.data.envs`` / ``nida.bench.envs``): checked, never installed.
_SINGLE_ENV_PINS = {"transformers": "4.57.6"}
_SINGLE_ENV_ASR_PINS = {"qwen-asr": "0.0.6"}


def _single_env_report(with_asr: bool) -> dict[str, object]:
    """Interpreter and package versions a ``--single-env`` run uses, with every difference

    from the worker envs' pins logged as a warning and listed (results may then differ

    from Colab runs)."""
    from importlib import metadata

    pins = {**_SINGLE_ENV_PINS, **(_SINGLE_ENV_ASR_PINS if with_asr else {})}
    versions: dict[str, str | None] = {}
    for name in ("torch", "torchaudio", *pins):
        try:
            versions[name] = metadata.version(name)
        except metadata.PackageNotFoundError:
            versions[name] = None
    mismatches = {
        name: {"pinned": pin, "installed": versions[name]}
        for name, pin in pins.items()
        if versions[name] != pin
    }
    for name, diff in mismatches.items():
        logging.getLogger(__name__).warning(
            "say --single-env: %s %s installed, worker env pins %s",
            name,
            diff["installed"],
            diff["pinned"],
        )
    return {
        "single_env": True,
        "python": sys.executable,
        "versions": versions,
        "pin_mismatches": mismatches,
    }


def _cmd_say(args: argparse.Namespace) -> int:
    import torch

    from nida.bench import envs as bench_envs
    from nida.bench.b1 import _stream_runner
    from nida.data.envs import data_env_for_worker
    from nida.infer.say import (
        SaySettings,
        make_say_page,
        pick_prompts,
        prompt_candidates,
        rank_prompts,
    )

    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    settings = SaySettings(
        steps=args.steps,
        w_text=args.w_text,
        w_emo=args.w_emo,
        seed=args.seed,
        method=args.method,
        sway=args.sway,
        speed=args.speed,
        normalize_text=not args.no_normalize,
        strict_normalize=args.strict_normalize,
    )
    text = args.text if args.text is not None else args.text_file.read_text(encoding="utf-8")
    device = _resolve_device(args.device, torch.cuda.is_available())
    if args.data is None and (args.prompt_list is None or args.prompt_id or args.rank_prompts):
        raise SystemExit("say: --data is required unless every prompt comes from --prompt-list")
    if args.rank_prompts and (args.prompt_list is not None or args.prompt_id):
        logging.getLogger(__name__).warning(
            "say: --rank-prompts is ignored when --prompt-list or --prompt-id gives the prompts"
        )
    # make_say_page reads data_dir only for dataset prompt ids, which the guard above rules
    # out when --data is missing.
    data_dir: Path = args.data if args.data is not None else args.out / "no-dataset"
    worker_device = "cuda" if torch.device(device).type == "cuda" else "cpu"
    if args.single_env:
        args.out.mkdir(parents=True, exist_ok=True)
        atomic_write_json(args.out / "environment.json", _single_env_report(not args.no_asr))
    else:
        (args.out / "environment.json").unlink(missing_ok=True)  # never a stale single-env report

    def env_python(spec: bench_envs.EnvSpec, uv: str | None) -> Path:
        if args.single_env:
            return Path(sys.executable)
        assert uv is not None
        return bench_envs.ensure_env(spec, args.envs, args.repo, uv=uv)

    def execute() -> Path:
        uv = None if args.single_env else bench_envs.find_uv(bench_envs.subprocess_runner)
        python = env_python(data_env_for_worker("vae_decode"), uv)
        external = []
        if args.prompt_list is not None:
            from nida.data.registry import DATA_MODELS
            from nida.infer.external_prompt import build_external_prompts, read_prompt_list

            ctc_python = env_python(data_env_for_worker(DATA_MODELS["mpoyraz_ctc_tr"].worker), uv)
            external = build_external_prompts(
                read_prompt_list(args.prompt_list),
                args.out / "external",
                vae_python=env_python(data_env_for_worker("vae_encode"), uv),
                ctc_python=ctc_python,
                runner=_stream_runner,
                device=worker_device,
            )
        if args.prompt_id:
            prompt_ids = args.prompt_id
        elif external:
            prompt_ids = []
        elif args.rank_prompts > 0:
            ranked = rank_prompts(
                args.data,
                prompt_candidates(args.data, args.rank_prompts),
                args.out,
                decoder_python=python,
                runner=_stream_runner,
                device=worker_device,
            )
            prompt_ids = [str(r["segment_id"]) for r in ranked[: args.n_prompts]]
        else:
            prompt_ids = pick_prompts(args.data, args.n_prompts)
        asr_python = None
        if not args.no_asr:
            asr_python = env_python(bench_envs.env_for_worker("asr_qwen"), uv)
        return make_say_page(
            args.checkpoint,
            data_dir,
            prompt_ids,
            text,
            args.out,
            settings,
            decoder_python=python,
            runner=_stream_runner,
            device=torch.device(device),
            asr_python=asr_python,
            external=external,
        )

    if args.runs_root is None:
        print(execute())
        return 0
    with RunContext(args.runs_root, "say", run_id=args.run_id, git_commit=args.git_commit) as run:
        page = execute()
        settings_json = (args.out / "settings.json").read_text(encoding="utf-8")
        atomic_write_json(run.dir / "metrics.json", {"page": str(page)})
        atomic_write_text(
            run.dir / "report.md", f"# say\n\npage: {page}\n\n```json\n{settings_json}\n```\n"
        )
        print(f"run_id={run.run_id}")
    return 0


def _cmd_queue(args: argparse.Namespace) -> int:
    from nida.jobqueue import (
        run_git,
        run_queue,
        state_path_for,
        stream_to_log,
        summary_markdown,
    )

    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    if args.runs_root is None or args.git_commit is None:
        print("queue needs --runs-root and --git-commit (the Colab runner passes both)")
        return 2
    bundle_dir = args.bundle_dir or args.runs_root.parent / "code"
    with RunContext(args.runs_root, "queue", run_id=args.run_id, git_commit=args.git_commit) as run:
        summary = run_queue(
            args.file,
            runs_root=args.runs_root,
            git_commit=args.git_commit,
            repo_dir=args.repo,
            bundle_dir=bundle_dir,
            log_dir=args.log_dir,
            runner=stream_to_log,
            git=run_git,
            wait_minutes=args.wait_minutes,
        )
        state_file = state_path_for(args.file)
        state = json.loads(state_file.read_text(encoding="utf-8")) if state_file.exists() else {}
        atomic_write_json(
            run.dir / "metrics.json",
            {
                "jobs": len(summary.outcomes),
                "failed": len(summary.failed),
                "interrupted": summary.interrupted,
            },
        )
        atomic_write_text(run.dir / "report.md", summary_markdown(summary, state))
        if summary.interrupted:
            run.status.extra["interrupted"] = str(summary.interrupted_key)
        print(f"run_id={run.run_id}")
    return EXIT_TRAIN_INTERRUPTED if summary.interrupted else 0


def main(argv: Sequence[str] | None = None) -> int:
    """Entry point. Returns a process exit code."""
    args = _build_parser().parse_args(argv)
    if args.command == "env-report":
        return _cmd_env_report(args)
    if args.command == "bench-phase0":
        return _cmd_bench_phase0(args)
    if args.command == "data-pilot":
        return _cmd_data_pilot(args)
    if args.command == "data-ears":
        return _cmd_data_ears(args)
    if args.command == "data-import":
        return _cmd_data_import(args)
    if args.command == "train":
        return _cmd_train(args)
    if args.command == "synth-samples":
        return _cmd_synth_samples(args)
    if args.command == "say":
        return _cmd_say(args)
    if args.command == "queue":
        return _cmd_queue(args)
    raise AssertionError(f"unhandled command {args.command!r}")  # argparse prevents this


if __name__ == "__main__":
    raise SystemExit(main())