feat: publish downloadable wakeword trainer source

scripts/build-wakeword-training-image

@@ -0,0 +1,34 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+# Build the isolated training image from the reviewed, complete hash lock. This
+# is an operator preparation command, never a CI step or training invocation.
+# It intentionally does not download corpora or rewrite the reviewed lock.
+
+readonly build_dir="deploy/wakeword-training"
+readonly lock_file="${build_dir}/requirements.wheelhouse.lock"
+readonly base_image="docker.io/pytorch/pytorch@sha256:639b8229ccfd8a3aa803cf49c33d6d6fe406750d79aaf723fe8c0eb1060d8cff"
+
+if [[ $# -ne 1 ]]; then
+  echo "usage: $0 <output-image-tag>" >&2
+  exit 64
+fi
+if [[ ! -s "$lock_file" ]]; then
+  echo "required reviewed dependency lock is missing: $lock_file" >&2
+  exit 66
+fi
+if ! grep -F "livekit-wakeword" "$lock_file" >/dev/null \
+  || ! grep -F "0.2.1" "$lock_file" >/dev/null; then
+  echo "compiled lock did not retain livekit-wakeword 0.2.1" >&2
+  exit 65
+fi
+if ! grep -F "cf2d9cf4867812c06788f64c15e49abd909d9d6291f0a13f1c3f9cb649fa6127" "$lock_file" >/dev/null; then
+  echo "compiled lock did not retain the attested LiveKit source hash" >&2
+  exit 65
+fi
+if ! grep -F "torch==2.7.1+cu126" "$lock_file" >/dev/null \
+  || ! grep -F "torchaudio==2.7.1+cu126" "$lock_file" >/dev/null; then
+  echo "compiled lock must match the base image's PyTorch/TorchAudio 2.7.1 CUDA 12.6 pair" >&2
+  exit 65
+fi
+docker build --pull=false --build-arg "BASE_IMAGE=$base_image" -f "${build_dir}/Dockerfile" -t "$1" .

scripts/jrich-wakeword-trainer

@@ -0,0 +1,8 @@
+#!/usr/bin/env python3
+"""Executable entrance for the downloadable JRich Wakeword Trainer."""
+
+from wakeword_trainer import main
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/run-wakeword-training

@@ -0,0 +1,112 @@
+#!/usr/bin/env python3
+"""Start one sealed offline Wakeword Training attempt.
+
+The command is deliberately operator-only.  It has no gateway/API-key path and
+refuses to start unless both ``--enable`` and an active maintenance window are
+explicitly supplied.  See docs/wakeword-training-operator-runner.md.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+from pathlib import Path
+
+from wakeword_training import (
+    DockerTrainingExecutor,
+    FakeTrainingExecutor,
+    NvidiaSmiGpuEvidenceSampler,
+    NvidiaSmiProcessInspector,
+    RunnerSettings,
+    StaticGpuProcessInspector,
+    StaticMaintenanceWindow,
+    WakewordTrainingRequest,
+    WakewordTrainingRunner,
+    terminal_exit_code,
+)
+
+
+def _parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--request", required=True, type=Path, help="immutable request JSON")
+    parser.add_argument("--work-root", required=True, type=Path, help="operator run root")
+    parser.add_argument("--enable", action="store_true", help="explicitly permit this invocation")
+    parser.add_argument(
+        "--maintenance-window-active",
+        action="store_true",
+        help="attest that the configured GPU maintenance window is active",
+    )
+    parser.add_argument("--retry-of", help="terminal matching run ID for a new immutable attempt")
+    parser.add_argument("--timeout-s", type=float, default=6 * 60 * 60)
+    parser.add_argument(
+        "--fake", action="store_true", help="CI/local quarantined contract executor"
+    )
+    parser.add_argument(
+        "--image-ref", help="digest-pinned training image, e.g. registry/image@sha256:…"
+    )
+    parser.add_argument("--gpu-uuid", help="stable configured RTX UUID, never a mutable ordinal")
+    parser.add_argument(
+        "--approved-inputs-dir",
+        type=Path,
+        help="hash-locked data mount; mounted read-only at /inputs",
+    )
+    parser.add_argument(
+        "--allow-active-gpu-processes",
+        action="store_true",
+        help="explicit shared-GPU override; never terminates another process",
+    )
+    parser.add_argument(
+        "--stream-output", action="store_true", help="mirror container progress to stderr"
+    )
+    return parser
+
+
+async def _main(args: argparse.Namespace) -> int:
+    request = WakewordTrainingRequest.from_file(args.request)
+    if args.fake:
+        executor = FakeTrainingExecutor()
+    else:
+        if not (args.image_ref and args.gpu_uuid and args.approved_inputs_dir):
+            raise SystemExit(
+                "a real run requires --image-ref, --gpu-uuid, and --approved-inputs-dir; "
+                "use --fake only for a quarantined contract smoke"
+            )
+        executor = DockerTrainingExecutor(
+            image_ref=args.image_ref,
+            gpu_uuid=args.gpu_uuid,
+            approved_inputs_dir=args.approved_inputs_dir,
+            stream_output=args.stream_output,
+        )
+    runner = WakewordTrainingRunner(
+        settings=RunnerSettings(
+            work_root=args.work_root,
+            enabled=args.enable,
+            maintenance_window=StaticMaintenanceWindow(args.maintenance_window_active),
+            process_inspector=(
+                StaticGpuProcessInspector(active_unapproved_processes=False)
+                if args.fake or args.allow_active_gpu_processes
+                else NvidiaSmiProcessInspector(args.gpu_uuid)
+            ),
+            gpu_uuid=None if args.fake else args.gpu_uuid,
+            gpu_sampler=None if args.fake else NvidiaSmiGpuEvidenceSampler(args.gpu_uuid),
+            timeout_s=args.timeout_s,
+        ),
+        executor=executor,
+    )
+    result = await runner.run(request, retry_of=args.retry_of)
+    print(
+        json.dumps(
+            {
+                "run_dir": str(result.run_dir),
+                "run_id": result.state.run_id,
+                "status": result.state.status,
+            },
+            sort_keys=True,
+        )
+    )
+    return terminal_exit_code(result.state.status)
+
+
+if __name__ == "__main__":
+    raise SystemExit(asyncio.run(_main(_parser().parse_args())))

scripts/wakeword_trainer.py

@@ -0,0 +1,603 @@
+"""Standard-library host launcher for the downloadable JRich Wakeword Trainer."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import re
+import shutil
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+RUNNER_VERSION = "0.2.1-jrich-2"
+DEFAULT_IMAGE = (
+    "git.jimandkrista.com/jr-public/jrich-wakeword-training"
+    "@sha256:cef5b1e769a6160bbde69f67df76f0b865f02c27794a3da096398c18cc9a3c2f"
+)
+DEFAULT_CACHE = Path.home() / ".cache" / "jrich-wakeword" / "livekit-0.2.1"
+DEFAULT_WORK_ROOT = Path.cwd() / "jrich-wakeword-runs"
+MIN_FREE_BYTES = 40 * 1024**3
+PRESET_PATH = (
+    Path(__file__).resolve().parents[1]
+    / "deploy"
+    / "wakeword-training"
+    / "presets"
+    / "manual-poc-v1.json"
+)
+_SLUG = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
+_MODEL_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]*$")
+
+
+def _canonical(value: object) -> str:
+    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+
+
+def _sha256(value: bytes) -> str:
+    return hashlib.sha256(value).hexdigest()
+
+
+def _deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]:
+    result = dict(base)
+    for key, value in override.items():
+        if key not in base:
+            raise ValueError(f"unknown pipeline override: {key}")
+        if isinstance(base[key], dict) and isinstance(value, dict):
+            result[key] = _deep_merge(base[key], value)
+        else:
+            result[key] = value
+    return result
+
+
+def default_pipeline_config(
+    *,
+    model_name: str,
+    family_slug: str,
+    version: int,
+    phrases: tuple[str, ...],
+    near_misses: tuple[str, ...],
+) -> dict[str, Any]:
+    """Return the measured, versioned manual-POC profile before user overrides."""
+
+    preset = json.loads(PRESET_PATH.read_text(encoding="utf-8"))
+    if not isinstance(preset, dict) or preset.get("preset_version") != "manual-poc-v1":
+        raise ValueError("the bundled manual POC preset is malformed")
+    config = preset.get("pipeline_config")
+    if not isinstance(config, dict):
+        raise ValueError("the bundled manual POC preset has no pipeline_config")
+    return {
+        "model_name": model_name,
+        "family_slug": family_slug,
+        "version": version,
+        "target_phrases": list(phrases),
+        "custom_negative_phrases": list(near_misses),
+        **config,
+    }
+
+
+def build_setup_command(
+    *, image_ref: str, gpu_uuid: str, cache_dir: Path, config_path: Path
+) -> tuple[str, ...]:
+    """Build the explicit networked cache-population command; training stays offline."""
+
+    cache_dir.mkdir(parents=True, exist_ok=True)
+    return (
+        "docker",
+        "run",
+        "--rm",
+        "--init",
+        "--gpus",
+        f"device={gpu_uuid}",
+        "--read-only",
+        "--cap-drop",
+        "ALL",
+        "--security-opt",
+        "no-new-privileges",
+        "--mount",
+        f"type=bind,src={cache_dir.resolve()},dst=/data",
+        "--mount",
+        f"type=bind,src={config_path.resolve()},dst=/config.json,readonly",
+        "--tmpfs",
+        "/tmp:rw,noexec,nosuid,size=4g",
+        image_ref,
+        "livekit-wakeword",
+        "setup",
+        "--config",
+        "/config.json",
+    )
+
+
+def _run(
+    command: tuple[str, ...] | list[str], *, capture: bool = False
+) -> subprocess.CompletedProcess:
+    return subprocess.run(command, check=False, capture_output=capture, text=True)
+
+
+def _inspect_image(image_ref: str) -> dict[str, Any] | None:
+    completed = _run(
+        ["docker", "image", "inspect", "--format", "{{json .}}", image_ref],
+        capture=True,
+    )
+    if completed.returncode != 0:
+        return None
+    try:
+        value = json.loads(completed.stdout)
+    except json.JSONDecodeError as exc:
+        raise ValueError("Docker returned malformed image metadata") from exc
+    if not isinstance(value, dict):
+        raise ValueError("Docker returned malformed image metadata")
+    return value
+
+
+def _image_identity(image_ref: str) -> str:
+    if image_ref.startswith("sha256:") and len(image_ref) == 71:
+        return image_ref.lower()
+
+    inspected = _inspect_image(image_ref)
+    if inspected is None:
+        print(f"Downloading pinned training runtime: {image_ref}", file=sys.stderr)
+        pulled = _run(["docker", "pull", image_ref])
+        if pulled.returncode != 0:
+            raise ValueError(f"training image download failed: {image_ref}")
+        inspected = _inspect_image(image_ref)
+    if inspected is None:
+        raise ValueError(
+            f"training image is unavailable after download: {image_ref}"
+        )
+
+    if "@sha256:" in image_ref:
+        expected = image_ref.lower()
+        repo_digests = inspected.get("RepoDigests", [])
+        if not isinstance(repo_digests, list) or expected not in {
+            str(value).lower() for value in repo_digests
+        }:
+            raise ValueError("Docker did not retain the requested registry manifest digest")
+
+    digest = str(inspected.get("Id", "")).lower()
+    if not digest.startswith("sha256:") or len(digest) != 71:
+        raise ValueError("Docker returned an invalid immutable image ID")
+    return digest
+
+
+def _default_gpu_uuid() -> str:
+    completed = _run(
+        ["nvidia-smi", "--query-gpu=uuid", "--format=csv,noheader"], capture=True
+    )
+    values = [line.strip() for line in completed.stdout.splitlines() if line.strip()]
+    if completed.returncode != 0 or not values:
+        raise ValueError("no NVIDIA GPU UUID was found; verify the driver and nvidia-smi")
+    if len(values) > 1:
+        raise ValueError("multiple NVIDIA GPUs found; choose one with --gpu-uuid")
+    return values[0]
+
+
+def _prompt(label: str, default: str | None = None) -> str:
+    suffix = f" [{default}]" if default else ""
+    value = input(f"{label}{suffix}: ").strip()
+    return value or (default or "")
+
+
+def _slug(value: str) -> str:
+    candidate = re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-")
+    if not _SLUG.fullmatch(candidate):
+        raise ValueError("family slug must use lowercase letters, numbers, and single hyphens")
+    return candidate
+
+
+def _interactive_values(args: argparse.Namespace) -> None:
+    if args.phrase:
+        return
+    print("JRich Wakeword Trainer — the defaults below match the measured RTX 3090 POC.")
+    primary = _prompt("Primary spoken phrase", "Hey Jay Rich")
+    variants = _prompt("Additional spoken phrases, comma-separated", "Yo Jay Rich")
+    near_misses = _prompt("Near-miss phrases that must not activate, comma-separated", "Jay Rich")
+    label = _prompt("Model name", "jrich")
+    args.phrase = [primary, *[item.strip() for item in variants.split(",") if item.strip()]]
+    args.near_miss = [item.strip() for item in near_misses.split(",") if item.strip()]
+    args.model_name = args.model_name or label
+    args.family_slug = args.family_slug or _slug(label)
+
+
+def _load_override(path: Path | None) -> dict[str, Any]:
+    if path is None:
+        return {}
+    value = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(value, dict):
+        raise ValueError("pipeline override must be a JSON object")
+    return value
+
+
+def _file_sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as file:
+        for block in iter(lambda: file.read(1024 * 1024), b""):
+            digest.update(block)
+    return digest.hexdigest()
+
+
+def write_local_source_lock(work_root: Path, cache_dir: Path) -> tuple[Path, str]:
+    """Write a path-independent, content-addressed inventory of the local cache."""
+
+    request_root = work_root / "requests"
+    request_root.mkdir(parents=True, exist_ok=True)
+    cache_dir.mkdir(parents=True, exist_ok=True)
+    files: list[dict[str, object]] = []
+    for path in sorted(cache_dir.rglob("*")):
+        if path.is_symlink():
+            raise ValueError(f"training cache must not contain symlinks: {path}")
+        if not path.is_file():
+            continue
+        files.append(
+            {
+                "relative_path": path.relative_to(cache_dir).as_posix(),
+                "size_bytes": path.stat().st_size,
+                "sha256": _file_sha256(path),
+            }
+        )
+    value = {
+        "schema_version": 1,
+        "run_kind": "local_experiment",
+        "sources": [
+            {
+                "source_id": "livekit-setup-cache-v0.2.1",
+                "kind": "user-local-cache",
+                "publication_approved": False,
+                "files": files,
+            }
+        ],
+    }
+    encoded = (_canonical(value) + "\n").encode("utf-8")
+    digest = _sha256(encoded)
+    path = request_root / f"sources-{digest}.lock.json"
+    if not path.exists():
+        path.write_bytes(encoded)
+        path.chmod(0o444)
+    return path, digest
+
+
+def resolve_plan(args: argparse.Namespace) -> dict[str, Any]:
+    phrases = tuple(" ".join(value.split()) for value in (args.phrase or ()) if value.strip())
+    if not phrases:
+        raise ValueError("at least one --phrase is required in non-interactive mode")
+    near_misses = tuple(
+        " ".join(value.split()) for value in (args.near_miss or ()) if value.strip()
+    )
+    model_name = (args.model_name or _slug(phrases[0]).replace("-", "_")).strip()
+    if not _MODEL_NAME.fullmatch(model_name):
+        raise ValueError(
+            "model name must start with a letter or number and contain only letters, "
+            "numbers, underscores, or hyphens"
+        )
+    if args.version < 1:
+        raise ValueError("version must be a positive integer")
+    family_slug = _slug(args.family_slug or model_name)
+    image_digest = _image_identity(args.image_ref)
+    gpu_uuid = args.gpu_uuid or _default_gpu_uuid()
+    cache_dir = args.cache_dir.resolve()
+    work_root = args.work_root.resolve()
+    pipeline = default_pipeline_config(
+        model_name=model_name,
+        family_slug=family_slug,
+        version=args.version,
+        phrases=phrases,
+        near_misses=near_misses,
+    )
+    pipeline = _deep_merge(pipeline, _load_override(args.pipeline_config))
+    identity = {
+        "model_name": model_name,
+        "family_slug": family_slug,
+        "version": args.version,
+        "target_phrases": list(phrases),
+        "custom_negative_phrases": list(near_misses),
+    }
+    divergent = [name for name, value in identity.items() if pipeline.get(name) != value]
+    if divergent:
+        raise ValueError(
+            "pipeline override cannot change request identity fields: " + ", ".join(divergent)
+        )
+    source_lock, source_digest = write_local_source_lock(work_root, cache_dir)
+    request = {
+        "schema_version": 1,
+        "primary_phrase": phrases[0],
+        "variants": list(phrases[1:]),
+        "near_misses": list(near_misses),
+        "evaluation_profile": "local-balanced-v1",
+        "source_lock_path": str(source_lock),
+        "source_lock_sha256": source_digest,
+        "intended_release_id": f"{family_slug}-v{args.version}",
+        "runner_image_digest": image_digest,
+        "pipeline_config": pipeline,
+        "run_kind": "local_experiment",
+        "derived_from_run_id": args.reuse_from,
+    }
+    # Keep the local wizard and future API worker on the exact same fail-closed
+    # validation boundary. Import shape differs only between the extracted
+    # archive (sibling module) and repository tests (scripts package).
+    try:
+        from wakeword_training import RunKind, WakewordTrainingRequest
+    except ModuleNotFoundError:
+        from scripts.wakeword_training import RunKind, WakewordTrainingRequest
+
+    WakewordTrainingRequest(
+        primary_phrase=request["primary_phrase"],
+        variants=tuple(request["variants"]),
+        near_misses=tuple(request["near_misses"]),
+        evaluation_profile=request["evaluation_profile"],
+        source_lock_path=Path(request["source_lock_path"]),
+        source_lock_sha256=request["source_lock_sha256"],
+        intended_release_id=request["intended_release_id"],
+        runner_image_digest=request["runner_image_digest"],
+        pipeline_config=request["pipeline_config"],
+        run_kind=RunKind.LOCAL_EXPERIMENT,
+        derived_from_run_id=request["derived_from_run_id"],
+    ).canonical_payload()
+    return {
+        "schema_version": 1,
+        "runner_version": RUNNER_VERSION,
+        "request": request,
+        "runtime": {
+            "image_ref": image_digest,
+            "image_source": args.image_ref,
+            "gpu_uuid": gpu_uuid,
+            "cache_dir": str(cache_dir),
+            "work_root": str(work_root),
+            "hardware_profile": "shared-conservative"
+            if args.allow_active_gpu_processes
+            else "exclusive",
+        },
+    }
+
+
+def _preflight(plan: dict[str, Any]) -> None:
+    runtime = plan["runtime"]
+    for executable in ("docker", "nvidia-smi"):
+        if shutil.which(executable) is None:
+            raise ValueError(f"required host command is missing: {executable}")
+    docker = _run(["docker", "info"], capture=True)
+    if docker.returncode != 0:
+        raise ValueError("Docker daemon is unavailable to this user")
+    gpu = _run(
+        [
+            "nvidia-smi",
+            f"--id={runtime['gpu_uuid']}",
+            "--query-gpu=uuid,name,memory.total,memory.free,driver_version",
+            "--format=csv,noheader,nounits",
+        ],
+        capture=True,
+    )
+    if gpu.returncode != 0:
+        raise ValueError("the selected GPU UUID is unavailable")
+    container_gpu = _run(
+        [
+            "docker",
+            "run",
+            "--rm",
+            "--gpus",
+            f"device={runtime['gpu_uuid']}",
+            "--network",
+            "none",
+            "--read-only",
+            "--cap-drop",
+            "ALL",
+            "--security-opt",
+            "no-new-privileges",
+            runtime["image_ref"],
+            "python",
+            "-c",
+            "import torch; assert torch.cuda.is_available(); print(torch.cuda.get_device_name())",
+        ],
+        capture=True,
+    )
+    if container_gpu.returncode != 0:
+        raise ValueError("NVIDIA Container Toolkit did not expose the selected GPU to the image")
+    for key in ("cache_dir", "work_root"):
+        path = Path(runtime[key])
+        path.mkdir(parents=True, exist_ok=True)
+        free = shutil.disk_usage(path).free
+        if free < MIN_FREE_BYTES:
+            raise ValueError(f"{path} has less than the recommended 40 GiB free")
+    print(f"GPU: {gpu.stdout.strip()}", file=sys.stderr)
+    print(f"Container CUDA: {container_gpu.stdout.strip()}", file=sys.stderr)
+    print(f"Host CPUs: {os.cpu_count() or 'unknown'}", file=sys.stderr)
+    print("Preflight passed; training will run with network disabled.", file=sys.stderr)
+
+
+def _write_request(plan: dict[str, Any]) -> Path:
+    work_root = Path(plan["runtime"]["work_root"])
+    encoded = (_canonical(plan["request"]) + "\n").encode("utf-8")
+    path = work_root / "requests" / f"request-{_sha256(encoded)}.json"
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_bytes(encoded)
+    return path
+
+
+def _setup(plan: dict[str, Any]) -> None:
+    runtime = plan["runtime"]
+    config = dict(plan["request"]["pipeline_config"])
+    config["data_dir"] = "/data"
+    config["output_dir"] = "/tmp/setup-output"
+    setup_path = Path(runtime["work_root"]) / "requests" / "setup-config.json"
+    setup_path.write_text(_canonical(config) + "\n", encoding="utf-8")
+    command = build_setup_command(
+        image_ref=runtime["image_ref"],
+        gpu_uuid=runtime["gpu_uuid"],
+        cache_dir=Path(runtime["cache_dir"]),
+        config_path=setup_path,
+    )
+    completed = _run(command)
+    if completed.returncode != 0:
+        raise ValueError("LiveKit asset setup failed")
+
+
+def _train(plan: dict[str, Any], request_path: Path, args: argparse.Namespace) -> int:
+    runtime = plan["runtime"]
+    command = [
+        sys.executable,
+        str(Path(__file__).with_name("run-wakeword-training")),
+        "--request",
+        str(request_path),
+        "--work-root",
+        runtime["work_root"],
+        "--enable",
+        "--maintenance-window-active",
+        "--image-ref",
+        runtime["image_ref"],
+        "--gpu-uuid",
+        runtime["gpu_uuid"],
+        "--approved-inputs-dir",
+        runtime["cache_dir"],
+        "--timeout-s",
+        str(args.timeout_s),
+        "--stream-output",
+    ]
+    if args.allow_active_gpu_processes:
+        command.append("--allow-active-gpu-processes")
+    if args.retry_of:
+        command.extend(("--retry-of", args.retry_of))
+    return _run(command).returncode
+
+
+def _find_run(work_root: Path, run_id: str) -> tuple[Path, dict[str, Any]]:
+    if len(run_id) != 32 or any(character not in "0123456789abcdef" for character in run_id):
+        raise ValueError("run ID must be 32 lowercase hexadecimal characters")
+    matches: list[tuple[Path, dict[str, Any]]] = []
+    for run_dir in (work_root / "runs").glob(f"*-{run_id}"):
+        state_path = run_dir / "state.json"
+        if not state_path.is_file():
+            continue
+        state = json.loads(state_path.read_text(encoding="utf-8"))
+        if isinstance(state, dict) and state.get("run_id") == run_id:
+            matches.append((run_dir, state))
+    if len(matches) != 1:
+        raise ValueError("run ID was not found under the selected work root")
+    return matches[0]
+
+
+def run_local_operation(args: argparse.Namespace) -> int | None:
+    operation = next(
+        (
+            (name, run_id)
+            for name, run_id in (
+                ("status", args.status),
+                ("logs", args.logs),
+                ("cancel", args.cancel),
+            )
+            if run_id is not None
+        ),
+        None,
+    )
+    if operation is None:
+        return None
+    name, run_id = operation
+    run_dir, state = _find_run(args.work_root.resolve(), run_id)
+    if name == "status":
+        result_path = run_dir / "training-result.json"
+        result = (
+            json.loads(result_path.read_text(encoding="utf-8"))
+            if result_path.is_file()
+            else None
+        )
+        print(json.dumps({"run_dir": str(run_dir), "state": state, "result": result}, indent=2))
+        return 0
+    if name == "logs":
+        for label in ("stdout.log", "stderr.log"):
+            print(f"== {label} ==")
+            path = run_dir / label
+            if path.is_file():
+                print(path.read_text(encoding="utf-8", errors="replace"), end="")
+        return 0
+    if state.get("status") in {
+        "completed",
+        "pending_publication",
+        "failed",
+        "canceled",
+        "timed_out",
+        "abandoned",
+    }:
+        raise ValueError("terminal run cannot be canceled")
+    (run_dir / "cancel-requested").write_text("requested\n", encoding="utf-8")
+    print(f"Cancellation requested for {run_id}; the runner will stop its own container.")
+    return 0
+
+
+def _parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description="Train a LiveKit Wakeword model in pinned Docker")
+    parser.add_argument("--non-interactive", action="store_true")
+    parser.add_argument("--plan", action="store_true", help="print resolved JSON without running")
+    operation = parser.add_mutually_exclusive_group()
+    operation.add_argument("--status", metavar="RUN_ID", help="show one local attempt")
+    operation.add_argument("--logs", metavar="RUN_ID", help="show one local attempt's logs")
+    operation.add_argument("--cancel", metavar="RUN_ID", help="request safe cancellation")
+    parser.add_argument("--phrase", action="append", help="spoken phrase; repeat for alternatives")
+    parser.add_argument("--near-miss", action="append", help="phrase that must not activate")
+    parser.add_argument("--model-name")
+    parser.add_argument("--family-slug")
+    parser.add_argument("--version", type=int, default=1)
+    parser.add_argument("--pipeline-config", type=Path, help="JSON deep overrides")
+    parser.add_argument("--image-ref", default=os.getenv("JR_WAKEWORD_IMAGE", DEFAULT_IMAGE))
+    parser.add_argument("--gpu-uuid")
+    parser.add_argument("--cache-dir", type=Path, default=DEFAULT_CACHE)
+    parser.add_argument("--work-root", type=Path, default=DEFAULT_WORK_ROOT)
+    parser.add_argument("--skip-setup", action="store_true")
+    parser.add_argument(
+        "--setup-only",
+        action="store_true",
+        help="populate the pinned input cache, print its source-lock digest, and exit",
+    )
+    parser.add_argument(
+        "--reuse-from",
+        help="terminal local run ID whose sealed generated corpus seeds a new attempt",
+    )
+    parser.add_argument(
+        "--retry-of", help="matching terminal run ID to retry as a new immutable attempt"
+    )
+    parser.add_argument("--allow-active-gpu-processes", action="store_true")
+    parser.add_argument("--timeout-s", type=float, default=6 * 60 * 60)
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = _parser().parse_args(argv)
+    try:
+        if getattr(args, "setup_only", False) and args.skip_setup:
+            raise ValueError("--setup-only and --skip-setup cannot be combined")
+        operation_result = run_local_operation(args)
+        if operation_result is not None:
+            return operation_result
+        if not args.non_interactive:
+            _interactive_values(args)
+        plan = resolve_plan(args)
+        if args.plan:
+            print(json.dumps(plan, indent=2, sort_keys=True))
+            return 0
+        print(json.dumps(plan, indent=2, sort_keys=True))
+        if not args.non_interactive and _prompt("Continue with this configuration", "yes") != "yes":
+            return 0
+        _preflight(plan)
+        if not args.skip_setup:
+            _setup(plan)
+            refreshed = resolve_plan(args)
+            if refreshed["request"]["source_lock_sha256"] != plan["request"][
+                "source_lock_sha256"
+            ]:
+                plan = refreshed
+                print("Final content-locked request after setup:", file=sys.stderr)
+                print(json.dumps(plan, indent=2, sort_keys=True))
+        if getattr(args, "setup_only", False):
+            print(
+                "WAKEWORD_TRAINING_SOURCE_LOCK_SHA256="
+                + plan["request"]["source_lock_sha256"]
+            )
+            return 0
+        request_path = _write_request(plan)
+        return _train(plan, request_path, args)
+    except (OSError, ValueError, json.JSONDecodeError) as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/wakeword_training_image.py

@@ -0,0 +1,395 @@
+"""Pinned-image input and provenance contract for Wakeword Training (#163).
+
+This module deliberately does not implement wakeword generation, augmentation,
+feature extraction, training, export, or evaluation.  It validates the
+commercial source lock, materializes only its declared inputs, and leaves the
+actual ML stages to the released LiveKit CLI inside the isolated image.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import re
+import shutil
+from dataclasses import dataclass
+from enum import StrEnum
+from pathlib import Path
+from typing import Any
+
+LIVEKIT_WAKEWORD_VERSION = "0.2.1"
+LIVEKIT_WAKEWORD_REVISION = "1ec7f680df30ff4ca0ebae6b5983441e94b10980"
+LIVEKIT_WAKEWORD_SDIST_SHA256 = "cf2d9cf4867812c06788f64c15e49abd909d9d6291f0a13f1c3f9cb649fa6127"
+
+_COMMERCIAL_PURPOSES = frozenset({"wakeword_training", "classifier_distribution"})
+_BLOCKED_LICENSE_MARKERS = ("unknown", "nc", "non-commercial", "research-only")
+_BLOCKED_SOURCE_MARKERS = ("acav", "unknown_rir_mirror")
+_SYNTHETIC_SOURCE_KINDS = frozenset({"piper_output", "voxcpm_output"})
+_SAFE_TOKEN = re.compile(r"^[a-z0-9][a-z0-9._-]{0,127}$")
+_ALLOWED_PURPOSES = frozenset(
+    {
+        "positive",
+        "adversarial_negative",
+        "near_miss",
+        "general_negative_speech",
+        "background",
+        "rir",
+    }
+)
+_ALLOWED_PARTITIONS = frozenset({"train", "validation", "calibration", "release_test"})
+_REQUIRED_PURPOSE_PARTITIONS = frozenset(
+    {
+        ("positive", "train"),
+        ("positive", "release_test"),
+        ("general_negative_speech", "train"),
+        ("general_negative_speech", "validation"),
+        ("background", "train"),
+        ("rir", "train"),
+    }
+)
+
+
+class SourceLockError(ValueError):
+    """A source lock cannot safely enter the requested training lane."""
+
+
+class RunClassification(StrEnum):
+    COMMERCIAL_REVIEW_REQUIRED = "commercial_review_required"
+    LOCAL_ONLY = "local_only"
+    QUARANTINED = "quarantined"
+
+    @property
+    def commercial_promotion_eligible(self) -> bool:
+        return self is RunClassification.COMMERCIAL_REVIEW_REQUIRED
+
+
+def classify_run(run_kind: str) -> RunClassification:
+    if run_kind == "commercial":
+        return RunClassification.COMMERCIAL_REVIEW_REQUIRED
+    if run_kind == "local_experiment":
+        return RunClassification.LOCAL_ONLY
+    if run_kind == "quarantined_smoke":
+        return RunClassification.QUARANTINED
+    raise SourceLockError("run_kind must be commercial, local_experiment, or quarantined_smoke")
+
+
+def _sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as file:
+        for block in iter(lambda: file.read(1024 * 1024), b""):
+            digest.update(block)
+    return digest.hexdigest()
+
+
+def _require_mapping(value: object, label: str) -> dict[str, Any]:
+    if not isinstance(value, dict):
+        raise SourceLockError(f"{label} must be an object")
+    return value
+
+
+def _require_string(value: object, label: str) -> str:
+    if not isinstance(value, str) or not value.strip():
+        raise SourceLockError(f"{label} must be a non-empty string")
+    return value.strip()
+
+
+def _require_sha256(value: object, label: str) -> str:
+    digest = _require_string(value, label).removeprefix("sha256:").lower()
+    if len(digest) != 64 or any(character not in "0123456789abcdef" for character in digest):
+        raise SourceLockError(f"{label} must be a SHA-256 digest")
+    return digest
+
+
+def _safe_input_path(root: Path, declared_path: str) -> Path:
+    candidate = (root / declared_path).resolve(strict=True)
+    if not candidate.is_file() or os.path.commonpath((root, candidate)) != str(root):
+        raise SourceLockError("source lock file path must identify a regular file below input_root")
+    return candidate
+
+
+@dataclass(frozen=True)
+class MaterializedInputs:
+    provenance_manifest_path: Path
+    recorded_manifest_path: Path
+
+
+class InputMaterializer:
+    """Verify and copy a source lock's declared inputs into one sealed run.
+
+    The input root is read-only when this code runs in the training image.  The
+    run directory is the sole persistence target, so no upstream ``setup``
+    download can become an undeclared training input.
+    """
+
+    def __init__(self, lock_path: Path, input_root: Path, run_dir: Path, *, run_kind: str) -> None:
+        self.lock_path = lock_path.resolve(strict=True)
+        self.input_root = input_root.resolve(strict=True)
+        self.run_dir = run_dir.resolve()
+        self.run_kind = run_kind
+
+    def materialize(self) -> MaterializedInputs:
+        classification = classify_run(self.run_kind)
+        lock = self._read_lock()
+        if lock.get("run_kind") != self.run_kind:
+            raise SourceLockError("source lock run_kind does not match the requested run")
+        sources = lock.get("sources")
+        if not isinstance(sources, list) or not sources:
+            raise SourceLockError("source lock must contain at least one source")
+
+        materialized_root = self.run_dir / "materialized"
+        materialized_root.mkdir(parents=True, exist_ok=False)
+        source_manifest: list[dict[str, object]] = []
+        supplied_purposes: set[tuple[str, str]] = set()
+        seen_source_ids: set[str] = set()
+        materialized_destinations: set[Path] = set()
+        for index, raw_source in enumerate(sources):
+            source = _require_mapping(raw_source, f"sources[{index}]")
+            source_id, purpose, partition, file_records = self._validate_source(
+                source, index, commercial=classification.commercial_promotion_eligible
+            )
+            if source_id in seen_source_ids:
+                raise SourceLockError(f"duplicate source_id: {source_id}")
+            seen_source_ids.add(source_id)
+            supplied_purposes.add((purpose, partition))
+            target_root = materialized_root / partition / purpose / source_id
+            target_root.mkdir(parents=True)
+            materialized_files: list[dict[str, str]] = []
+            for record in file_records:
+                source_path = _safe_input_path(self.input_root, record["path"])
+                if _sha256_file(source_path) != record["sha256"]:
+                    raise SourceLockError(f"source hash mismatch: {record['path']}")
+                target = target_root / source_path.name
+                if target in materialized_destinations:
+                    raise SourceLockError(f"duplicate materialized destination: {target.name}")
+                materialized_destinations.add(target)
+                shutil.copyfile(source_path, target)
+                target.chmod(0o444)
+                materialized_files.append(
+                    {
+                        "source_path": record["path"],
+                        "materialized_path": target.relative_to(self.run_dir).as_posix(),
+                        "sha256": record["sha256"],
+                    }
+                )
+            source_manifest.append(
+                {
+                    "source_id": source_id,
+                    "purpose": purpose,
+                    "partition": partition,
+                    "files": materialized_files,
+                }
+            )
+
+        if classification.commercial_promotion_eligible:
+            missing = _REQUIRED_PURPOSE_PARTITIONS - supplied_purposes
+            if missing:
+                formatted = ", ".join(
+                    f"{purpose}:{partition}" for purpose, partition in sorted(missing)
+                )
+                raise SourceLockError(
+                    f"commercial source lock is missing required inputs: {formatted}"
+                )
+
+        provenance_path = self.run_dir / "sources.lock.json"
+        if self.lock_path != provenance_path:
+            shutil.copyfile(self.lock_path, provenance_path)
+        provenance_path.chmod(0o444)
+        attribution_dir = self.run_dir / "attribution"
+        attribution_dir.mkdir(exist_ok=True)
+        self._materialize_attribution(lock["sources"], attribution_dir)
+        recorded_manifest_path = self.run_dir / "recorded-inputs.json"
+        recorded_manifest_path.write_text(
+            json.dumps(
+                {
+                    "schema_version": 1,
+                    "classification": classification.value,
+                    "sources": source_manifest,
+                },
+                sort_keys=True,
+                separators=(",", ":"),
+            )
+            + "\n",
+            encoding="utf-8",
+        )
+        recorded_manifest_path.chmod(0o444)
+        return MaterializedInputs(
+            provenance_manifest_path=provenance_path,
+            recorded_manifest_path=recorded_manifest_path,
+        )
+
+    def _read_lock(self) -> dict[str, Any]:
+        try:
+            lock = json.loads(self.lock_path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError) as exc:
+            raise SourceLockError("source lock must be readable JSON") from exc
+        lock = _require_mapping(lock, "source lock")
+        if lock.get("schema_version") != 1:
+            raise SourceLockError("source lock schema_version must be 1")
+        return lock
+
+    def _validate_source(
+        self, source: dict[str, Any], index: int, *, commercial: bool
+    ) -> tuple[str, str, str, list[dict[str, str]]]:
+        source_id = _require_string(source.get("source_id"), f"sources[{index}].source_id")
+        purpose = _require_string(source.get("purpose"), f"sources[{index}].purpose")
+        partition = _require_string(source.get("partition"), f"sources[{index}].partition")
+        source_kind = _require_string(source.get("source_kind"), f"sources[{index}].source_kind")
+        if not _SAFE_TOKEN.fullmatch(source_id):
+            raise SourceLockError(f"sources[{index}].source_id must be a safe token")
+        if purpose not in _ALLOWED_PURPOSES:
+            raise SourceLockError(f"sources[{index}].purpose is not an allowed purpose")
+        if partition not in _ALLOWED_PARTITIONS:
+            raise SourceLockError(f"sources[{index}].partition is not an allowed partition")
+        publisher = _require_mapping(
+            source.get("original_publisher"), f"sources[{index}].original_publisher"
+        )
+        _require_string(publisher.get("name"), f"sources[{index}].original_publisher.name")
+        _require_string(publisher.get("url"), f"sources[{index}].original_publisher.url")
+        _require_string(source.get("version"), f"sources[{index}].version")
+        _require_string(source.get("revision"), f"sources[{index}].revision")
+        _require_string(source.get("attribution"), f"sources[{index}].attribution")
+        license_data = _require_mapping(source.get("license"), f"sources[{index}].license")
+        license_id = _require_string(
+            license_data.get("spdx_id"), f"sources[{index}].license.spdx_id"
+        )
+        _require_string(license_data.get("terms_url"), f"sources[{index}].license.terms_url")
+        captured_text = _require_string(
+            license_data.get("captured_text_path"),
+            f"sources[{index}].license.captured_text_path",
+        )
+        captured_hash = _require_sha256(
+            license_data.get("captured_text_sha256"),
+            f"sources[{index}].license.captured_text_sha256",
+        )
+        license_path = _safe_input_path(self.input_root, captured_text)
+        if _sha256_file(license_path) != captured_hash:
+            raise SourceLockError(f"captured license hash mismatch: {captured_text}")
+        files = source.get("files")
+        if not isinstance(files, list) or not files:
+            raise SourceLockError(f"sources[{index}].files must be a non-empty list")
+        records: list[dict[str, str]] = []
+        for file_index, raw_file in enumerate(files):
+            file_data = _require_mapping(raw_file, f"sources[{index}].files[{file_index}]")
+            path = _require_string(
+                file_data.get("path"), f"sources[{index}].files[{file_index}].path"
+            )
+            digest = _require_sha256(
+                file_data.get("sha256"), f"sources[{index}].files[{file_index}].sha256"
+            )
+            size = file_data.get("size_bytes")
+            if not isinstance(size, int) or size < 1:
+                raise SourceLockError(
+                    f"sources[{index}].files[{file_index}].size_bytes must be positive"
+                )
+            if _safe_input_path(self.input_root, path).stat().st_size != size:
+                raise SourceLockError(f"source size mismatch: {path}")
+            records.append({"path": path, "sha256": digest})
+        if commercial:
+            self._validate_commercial_source(source_id, source_kind, license_id, source, index)
+        return source_id, purpose, partition, records
+
+    @staticmethod
+    def _validate_commercial_source(
+        source_id: str, source_kind: str, license_id: str, source: dict[str, Any], index: int
+    ) -> None:
+        normalized_id = source_id.lower()
+        normalized_license = license_id.lower()
+        if any(marker in normalized_id for marker in _BLOCKED_SOURCE_MARKERS) or any(
+            marker in normalized_license for marker in _BLOCKED_LICENSE_MARKERS
+        ):
+            raise SourceLockError(f"blocked source for commercial mode: {source_id}")
+        commercial_data = _require_mapping(source.get("commercial"), f"sources[{index}].commercial")
+        if commercial_data.get("disposition") != "approved":
+            raise SourceLockError(f"commercial source is not approved: {source_id}")
+        allowed_purposes = commercial_data.get("allowed_purposes")
+        if not isinstance(allowed_purposes, list) or not _COMMERCIAL_PURPOSES.issubset(
+            set(allowed_purposes)
+        ):
+            raise SourceLockError(f"commercial source has incomplete allowed_purposes: {source_id}")
+        _require_string(commercial_data.get("approver"), f"sources[{index}].commercial.approver")
+        _require_string(
+            commercial_data.get("approved_at"), f"sources[{index}].commercial.approved_at"
+        )
+        if (
+            source_kind in _SYNTHETIC_SOURCE_KINDS
+            and commercial_data.get("tts_output_approved") is not True
+        ):
+            raise SourceLockError(f"unapproved synthetic source for commercial mode: {source_id}")
+
+    def _materialize_attribution(self, sources: object, attribution_dir: Path) -> None:
+        assert isinstance(sources, list)
+        for raw_source in sources:
+            source = _require_mapping(raw_source, "source")
+            source_id = _require_string(source.get("source_id"), "source.source_id")
+            license_data = _require_mapping(source.get("license"), "source.license")
+            license_path = _safe_input_path(
+                self.input_root,
+                _require_string(
+                    license_data.get("captured_text_path"), "source.license.captured_text_path"
+                ),
+            )
+            target = attribution_dir / f"{source_id}-license.txt"
+            shutil.copyfile(license_path, target)
+            target.chmod(0o444)
+
+
+def build_stage_command(run_kind: str) -> tuple[str, ...]:
+    """Return the only command an isolated training container may execute."""
+
+    classification = classify_run(run_kind)
+    if classification is RunClassification.COMMERCIAL_REVIEW_REQUIRED:
+        return (
+            "python",
+            "/opt/jr-wakeword/run-commercial-pipeline.py",
+            "/run/resolved-config.yaml",
+        )
+    if classification is RunClassification.LOCAL_ONLY:
+        return (
+            "python",
+            "/opt/jr-wakeword/run-local-experiment.py",
+            "/run/resolved-config.yaml",
+        )
+    return (
+        "python",
+        "/opt/jr-wakeword/run-quarantined-smoke.py",
+        "/run/resolved-config.yaml",
+    )
+
+
+def collect_image_provenance(
+    *,
+    dependency_lock_path: Path,
+    frontend_asset_paths: list[Path],
+    installed_distributions: dict[str, str],
+    image_digest: str,
+    python_version: str,
+    cuda_version: str,
+    os_packages: list[str],
+) -> dict[str, object]:
+    """Return manifest-safe image evidence; callers persist it in the run bundle."""
+
+    if not image_digest.startswith("sha256:"):
+        raise ValueError("image_digest must be sha256-pinned")
+    lock_path = dependency_lock_path.resolve(strict=True)
+    assets = []
+    for path in frontend_asset_paths:
+        resolved = path.resolve(strict=True)
+        assets.append({"path": resolved.name, "sha256": _sha256_file(resolved)})
+    return {
+        "schema_version": 1,
+        "image_digest": image_digest,
+        "python_version": python_version,
+        "cuda_version": cuda_version,
+        "os_packages": sorted(os_packages),
+        "livekit_wakeword": {
+            "distribution": "livekit-wakeword",
+            "version": LIVEKIT_WAKEWORD_VERSION,
+            "source_revision": LIVEKIT_WAKEWORD_REVISION,
+            "source_distribution_sha256": LIVEKIT_WAKEWORD_SDIST_SHA256,
+            "installed_version": installed_distributions.get("livekit-wakeword"),
+        },
+        "installed_distributions": dict(sorted(installed_distributions.items())),
+        "dependency_lock": {"path": lock_path.name, "sha256": _sha256_file(lock_path)},
+        "frontend_assets": assets,
+    }