--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).
This commit is contained in:
Lap Tran
@@ -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"
|
||||
|
||||
+162
-51
@@ -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"<?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)"
|
||||
)
|
||||
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user