Lap Tran
715f6f9495
Fixes from functional correctness review: - PHASE_PRIORITY: add code_review (was missing, caused wrong task selection) - cmd_drive: implement→code_review (was illegal implement→bug_find) - cmd_drive: add handlers for code_review and code_review:approved - _all_tasks: skip tasks/complete/ (was showing completed tasks as phase=None) - _all_tasks: recurse into subtasks/ (subtasks were invisible to autopilot) - _all_tasks: use .automaton/tasks for non-framework projects (was project/tasks) - is_terminal: only True for complete (human_intervention has legal transitions) - needs_user_input: flag human_intervention as blocked (was auto-driving it) Tests cover all 8 bugs: PHASE_PRIORITY ordering, every phase transition, code_review handlers, complete/ exclusion, subtask recursion, project path, terminal state, user-input detection, human_intervention handling.
463 lines
16 KiB
Python
Executable File
463 lines
16 KiB
Python
Executable File
#!/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 . --loop Run continuous loop
|
|
python autopilot.py --project . --detect-stuck Detect stuck tasks
|
|
python autopilot.py --project . --summary Show autopilot summary
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
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 _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 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"]
|
|
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}"
|
|
)
|
|
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}"
|
|
)
|
|
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}"
|
|
)
|
|
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}"
|
|
)
|
|
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}"
|
|
)
|
|
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}"
|
|
)
|
|
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}"
|
|
)
|
|
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}"
|
|
)
|
|
elif phase == "test_design:approved":
|
|
print("→ Transition to implement:")
|
|
print(
|
|
f" python ~/.automaton/scripts/status.py --transition implement --task {task['name']} --project {project_dir}"
|
|
)
|
|
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}"
|
|
)
|
|
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}"
|
|
)
|
|
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}"
|
|
)
|
|
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}"
|
|
)
|
|
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}"
|
|
)
|
|
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}"
|
|
)
|
|
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}"
|
|
)
|
|
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}"
|
|
)
|
|
|
|
return 0
|
|
|
|
|
|
def cmd_loop(args):
|
|
"""Run the drive loop continuously until blocked or complete."""
|
|
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 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("--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(
|
|
"--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.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())
|