scripts/run_captured_continuation.py

#!/usr/bin/env python3
"""Run or verify Tier-A captured-state continuation."""

from __future__ import annotations

import argparse
import json
from pathlib import Path

from music3lab.editing.continue_capture import ArbitraryWavContinuationUnavailableError
from music3lab.editing.music3_continuation import (
    run_captured_continuation,
    verify_continuation_bundle,
)


def parser() -> argparse.ArgumentParser:
    result = argparse.ArgumentParser()
    commands = result.add_subparsers(dest="command", required=True)
    run = commands.add_parser("run")
    for name in (
        "config",
        "snapshot",
        "base-manifest",
        "diffusers-root",
        "phase0-artifacts",
        "output-root",
    ):
        run.add_argument(f"--{name}", required=True)
    verify = commands.add_parser("verify")
    verify.add_argument("--output-root", required=True)
    blocked = commands.add_parser("arbitrary-wav")
    blocked.add_argument("--wav", required=True)
    return result


def main(argv: list[str] | None = None) -> int:
    args = parser().parse_args(argv)
    if args.command == "run":
        verified = run_captured_continuation(
            config_path=Path(args.config),
            snapshot=Path(args.snapshot),
            base_manifest=Path(args.base_manifest),
            diffusers_root=Path(args.diffusers_root),
            phase0_artifacts=Path(args.phase0_artifacts),
            output_root=Path(args.output_root),
        )
        print(json.dumps({
            "captured_state_continuation_pass": verified["metrics"][
                "captured_state_continuation_pass"
            ],
            "metrics_semantic_digest": verified["metrics"]["semantic_digest"],
            "output_root": str(Path(args.output_root).absolute()),
        }, sort_keys=True))
        return 0
    if args.command == "verify":
        verified = verify_continuation_bundle(Path(args.output_root))
        print(json.dumps({
            "captured_state_continuation_pass": True,
            "metrics_semantic_digest": verified["metrics"]["semantic_digest"],
        }, sort_keys=True))
        return 0
    try:
        raise ArbitraryWavContinuationUnavailableError(
            "arbitrary WAV continuation is BLOCKED; a verified Phase-0 "
            "native capture is required"
        )
    except ArbitraryWavContinuationUnavailableError as error:
        print(json.dumps({
            "capability": "arbitrary_wav_continuation",
            "status": "BLOCKED",
            "detail": str(error),
            "wav": args.wav,
        }, sort_keys=True))
        return 2


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