gitea
3480e4ecba
CI / build (push) Has been cancelled
- Fix decomposition:approved and human_intervention dead-end phases - Add scripts/autopilot.py: real drive_all() implementation - Fix guard plugin: throw Error instead of injecting user messages - Fix status.py: double continue, _require_state, --list-states - Add Category 3 (git-based modification) audit - Add Category 5 (stuck-task detection) audit - All 206 tests pass
346 lines
14 KiB
Python
Executable File
346 lines
14 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": 10,
|
|
"adversarial_bug_find": 8,
|
|
"bug_find": 6,
|
|
"implement": 5,
|
|
"test_design": 4,
|
|
"design": 3,
|
|
"decomposition": 2,
|
|
"research": 1,
|
|
"new": 0,
|
|
}
|
|
|
|
|
|
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]]:
|
|
tasks_dir = project_dir / "tasks"
|
|
if not tasks_dir.is_dir():
|
|
return []
|
|
result = []
|
|
for subdir in sorted(tasks_dir.iterdir()):
|
|
if subdir.is_dir():
|
|
result.append((subdir.name, subdir))
|
|
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."""
|
|
phase = task.get("phase")
|
|
if phase is None:
|
|
return False
|
|
return phase in ("complete", "human_intervention")
|
|
|
|
|
|
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
|
|
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}")
|
|
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 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(f" 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())
|