Astral Omnidominion

"""Plan-first Python entry point for the Astral Omnidominion demo.

Plan and help are rendered locally without invoking npm or inspecting the
runtime. Only an explicit, fully specified execute request reaches the injected
runner; valid execution arguments are forwarded unchanged.
"""
from __future__ import annotations

import os
import subprocess
import sys
from pathlib import Path
from typing import Callable, Iterable, NamedTuple, Protocol


class _Runner(Protocol):
    def __call__(self, cmd: list[str], *, check: bool, cwd: str | os.PathLike[str]) -> subprocess.CompletedProcess:
        ...


REPO_ROOT = Path(__file__).resolve().parents[2]
_HELP_FLAGS = frozenset({"-h", "-?", "--help"})
_VALUE_FLAGS = frozenset({"--mode", "--network"})
_BOOLEAN_FLAGS = frozenset(
    {"--dry-run", "--yes", "--non-interactive", "--compose", "--no-compose", "--skip-deploy"}
)


class _Options(NamedTuple):
    mode: str = "plan"
    network: str | None = None
    compose: bool | None = None
    skip_deploy: bool = False
    yes: bool = False
    help: bool = False


def _parse_args(args: list[str]) -> _Options:
    """Validate the complete argument list before any operational work."""

    if any(arg in _HELP_FLAGS for arg in args):
        if len(args) != 1:
            raise ValueError("Help (-h, -?, --help) must be used alone.")
        return _Options(help=True)

    values: dict[str, str | bool] = {}
    index = 0
    while index < len(args):
        flag = args[index]
        if flag not in _VALUE_FLAGS and flag not in _BOOLEAN_FLAGS:
            raise ValueError(f"Unknown argument: {flag!r}. Use --help for supported options.")
        if flag in values:
            raise ValueError(f"Duplicate option: {flag}.")
        if flag in _VALUE_FLAGS:
            if index + 1 >= len(args) or args[index + 1].startswith("-"):
                raise ValueError(f"{flag} requires a value.")
            values[flag] = args[index + 1]
            index += 2
        else:
            values[flag] = True
            index += 1

    if "--mode" in values and "--dry-run" in values:
        raise ValueError("--dry-run is an alias for plan; do not combine it with --mode.")
    mode = values.get("--mode", "plan")
    if mode not in {"plan", "execute"}:
        raise ValueError("--mode must be plan or execute.")
    network = values.get("--network")
    if network is not None and network not in {"localhost", "sepolia"}:
        raise ValueError("--network must be localhost or sepolia.")
    if "--compose" in values and "--no-compose" in values:
        raise ValueError("--compose and --no-compose cannot be combined.")
    if "--skip-deploy" in values and "--compose" in values:
        raise ValueError("--skip-deploy cannot be combined with --compose.")

    compose = True if "--compose" in values else False if "--no-compose" in values else None
    yes = "--yes" in values
    if mode == "execute":
        if network is None:
            raise ValueError("Execute requires an explicit --network localhost or --network sepolia.")
        if compose is None:
            raise ValueError("Execute requires exactly one of --compose or --no-compose.")
        if not yes:
            raise ValueError("Execute requires --yes; --non-interactive, environment variables and TTY state are not consent.")

    return _Options(
        mode=str(mode),
        network=str(network) if network is not None else None,
        compose=compose,
        skip_deploy="--skip-deploy" in values,
        yes=yes,
    )


def build_command(args: Iterable[str]) -> list[str]:
    """Return exact npm argv after validation; raise ValueError on invalid input.

    This function only constructs a command. ``run`` handles plan and help
    locally and never submits those commands to a runner.
    """

    original_args = list(args)
    _parse_args(original_args)
    return ["npm", "run", "demo:agi-os:first-class", "--", *original_args]


def _print_help() -> None:
    print(
        "Astral Omnidominion Operating System Demo\n"
        "\nUsage: python run_demo.py [options]\n"
        "\nOptions:\n"
        "  --mode plan|execute       Default: plan (effect-free preview).\n"
        "  --dry-run                 Exact alias for plan; cannot accompany --mode.\n"
        "  --network localhost|sepolia\n"
        "                            Required explicitly for execute.\n"
        "  --compose | --no-compose  Exactly one is required for execute.\n"
        "  --yes                     Explicit consent required for execute.\n"
        "  --non-interactive         Never substitutes for --yes.\n"
        "  --skip-deploy             Skip only the first-class deployment wizard.\n"
        "                            Incompatible with --compose.\n"
        "  -h, -?, --help            Show help; must be used alone.\n"
        "\nUnknown, positional, duplicate and --option=value arguments are rejected.\n"
        "CI, ASTRAL_OS_AUTO and terminal state do not authorize execution.\n"
        "--auto and --auto-yes are unsupported.\n"
        "Plan/help do not launch subprocesses, use RPC, deploy or create reports.\n"
        "--skip-deploy skips only the first-class deployment wizard; owner/runtime commands still run.\n"
        "The bundled rehearsal is a separate in-memory Hardhat fixture; it does not\n"
        "verify the selected localhost/Sepolia deployment.\n"
        "The Sepolia outer path is available in source but remains operationally unqualified.\n"
        "\nExamples:\n"
        "  python run_demo.py\n"
        "  python run_demo.py --dry-run --network sepolia --no-compose\n"
        "  python run_demo.py --mode execute --network localhost --no-compose --yes"
    )


def _print_plan(options: _Options) -> None:
    network = options.network or "not selected (required for execute)"
    compose = {True: "enabled", False: "disabled", None: "not selected (required for execute)"}[options.compose]
    deployment = "first-class wizard skipped by --skip-deploy" if options.skip_deploy else "would run the first-class deployment wizard"
    print(
        "Astral Omnidominion Operating System: plan (effect-free preview)\n"
        f"Network: {network}\n"
        f"Docker Compose: {compose}\n"
        f"Deployment: {deployment}\n"
        "Execution would perform preflight checks, run the bundled rehearsal,\n"
        "run selected-target owner commands, and generate mission reports and manifests.\n"
        "The bundled rehearsal is a separate in-memory Hardhat fixture; it does not\n"
        "verify the selected localhost/Sepolia deployment.\n"
        "The Sepolia outer path is available in source but remains operationally unqualified.\n"
        "No subprocesses, RPC calls, deployment, filesystem writes or reports are performed in plan mode.\n"
        "Runtime prerequisites and network connectivity have not been checked.\n"
        "To execute, explicitly provide --mode execute, --network localhost|sepolia,\n"
        "exactly one of --compose/--no-compose, and --yes.\n"
        "--skip-deploy skips only the first-class deployment wizard; owner/runtime commands still run."
    )


def run(
    argv: Iterable[str] | None = None,
    *,
    runner: _Runner | None = None,
    is_interactive: Callable[[], bool] | None = None,
) -> int:
    """Render plan/help locally or execute an explicitly authorized command.

    ``runner`` keeps execution injectable for tests. ``is_interactive`` remains
    accepted for compatibility, but is never called or used as consent.
    Invalid arguments return 2; execution returns the runner's exit code.
    """

    args = list(argv) if argv is not None else sys.argv[1:]
    try:
        options = _parse_args(args)
    except ValueError as error:
        print(f"Error: {error}", file=sys.stderr)
        return 2

    if options.help:
        _print_help()
        return 0
    if options.mode == "plan":
        _print_plan(options)
        return 0

    cmd = build_command(args)
    execute = runner if runner is not None else subprocess.run
    result = execute(cmd, check=False, cwd=str(REPO_ROOT))
    return result.returncode


def main() -> None:
    raise SystemExit(run())


if __name__ == "__main__":
    main()