From 715f6f9495ae84da58c6942240fc6257e43ce2e9 Mon Sep 17 00:00:00 2001 From: Lap Tran Date: Thu, 25 Jun 2026 07:11:44 -0400 Subject: [PATCH] Fix 7 bugs in autopilot.py, add 43 tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- scripts/autopilot.py | 76 +++++++--- tests/test_autopilot.py | 316 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 375 insertions(+), 17 deletions(-) create mode 100644 tests/test_autopilot.py diff --git a/scripts/autopilot.py b/scripts/autopilot.py index f626bcb..3bce24d 100755 --- a/scripts/autopilot.py +++ b/scripts/autopilot.py @@ -24,15 +24,17 @@ 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, + "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, } @@ -60,13 +62,28 @@ def _read_state(task_path: Path) -> Optional[str]: def _all_tasks(project_dir: Path) -> list[tuple[str, Path]]: - tasks_dir = project_dir / "tasks" - if not tasks_dir.is_dir(): + """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 subdir in sorted(tasks_dir.iterdir()): - if subdir.is_dir(): - result.append((subdir.name, subdir)) + 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 @@ -87,11 +104,16 @@ def scan_all_tasks(project_dir: Path) -> list[dict]: def is_terminal(task: dict) -> bool: - """Check if a task is in a terminal state.""" + """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 in ("complete", "human_intervention") + return phase == "complete" def needs_user_input(task: dict) -> bool: @@ -101,6 +123,8 @@ def needs_user_input(task: dict) -> bool: 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(): @@ -246,6 +270,13 @@ def cmd_drive(args): 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 @@ -313,8 +344,19 @@ def cmd_drive(args): 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}" + 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( diff --git a/tests/test_autopilot.py b/tests/test_autopilot.py new file mode 100644 index 0000000..d2ddadd --- /dev/null +++ b/tests/test_autopilot.py @@ -0,0 +1,316 @@ +"""Tests for scripts/autopilot.py — functional correctness review. + +Covers bugs found during 2026-06-25 review: + 1. PHASE_PRIORITY missing code_review + 2. cmd_drive suggested implement→bug_find (illegal; should be implement→code_review) + 3. cmd_drive had no handler for code_review / code_review:approved + 4. _all_tasks didn't skip tasks/complete/ archive dir + 5. _all_tasks didn't recurse into subtasks/ + 6. _all_tasks used wrong path for non-framework projects + 7. is_terminal treated human_intervention as terminal (dead-code handler) + 8. needs_user_input didn't flag human_intervention as blocked +""" + +import importlib.util +import sys +from pathlib import Path +from typing import Optional + +import pytest + +_AP_PATH = Path.home() / ".automaton" / "scripts" / "autopilot.py" +_spec = importlib.util.spec_from_file_location("autopilot_mod", _AP_PATH) +ap = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(ap) + + +def _make_task(project: Path, name: str, phase: str = "new") -> Path: + """Create a task dir + .state (no subprocess).""" + if project == ap.AUTOMATON_DIR: + tp = project / "tasks" / name + else: + tp = project / ".automaton" / "tasks" / name + tp.mkdir(parents=True, exist_ok=True) + (tp / ".state").write_text(phase + "\n") + return tp + + +def _make_subtask(project: Path, parent: str, name: str, phase: str = "new") -> Path: + if project == ap.AUTOMATON_DIR: + sp = project / "tasks" / parent / "subtasks" / name + else: + sp = project / ".automaton" / "tasks" / parent / "subtasks" / name + sp.mkdir(parents=True, exist_ok=True) + (sp / ".state").write_text(phase + "\n") + return sp + + +def _write_verdict(task_path: Path, first_line: str) -> None: + (task_path / "VERDICT.md").write_text(first_line + "\nrest of verdict\n") + + +class _Args: + """Minimal args object for cmd_* functions.""" + def __init__(self, **kwargs): + self.project = kwargs.get("project") + self.max_iterations = kwargs.get("max_iterations") + self.delay = kwargs.get("delay") + self.threshold = kwargs.get("threshold") + + +@pytest.fixture +def tmp_project(tmp_path): + (tmp_path / ".automaton" / "tasks").mkdir(parents=True) + return tmp_path + + +class TestPhasePriority: + """Bug 1: code_review was missing from PHASE_PRIORITY.""" + + def test_code_review_in_priority(self): + assert "code_review" in ap.PHASE_PRIORITY + + def test_code_review_priority_between_implement_and_bug_find(self): + assert ap.PHASE_PRIORITY["code_review"] > ap.PHASE_PRIORITY["implement"] + assert ap.PHASE_PRIORITY["code_review"] < ap.PHASE_PRIORITY["bug_find"] + + def test_human_intervention_in_priority(self): + assert "human_intervention" in ap.PHASE_PRIORITY + + def test_priorities_monotonically_increase_along_pipeline(self): + pipeline = [ + "new", "research", "decomposition", "design", "test_design", + "implement", "code_review", "bug_find", "adversarial_bug_find", + "doc_review", "referee", + ] + for i in range(len(pipeline) - 1): + assert ap.PHASE_PRIORITY[pipeline[i]] < ap.PHASE_PRIORITY[pipeline[i + 1]], \ + f"{pipeline[i]} ({ap.PHASE_PRIORITY[pipeline[i]]}) should be < {pipeline[i+1]} ({ap.PHASE_PRIORITY[pipeline[i+1]]})" + + +class TestCmdDriveCodeReview: + """Bugs 2+3: implement→bug_find was illegal; code_review handlers missing.""" + + def test_implement_suggests_code_review_not_bug_find(self, tmp_project, capsys): + _make_task(tmp_project, "task-a", "implement") + args = _Args(project=str(tmp_project)) + ap.cmd_drive(args) + out = capsys.readouterr().out + assert "--transition code_review" in out + assert "--transition bug_find" not in out + + def test_code_review_suggests_awaiting_approval(self, tmp_project, capsys): + _make_task(tmp_project, "task-cr", "code_review") + args = _Args(project=str(tmp_project)) + ap.cmd_drive(args) + out = capsys.readouterr().out + assert "--transition code_review:awaiting_approval" in out + + def test_code_review_approved_suggests_bug_find(self, tmp_project, capsys): + _make_task(tmp_project, "task-cr2", "code_review:approved") + args = _Args(project=str(tmp_project)) + ap.cmd_drive(args) + out = capsys.readouterr().out + assert "--transition bug_find" in out + + def test_implement_task_not_stalled(self, tmp_project, capsys): + """Ensure cmd_drive produces a transition command for implement phase.""" + _make_task(tmp_project, "task-impl", "implement") + args = _Args(project=str(tmp_project)) + rc = ap.cmd_drive(args) + out = capsys.readouterr().out + assert rc == 0 + assert "status.py --transition" in out + + +class TestAllTasksSkipComplete: + """Bug 4: _all_tasks included tasks/complete/ as a task.""" + + def test_complete_dir_excluded(self, tmp_project): + _make_task(tmp_project, "active-task", "implement") + complete_dir = tmp_project / ".automaton" / "tasks" / "complete" + complete_dir.mkdir(parents=True) + (complete_dir / "old-task").mkdir() + (complete_dir / "old-task" / ".state").write_text("complete\n") + + tasks = ap._all_tasks(tmp_project) + names = [t[0] for t in tasks] + assert "active-task" in names + assert "complete" not in names + + +class TestAllTasksSubtasks: + """Bug 5: _all_tasks didn't recurse into subtasks/.""" + + def test_subtask_found(self, tmp_project): + _make_task(tmp_project, "parent-task", "decomposition:approved") + _make_subtask(tmp_project, "parent-task", "sub-a", "implement") + _make_subtask(tmp_project, "parent-task", "sub-b", "new") + + tasks = ap._all_tasks(tmp_project) + names = [t[0] for t in tasks] + assert "parent-task" in names + assert "parent-task/sub-a" in names + assert "parent-task/sub-b" in names + + def test_subtask_scan_all_tasks(self, tmp_project): + _make_task(tmp_project, "parent-task", "decomposition:approved") + _make_subtask(tmp_project, "parent-task", "sub-a", "implement") + + scanned = ap.scan_all_tasks(tmp_project) + names = [t["name"] for t in scanned] + assert "parent-task" in names + assert "parent-task/sub-a" in names + + +class TestAllTasksProjectPath: + """Bug 6: _all_tasks used project_dir/tasks instead of project_dir/.automaton/tasks.""" + + def test_non_framework_project_path(self, tmp_project): + _make_task(tmp_project, "proj-task", "implement") + tasks = ap._all_tasks(tmp_project) + names = [t[0] for t in tasks] + assert "proj-task" in names + + def test_wrong_path_returns_empty(self, tmp_path): + """If .automaton/tasks doesn't exist, return empty (not crash).""" + tasks = ap._all_tasks(tmp_path) + assert tasks == [] + + +class TestIsTerminal: + """Bug 7: human_intervention was treated as terminal.""" + + def test_complete_is_terminal(self): + assert ap.is_terminal({"phase": "complete"}) is True + + def test_human_intervention_not_terminal(self): + assert ap.is_terminal({"phase": "human_intervention"}) is False + + def test_none_phase_not_terminal(self): + assert ap.is_terminal({"phase": None}) is False + + def test_new_not_terminal(self): + assert ap.is_terminal({"phase": "new"}) is False + + +class TestNeedsUserInput: + """Bug 8: human_intervention wasn't flagged as needing user input.""" + + def test_awaiting_approval_needs_input(self, tmp_project): + tp = _make_task(tmp_project, "task-r", "research:awaiting_approval") + assert ap.needs_user_input({"phase": "research:awaiting_approval", "path": str(tp)}) is True + + def test_human_intervention_needs_input(self, tmp_project): + tp = _make_task(tmp_project, "task-hi", "human_intervention") + assert ap.needs_user_input({"phase": "human_intervention", "path": str(tp)}) is True + + def test_implement_no_input(self, tmp_project): + tp = _make_task(tmp_project, "task-i", "implement") + assert ap.needs_user_input({"phase": "implement", "path": str(tp)}) is False + + def test_verdict_fail_needs_input(self, tmp_project): + tp = _make_task(tmp_project, "task-v", "referee") + _write_verdict(tp, "FAIL: bugs found") + assert ap.needs_user_input({"phase": "referee", "path": str(tp)}) is True + + def test_verdict_pass_no_input(self, tmp_project): + tp = _make_task(tmp_project, "task-vp", "referee") + _write_verdict(tp, "PASS: all good") + assert ap.needs_user_input({"phase": "referee", "path": str(tp)}) is False + + +class TestSortByAdvancement: + """Verify sort orders code_review correctly after the fix.""" + + def test_code_review_more_advanced_than_implement(self): + tasks = [ + {"name": "impl-task", "base_phase": "implement"}, + {"name": "cr-task", "base_phase": "code_review"}, + ] + sorted_tasks = ap.sort_by_advancement(tasks) + assert sorted_tasks[0]["name"] == "cr-task" + + def test_bug_find_more_advanced_than_code_review(self): + tasks = [ + {"name": "cr-task", "base_phase": "code_review"}, + {"name": "bf-task", "base_phase": "bug_find"}, + ] + sorted_tasks = ap.sort_by_advancement(tasks) + assert sorted_tasks[0]["name"] == "bf-task" + + +class TestCmdDriveHumanIntervention: + """Verify human_intervention is now drivable (not dead code).""" + + def test_human_intervention_shows_in_blocked(self, tmp_project, capsys): + _make_task(tmp_project, "task-hi", "human_intervention") + args = _Args(project=str(tmp_project)) + ap.cmd_drive(args) + out = capsys.readouterr().out + assert "ORCHESTRATION_BLOCKED" in out + assert "task-hi" in out + assert "--transition referee" in out + + def test_human_intervention_not_in_unblocked(self, tmp_project, capsys): + """human_intervention should be blocked, not selected as NEXT_TASK.""" + _make_task(tmp_project, "task-hi", "human_intervention") + _make_task(tmp_project, "task-active", "implement") + args = _Args(project=str(tmp_project)) + ap.cmd_drive(args) + out = capsys.readouterr().out + assert "NEXT_TASK: task-active" in out + assert "NEXT_TASK: task-hi" not in out + + +class TestCmdDriveSummary: + """Integration: cmd_summary with mixed states.""" + + def test_summary_counts(self, tmp_project, capsys): + _make_task(tmp_project, "t-new", "new") + _make_task(tmp_project, "t-impl", "implement") + _make_task(tmp_project, "t-blocked", "research:awaiting_approval") + _make_task(tmp_project, "t-hi", "human_intervention") + complete_dir = tmp_project / ".automaton" / "tasks" / "complete" + complete_dir.mkdir() + (complete_dir / "t-done").mkdir() + (complete_dir / "t-done" / ".state").write_text("complete\n") + + args = _Args(project=str(tmp_project)) + ap.cmd_summary(args) + out = capsys.readouterr().out + assert "3" in out # 3 non-terminal (new, implement, blocked, hi... wait, hi is now non-terminal too) + assert "READY TO DRIVE" in out + assert "AWAITING USER" in out + assert "t-done" not in out # complete dir should not appear + assert "complete" not in out.split("READY")[0] # no "complete" as a task name + + +class TestCmdDriveAllPhases: + """Every phase produces a transition command — no silent stalls.""" + + @pytest.mark.parametrize("phase,expected_fragment", [ + ("new", "--transition research"), + ("research", "--transition research:awaiting_approval"), + ("research:approved", "--transition decomposition"), + ("decomposition", "--transition decomposition:awaiting_approval"), + ("decomposition:approved", "--transition complete"), + ("design", "--transition design:awaiting_approval"), + ("design:approved", "--transition test_design"), + ("test_design", "--transition test_design:awaiting_approval"), + ("test_design:approved", "--transition implement"), + ("implement", "--transition code_review"), + ("code_review", "--transition code_review:awaiting_approval"), + ("code_review:approved", "--transition bug_find"), + ("bug_find", "--transition adversarial_bug_find"), + ("adversarial_bug_find", "--transition doc_review"), + ("doc_review", "--transition referee"), + ("referee", "--transition complete"), + ]) + def test_phase_produces_transition(self, tmp_project, capsys, phase, expected_fragment): + _make_task(tmp_project, "test-task", phase) + args = _Args(project=str(tmp_project)) + rc = ap.cmd_drive(args) + out = capsys.readouterr().out + assert rc == 0 + assert expected_fragment in out, f"Phase '{phase}' should suggest '{expected_fragment}', got:\n{out}"