src/lumi_eggcracker/cli.py

"""Public CLI for protected workloads and autonomous AI-runtime enforcement."""

from __future__ import annotations

import argparse
import json
import math
import os
import re
import sys
import time
from pathlib import Path

from . import __version__
from .brokered.compare_detection_receipts import main as compare_detection_receipts_main
from .client import RequestDeadlineExceeded, request, request_before
from .gate import main as gate_main
from .jsonio import JsonInputError, write_new_json
from .kill_receipt import main as validate_kill_receipt_main
from .kill_receipt import prepare_handoff_main
from .supervisor import main as supervisor_main
from .support_bundle import main as support_bundle_main
from .support_bundle import validate_main as validate_support_bundle_main
from .watchdog import main as watchdog_main

_DOCTOR_SUMMARY_SCHEMA = "lumi-eggcracker.doctor-summary.v1"
_DOCTOR_SUMMARY_ERROR = "eggcracker: doctor summary unavailable"
_CHECK_START_SCHEMA = "lumi-eggcracker.check-start.v1"
_CHECK_START_ERROR = "eggcracker: check-start response unavailable"
_CHECK_START_REASONS = frozenset(
    {
        "APPROVAL_MISSING",
        "EXACT_APPROVAL_MATCHED",
        "EXECUTION_POLICY_INVALID",
        "INVALID_ARGUMENTS",
        "LOCKDOWN",
        "NAME_UNAVAILABLE",
        "STATE_UNAVAILABLE",
    }
)
_CHECK_START_NEXT_ACTIONS = {
    "EXACT_APPROVAL_MATCHED": (
        "run start --require-approval immediately; admission is revalidated"
    ),
    "APPROVAL_MISSING": (
        "create or restore the exact root approval for this tuple, then check again"
    ),
    "EXECUTION_POLICY_INVALID": (
        "use a valid root-created execution policy or omit it, then check again"
    ),
    "INVALID_ARGUMENTS": "correct the launch tuple, then check again",
    "LOCKDOWN": "clear the exact local lockdown through root recovery, then check again",
    "NAME_UNAVAILABLE": "choose an unused workload name, then check again",
    "STATE_UNAVAILABLE": "repair unavailable supervisor state, then check again",
}
_CHECK_START_RESULTS = frozenset({"READY_TO_ATTEMPT", "NOT_READY"})
_DOCTOR_SUMMARY_INSTALLATION_STATES = frozenset(
    {"HEALTHY", "DRIFT", "RECOVERY_REQUIRED", "NOT_INSTALLED"}
)
_DOCTOR_SUMMARY_NETWORK_MODES = frozenset({"offline", "unsupported"})
_DOCTOR_RECOVERY_PLAN_SCHEMA = "lumi-eggcracker.doctor-recovery-plan.v1"
_DOCTOR_RECOVERY_PLAN_ERROR = "eggcracker: doctor recovery plan unavailable"
_DOCTOR_RECOVERY_PLAN_ACTIONS = {
    "cgroup_v2": "MANUAL_HOST_REVIEW",
    "discovery": "MANUAL_SUPERVISOR_REVIEW",
    "execution_boundary": "MANUAL_POLICY_REVIEW",
    "incidents": "MANUAL_INCIDENT_REVIEW",
    "network_cleanup": "MANUAL_SUPERVISOR_REVIEW",
    "network_mode": "MANUAL_POLICY_REVIEW",
    "network_primitives": "MANUAL_HOST_REVIEW",
    "pidfd": "MANUAL_HOST_REVIEW",
    "receipt_persistence": "MANUAL_SUPERVISOR_REVIEW",
    "supervisor_readiness": "MANUAL_SUPERVISOR_REVIEW",
    "workload_identity": "MANUAL_IDENTITY_REVIEW",
}
_DOCTOR_RECOVERY_PLAN_INSTALLATION_ACTIONS = {
    "NOT_INSTALLED": "MANUAL_INSTALLATION_REQUIRED",
    "DRIFT": "MANUAL_INSTALLATION_REPAIR",
    "RECOVERY_REQUIRED": "MANUAL_INSTALLATION_RECOVERY",
}
_WAIT_PUBLIC_STATUS_KEYS = frozenset(
    {
        "exec_policy_digest",
        "exec_policy_id",
        "incident_lockdown",
        "name",
        "network_mode",
        "run_id",
        "state",
        "unit",
        "workload_uid",
    }
)
_WAIT_ACTIVE_STATES = frozenset({"STARTING", "RUNNING"})
_WAIT_TERMINAL_STATES = frozenset(
    {
        "COMPLETED_ALLOWED",
        "CONTAINED_RECEIPT_FAILED",
        "CONTAINMENT_FAILED",
        "TERMINATED",
    }
)
_WAIT_NAME = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,63}\Z")
_WAIT_RUN_ID = re.compile(r"[0-9a-f]{24}\Z")
_WAIT_UNIT_PREFIX = "lumi-eggcracker-workload-"
_WAIT_HEX = re.compile(r"[0-9a-f]{64}\Z")
_WAIT_POLL_CADENCE_SECONDS = 0.1
_WAIT_ERROR = "ERROR"
_WAIT_TERMINAL = "TERMINAL"
_WAIT_TIMED_OUT = "TIMED_OUT"
_WAIT_EXPECTATION_MISMATCH = 7


class _DoctorSummaryInvalid(ValueError):
    """The supervisor response cannot be safely reduced to a doctor summary."""


def _doctor_summary(value: object) -> dict[str, object]:
    """Reduce a doctor response to a fixed, nonidentifying health object."""

    if type(value) is not dict:
        raise _DoctorSummaryInvalid

    result = value.get("result")
    if type(result) is not str or result not in {"PASS", "UNSUPPORTED"}:
        raise _DoctorSummaryInvalid

    def required_bool(container: object, key: str) -> bool:
        if type(container) is not dict or key not in container:
            raise _DoctorSummaryInvalid
        item = container[key]
        if type(item) is not bool:
            raise _DoctorSummaryInvalid
        return item

    autonomous_discovery = required_bool(value, "autonomous_discovery")
    cgroup_v2 = required_bool(value, "cgroup_v2")
    pidfd = required_bool(value, "pidfd")

    network = value.get("network")
    if type(network) is not dict:
        raise _DoctorSummaryInvalid
    network_mode = network.get("mode")
    if type(network_mode) is not str or network_mode not in _DOCTOR_SUMMARY_NETWORK_MODES:
        raise _DoctorSummaryInvalid
    network_primitives = network.get("primitives")
    primitives_supported = required_bool(network_primitives, "supported")
    network_cleanup = required_bool(network, "cleanup_healthy")

    execution_boundary = value.get("execution_boundary")
    execution_supported = required_bool(execution_boundary, "supported")

    discovery = value.get("discovery")
    discovery_healthy = required_bool(discovery, "healthy")
    receipt_persistence = required_bool(discovery, "receipt_persistence_healthy")

    incidents = value.get("incidents")
    incidents_healthy = required_bool(incidents, "healthy")
    incidents_lockdown = required_bool(incidents, "lockdown")
    if not incidents_healthy and not incidents_lockdown:
        raise _DoctorSummaryInvalid

    installation = value.get("installation")
    if type(installation) is not dict or "state" not in installation:
        raise _DoctorSummaryInvalid
    installation_state = installation["state"]
    if (
        type(installation_state) is not str
        or installation_state not in _DOCTOR_SUMMARY_INSTALLATION_STATES
    ):
        raise _DoctorSummaryInvalid

    workload_identity = value.get("workload_identity")
    workload_identity_healthy = required_bool(workload_identity, "healthy")

    failed_checks: set[str] = set()
    if not cgroup_v2:
        failed_checks.add("cgroup_v2")
    if not pidfd:
        failed_checks.add("pidfd")
    if network_mode != "offline":
        failed_checks.add("network_mode")
    if not primitives_supported:
        failed_checks.add("network_primitives")
    if not network_cleanup:
        failed_checks.add("network_cleanup")
    if not execution_supported:
        failed_checks.add("execution_boundary")
    if not discovery_healthy:
        failed_checks.add("discovery")
    if not receipt_persistence:
        failed_checks.add("receipt_persistence")
    if not incidents_healthy:
        failed_checks.add("incidents")
    if installation_state != "HEALTHY":
        failed_checks.add("installation")
    if not workload_identity_healthy:
        failed_checks.add("workload_identity")

    if result == "PASS":
        if not autonomous_discovery or failed_checks:
            raise _DoctorSummaryInvalid
        summary_result = "READY"
    else:
        if autonomous_discovery:
            raise _DoctorSummaryInvalid
        if not failed_checks:
            # The supervisor also gates readiness on quarantine_root, which is
            # deliberately not exposed by its public doctor response.
            failed_checks.add("supervisor_readiness")
        summary_result = "NOT_READY"

    return {
        "failed_checks": sorted(failed_checks),
        "next_action": (
            "NONE" if summary_result == "READY" else "REVIEW_SUPERVISOR_READINESS"
        ),
        "result": summary_result,
        "schema": _DOCTOR_SUMMARY_SCHEMA,
    }


def _doctor_recovery_plan(value: object) -> dict[str, object]:
    """Reduce a doctor response to a fixed, nonidentifying recovery plan."""

    summary = _doctor_summary(value)
    installation = value["installation"]
    installation_state = installation["state"]
    failed_checks = summary["failed_checks"]

    actions = {
        _DOCTOR_RECOVERY_PLAN_ACTIONS[check]
        for check in failed_checks
        if check != "installation"
    }
    if installation_state != "HEALTHY":
        actions.add(_DOCTOR_RECOVERY_PLAN_INSTALLATION_ACTIONS[installation_state])

    return {
        "actions": sorted(actions),
        "failed_checks": failed_checks,
        "installation_state": installation_state,
        "result": summary["result"],
        "schema": _DOCTOR_RECOVERY_PLAN_SCHEMA,
    }


def _check_start_projection(value: object) -> dict[str, str]:
    """Accept only the supervisor's fixed, nonidentifying check result."""
    if type(value) is not dict or set(value) != {
        "next_action",
        "reason_code",
        "result",
        "schema",
    }:
        raise ValueError
    schema = value["schema"]
    reason_code = value["reason_code"]
    result = value["result"]
    next_action = value["next_action"]
    if (
        type(schema) is not str
        or schema != _CHECK_START_SCHEMA
        or type(reason_code) is not str
        or reason_code not in _CHECK_START_REASONS
        or type(result) is not str
        or result not in _CHECK_START_RESULTS
        or type(next_action) is not str
        or next_action != _CHECK_START_NEXT_ACTIONS[reason_code]
        or (result == "READY_TO_ATTEMPT") != (reason_code == "EXACT_APPROVAL_MATCHED")
    ):
        raise ValueError
    return {
        "next_action": next_action,
        "reason_code": reason_code,
        "result": result,
        "schema": schema,
    }


def _wait_status_projection(value: object, *, name: str) -> dict[str, object]:
    """Validate and close the supervisor's public run projection."""

    if type(value) is not dict or set(value) != _WAIT_PUBLIC_STATUS_KEYS:
        raise ValueError
    if (
        type(value["name"]) is not str
        or value["name"] != name
        or _WAIT_NAME.fullmatch(value["name"]) is None
    ):
        raise ValueError
    run_id = value["run_id"]
    if type(run_id) is not str or _WAIT_RUN_ID.fullmatch(run_id) is None:
        raise ValueError
    state = value["state"]
    if type(state) is not str or state not in _WAIT_ACTIVE_STATES | _WAIT_TERMINAL_STATES:
        raise ValueError
    unit = value["unit"]
    if type(unit) is not str or unit != f"{_WAIT_UNIT_PREFIX}{run_id}.service":
        raise ValueError
    network_mode = value["network_mode"]
    if type(network_mode) is not str or network_mode not in {"offline", "none"}:
        raise ValueError
    exec_policy_id = value["exec_policy_id"]
    if type(exec_policy_id) is not str or _WAIT_NAME.fullmatch(exec_policy_id) is None:
        raise ValueError
    exec_policy_digest = value["exec_policy_digest"]
    if type(exec_policy_digest) is not str or _WAIT_HEX.fullmatch(exec_policy_digest) is None:
        raise ValueError
    if type(value["incident_lockdown"]) is not bool:
        raise ValueError
    workload_uid = value["workload_uid"]
    if type(workload_uid) is not int or workload_uid < 0:
        raise ValueError
    return {
        "exec_policy_digest": exec_policy_digest,
        "exec_policy_id": exec_policy_id,
        "incident_lockdown": value["incident_lockdown"],
        "name": value["name"],
        "network_mode": network_mode,
        "run_id": run_id,
        "state": state,
        "unit": unit,
        "workload_uid": workload_uid,
    }


class _WaitClock:
    """Read one monotonic clock and reject non-monotonic or invalid samples."""

    def __init__(self) -> None:
        self._last: float | None = None

    def now(self) -> float:
        value = time.monotonic()
        if (
            isinstance(value, bool)
            or not isinstance(value, (int, float))
            or not math.isfinite(value)
            or self._last is not None
            and value < self._last
        ):
            raise ValueError
        self._last = float(value)
        return self._last


def _wait_output(marker: str, state: dict[str, object] | None = None) -> None:
    print(marker)
    if state is not None:
        print(json.dumps(state, sort_keys=True))


def _wait_error() -> int:
    print(_WAIT_ERROR, file=sys.stderr)
    return 4


def _wait_main(
    *, name: str, timeout_seconds: int, expect_state: str | None = None
) -> int:
    clock = _WaitClock()
    try:
        started = clock.now()
        deadline = started + timeout_seconds
        if not math.isfinite(deadline):
            raise ValueError
        last_state: dict[str, object] | None = None
        pinned_run_id: str | None = None
        while True:
            remaining = deadline - clock.now()
            if remaining <= 0:
                _wait_output(_WAIT_TIMED_OUT, last_state)
                return 6
            try:
                value = request_before(
                    "status", deadline_monotonic=deadline, name=name
                )
            except (RequestDeadlineExceeded, TimeoutError):
                _wait_output(_WAIT_TIMED_OUT, last_state)
                return 6
            except Exception:  # noqa: BLE001 - wait failures are constant and closed
                return _wait_error()
            # A response completing at or after the absolute deadline cannot be
            # treated as a validated observation for this bounded wait.
            if clock.now() >= deadline:
                _wait_output(_WAIT_TIMED_OUT, last_state)
                return 6
            try:
                state = _wait_status_projection(value, name=name)
            except Exception:  # noqa: BLE001 - malformed status is constant
                return _wait_error()
            run_id = state["run_id"]
            if pinned_run_id is None:
                pinned_run_id = run_id
            elif run_id != pinned_run_id:
                return _wait_error()
            last_state = state
            if state["state"] in _WAIT_TERMINAL_STATES:
                _wait_output(_WAIT_TERMINAL, state)
                if expect_state is None or state["state"] == expect_state:
                    return 0
                return _WAIT_EXPECTATION_MISMATCH
            remaining = deadline - clock.now()
            if remaining <= 0:
                _wait_output(_WAIT_TIMED_OUT, last_state)
                return 6
            time.sleep(min(_WAIT_POLL_CADENCE_SECONDS, remaining))
    except Exception:  # noqa: BLE001 - clock and sleep failures are constant
        return _wait_error()


class _RejectDuplicateIdentifier(argparse.Action):
    """Accept one identifier option and fail closed on a repeated spelling."""

    def __call__(
        self,
        parser: argparse.ArgumentParser,
        namespace: argparse.Namespace,
        values: object,
        option_string: str | None = None,
    ) -> None:
        marker = f"_eggcracker_seen_{self.dest}"
        if getattr(namespace, marker, False):
            raise argparse.ArgumentError(
                self,
                f"{option_string or self.dest} may only be specified once",
            )
        setattr(namespace, marker, True)
        setattr(namespace, self.dest, values)


class _CheckStartParseFailure(Exception):
    """Internal marker for one closed check-start parse failure."""


class _CheckStartHelp(Exception):
    """Internal marker for successfully rendered check-start help."""


class _CheckStartParser(argparse.ArgumentParser):
    """Keep check-start parser failures fixed and nonidentifying."""

    def error(self, message: str) -> None:
        raise _CheckStartParseFailure from None

    def exit(self, status: int = 0, message: str | None = None) -> None:
        if status == 0:
            raise _CheckStartHelp
        raise _CheckStartParseFailure from None


def _add_check_start_arguments(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--name", required=True, action=_RejectDuplicateIdentifier)
    parser.add_argument("--max-pids", required=True, type=int)
    parser.add_argument("--max-memory-mib", default=2048, type=int)
    parser.add_argument("--cpu-quota-percent", default=400, type=int)
    parser.add_argument(
        "--exec-policy",
        action=_RejectDuplicateIdentifier,
        help="root-created sealed-exec policy identifier",
    )
    parser.add_argument("argv", nargs=argparse.REMAINDER)


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="eggcracker")
    commands = parser.add_subparsers(dest="command", required=True)
    start = commands.add_parser("start", help="launch an explicitly selected protected workload")
    start.add_argument("--name", required=True, action=_RejectDuplicateIdentifier)
    start.add_argument("--max-pids", required=True, type=int)
    start.add_argument("--max-memory-mib", default=2048, type=int)
    start.add_argument("--cpu-quota-percent", default=400, type=int)
    start.add_argument(
        "--exec-policy",
        action=_RejectDuplicateIdentifier,
        help="root-created sealed-exec policy identifier",
    )
    start.add_argument(
        "--require-approval",
        action="store_true",
        help="reject the launch unless an exact approval exists at admission time",
    )
    start.add_argument("argv", nargs=argparse.REMAINDER)
    check_start = commands.add_parser(
        "check-start",
        help="check one exact approved launch without reserving or starting it",
    )
    _add_check_start_arguments(check_start)
    kill = commands.add_parser("kill", help="terminate one protected workload")
    kill.add_argument("--name", required=True, action=_RejectDuplicateIdentifier)
    kill.add_argument("--receipt", required=True, type=Path)
    status = commands.add_parser("status", help="show one protected workload")
    status.add_argument("--name", required=True, action=_RejectDuplicateIdentifier)
    wait = commands.add_parser(
        "wait",
        help="wait for one protected workload to reach a terminal state",
        description=(
            "Wait for one protected workload to reach a terminal state. With "
            "--expect-state, a mismatch exits 7. COMPLETED_ALLOWED means an "
            "observed terminal/cgroup-empty completion; it does not mean "
            "application exit-code success or provide a safety guarantee."
        ),
    )
    wait.add_argument("--name", required=True, action=_RejectDuplicateIdentifier)
    wait.add_argument(
        "--timeout-seconds",
        required=True,
        type=_wait_timeout,
        action=_RejectDuplicateIdentifier,
        help="maximum wait duration in seconds (1-300)",
    )
    wait.add_argument(
        "--expect-state",
        choices=sorted(_WAIT_TERMINAL_STATES),
        action=_RejectDuplicateIdentifier,
        help=(
            "expected terminal state; mismatch exits 7 (COMPLETED_ALLOWED is "
            "observed cgroup-empty completion, not application success or a "
            "safety guarantee)"
        ),
    )
    commands.add_parser("list", help="list protected workloads")
    history = commands.add_parser(
        "history",
        help="list retained terminal records in recorded monotonic order",
    )
    history.add_argument(
        "--limit",
        default=20,
        type=_history_limit,
        action=_RejectDuplicateIdentifier,
        help="number of retained terminal records to return (1-20; default: 20)",
    )
    history.add_argument(
        "--state",
        choices=sorted(_WAIT_TERMINAL_STATES),
        action=_RejectDuplicateIdentifier,
        help="show only records in the selected terminal state",
    )
    approve = commands.add_parser("approve", help="approve one exact AI runtime invocation")
    approve.add_argument("--name", required=True, action=_RejectDuplicateIdentifier)
    approve.add_argument("--uid", required=True, type=int)
    approve.add_argument("--max-pids", default=64, type=int)
    approve.add_argument("--max-memory-mib", default=2048, type=int)
    approve.add_argument("--cpu-quota-percent", default=400, type=int)
    approve.add_argument(
        "--allow-interface-discovery",
        action="store_true",
        help="allow unprivileged interface lookup inside the offline namespace",
    )
    approve.add_argument("argv", nargs=argparse.REMAINDER)
    revoke = commands.add_parser("revoke", help="revoke one exact AI runtime approval")
    revoke.add_argument("--name", required=True, action=_RejectDuplicateIdentifier)
    commands.add_parser("approvals", help="list exact AI runtime approvals")
    exec_policy = commands.add_parser("exec-policy", help="manage root-created executable policies")
    exec_commands = exec_policy.add_subparsers(dest="exec_action", required=True)
    create_policy = exec_commands.add_parser("create", help="create an immutable executable policy")
    create_policy.add_argument(
        "--name", required=True, action=_RejectDuplicateIdentifier
    )
    create_policy.add_argument("paths", nargs=argparse.REMAINDER)
    revoke_policy = exec_commands.add_parser("revoke", help="revoke an executable policy")
    revoke_policy.add_argument(
        "--policy-id", required=True, action=_RejectDuplicateIdentifier
    )
    exec_commands.add_parser("list", help="list executable policies")
    commands.add_parser("detections", help="list autonomous containment summaries")
    commands.add_parser("incidents", help="list bounded local lockdown incidents")
    incident = commands.add_parser("incident", help="inspect or clear one local lockdown incident")
    incident_commands = incident.add_subparsers(dest="incident_action", required=True)
    for action, help_text in (
        ("show", "show one root-only incident detail"),
        ("acknowledge", "acknowledge one incident as root"),
        ("clear", "clear one local lockdown as root"),
    ):
        command = incident_commands.add_parser(action, help=help_text)
        command.add_argument("incident_id")
    doctor = commands.add_parser("doctor", help="check the installed protected supervisor")
    doctor_projection = doctor.add_mutually_exclusive_group()
    doctor_projection.add_argument(
        "--summary",
        action="store_true",
        help="emit a bounded readiness summary instead of the full doctor response",
    )
    doctor_projection.add_argument(
        "--recovery-plan",
        action="store_true",
        help="emit a bounded manual recovery plan instead of the full doctor response",
    )
    support = commands.add_parser(
        "support-bundle", help="write a local redacted health and receipt bundle"
    )
    support.add_argument("--output", required=True, type=Path)
    validate_support = commands.add_parser(
        "validate-support-bundle",
        help="offline structural review of a support bundle",
    )
    validate_support.add_argument("FILE", type=Path)
    validate_kill_receipt = commands.add_parser(
        "validate-kill-receipt",
        help="offline structural review of one manual kill receipt",
    )
    validate_kill_receipt.add_argument("FILE", type=Path)
    handoff = commands.add_parser(
        "prepare-kill-receipt-handoff",
        help="prepare a local redacted manual kill receipt handoff",
    )
    handoff.add_argument("RECEIPT", type=Path)
    handoff.add_argument(
        "--route",
        required=True,
        choices=("q-and-a", "bug-report", "private-security"),
    )
    handoff.add_argument("--output", required=True, type=Path)
    commands.add_parser(
        "compare-detection-receipts",
        help="compare two redacted detection receipt exports",
    )
    commands.add_parser("version", help="print the public version")
    return parser


def _command(argv: list[str]) -> list[str]:
    return argv[1:] if argv[:1] == ["--"] else argv


def _history_limit(value: str) -> int:
    if not isinstance(value, str) or not value.isascii() or not value.isdecimal():
        raise argparse.ArgumentTypeError("history limit must be an integer from 1 to 20")
    result = int(value)
    if not 1 <= result <= 20:
        raise argparse.ArgumentTypeError("history limit must be an integer from 1 to 20")
    return result


def _wait_timeout(value: str) -> int:
    if not isinstance(value, str) or not value.isascii() or not value.isdecimal():
        raise argparse.ArgumentTypeError("wait timeout must be an integer from 1 to 300")
    result = int(value)
    if not 1 <= result <= 300:
        raise argparse.ArgumentTypeError("wait timeout must be an integer from 1 to 300")
    return result


def _check_start_main(values: list[str]) -> int:
    parser = _CheckStartParser(prog="eggcracker check-start")
    _add_check_start_arguments(parser)
    try:
        args = parser.parse_args(values)
        check_args = {
            "name": args.name,
            "max_pids": args.max_pids,
            "max_memory_mib": args.max_memory_mib,
            "cpu_quota_percent": args.cpu_quota_percent,
            "argv": _command(args.argv),
        }
        if args.exec_policy is not None:
            check_args["exec_policy"] = args.exec_policy
        value = _check_start_projection(request("check_start", **check_args))
    except _CheckStartHelp:
        return 0
    except Exception:  # noqa: BLE001 - parse/transport/shape failures are constant
        print(_CHECK_START_ERROR, file=sys.stderr)
        return 4
    print(json.dumps(value, sort_keys=True))
    return 0 if value["result"] == "READY_TO_ATTEMPT" else 6


def main(argv: list[str] | None = None) -> int:
    values = sys.argv[1:] if argv is None else argv
    if values[:1] == ["check-start"]:
        return _check_start_main(values[1:])
    if values[:1] == ["compare-detection-receipts"]:
        return compare_detection_receipts_main(
            values[1:], prog="eggcracker compare-detection-receipts"
        )
    if values[:1] == ["validate-support-bundle"]:
        if values[1:] == ["--help"]:
            _parser().parse_args(values)
            return 0
        return validate_support_bundle_main(values[1:])
    if values[:1] == ["validate-kill-receipt"]:
        if values[1:] == ["--help"]:
            _parser().parse_args(values)
            return 0
        return validate_kill_receipt_main(values[1:])
    if values[:1] == ["prepare-kill-receipt-handoff"]:
        if values[1:] == ["--help"]:
            _parser().parse_args(values)
            return 0
        return prepare_handoff_main(values[1:])
    if values[:1] == ["_supervisor"]:
        return supervisor_main(values[1:])
    if values[:1] == ["_gate"]:
        return gate_main(values[1:])
    if values[:1] == ["_watchdog"]:
        return watchdog_main(values[1:])
    args = _parser().parse_args(values)
    if args.command == "version":
        print(__version__)
        return 0
    if args.command == "support-bundle":
        return support_bundle_main(args.output)
    if args.command == "wait":
        return _wait_main(
            name=args.name,
            timeout_seconds=args.timeout_seconds,
            expect_state=args.expect_state,
        )
    try:
        if args.command == "doctor":
            if args.summary:
                try:
                    summary = _doctor_summary(request("doctor"))
                except Exception:  # noqa: BLE001 - summary failures are intentionally constant
                    print(_DOCTOR_SUMMARY_ERROR, file=sys.stderr)
                    return 6
                print(json.dumps(summary, sort_keys=True))
                return 0 if summary["result"] == "READY" else 6
            if args.recovery_plan:
                try:
                    recovery_plan = _doctor_recovery_plan(request("doctor"))
                except Exception:  # noqa: BLE001 - plan failures are intentionally constant
                    print(_DOCTOR_RECOVERY_PLAN_ERROR, file=sys.stderr)
                    return 6
                print(json.dumps(recovery_plan, sort_keys=True))
                return 0 if recovery_plan["result"] == "READY" else 6
            value = request("doctor")
            print(json.dumps(value, sort_keys=True))
            return 0 if value.get("result") == "PASS" else 6
        if args.command == "start":
            start_args = {
                "action": "start",
                "name": args.name,
                "max_pids": args.max_pids,
                "max_memory_mib": args.max_memory_mib,
                "cpu_quota_percent": args.cpu_quota_percent,
                "argv": _command(args.argv),
            }
            if args.exec_policy is not None:
                start_args["exec_policy"] = args.exec_policy
            if args.require_approval:
                start_args["require_approval"] = True
            action = start_args.pop("action")
            value = request(action, **start_args)
        elif args.command == "status":
            value = request("status", name=args.name)
        elif args.command == "list":
            value = request("list")
        elif args.command == "history":
            history_args: dict[str, object] = {"limit": args.limit}
            if args.state is not None:
                history_args["state"] = args.state
            value = request("history", **history_args)
        elif args.command == "approve":
            if os.geteuid() != 0:
                raise JsonInputError("approve requires root administrative authority; use sudo")
            value = request(
                "approve",
                name=args.name,
                uid=args.uid,
                max_pids=args.max_pids,
                max_memory_mib=args.max_memory_mib,
                cpu_quota_percent=args.cpu_quota_percent,
                allow_interface_discovery=args.allow_interface_discovery,
                argv=_command(args.argv),
            )
        elif args.command == "revoke":
            if os.geteuid() != 0:
                raise JsonInputError("revoke requires root administrative authority; use sudo")
            value = request("revoke", name=args.name)
        elif args.command == "approvals":
            value = request("approvals")
        elif args.command == "exec-policy":
            if args.exec_action == "list":
                value = request("exec_policies")
            elif args.exec_action == "create":
                if os.geteuid() != 0:
                    raise JsonInputError("execution policy creation requires root administrative authority; use sudo")
                value = request("exec_policy_create", name=args.name, paths=_command(args.paths))
            else:
                if os.geteuid() != 0:
                    raise JsonInputError("execution policy revocation requires root administrative authority; use sudo")
                value = request("exec_policy_revoke", policy_id=args.policy_id)
        elif args.command == "detections":
            value = request("detections")
        elif args.command == "incidents":
            value = request("incidents")
        elif args.command == "incident":
            if os.geteuid() != 0:
                raise JsonInputError("incident administration requires root administrative authority; use sudo")
            action = {
                "show": "incident_show",
                "acknowledge": "incident_acknowledge",
                "clear": "incident_clear",
            }[args.incident_action]
            value = request(action, incident_id=args.incident_id)
        else:
            # This validation is deliberately read-only: no receipt reservation before kill.
            if args.receipt.exists() or args.receipt.is_symlink() or not args.receipt.parent.is_dir():
                raise JsonInputError("receipt must be a new file under an existing directory")
            value = request("kill", name=args.name)
            write_new_json(args.receipt, value)
        print(json.dumps(value, sort_keys=True))
        return 0
    except (JsonInputError, OSError) as error:
        print(f"eggcracker: {error}", file=sys.stderr)
        return 4