pyguard.py

"""Command-line entry point for Python Guardian supervision."""
from __future__ import annotations

import argparse
import json
import shutil
import sys
import time
from pathlib import Path

from guardian_client import job_status, preflight, submit


TERMINAL_STATES = {"success", "failed_analyzed", "broker_error", "interrupted"}


def interpreter_candidates(script: Path) -> list[Path]:
    found: list[Path] = []
    for parent in (script.parent, *list(script.parents)[:7]):
        for environment in (".venv", "venv", f"{parent.name}_venv"):
            candidate = parent / environment / "Scripts" / "python.exe"
            if candidate.is_file():
                found.append(candidate)
    base = Path(getattr(sys, "_base_executable", sys.executable))
    if base.name.casefold() == "pythonw.exe":
        base = base.with_name("python.exe")
    if base.is_file():
        found.append(base)
    on_path = shutil.which("python")
    if on_path:
        found.append(Path(on_path))
    result: list[Path] = []
    seen: set[str] = set()
    for path in found:
        resolved = path.resolve()
        key = str(resolved).casefold()
        if key not in seen:
            seen.add(key)
            result.append(resolved)
    return result


def main() -> int:
    parser = argparse.ArgumentParser(
        prog="PyGuard",
        description="Run a Python file under resident Codex/PANIC supervision.",
    )
    parser.add_argument("--version", action="version", version="PyGuard 0.1.0")
    parser.add_argument("-p", "--python", dest="interpreter",
                        help="target Python interpreter (nearby virtual environments are preferred)")
    parser.add_argument("--keep", choices=("always", "error", "never"), default="error",
                        help="when the target console remains open (default: error)")
    parser.add_argument("-y", "--yes", action="store_true",
                        help="non-interactive approval after preflight; intended for automation")
    parser.add_argument("--no-preflight", action="store_true",
                        help="submit immediately without a Codex preflight")
    parser.add_argument("--detach", action="store_true",
                        help="print the job ID and return without waiting")
    parser.add_argument("--timeout", type=float, default=0,
                        help="maximum seconds to wait; zero waits indefinitely")
    parser.add_argument("script", type=Path)
    parser.add_argument("arguments", nargs=argparse.REMAINDER,
                        help="arguments passed unchanged to the target script")
    options = parser.parse_args()

    script = options.script.resolve()
    if not script.is_file() or script.suffix.casefold() not in (".py", ".pyw"):
        parser.error(f"not a Python file: {script}")
    if options.interpreter:
        interpreter = Path(options.interpreter.strip('"')).resolve()
    else:
        candidates = interpreter_candidates(script)
        if not candidates:
            parser.error("no Python interpreter was found")
        interpreter = candidates[0]
    if not interpreter.is_file():
        parser.error(f"Python interpreter not found: {interpreter}")

    if not options.no_preflight:
        print(f"Python Guardian preflight: {script}\n")
        try:
            review = preflight(str(script))
            if not review.get("ok"):
                raise RuntimeError(review.get("error", "preflight failed"))
            print(review.get("text", "(No preflight text returned.)"))
        except Exception as exc:  # noqa: BLE001 - CLI makes the fallback explicit
            print(f"\n[Guardian] Preflight unavailable: {type(exc).__name__}: {exc}", file=sys.stderr)
        if not options.yes:
            try:
                answer = input("\nRun under Guardian supervision? [Y/n] ").strip().casefold()
            except (EOFError, KeyboardInterrupt):
                print("\nCancelled.")
                return 130
            if answer not in ("", "y", "yes"):
                print("Cancelled.")
                return 0

    try:
        queued = submit(str(script), str(interpreter), list(options.arguments), options.keep)
    except Exception as exc:  # noqa: BLE001 - concise command-line failure
        print(f"Guardian could not accept the job: {type(exc).__name__}: {exc}", file=sys.stderr)
        return 1
    if not queued.get("ok"):
        print(json.dumps(queued, indent=2), file=sys.stderr)
        return 1
    job_id = queued["job_id"]
    print(f"\nGuardian job queued: {job_id}")
    print(f"Interpreter: {interpreter}")
    if options.detach:
        return 0

    deadline = time.monotonic() + options.timeout if options.timeout > 0 else None
    last_state = ""
    try:
        while True:
            try:
                state = job_status(job_id)
            except Exception as exc:  # noqa: BLE001
                print(f"Guardian connection failed while waiting: {type(exc).__name__}: {exc}",
                      file=sys.stderr)
                return 1
            status = str(state.get("status", "unknown"))
            if status != last_state:
                print(f"Guardian status: {status}")
                last_state = status
            if status in TERMINAL_STATES:
                if state.get("report_path"):
                    print(f"Report: {state['report_path']}")
                if state.get("error"):
                    print(f"Guardian error: {state['error']}", file=sys.stderr)
                if status == "success":
                    return 0
                target_code = state.get("exit_code")
                return target_code if isinstance(target_code, int) and 0 < target_code < 256 else 1
            if deadline is not None and time.monotonic() >= deadline:
                print(f"Timed out waiting; job {job_id} continues in Guardian.", file=sys.stderr)
                return 124
            time.sleep(0.5)
    except KeyboardInterrupt:
        print(f"\nStopped waiting; job {job_id} continues in Guardian.")
        return 130


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