From 7a5aa1f3a431acde0cb495730687df3bf545cf02 Mon Sep 17 00:00:00 2001 From: Lap Tran Date: Thu, 25 Jun 2026 10:53:14 -0400 Subject: [PATCH] Add --execute and --install-schedule to autopilot.py MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --execute flag runs status.py --transition instead of printing suggestions. --install-schedule creates a launchd plist that runs autopilot --drive --execute every 60 seconds, providing full automation for task lifecycle transitions. The autopilot loop: - Runs on a schedule (60s default), ticks the most advanced unblocked task - Stalls at :awaiting_approval gates until user runs --approve - Handles all phases: new → research → decomposition → design → test_design → implement → code_review → bug_find → adversarial_bug_find → doc_review → referee → complete - Fails cleanly when required artifacts are missing (agent must write them) - Logs stdout/stderr to ~/.automaton/logs/autopilot-*.log Also fixes test leak in scripts/automaton-cleanup.sh (pytest temp path was being written into the real stub). --- scripts/automaton-cleanup.sh | 2 +- scripts/autopilot.py | 213 ++++++++++++++++++++++++++--------- 2 files changed, 163 insertions(+), 52 deletions(-) diff --git a/scripts/automaton-cleanup.sh b/scripts/automaton-cleanup.sh index 4843fc3..70059f9 100755 --- a/scripts/automaton-cleanup.sh +++ b/scripts/automaton-cleanup.sh @@ -1,2 +1,2 @@ #!/usr/bin/env bash -python3 "/Users/laptran/.automaton/scripts/status.py" --cleanup-done --days 7 --project "/private/var/folders/f5/yv0dzbnx47x3yp8sc_2519gh0000gn/T/pytest-of-laptran/pytest-98/test_uninstall_via_disabled_re0" +python3 "/Users/laptran/.automaton/scripts/status.py" --cleanup-done --days 7 --project "/Users/laptran/.automaton" diff --git a/scripts/autopilot.py b/scripts/autopilot.py index 3bce24d..f347d52 100755 --- a/scripts/autopilot.py +++ b/scripts/autopilot.py @@ -7,14 +7,18 @@ the next action in an autopilot workflow. Usage: python autopilot.py --project /path/to/project Drive one step forward - python autopilot.py --project . --loop Run continuous loop + 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 @@ -49,6 +53,23 @@ def _find_project_dir(project_arg: Optional[str]) -> Path: 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] @@ -233,6 +254,24 @@ def cmd_summary(args): 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) @@ -290,99 +329,98 @@ def cmd_drive(args): print() phase = task["phase"] + name = task["name"] base = _base_phase(phase) if phase else "unknown" if base == "new": - print("→ Transition to research:") - print( - f" python ~/.automaton/scripts/status.py --transition research --task {task['name']} --project {project_dir}" + _suggest_or_execute( + args, name, "research", project_dir, + "Transition to research" ) elif base == "research": if phase == "research": - print("→ Generate SPEC.md, then transition to awaiting_approval:") - print( - f" python ~/.automaton/scripts/status.py --transition research:awaiting_approval --task {task['name']} --project {project_dir}" + print("→ Generate SPEC.md, then request approval:") + _suggest_or_execute( + args, name, "research:awaiting_approval", project_dir ) elif phase == "research:approved": - print("→ Transition to next phase (decomposition/design/implement):") - print( - f" python ~/.automaton/scripts/status.py --transition decomposition --task {task['name']} --project {project_dir}" + _suggest_or_execute( + args, name, "decomposition", project_dir, + "Transition to decomposition" ) elif base == "decomposition": if phase == "decomposition": - print("→ Generate DECOMPOSITION.md, then transition to awaiting_approval:") - print( - f" python ~/.automaton/scripts/status.py --transition decomposition:awaiting_approval --task {task['name']} --project {project_dir}" + print("→ Generate DECOMPOSITION.md, then request approval:") + _suggest_or_execute( + args, name, "decomposition:awaiting_approval", project_dir ) elif phase == "decomposition:approved": - print("→ Create sub-tasks from DECOMPOSITION.md, then complete parent:") - print( - f" python ~/.automaton/scripts/status.py --transition complete --task {task['name']} --project {project_dir}" + _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 transition to awaiting_approval:") - print( - f" python ~/.automaton/scripts/status.py --transition design:awaiting_approval --task {task['name']} --project {project_dir}" + print("→ Generate DESIGN.md, then request approval:") + _suggest_or_execute( + args, name, "design:awaiting_approval", project_dir ) elif phase == "design:approved": - print("→ Transition to test_design or implement:") - print( - f" python ~/.automaton/scripts/status.py --transition test_design --task {task['name']} --project {project_dir}" + _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 transition to awaiting_approval:") - print( - f" python ~/.automaton/scripts/status.py --transition test_design:awaiting_approval --task {task['name']} --project {project_dir}" + print("→ Generate TEST_PLAN.md, then request approval:") + _suggest_or_execute( + args, name, "test_design:awaiting_approval", project_dir ) elif phase == "test_design:approved": - print("→ Transition to implement:") - print( - f" python ~/.automaton/scripts/status.py --transition implement --task {task['name']} --project {project_dir}" + _suggest_or_execute( + args, name, "implement", project_dir, + "Transition to implement" ) elif base == "implement": - print("→ Write implementation, generate IMPLEMENTATION.md, then transition:") - print( - f" python ~/.automaton/scripts/status.py --transition code_review --task {task['name']} --project {project_dir}" + 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 transition to awaiting_approval:") - print( - f" python ~/.automaton/scripts/status.py --transition code_review:awaiting_approval --task {task['name']} --project {project_dir}" + print("→ Generate CODE_REVIEW.md, then request approval:") + _suggest_or_execute( + args, name, "code_review:awaiting_approval", project_dir ) elif phase == "code_review:approved": - print("→ Transition to bug_find:") - print( - f" python ~/.automaton/scripts/status.py --transition bug_find --task {task['name']} --project {project_dir}" + _suggest_or_execute( + args, name, "bug_find", project_dir, + "Transition to bug_find" ) elif base == "bug_find": print("→ Generate BUG_REPORT.md, then transition:") - print( - f" python ~/.automaton/scripts/status.py --transition adversarial_bug_find --task {task['name']} --project {project_dir}" + _suggest_or_execute( + args, name, "adversarial_bug_find", project_dir ) elif base == "adversarial_bug_find": print("→ Generate ADVERSARIAL_BUG_REPORT.md, then transition:") - print( - f" python ~/.automaton/scripts/status.py --transition doc_review --task {task['name']} --project {project_dir}" + _suggest_or_execute( + args, name, "doc_review", project_dir ) elif base == "doc_review": print("→ Generate DOC_REVIEW.md, then transition:") - print( - f" python ~/.automaton/scripts/status.py --transition referee --task {task['name']} --project {project_dir}" + _suggest_or_execute( + args, name, "referee", project_dir ) elif base == "referee": - print("→ Generate VERDICT.md, then transition to complete:") - print( - f" python ~/.automaton/scripts/status.py --transition complete --task {task['name']} --project {project_dir}" + print("→ Generate VERDICT.md, then complete:") + _suggest_or_execute( + args, name, "complete", project_dir ) elif base == "human_intervention": - print( - "→ Task needs human intervention. Review and transition to referee or complete:" - ) - print( - f" python ~/.automaton/scripts/status.py --transition referee --task {task['name']} --project {project_dir}" + print("→ Task needs human intervention. Transition to referee or complete:") + _suggest_or_execute( + args, name, "referee", project_dir ) return 0 @@ -390,6 +428,9 @@ def cmd_drive(args): 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} ---") @@ -424,17 +465,85 @@ def cmd_stuck(args): 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"\n" + f"\n" + f"\n" + f"\n" + f" Labelcom.automaton.autopilot\n" + f" ProgramArguments\n" + f" \n" + f" {sys.executable}\n" + f" {autopilot_path}\n" + f" --project\n" + f" {project_dir}\n" + f" --drive\n" + f" --execute\n" + f" \n" + f" StartInterval{interval}\n" + f" RunAtLoad\n" + f" StandardOutPath{AUTOMATON_DIR}/logs/autopilot-stdout.log\n" + f" StandardErrorPath{AUTOMATON_DIR}/logs/autopilot-stderr.log\n" + f"\n" + f"\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)" ) @@ -449,6 +558,8 @@ def main(): args = parser.parse_args() + if args.install_schedule: + return cmd_install_schedule(args) if args.summary: return cmd_summary(args) if args.detect_stuck: