automaton/scripts/autopilot.py

#!/usr/bin/env python3
"""Autopilot runtime for Automaton framework.

Replaces the pseudocode drive_all() loop in prompt/orchestrate.md
with a functioning CLI tool that agents can invoke to determine
the next action in an autopilot workflow.

Usage:
    python autopilot.py --project /path/to/project       Drive one step forward
    python autopilot.py --project . --execute             Drive and execute transitions
    python autopilot.py --project . --loop               Run continuous drive loop
    python autopilot.py --project . --detect-stuck        Detect stuck tasks
    python autopilot.py --project . --summary              Show autopilot summary
    python autopilot.py --install-schedule [--interval N] Install OS scheduler unit
"""

from __future__ import annotations

import argparse
import platform
import subprocess
import sys
import time
from pathlib import Path
from typing import Optional

AUTOMATON_DIR = Path.home() / ".automaton"

PHASE_PRIORITY = {
    "referee": 12,
    "doc_review": 11,
    "adversarial_bug_find": 10,
    "bug_find": 9,
    "code_review": 8,
    "implement": 7,
    "test_design": 6,
    "design": 5,
    "decomposition": 4,
    "research": 3,
    "new": 2,
    "human_intervention": 1,
}


def _find_project_dir(project_arg: Optional[str]) -> Path:
    if project_arg:
        p = Path(project_arg).expanduser().resolve()
        if p.is_dir():
            return p
    cwd = Path.cwd()
    if (cwd / ".automaton").is_dir():
        return cwd
    return AUTOMATON_DIR


def _run_transition(task_name: str, target: str, project_dir: Path) -> int:
    """Execute a status.py --transition and print output."""
    status_py = str(AUTOMATON_DIR / "scripts" / "status.py")
    cmd = [
        sys.executable, status_py,
        "--transition", target,
        "--task", task_name,
        "--project", str(project_dir),
    ]
    res = subprocess.run(cmd, capture_output=True, text=True)
    if res.stdout:
        print(res.stdout, end="")
    if res.stderr:
        print(res.stderr, end="", file=sys.stderr)
    return res.returncode


def _base_phase(phase: str) -> str:
    return phase.split(":")[0]


def _read_state(task_path: Path) -> Optional[str]:
    state_file = task_path / ".state"
    if not state_file.exists():
        return None
    content = state_file.read_text().strip()
    return content if content else None


def _all_tasks(project_dir: Path) -> list[tuple[str, Path]]:
    """Scan all tasks (including subtasks) for a project.

    Mirrors ``status.py:_all_task_dirs``: skips ``tasks/complete/``,
    recurses into ``subtasks/``, uses ``.automaton/tasks`` for non-framework
    projects.
    """
    if project_dir == AUTOMATON_DIR:
        base = AUTOMATON_DIR / "tasks"
    else:
        base = project_dir / ".automaton" / "tasks"
    if not base.is_dir():
        return []
    result = []
    for entry in sorted(base.iterdir()):
        if not entry.is_dir() or entry.name.startswith(".") or entry.name == "complete":
            continue
        result.append((entry.name, entry))
        subtasks = entry / "subtasks"
        if subtasks.exists():
            for sub in sorted(subtasks.iterdir()):
                if sub.is_dir() and not sub.name.startswith("."):
                    result.append((f"{entry.name}/{sub.name}", sub))
    return result


def scan_all_tasks(project_dir: Path) -> list[dict]:
    """Scan all tasks and return their state info."""
    tasks = []
    for name, path in _all_tasks(project_dir):
        phase = _read_state(path)
        tasks.append(
            {
                "name": name,
                "path": str(path),
                "phase": phase,
                "base_phase": _base_phase(phase) if phase else None,
            }
        )
    return tasks


def is_terminal(task: dict) -> bool:
    """Check if a task is in a terminal state.

    Only ``complete`` is terminal. ``human_intervention`` has legal
    transitions (→ referee, → complete) so the autopilot can still
    suggest next steps for it.
    """
    phase = task.get("phase")
    if phase is None:
        return False
    return phase == "complete"


def needs_user_input(task: dict) -> bool:
    """Check if a task is blocked waiting for user input."""
    phase = task.get("phase")
    if phase is None:
        return False
    if phase.endswith(":awaiting_approval"):
        return True
    if phase == "human_intervention":
        return True
    task_path = Path(task["path"])
    verdict_file = task_path / "VERDICT.md"
    if verdict_file.exists():
        content = verdict_file.read_text()
        first_line = content.split("\n")[0] if content else ""
        if any(
            kw in first_line.upper() for kw in ("FAIL", "NEEDS_REVIEW", "TIE-BREAK")
        ):
            return True
    return False


def sort_by_advancement(tasks: list[dict]) -> list[dict]:
    """Sort tasks by how close they are to completion (most advanced first)."""
    return sorted(
        tasks,
        key=lambda t: PHASE_PRIORITY.get(t.get("base_phase", ""), 0),
        reverse=True,
    )


def detect_stuck_tasks(project_dir: Path, threshold_minutes: int = 60) -> list[dict]:
    """Detect tasks that appear stuck (in same non-terminal phase too long)."""
    stuck = []
    now = time.time()
    for name, path in _all_tasks(project_dir):
        state_file = path / ".state"
        if not state_file.exists():
            continue
        phase = _read_state(path)
        if phase is None or is_terminal({"phase": phase}):
            continue
        mtime = state_file.stat().st_mtime
        age_minutes = (now - mtime) / 60
        if age_minutes > threshold_minutes:
            stuck.append(
                {
                    "name": name,
                    "phase": phase,
                    "age_minutes": round(age_minutes),
                    "path": str(path),
                }
            )
    return sorted(stuck, key=lambda t: t["age_minutes"], reverse=True)


def cmd_summary(args):
    """Show autopilot summary of all tasks."""
    project_dir = _find_project_dir(args.project)
    all_tasks = scan_all_tasks(project_dir)

    if not all_tasks:
        print("No tasks found. Create one with:")
        print(
            "  python ~/.automaton/scripts/status.py --create-task <name> --project",
            project_dir,
        )
        return 0

    terminal = [t for t in all_tasks if is_terminal(t)]
    non_terminal = [t for t in all_tasks if not is_terminal(t)]
    blocked = [t for t in non_terminal if needs_user_input(t)]
    unblocked = [t for t in non_terminal if not needs_user_input(t)]
    stuck = detect_stuck_tasks(project_dir)

    print(f"Project: {project_dir}")
    print(
        f"Tasks: {len(all_tasks)} total | {len(terminal)} done | {len(non_terminal)} in progress"
    )
    print(f"  Blocked (awaiting user): {len(blocked)}")
    print(f"  Unblocked (ready to drive): {len(unblocked)}")
    print(f"  Stuck (>60 min): {len(stuck)}")
    print()

    if stuck:
        print("=== STUCK TASKS ===")
        for t in stuck:
            print(f"  {t['name']}: stuck at '{t['phase']}' for {t['age_minutes']} min")
        print()

    if unblocked:
        print("=== READY TO DRIVE ===")
        for t in sort_by_advancement(unblocked):
            print(
                f"  {t['name']}: {t['phase']} (priority: {PHASE_PRIORITY.get(t.get('base_phase', ''), 0)})"
            )
        print()

    if blocked:
        print("=== AWAITING USER ===")
        for t in blocked:
            approval = t["phase"].endswith(":awaiting_approval")
            print(
                f"  {t['name']}: {t['phase']}{' (needs --approve)' if approval else ' (needs VERDICT review)'}"
            )
        print()

    if not unblocked and not non_terminal:
        print("ALL TASKS TERMINAL — nothing to drive.")
        print("Create a new task to continue:")
        print(
            "  python ~/.automaton/scripts/status.py --create-task <name> --project",
            project_dir,
        )

    return 0


def _suggest_or_execute(args, task_name: str, target: str, project_dir: Path,
                        description: str = "") -> int:
    """Print the suggested transition, or execute it if --execute is set."""
    if description:
        print(f"→ {description}")
    if getattr(args, "execute", False):
        rc = _run_transition(task_name, target, project_dir)
        if rc == 0:
            print(f"✓ {task_name}: transitioned to {target}")
        else:
            print(f"✗ {task_name}: transition to {target} failed")
        return rc
    print(
        f"  python ~/.automaton/scripts/status.py --transition {target} --task {task_name} --project {project_dir}"
    )
    return 0


def cmd_drive(args):
    """Drive one step: find the best task to advance and output instructions."""
    project_dir = _find_project_dir(args.project)
    all_tasks = scan_all_tasks(project_dir)

    if not all_tasks:
        print("NO_TASKS: No tasks found. Create one with:")
        print(
            "  python ~/.automaton/scripts/status.py --create-task <name> --project",
            project_dir,
        )
        return 1

    non_terminal = [t for t in all_tasks if not is_terminal(t)]

    if not non_terminal:
        print("ORCHESTRATION_COMPLETE: all tasks done.")
        print("Nothing to drive. Create a new task:")
        print(
            "  python ~/.automaton/scripts/status.py --create-task <name> --project",
            project_dir,
        )
        return 0

    unblocked = [t for t in non_terminal if not needs_user_input(t)]

    if not unblocked:
        print("ORCHESTRATION_BLOCKED: all non-terminal tasks are awaiting user review.")
        for t in non_terminal:
            print(f"  {t['name']}: {t['phase']}")
        print()
        print("To proceed:")
        for t in non_terminal:
            if t["phase"].endswith(":awaiting_approval"):
                print(
                    f"  python ~/.automaton/scripts/status.py --approve --task {t['name']} --project {project_dir}"
                )
            elif t["phase"] == "human_intervention":
                print(
                    f"  Review {t['name']} — transition to referee or complete:"
                )
                print(
                    f"  python ~/.automaton/scripts/status.py --transition referee --task {t['name']} --project {project_dir}"
                )
            else:
                print(f"  Review {t['name']}/VERDICT.md and take action")
        return 0

    sorted_tasks = sort_by_advancement(unblocked)
    task = sorted_tasks[0]

    print(f"NEXT_TASK: {task['name']}")
    print(f"PHASE: {task['phase']}")
    print(f"PRIORITY: {PHASE_PRIORITY.get(task.get('base_phase', ''), 0)}")
    print()

    phase = task["phase"]
    name = task["name"]
    base = _base_phase(phase) if phase else "unknown"

    if base == "new":
        _suggest_or_execute(
            args, name, "research", project_dir,
            "Transition to research"
        )
    elif base == "research":
        if phase == "research":
            print("→ Generate SPEC.md, then request approval:")
            _suggest_or_execute(
                args, name, "research:awaiting_approval", project_dir
            )
        elif phase == "research:approved":
            _suggest_or_execute(
                args, name, "decomposition", project_dir,
                "Transition to decomposition"
            )
    elif base == "decomposition":
        if phase == "decomposition":
            print("→ Generate DECOMPOSITION.md, then request approval:")
            _suggest_or_execute(
                args, name, "decomposition:awaiting_approval", project_dir
            )
        elif phase == "decomposition:approved":
            _suggest_or_execute(
                args, name, "complete", project_dir,
                "All subtasks complete — mark parent done"
            )
    elif base == "design":
        if phase == "design":
            print("→ Generate DESIGN.md, then request approval:")
            _suggest_or_execute(
                args, name, "design:awaiting_approval", project_dir
            )
        elif phase == "design:approved":
            _suggest_or_execute(
                args, name, "test_design", project_dir,
                "Transition to test_design"
            )
    elif base == "test_design":
        if phase == "test_design":
            print("→ Generate TEST_PLAN.md, then request approval:")
            _suggest_or_execute(
                args, name, "test_design:awaiting_approval", project_dir
            )
        elif phase == "test_design:approved":
            _suggest_or_execute(
                args, name, "implement", project_dir,
                "Transition to implement"
            )
    elif base == "implement":
        print("→ Write implementation, then request code review:")
        _suggest_or_execute(
            args, name, "code_review", project_dir
        )
    elif base == "code_review":
        if phase == "code_review":
            print("→ Generate CODE_REVIEW.md, then request approval:")
            _suggest_or_execute(
                args, name, "code_review:awaiting_approval", project_dir
            )
        elif phase == "code_review:approved":
            _suggest_or_execute(
                args, name, "bug_find", project_dir,
                "Transition to bug_find"
            )
    elif base == "bug_find":
        print("→ Generate BUG_REPORT.md, then transition:")
        _suggest_or_execute(
            args, name, "adversarial_bug_find", project_dir
        )
    elif base == "adversarial_bug_find":
        print("→ Generate ADVERSARIAL_BUG_REPORT.md, then transition:")
        _suggest_or_execute(
            args, name, "doc_review", project_dir
        )
    elif base == "doc_review":
        print("→ Generate DOC_REVIEW.md, then transition:")
        _suggest_or_execute(
            args, name, "referee", project_dir
        )
    elif base == "referee":
        print("→ Generate VERDICT.md, then complete:")
        _suggest_or_execute(
            args, name, "complete", project_dir
        )
    elif base == "human_intervention":
        print("→ Task needs human intervention. Transition to referee or complete:")
        _suggest_or_execute(
            args, name, "referee", project_dir
        )

    return 0


def cmd_loop(args):
    """Run the drive loop continuously until blocked or complete."""
    if not getattr(args, "execute", False):
        args.execute = True
        print("(auto-enabling --execute for loop mode)")
    max_iterations = args.max_iterations or 100
    for iteration in range(1, max_iterations + 1):
        print(f"--- Iteration {iteration}/{max_iterations} ---")
        result = cmd_drive(args)
        if result != 0:
            return result
        if args.delay:
            import time as tm

            tm.sleep(args.delay)
    print(f"Reached max iterations ({max_iterations}).")
    return 0


def cmd_stuck(args):
    """Detect and report stuck tasks."""
    project_dir = _find_project_dir(args.project)
    threshold = args.threshold or 60
    stuck = detect_stuck_tasks(project_dir, threshold)

    if not stuck:
        print(f"No stuck tasks detected (threshold: {threshold} min).")
        return 0

    print(f"=== STUCK TASKS (>{threshold} min in same phase) ===")
    for t in stuck:
        print(f"  {t['name']}: stuck at '{t['phase']}' for {t['age_minutes']} min")
        print(f"    Path: {t['path']}")
        print("     Action: Review and transition manually or mark complete")
        print()
    print(f"Total stuck: {len(stuck)}")
    return 0


def cmd_install_schedule(args):
    """Install an OS scheduler unit that runs autopilot --drive --execute periodically."""
    project_dir = _find_project_dir(args.project)
    interval = getattr(args, "interval", None) or 60
    autopilot_path = AUTOMATON_DIR / "scripts" / "autopilot.py"
    system = platform.system()

    if system == "Darwin":
        plist_dir = Path.home() / "Library" / "LaunchAgents"
        plist_dir.mkdir(parents=True, exist_ok=True)
        plist_path = plist_dir / "com.automaton.autopilot.plist"
        plist = (
            f"<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n"
            f"<!DOCTYPE plist PUBLIC \"-//Apple//DTD PLIST 1.0//EN\" "
            f"\"http://www.apple.com/DTDs/PropertyList-1.0.dtd\">\n"
            f"<plist version=\"1.0\">\n"
            f"<dict>\n"
            f"  <key>Label</key><string>com.automaton.autopilot</string>\n"
            f"  <key>ProgramArguments</key>\n"
            f"  <array>\n"
            f"    <string>{sys.executable}</string>\n"
            f"    <string>{autopilot_path}</string>\n"
            f"    <string>--project</string>\n"
            f"    <string>{project_dir}</string>\n"
            f"    <string>--drive</string>\n"
            f"    <string>--execute</string>\n"
            f"  </array>\n"
            f"  <key>StartInterval</key><integer>{interval}</integer>\n"
            f"  <key>RunAtLoad</key><true/>\n"
            f"  <key>StandardOutPath</key><string>{AUTOMATON_DIR}/logs/autopilot-stdout.log</string>\n"
            f"  <key>StandardErrorPath</key><string>{AUTOMATON_DIR}/logs/autopilot-stderr.log</string>\n"
            f"</dict>\n"
            f"</plist>\n"
        )
        plist_path.write_text(plist)
        (AUTOMATON_DIR / "logs").mkdir(parents=True, exist_ok=True)
        print(f"Installed launchd unit: {plist_path}")
        print(f"Interval: {interval}s")
        print(f"Logs: {AUTOMATON_DIR / 'logs' / 'autopilot-*.log'}")
        print("Load with: launchctl load ~/Library/LaunchAgents/com.automaton.autopilot.plist")
        print("Unload with: launchctl unload ~/Library/LaunchAgents/com.automaton.autopilot.plist")
        return 0
    elif system == "Linux":
        cron_line = f"*/{max(1, interval // 60)} * * * * {sys.executable} {autopilot_path} --project {project_dir} --drive --execute >> {AUTOMATON_DIR}/logs/autopilot.log 2>&1"
        (AUTOMATON_DIR / "logs").mkdir(parents=True, exist_ok=True)
        print("Add to crontab:")
        print(f"  {cron_line}")
        return 0
    elif system == "Windows":
        print("Windows: use Task Scheduler to run:")
        print(f"  {sys.executable} {autopilot_path} --project {project_dir} --drive --execute")
        return 0
    print(f"ERROR: unsupported platform '{system}'")
    return 1


def main():
    parser = argparse.ArgumentParser(description="Automaton autopilot runtime")
    parser.add_argument("--project", help="Project root directory")
    parser.add_argument(
        "--drive", action="store_true", help="Drive one step forward (default)"
    )
    parser.add_argument(
        "--execute", action="store_true",
        help="Execute transitions instead of printing suggestions"
    )
    parser.add_argument("--summary", action="store_true", help="Show autopilot summary")
    parser.add_argument(
        "--stuck", action="store_true", dest="detect_stuck", help="Detect stuck tasks"
    )
    parser.add_argument("--loop", action="store_true", help="Run continuous drive loop")
    parser.add_argument(
        "--install-schedule", action="store_true",
        help="Install OS scheduler unit that runs --drive --execute periodically"
    )
    parser.add_argument(
        "--interval", type=int, default=60,
        help="Tick interval in seconds for --install-schedule (default: 60)"
    )
    parser.add_argument(
        "--max-iterations", type=int, help="Max iterations for --loop (default: 100)"
    )
    parser.add_argument(
        "--delay", type=int, help="Delay seconds between iterations for --loop"
    )
    parser.add_argument(
        "--threshold",
        type=int,
        help="Stuck detection threshold in minutes (default: 60)",
    )

    args = parser.parse_args()

    if args.install_schedule:
        return cmd_install_schedule(args)
    if args.summary:
        return cmd_summary(args)
    if args.detect_stuck:
        return cmd_stuck(args)
    if args.loop:
        return cmd_loop(args)
    return cmd_drive(args)


if __name__ == "__main__":
    sys.exit(main())