#!/usr/bin/env python3 """Automaton status and enforcement script. Manages task state, validates phase transitions, enforces approval gates, and provides enforcement hooks for agent tool integrations. Integration Contract for Harnesses ------------------------------------ Before allowing any file edit, a harness MUST call: python ~/.automaton/scripts/status.py --can-edit --project {project} [--file {path}] [--json] Exit code 0 = ALLOWED, exit code 1 = DENIED, exit code 2 = ERROR. Modes: 1. --can-edit --project {p} Is editing allowed on this project at all? Checks that at least one task is in implement or doc_review phase. 2. --can-edit --project {p} --file {path} Same as (1) but also checks that the file is within the project scope. 3. --can-edit --project {p} --task {t} Is this specific task in an edit-allowed phase? 4. --can-edit --project {p} --task {t} --file {path} Same as (3) but also checks file scope. Add --json for machine-readable output on the last line. Usage: python status.py --task {name} Show task status python status.py --list List all tasks python status.py --create-task {name} Create a new task python status.py --transition {phase} --task {n} Transition task phase python status.py --approve --task {name} Approve current phase python status.py --validate-folder --task {name} Validate task folder python status.py --audit Audit all tasks python status.py --claim --task {name} --agent {id} Claim task (multi-agent) python status.py --release --task {name} --agent {id} Release task (multi-agent) python status.py --next-available --agent {id} Find available work python status.py --available --agent {id} List available work python status.py --can-edit --project {p} [--task {name}] [--file {path}] Check if edits allowed (harness hook) python status.py --scope-check --task {name} --file {path} Check file scope python status.py --cleanup-done [--days 7] [--dry-run] Archive complete tasks older than N days into tasks/complete/ python status.py --install-cleanup-schedule [--days 7] [--interval 86400] Install daily cleanup job """ from __future__ import annotations import argparse import contextlib import json import os import platform import re import subprocess import sys from datetime import datetime, timezone from pathlib import Path from typing import Optional AUTOMATON_DIR = Path.home() / ".automaton" VALID_PHASES = [ "new", "research", "research:awaiting_approval", "research:approved", "decomposition", "decomposition:awaiting_approval", "decomposition:approved", "design", "design:awaiting_approval", "design:approved", "test_design", "test_design:awaiting_approval", "test_design:approved", "implement", "code_review", "code_review:awaiting_approval", "code_review:approved", "bug_find", "adversarial_bug_find", "doc_review", "referee", "complete", "human_intervention", ] BASE_PHASES = [ "new", "research", "decomposition", "design", "test_design", "implement", "code_review", "bug_find", "adversarial_bug_find", "doc_review", "referee", "complete", "human_intervention", ] APPROVAL_PHASES = {"research", "decomposition", "design", "test_design", "code_review"} LEGAL_TRANSITIONS = { "new": ["research"], "research": ["research:awaiting_approval", "decomposition", "design", "implement"], "research:awaiting_approval": ["research:approved"], "research:approved": ["decomposition", "design", "implement"], "decomposition": ["decomposition:awaiting_approval"], "decomposition:awaiting_approval": ["decomposition:approved"], "decomposition:approved": ["complete"], "design": ["design:awaiting_approval", "test_design", "implement"], "design:awaiting_approval": ["design:approved"], "design:approved": ["test_design", "implement"], "test_design": ["test_design:awaiting_approval", "implement"], "test_design:awaiting_approval": ["test_design:approved"], "test_design:approved": ["implement"], "implement": ["code_review"], "code_review": ["code_review:awaiting_approval"], "code_review:awaiting_approval": ["code_review:approved"], "code_review:approved": ["bug_find"], "bug_find": ["adversarial_bug_find"], "adversarial_bug_find": ["doc_review"], "doc_review": ["referee"], "referee": ["complete", "human_intervention"], "human_intervention": ["referee", "complete"], } PHASE_REQUIRED_ARTIFACTS = { "research": "SPEC.md", "decomposition": "DECOMPOSITION.md", "design": "DESIGN.md", "test_design": "TEST_PLAN.md", "implement": "IMPLEMENTATION.md", "code_review": "CODE_REVIEW.md", "bug_find": "BUG_REPORT.md", "adversarial_bug_find": "ADVERSARIAL_BUG_REPORT.md", "doc_review": "DOC_REVIEW.md", "referee": "VERDICT.md", } FORBIDDEN_ARTIFACTS = { "new": ["SPEC.md", "DESIGN.md", "DECOMPOSITION.md", "TEST_PLAN.md", "IMPLEMENTATION.md", "CODE_REVIEW.md", "BUG_REPORT.md", "ADVERSARIAL_BUG_REPORT.md", "DOC_REVIEW.md", "VERDICT.md"], "research": ["DESIGN.md", "DECOMPOSITION.md", "TEST_PLAN.md", "IMPLEMENTATION.md", "CODE_REVIEW.md", "BUG_REPORT.md", "ADVERSARIAL_BUG_REPORT.md", "DOC_REVIEW.md", "VERDICT.md"], "decomposition": ["DESIGN.md", "TEST_PLAN.md", "IMPLEMENTATION.md", "CODE_REVIEW.md", "BUG_REPORT.md", "ADVERSARIAL_BUG_REPORT.md", "DOC_REVIEW.md", "VERDICT.md"], "design": ["DECOMPOSITION.md", "TEST_PLAN.md", "IMPLEMENTATION.md", "CODE_REVIEW.md", "BUG_REPORT.md", "ADVERSARIAL_BUG_REPORT.md", "DOC_REVIEW.md", "VERDICT.md"], "test_design": ["DECOMPOSITION.md", "IMPLEMENTATION.md", "CODE_REVIEW.md", "BUG_REPORT.md", "ADVERSARIAL_BUG_REPORT.md", "DOC_REVIEW.md", "VERDICT.md"], "implement": ["CODE_REVIEW.md", "BUG_REPORT.md", "ADVERSARIAL_BUG_REPORT.md", "DOC_REVIEW.md", "VERDICT.md"], "code_review": ["BUG_REPORT.md", "ADVERSARIAL_BUG_REPORT.md", "DOC_REVIEW.md", "VERDICT.md"], "bug_find": ["ADVERSARIAL_BUG_REPORT.md", "DOC_REVIEW.md", "VERDICT.md"], "adversarial_bug_find": ["DOC_REVIEW.md", "VERDICT.md"], "doc_review": ["VERDICT.md"], "referee": [], "complete": [], "human_intervention": [], } NON_ARTIFACT_FILES = {".state", ".state.tmp", ".state.lock", ".state.approvals", ".state.implementer", ".state.lastedit", "VRAM_CONFIG.md", "PARENT_SPEC.md", "REVIEW.md"} 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, } ALLOWED_ACTIONS_MAP = { "research": ["Read project files", "Ask clarifying questions", "Write SPEC.md"], "decomposition": ["Read SPEC.md", "Ask decomposition questions", "Write DECOMPOSITION.md", "Run VRAM detection"], "design": ["Read SPEC.md", "Ask design questions", "Write DESIGN.md"], "test_design": ["Read SPEC.md and DESIGN.md", "Ask test questions", "Write TEST_PLAN.md"], "implement": ["Edit code", "Write tests", "Create IMPLEMENTATION.md", "Run test suite"], "code_review": ["Read code", "Read SPEC.md", "Read DESIGN.md", "Read IMPLEMENTATION.md", "Write CODE_REVIEW.md"], "bug_find": ["Read code", "Read SPEC.md", "Read IMPLEMENTATION.md", "Write BUG_REPORT.md"], "adversarial_bug_find": ["Read code", "Read SPEC.md", "Read BUG_REPORT.md", "Write ADVERSARIAL_BUG_REPORT.md"], "doc_review": ["Read DESIGN.md", "Read code", "Read docs", "Write DOC_REVIEW.md", "Update documentation"], "referee": ["Read all artifacts", "Write VERDICT.md"], "new": ["Start research"], } FORBIDDEN_ACTIONS_MAP = { "research": ["Edit code", "Create IMPLEMENTATION.md", "Create DESIGN.md", "Create any artifact other than SPEC.md", "Skip to implementation"], "decomposition": ["Edit code", "Create IMPLEMENTATION.md", "Modify SPEC.md", "Create sub-task folders (Orchestrator does this)"], "design": ["Edit code", "Create IMPLEMENTATION.md", "Modify SPEC.md", "Skip to implementation"], "test_design": ["Edit code", "Write test implementations", "Create IMPLEMENTATION.md", "Modify SPEC.md or DESIGN.md"], "implement": ["Create new tasks", "Modify SPEC.md or DESIGN.md", "Transition to bug-find phase (Orchestrator does this)"], "code_review": ["Edit code", "Fix bugs or issues", "Modify SPEC.md", "Modify DESIGN.md", "Modify IMPLEMENTATION.md", "Create any artifact other than CODE_REVIEW.md"], "bug_find": ["Edit code", "Fix bugs (separate implementation task)", "Modify SPEC.md"], "adversarial_bug_find": ["Edit code", "Fix bugs", "Modify SPEC.md or BUG_REPORT.md"], "doc_review": ["Edit non-documentation code", "Modify SPEC.md", "Modify DESIGN.md"], "referee": ["Edit code", "Modify any artifact other than VERDICT.md"], "new": ["Edit code", "Create any artifact"], } NEXT_PHASE_MAP = { "new": "research", "research": "design or implement", "decomposition": "sub-task research", "design": "test_design or implement", "test_design": "implement", "implement": "code_review", "code_review": "bug_find", "bug_find": "adversarial_bug_find", "adversarial_bug_find": "doc_review", "doc_review": "referee", "referee": "complete or human_intervention", } def _base_phase(phase: str) -> str: if ":" in phase: return phase.split(":")[0] return phase def _find_project_dir(project: Optional[str] = None) -> Path: if project: p = Path(project).resolve() if (p / ".automaton").exists() or p == AUTOMATON_DIR: return p print(f"WARNING: '{project}' has no .automaton/ directory. Tasks will be stored at {p / '.automaton' / 'tasks'}.", file=sys.stderr) return p cwd = Path.cwd().resolve() if cwd == AUTOMATON_DIR: return AUTOMATON_DIR if (cwd / ".automaton").exists(): return cwd if cwd.parent == AUTOMATON_DIR: return AUTOMATON_DIR print(f"ERROR: Not in an automaton project directory (cwd={cwd}). " f"Use --project to specify the project path, or run from a directory with .automaton/ or from ~/.automaton/.", file=sys.stderr) sys.exit(1) def _task_dir(task_name: str, project: Optional[str] = None) -> Path: project_dir = _find_project_dir(project) if project_dir == AUTOMATON_DIR: base = AUTOMATON_DIR / "tasks" else: base = project_dir / ".automaton" / "tasks" parts = task_name.split("/") if len(parts) > 1: parent = "/".join(parts[:-1]) return base / parent / "subtasks" / parts[-1] task_path = base / task_name if not task_path.exists(): completed = base / "complete" / task_name if completed.exists(): return completed return task_path def _all_task_dirs(project: Optional[str] = None) -> list[tuple[str, Path]]: project_dir = _find_project_dir(project) if project_dir == AUTOMATON_DIR: base = AUTOMATON_DIR / "tasks" else: base = project_dir / ".automaton" / "tasks" tasks = [] if not base.exists(): return tasks for entry in sorted(base.iterdir()): if entry.is_dir() and not entry.name.startswith(".") and entry.name != "complete": tasks.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("."): tasks.append((f"{entry.name}/{sub.name}", sub)) return tasks def _read_state(task_path: Path) -> Optional[str]: state_file = task_path / ".state" if state_file.exists(): content = state_file.read_text().strip() if content in VALID_PHASES: return content base = content.split(":")[0] if ":" in content else content if base in BASE_PHASES: return content return None def _write_state(task_path: Path, phase: str) -> None: tmp = task_path / ".state.tmp" tmp.write_text(f"{phase}\n") tmp.replace(task_path / ".state") def _read_state_approvals(task_path: Path) -> list[str]: approvals_file = task_path / ".state.approvals" if approvals_file.exists(): return approvals_file.read_text().strip().splitlines() return [] def _append_approval(task_path: Path, phase: str, approver: str) -> None: approvals_file = task_path / ".state.approvals" ts = datetime.now(timezone.utc).isoformat() line = f"{phase}|{ts}|{approver}\n" with open(approvals_file, "a") as f: f.write(line) def _parse_verdict_status_line(content: str) -> Optional[str]: """Parse verdict status from structured header lines only. Looks for ``## Status: PASS/FAIL/NEEDS_REVIEW`` or ``- **Status**: PASS/FAIL/NEEDS_REVIEW`` header lines. Returns None if no structured header is found. Deliberately does NOT do substring search across the full file, because body text may mention status keywords without reflecting the actual verdict. """ for line in content.splitlines(): stripped = line.strip() low = stripped.lower() if low.startswith("## status") or low.startswith("- **status**"): after_colon = stripped.split(":", 1)[-1].strip() if ":" in stripped else "" for label in ("PASS", "FAIL", "NEEDS_REVIEW"): if after_colon.upper() == label or label in after_colon.upper(): return label return None def _infer_state_from_artifacts(task_path: Path) -> Optional[str]: artifacts = {} for name in ["SPEC.md", "DECOMPOSITION.md", "DESIGN.md", "TEST_PLAN.md", "IMPLEMENTATION.md", "CODE_REVIEW.md", "BUG_REPORT.md", "ADVERSARIAL_BUG_REPORT.md", "DOC_REVIEW.md", "VERDICT.md"]: f = task_path / name if f.exists() and f.stat().st_size > 0: artifacts[name] = True if "VERDICT.md" in artifacts: content = (task_path / "VERDICT.md").read_text() verdict_status = _parse_verdict_status_line(content) if verdict_status == "PASS": return "complete" if verdict_status in ("FAIL", "NEEDS_REVIEW"): return "human_intervention" # Verdict exists but status header unparseable — fall back to human review return "human_intervention" if "DOC_REVIEW.md" in artifacts: return "referee" if "ADVERSARIAL_BUG_REPORT.md" in artifacts and "BUG_REPORT.md" in artifacts: return "doc_review" if "BUG_REPORT.md" in artifacts: return "adversarial_bug_find" if "CODE_REVIEW.md" in artifacts: return "bug_find" if "IMPLEMENTATION.md" in artifacts: return "code_review" if "TEST_PLAN.md" in artifacts: return "test_design" if "DESIGN.md" in artifacts: return "test_design" if "DECOMPOSITION.md" in artifacts and "SPEC.md" in artifacts: return "decomposition" if "SPEC.md" in artifacts: return "research" return "new" def _require_state(task_path: Path, task_name: str) -> Optional[str]: """Read .state file and refuse operations on tasks without one. Returns the phase string if .state exists, or prints an error and returns None. """ phase = _read_state(task_path) if phase is None: print(f"ERROR: Task '{task_name}' has no .state file. This task was likely created before v2.0 state enforcement.") print(f" Run: python ~/.automaton/scripts/status.py --upgrade --task {task_name} --project ") print(f" Or: python ~/.automaton/scripts/status.py --audit --project (to upgrade all tasks at once)") return None return phase def _is_kebab_case(name: str) -> bool: return bool(re.match(r'^[a-z0-9]+(-[a-z0-9]+)*$', name)) def _parse_agent_config(project: Optional[str] = None) -> dict: project_dir = _find_project_dir(project) agent_file = project_dir / ".automaton" / ".agent.md" if project_dir != AUTOMATON_DIR else AUTOMATON_DIR / ".agent.md" if not agent_file.exists(): agent_file = AUTOMATON_DIR / ".agent.md" config = {"mode": "single-agent", "agents": {}, "lock_timeout": "30m"} if not agent_file.exists(): return config content = agent_file.read_text() in_agent_section = False current_agent = None for line in content.splitlines(): stripped = line.strip() if stripped.startswith("## Agent Configuration"): in_agent_section = True continue if in_agent_section and stripped.startswith("## "): break if not in_agent_section: continue if stripped.lower().startswith("mode:"): config["mode"] = stripped.split(":", 1)[1].strip().lower() elif stripped.lower().startswith("lock timeout:"): config["lock_timeout"] = stripped.split(":", 1)[1].strip() elif stripped.startswith("- id:"): current_agent = stripped.split("id:")[1].strip() config["agents"][current_agent] = {"phases": [], "role": "worker"} elif current_agent and "phases:" in stripped.lower(): phases_str = stripped.split(":", 1)[1].strip() phases = [p.strip().strip("[]") for p in phases_str.split(",")] config["agents"][current_agent]["phases"] = phases elif current_agent and "role:" in stripped.lower(): config["agents"][current_agent]["role"] = stripped.split(":", 1)[1].strip() return config def _is_multi_agent(project: Optional[str] = None) -> bool: config = _parse_agent_config(project) return config.get("mode") == "multi-agent" def _lock_timeout_seconds(project: Optional[str] = None) -> int: config = _parse_agent_config(project) timeout_str = config.get("lock_timeout", "30m") match = re.match(r'(\d+)(m|h|s)', timeout_str) if not match: return 1800 val, unit = int(match.group(1)), match.group(2) if unit == "h": return val * 3600 if unit == "m": return val * 60 return val # --- Command implementations --- def cmd_show_task(args): task_path = _task_dir(args.task, args.project) if not task_path.exists(): print(f"ERROR: Task '{args.task}' not found in {task_path.parent}") return 2 phase = _require_state(task_path, args.task) if phase is None: return 1 base = _base_phase(phase) allowed = ALLOWED_ACTIONS_MAP.get(base, []) forbidden = FORBIDDEN_ACTIONS_MAP.get(base, []) next_phase = NEXT_PHASE_MAP.get(base, "—") next_artifact = PHASE_REQUIRED_ARTIFACTS.get(base, "—") print(f"Task: {args.task}") print(f"Phase: {phase} (from .state)") print(f"State file: {task_path / '.state'}") if phase in (f"{p}:awaiting_approval" for p in APPROVAL_PHASES): print(f"Approval: AWAITING — user sign-off required before proceeding") elif phase in (f"{p}:approved" for p in APPROVAL_PHASES): print(f"Approval: APPROVED — ready to transition to next phase") print(f"Allowed actions:") for a in allowed: print(f" - {a}") print(f"Forbidden actions:") for f in forbidden: print(f" - {f}") print(f"Next artifact needed: {next_artifact}") print(f"Next phase: {next_phase}") return 0 def cmd_list(args): tasks = _all_task_dirs(args.project) if not tasks: print("No tasks found.") return 0 print(f"{'Task':<35} {'Phase':<30} {'Next Step'}") print("-" * 80) has_untracked = False for name, path in tasks: phase = _read_state(path) if phase is None: print(f"{name:<35} {'UNTRACKED (no .state)':<30} Run --upgrade --task {name}") has_untracked = True continue base = _base_phase(phase) next_step = NEXT_PHASE_MAP.get(base, "—") print(f"{name:<35} {phase:<30} {next_step}") if has_untracked: print("\nNOTE: Tasks marked UNTRACKED were created before v2.0 state enforcement.") print(" Run --upgrade to bootstrap .state files, or --audit to see all violations.") return 0 def cmd_create_task(args): task_name = args.create_task if not _is_kebab_case(task_name): print(f"ERROR: Task name '{task_name}' must be kebab-case (lowercase, hyphens, no spaces)") return 2 task_path = _task_dir(task_name, args.project) if task_path.exists(): print(f"ERROR: Task '{task_name}' already exists at {task_path}") return 2 task_path.mkdir(parents=True) _write_state(task_path, "new") (task_path / ".state.approvals").write_text("") print(f"Created task '{task_name}' in state 'new'.") print(f"Use --transition research to begin.") return 0 def _auto_update_verdict_on_complete(task_path): """When transitioning human_intervention→complete, update VERDICT.md to PASS.""" verdict_file = task_path / "VERDICT.md" if not verdict_file.exists(): return import datetime content = verdict_file.read_text() lines = content.splitlines() new_lines = [] found_status = False for line in lines: low = line.strip().lower() if low.startswith("## status") or low.startswith("- **status**"): new_lines.append("## Status: PASS") found_status = True else: new_lines.append(line) if not found_status: new_lines.insert(0, "## Status: PASS") new_lines.append("") new_lines.append(f"*(Status auto-updated to PASS on human_intervention → complete transition at {datetime.datetime.now(datetime.timezone.utc).isoformat()})*") verdict_file.write_text("\n".join(new_lines) + "\n") def cmd_transition(args): task_path = _task_dir(args.task, args.project) if not task_path.exists(): print(f"ERROR: Task '{args.task}' not found in {task_path.parent}") return 2 current = _require_state(task_path, args.task) if current is None: return 1 # R8: refuse when a HALTED loop owns this task. A paused/running loop can # still have its task transitioned via the runner's own --transition calls # (which run as a subprocess and would just be re-checked against gates), # but a human must clear the halt first. owning_loop = _loop_owning_task(args.task, args.project) if owning_loop is not None: loop_state = _read_state_loop(owning_loop[1]) if loop_state and loop_state.get("status") == "halted": print(f"ERROR: Task '{args.task}' is owned by loop '{owning_loop[0]}' which is HALTED " f"(halt_reason={loop_state.get('halt_reason')}). Clear the halt with " f"`--approve --loop {owning_loop[0]}` before transitioning this task.") return 1 target = args.transition base_current = _base_phase(current) if target not in VALID_PHASES: print(f"ERROR: Unknown phase '{target}'. Valid phases: {', '.join(VALID_PHASES)}") return 2 allowed = LEGAL_TRANSITIONS.get(current, []) if target not in allowed: allowed_str = ", ".join(allowed) if allowed else "(no legal transitions)" print(f"ERROR: Cannot transition from '{current}' to '{target}'. Legal transitions from '{current}' are: {allowed_str}") return 1 if current.endswith(":awaiting_approval") and target != f"{base_current}:approved": print(f"ERROR: Cannot transition from '{current}' to '{target}'. Current phase is awaiting approval — use --approve to grant approval first.") return 1 for phase_base, artifact in PHASE_REQUIRED_ARTIFACTS.items(): if _base_phase(target) == phase_base or (target.startswith(phase_base + ":")): af = task_path / artifact if current != target and not af.exists() or af.exists() and af.stat().st_size == 0: pass forbidden_in_folder = _check_forbidden_artifacts(task_path, base_current) if forbidden_in_folder: print(f"ERROR: Cannot transition to {target} phase. Found out-of-order artifacts:") for art, belongs_to in forbidden_in_folder: print(f" - {art} (belongs to {belongs_to} phase)") print("Remove out-of-order artifacts before transitioning.") return 1 required = PHASE_REQUIRED_ARTIFACTS.get(base_current) if required and target != current: # only check required artifact when leaving a phase pass if base_current in PHASE_REQUIRED_ARTIFACTS and base_current != "new": req = PHASE_REQUIRED_ARTIFACTS.get(_base_phase(current)) if req: af = task_path / req if not af.exists() or af.stat().st_size == 0: print(f"ERROR: Cannot transition from '{current}' to '{target}'. Required artifact '{req}' is missing or empty in task folder.") return 1 if current == "human_intervention" and target == "complete": _auto_update_verdict_on_complete(task_path) if current == "implement" and target == "code_review": lock_file = task_path / ".state.lock" if lock_file.exists(): content = lock_file.read_text().strip() lines = dict(l.split(": ", 1) for l in content.splitlines() if ": " in l) implementer = lines.get("agent", "unknown") (task_path / ".state.implementer").write_text(f"{implementer}\n") if target == "complete": tasks_root = task_path.parent if task_path.name == "complete": tasks_root = tasks_root.parent complete_dir = tasks_root / "complete" complete_dir.mkdir(parents=True, exist_ok=True) dest = complete_dir / task_path.name if dest.exists(): print(f"ERROR: Completed task '{args.task}' already exists at {dest}.") return 1 task_path.rename(dest) _write_state(dest, target) print(f"Moved task directory to {dest}") else: _write_state(task_path, target) print(f"Transitioned task '{args.task}' from '{current}' to '{target}'.") return 0 def cmd_approve(args): task_path = _task_dir(args.task, args.project) if not task_path.exists(): print(f"ERROR: Task '{args.task}' not found in {task_path.parent}") return 2 current = _require_state(task_path, args.task) if current is None: return 1 base = _base_phase(current) if base not in APPROVAL_PHASES: print(f"This phase ({base}) does not require approval.") return 0 if not current.endswith(":awaiting_approval"): print(f"ERROR: Current phase is '{current}' (not awaiting approval). Current sub-state must be '{base}:awaiting_approval' before approval can be granted.") return 1 new_phase = f"{base}:approved" _write_state(task_path, new_phase) approver = args.agent if hasattr(args, "agent") and args.agent else "user" _append_approval(task_path, new_phase, approver) print(f"Approved task '{args.task}' — transitioned from '{current}' to '{new_phase}'.") print(f"Approval recorded by: {approver}") return 0 def cmd_validate_folder(args): task_path = _task_dir(args.task, args.project) if not task_path.exists(): print(f"ERROR: Task '{args.task}' not found in {task_path.parent}") return 2 state_file = task_path / ".state" if not state_file.exists(): phase = _infer_state_from_artifacts(task_path) if phase: print(f"Task: {args.task}") print(f"Phase: {phase} (inferred from artifacts — no .state file)") print(f"Folder validation: WARN — task has no .state file (pre-v2.0 task).") print(f" Run 'python ~/.automaton/scripts/status.py --upgrade --task {args.task} --project ' to bootstrap .state file.") return 1 else: print(f"Task: {args.task}") print(f"Phase: unknown (no .state file and no artifacts)") print(f"Folder validation: FAIL — task has no .state file.") print(f" Run 'python ~/.automaton/scripts/status.py --upgrade --task {args.task} --project ' to bootstrap .state file.") return 1 phase = _read_state(task_path) if phase is None: print(f"Task: {args.task}") print(f"Phase: unknown (corrupted .state file)") print(f"Folder validation: FAIL — .state file exists but is empty or unreadable.") print(f" Run 'python ~/.automaton/scripts/status.py --upgrade --task {args.task} --project ' to re-bootstrap .state file.") return 1 base = _base_phase(phase) forbidden = _check_forbidden_artifacts(task_path, base) print(f"Task: {args.task}") print(f"Phase: {phase} (from .state)") if forbidden: print(f"Folder validation: FAIL — found out-of-order artifacts:") for art, belongs_to in forbidden: print(f" - {art} (belongs to {belongs_to} phase, not yet reached)") print("These artifacts indicate phase-skipping. Remove them or revert to the correct phase.") return 1 print(f"Folder validation: PASS — no out-of-order artifacts found") return 0 def _check_forbidden_artifacts(task_path: Path, phase: str) -> list[tuple[str, str]]: """Returns list of (artifact_name, phase_it_belongs_to) for forbidden artifacts found.""" forbidden_names = FORBIDDEN_ARTIFACTS.get(phase, []) artifact_to_phase = { "SPEC.md": "research", "DECOMPOSITION.md": "decomposition", "DESIGN.md": "design", "TEST_PLAN.md": "test_design", "IMPLEMENTATION.md": "implement", "CODE_REVIEW.md": "code_review", "BUG_REPORT.md": "bug_find", "ADVERSARIAL_BUG_REPORT.md": "adversarial_bug_find", "DOC_REVIEW.md": "doc_review", "VERDICT.md": "referee", } found = [] for name in forbidden_names: f = task_path / name if f.exists() and f.stat().st_size > 0: belongs_to = artifact_to_phase.get(name, "unknown") found.append((name, belongs_to)) return found def _audit_category3_paths(project_dir, tasks): """Return changed paths outside task folders (data-only sibling of _audit_category3; used by --audit --json).""" import subprocess try: result = subprocess.run( ["git", "diff", "--name-only", "HEAD"], capture_output=True, text=True, cwd=str(project_dir), timeout=10 ) uncommitted = [f.strip() for f in result.stdout.splitlines() if f.strip()] except Exception: return [] try: result = subprocess.run( ["git", "diff", "--cached", "--name-only", "HEAD"], capture_output=True, text=True, cwd=str(project_dir), timeout=10 ) staged = [f.strip() for f in result.stdout.splitlines() if f.strip()] except Exception: staged = [] all_changed = set(uncommitted + staged) if not all_changed: return [] task_names = {name for name, _ in tasks} unauthorized = [] for changed_file in all_changed: parts = Path(changed_file).parts in_task = (len(parts) >= 2 and parts[0] == "tasks" and parts[1] in task_names) if not in_task: in_task = (len(parts) >= 3 and parts[0] == ".automaton" and parts[1] == "tasks" and parts[2] in task_names) if not in_task: unauthorized.append(changed_file) return sorted(unauthorized) def _audit_category3(project_dir, tasks): """Audit Category 3: Git-based unauthorized modification detection.""" import subprocess # Collect active edit-allowed tasks active_edit_tasks = set() for name, path in tasks: phase = _read_state(path) if phase and _base_phase(phase) in ("implement", "doc_review"): active_edit_tasks.add(name) # Get uncommitted changes (working tree + staged) try: result = subprocess.run( ["git", "diff", "--name-only", "HEAD"], capture_output=True, text=True, cwd=str(project_dir), timeout=10 ) uncommitted = [f.strip() for f in result.stdout.splitlines() if f.strip()] except Exception as e: print(f"[WARN] Failed to check git diff: {e}") return 0 try: result = subprocess.run( ["git", "diff", "--cached", "--name-only", "HEAD"], capture_output=True, text=True, cwd=str(project_dir), timeout=10 ) staged = [f.strip() for f in result.stdout.splitlines() if f.strip()] except Exception: staged = [] all_changed = set(uncommitted + staged) violations = 0 if not all_changed: print("[PASS] No uncommitted modifications detected") return violations # Exclude files inside task folders task_names = {name for name, _ in tasks} unauthorized = set() for changed_file in all_changed: parts = Path(changed_file).parts # Framework mode: paths like "tasks/mytask/..." is_in_task_folder = ( len(parts) >= 2 and parts[0] == "tasks" and parts[1] in task_names ) # Regular project mode: paths like ".automaton/tasks/mytask/..." if not is_in_task_folder: is_in_task_folder = ( len(parts) >= 3 and parts[0] == ".automaton" and parts[1] == "tasks" and parts[2] in task_names ) if not is_in_task_folder: unauthorized.add(changed_file) if not unauthorized: print("[PASS] All uncommitted changes are within task folders — no unauthorized modifications") return violations if not active_edit_tasks: print("[FAIL] No task in implement or doc_review phase, but uncommitted changes exist outside task folders:") for f in sorted(unauthorized): print(f" - {f}") violations += 1 print(f" To allow edits: create a task and transition to implement phase") else: print(f"[INFO] Active edit tasks: {', '.join(sorted(active_edit_tasks))}") print("[INFO] Uncommitted changes outside task folders exist (may be authorized if within active task scope):") for f in sorted(unauthorized): print(f" - {f}") return violations def _audit_loops_block(args) -> int: """R9: print the Loops audit block and return the violation count.""" loops = _all_loop_dirs(args.project) if not loops: print("[INFO] No loops found.") return 0 import time as _ltime now = _ltime.time() v = 0 for name, path in loops: state = _read_state_loop(path) if state is None: print(f"[FAIL] {name}: no .state.loop (UNTRACKED loop). Run --create-loop {name} or future --upgrade-loops.") v += 1 continue status = state.get("status", "unknown") if status == "halted": print(f"[WARN] {name}: HALTED (reason={state.get('halt_reason')}). " f"Clear with: --approve --loop {name}") v += 1 elif status == "paused": print(f"[INFO] {name}: paused. Resume with --resume-loop {name}") elif status == "running": last_tick = state.get("last_tick_at") if last_tick: try: age_min = (now - datetime.fromisoformat(last_tick).timestamp()) / 60 if age_min > 120: print(f"[WARN] {name}: running but last tick was {age_min:.0f} min ago (stale).") v += 1 else: print(f"[PASS] {name}: running (last tick {age_min:.0f} min ago).") except (ValueError, TypeError): print(f"[PASS] {name}: running (unparseable last_tick_at).") else: print(f"[PASS] {name}: running (no ticks yet).") current_task = state.get("current_task") if current_task: if not _task_dir(current_task, args.project).exists(): print(f"[FAIL] {name}: current_task '{current_task}' does not exist.") v += 1 return v def _audit_collect(args): """Pure data collection used by both --audit (human) and --audit --json. Returns a dict: {"violations": [...], "loops": [...], "total_tasks": int, "untracked_tasks": int, "human_violation_count": int} Each violation: {"category": int, "severity": "high"|"med"|"low", "task": str|None, "message": str, "resolved": False} Each loop entry: {"name": str, "status": str, "halt_reason": str|None, "current_task": str|None, "violation": bool, "message": str} """ project_dir = _find_project_dir(args.project) tasks = _all_task_dirs(args.project) violations = [] loops = [] untracked = 0 for name, path in tasks: state_file = path / ".state" if not state_file.exists(): untracked += 1 phase = _infer_state_from_artifacts(path) or "unknown" violations.append({"category": 4, "severity": "high", "task": name, "message": f"no .state file (manually created or pre-v2.0 task) - inferred phase: {phase}", "resolved": False}) continue phase = _read_state(path) or _infer_state_from_artifacts(path) or "unknown" base = _base_phase(phase) forbidden = _check_forbidden_artifacts(path, base) if forbidden: items_desc = ", ".join(f"{a} ({p} phase artifact)" for a, p in forbidden) violations.append({"category": 1, "severity": "high", "task": name, "message": f"out-of-order artifacts: {items_desc} (phase: {phase})", "resolved": False}) expected = PHASE_REQUIRED_ARTIFACTS.get(base) if expected: af = path / expected if base == "implement" and (path / "IMPLEMENTATION.md").exists() and (path / "IMPLEMENTATION.md").stat().st_size == 0: violations.append({"category": 2, "severity": "med", "task": name, "message": f"IMPLEMENTATION.md is empty but .state says {base}", "resolved": False}) elif base in ("bug_find", "adversarial_bug_find", "code_review", "doc_review", "referee") and not (path / "IMPLEMENTATION.md").exists(): violations.append({"category": 2, "severity": "high", "task": name, "message": f".state says {base} but IMPLEMENTATION.md is missing", "resolved": False}) git_dir = project_dir / ".git" if git_dir.exists(): cat3_paths = _audit_category3_paths(project_dir, tasks) for path in cat3_paths: violations.append({"category": 3, "severity": "med", "task": None, "message": f"uncommitted change outside task folders: {path}", "resolved": False}) import time as _time stuck_threshold = 60 for name, path in tasks: state_file = path / ".state" if not state_file.exists(): continue phase = _read_state(path) if phase is None or phase in ("complete", "human_intervention"): continue mtime = state_file.stat().st_mtime age_minutes = (_time.time() - mtime) / 60 if age_minutes > stuck_threshold: violations.append({"category": 5, "severity": "med", "task": name, "message": f"stuck at '{phase}' for {age_minutes:.0f} minutes (threshold: {stuck_threshold} min)", "resolved": False}) loop_dirs = _all_loop_dirs(args.project) for lname, lpath in loop_dirs: lstate = _read_state_loop(lpath) if lstate is None: loops.append({"name": lname, "status": "untracked", "halt_reason": None, "current_task": None, "violation": True, "message": "no .state.loop (UNTRACKED loop)"}) continue lstatus = lstate.get("status", "unknown") lhalt = lstate.get("halt_reason") ltask = lstate.get("current_task") is_violation = False msg = "" if lstatus == "halted": is_violation = True msg = f"HALTED (reason={lhalt})" elif lstatus == "paused": msg = "paused" elif lstatus == "running": last_tick = lstate.get("last_tick_at") if last_tick: try: age_min = (_time.time() - datetime.fromisoformat(last_tick).timestamp()) / 60 if age_min > 120: is_violation = True msg = f"running but last tick was {age_min:.0f} min ago (stale)" else: msg = f"running (last tick {age_min:.0f} min ago)" except (ValueError, TypeError): msg = "running (unparseable last_tick_at)" else: msg = "running (no ticks yet)" if ltask and not _task_dir(ltask, args.project).exists(): is_violation = True msg = f"current_task '{ltask}' does not exist" loops.append({"name": lname, "status": lstatus, "halt_reason": lhalt, "current_task": ltask, "violation": is_violation, "message": msg}) return {"violations": violations, "loops": loops, "total_tasks": len(tasks), "untracked_tasks": untracked} def cmd_audit(args): project_dir = _find_project_dir(args.project) tasks = _all_task_dirs(args.project) json_mode = bool(getattr(args, "json_output", False)) if json_mode: data = _audit_collect(args) print(json.dumps(data)) return 1 if data["violations"] else 0 if not tasks: # Still audit loops even when no tasks exist (R9). print("=== Category 6: Loops ===") violations = _audit_loops_block(args) print("\n=== Summary ===") print("0 tasks audited") if violations: print(f"{violations} violation(s) found") return 1 print("No violations found") return 0 violations = 0 print(f"Audit Report for {project_dir}\n") cat1_violations = [] cat2_violations = [] cat4_violations = [] for name, path in tasks: state_file = path / ".state" if not state_file.exists(): phase = _infer_state_from_artifacts(path) cat4_violations.append((name, phase or "unknown")) continue phase = _read_state(path) if phase is None: phase = _infer_state_from_artifacts(path) if phase is None: phase = "unknown" base = _base_phase(phase) forbidden = _check_forbidden_artifacts(path, base) if forbidden: cat1_violations.append((name, phase, forbidden)) expected = PHASE_REQUIRED_ARTIFACTS.get(base) if expected: af = path / expected inconsistency = False details = [] if base == "implement" and (path / "IMPLEMENTATION.md").exists() and (path / "IMPLEMENTATION.md").stat().st_size == 0: inconsistency = True details.append(f"IMPLEMENTATION.md is empty but .state says {base}") if base in ("bug_find", "adversarial_bug_find", "code_review", "doc_review", "referee") and not (path / "IMPLEMENTATION.md").exists(): inconsistency = True details.append(f".state says {base} but IMPLEMENTATION.md is missing") if inconsistency: cat2_violations.append((name, phase, details)) print("=== Category 1: Out-of-order Artifacts ===") if not cat1_violations: for name, path in tasks: phase = _read_state(path) or _infer_state_from_artifacts(path) or "unknown" print(f"[PASS] {name}: no violations") else: for name, path in tasks: phase = _read_state(path) or _infer_state_from_artifacts(path) or "unknown" found = [(n, p) for n, ph, items in cat1_violations if n == name for n, p in items] if any(n == name for n, _, _ in cat1_violations): phase_for_name = next(ph for n, ph, _ in cat1_violations if n == name) items = next(items for n, ph, items in cat1_violations if n == name) print(f"[FAIL] {name} (phase: {phase_for_name}): {', '.join(f'{a} ({p} phase artifact)' for a, p in items)}") violations += 1 else: print(f"[PASS] {name}: no violations") print("\n=== Category 2: State-Artifact Inconsistency ===") for name, path in tasks: state_file = path / ".state" if not state_file.exists(): phase = _infer_state_from_artifacts(path) or "unknown" else: phase = _read_state(path) or _infer_state_from_artifacts(path) or "unknown" inconsistencies = [d for n, ph, d in cat2_violations if n == name] if inconsistencies: for detail in inconsistencies[0]: print(f"[WARN] {name}: {detail}") violations += 1 else: base = _base_phase(phase) expected = PHASE_REQUIRED_ARTIFACTS.get(base) if expected: af = path / expected if af.exists() and af.stat().st_size > 0: print(f"[PASS] {name}: .state ({phase}) matches artifacts ({expected} exists)") else: print(f"[INFO] {name}: .state ({phase}) - expected artifact {expected} not yet produced") else: print(f"[PASS] {name}: .state ({phase}) - no artifact requirement for this phase") print("\n=== Category 4: Manually Created Tasks ===") if not cat4_violations: for name, path in tasks: if (path / ".state").exists(): print(f"[PASS] {name}: has .state file") else: for name, inferred_phase in cat4_violations: print(f"[FAIL] {name}: no .state file (manually created or pre-v2.0 task) - inferred phase: {inferred_phase}") print(f" Run 'python ~/.automaton/scripts/status.py --upgrade --task {name} --project ' to bootstrap .state file") violations += 1 print("\n=== Category 3: Unauthorized Modifications ===") git_dir = project_dir / ".git" if not git_dir.exists(): print("Skipped: not a git repository") else: violations += _audit_category3(project_dir, tasks) print("\n=== Category 5: Stuck Tasks ===") import time as _time stuck_threshold = 60 stuck_found = 0 for name, path in tasks: state_file = path / ".state" if not state_file.exists(): continue phase = _read_state(path) if phase is None or phase in ("complete", "human_intervention"): continue mtime = state_file.stat().st_mtime age_minutes = (_time.time() - mtime) / 60 if age_minutes > stuck_threshold: print(f"[WARN] {name}: stuck at '{phase}' for {age_minutes:.0f} minutes (threshold: {stuck_threshold} min)") stuck_found += 1 violations += 1 if stuck_found == 0: print(f"[PASS] No stuck tasks (threshold: {stuck_threshold} min)") print("\n=== Category 6: Loops ===") violations += _audit_loops_block(args) print(f"\n=== Summary ===") total = len(tasks) print(f"{total} tasks audited") if violations: print(f"{violations} violation(s) found") return 1 print("No violations found") return 0 def cmd_upgrade(args): if args.task: task_path = _task_dir(args.task, args.project) if not task_path.exists(): print(f"ERROR: Task '{args.task}' not found in {task_path.parent}") return 2 state_file = task_path / ".state" if state_file.exists(): current = _read_state(task_path) print(f"Task '{args.task}' already has .state file (phase: {current})") return 0 phase = _infer_state_from_artifacts(task_path) if phase: _write_state(task_path, phase) approvals_file = task_path / ".state.approvals" if not approvals_file.exists(): approvals_file.write_text("") print(f"Bootstrapped .state for task '{args.task}': phase '{phase}' (inferred from artifacts)") return 0 else: _write_state(task_path, "new") approvals_file = task_path / ".state.approvals" if not approvals_file.exists(): approvals_file.write_text("") print(f"Bootstrapped .state for task '{args.task}': phase 'new' (no artifacts found)") return 0 else: tasks = _all_task_dirs(args.project) if not tasks: print("No tasks found to upgrade.") return 0 bootstrapped = 0 skipped = 0 for name, path in tasks: state_file = path / ".state" if state_file.exists(): skipped += 1 continue phase = _infer_state_from_artifacts(path) if phase: _write_state(path, phase) approvals_file = path / ".state.approvals" if not approvals_file.exists(): approvals_file.write_text("") print(f" {name}: bootstrapped as '{phase}'") bootstrapped += 1 else: _write_state(path, "new") approvals_file = path / ".state.approvals" if not approvals_file.exists(): approvals_file.write_text("") print(f" {name}: bootstrapped as 'new' (no artifacts found)") bootstrapped += 1 if (path / "subtasks").exists(): for sub_dir in sorted((path / "subtasks").iterdir()): if not sub_dir.is_dir(): continue sub_state = sub_dir / ".state" if sub_state.exists(): continue sub_phase = _infer_state_from_artifacts(sub_dir) if sub_phase: _write_state(sub_dir, sub_phase) sub_approvals = sub_dir / ".state.approvals" if not sub_approvals.exists(): sub_approvals.write_text("") print(f" {name}/subtasks/{sub_dir.name}: bootstrapped as '{sub_phase}'") else: _write_state(sub_dir, "new") sub_approvals = sub_dir / ".state.approvals" if not sub_approvals.exists(): sub_approvals.write_text("") print(f" {name}/subtasks/{sub_dir.name}: bootstrapped as 'new'") print(f"\nUpgraded: {bootstrapped}, Skipped (already had .state): {skipped}") return 0 def _get_edit_timestamp(task_path: Path) -> float: """Return the mtime to use for stale-task detection. Uses ``.state.lastedit`` if it exists (updated by --can-edit on ALLOWED). Falls back to ``.state`` mtime for backward compatibility. """ lastedit = task_path / ".state.lastedit" if lastedit.exists(): try: return lastedit.stat().st_mtime except OSError: pass state_file = task_path / ".state" if state_file.exists(): try: return state_file.stat().st_mtime except OSError: pass return 0 def _touch_lastedit(task_path: Path) -> None: """Update .state.lastedit timestamp to mark edit activity.""" lastedit = task_path / ".state.lastedit" try: lastedit.touch() except OSError: pass def cmd_can_edit(args): project_dir = _find_project_dir(args.project) if getattr(args, "loop", None): return cmd_can_edit_loop(args) if not args.task: edit_tasks = [] tasks = _all_task_dirs(args.project) for name, path in tasks: phase = _read_state(path) if phase is None: continue base = _base_phase(phase) if base in ("implement", "doc_review"): edit_mtime = _get_edit_timestamp(path) edit_tasks.append((name, base, path, edit_mtime)) if not edit_tasks: print("DENIED: No tasks in implement or doc_review phase. Create a task and transition it to implement before editing files.") if args.json_output: print(json.dumps({"allowed": False, "reason": "no_edit_tasks", "tasks": []})) return 1 if args.file: file_path = Path(args.file).resolve() proj_str = str(project_dir.resolve()) scope_tasks = [] out_of_scope = [] for name, base, path, state_mtime in edit_tasks: if str(file_path).startswith(proj_str + os.sep) or str(file_path) == proj_str: scope_tasks.append({"task": name, "phase": base, "state_mtime": state_mtime, "path": path}) else: out_of_scope.append({"task": name, "phase": base, "file": str(file_path)}) if not scope_tasks: print(f"DENIED: File '{file_path}' is outside project '{project_dir}'. No task allows editing this file.") if args.json_output: print(json.dumps({"allowed": False, "reason": "out_of_scope", "out_of_scope": out_of_scope, "tasks": []})) return 1 primary = scope_tasks[0] import time as _time now = _time.time() max_state_age = max(t["state_mtime"] for t in scope_tasks) age_minutes = (now - max_state_age) / 60 if age_minutes > 30: latest_task = max(scope_tasks, key=lambda t: t["state_mtime"]) print(f"DENIED: Task '{latest_task['task']}' has been in {latest_task['phase']} phase for {age_minutes:.0f} minutes (stale). Create a new task for new work.") if args.json_output: print(json.dumps({"allowed": False, "reason": "stale_task", "stale_task": latest_task["task"], "stale_minutes": round(age_minutes), "all_edit_tasks": scope_tasks})) return 1 print(f"ALLOWED: Task '{primary['task']}' is in {primary['phase']} phase and file '{file_path}' is within project '{project_dir}'.") if args.json_output: print(json.dumps({"allowed": True, "reason": "edit_task_in_scope", "primary_task": {"task": primary["task"], "phase": primary["phase"]}, "all_edit_tasks": [{"task": t["task"], "phase": t["phase"]} for t in scope_tasks]})) _touch_lastedit(primary["path"]) return 0 import time as _time now = _time.time() max_state_age = max(mtime for _, _, _, mtime in edit_tasks) age_minutes = (now - max_state_age) / 60 latest = max(edit_tasks, key=lambda t: t[3]) if age_minutes > 30: print(f"DENIED: Task '{latest[0]}' has been in {latest[1]} phase for {age_minutes:.0f} minutes (stale). Create a new task for new work.") if args.json_output: print(json.dumps({"allowed": False, "reason": "stale_task", "stale_task": latest[0], "stale_minutes": round(age_minutes), "all_edit_tasks": [{"task": n, "phase": b} for n, b, _, _ in edit_tasks]})) return 1 primary = edit_tasks[0] print(f"ALLOWED: Task '{primary[0]}' is in {primary[1]} phase — code edits are permitted.") if args.json_output: print(json.dumps({"allowed": True, "reason": "edit_task", "primary_task": {"task": primary[0], "phase": primary[1]}, "all_edit_tasks": [{"task": n, "phase": b} for n, b, _, _ in edit_tasks]})) _touch_lastedit(primary[2]) return 0 task_path = _task_dir(args.task, args.project) if not task_path.exists(): print(f"ERROR: Task '{args.task}' not found") return 2 phase = _require_state(task_path, args.task) if phase is None: return 1 base = _base_phase(phase) if args.file: file_path = Path(args.file).resolve() proj_str = str(project_dir.resolve()) auto_str = str(AUTOMATON_DIR) if project_dir == AUTOMATON_DIR: if not (str(file_path).startswith(auto_str + os.sep) or str(file_path) == auto_str): print(f"OUT_OF_SCOPE: File '{file_path}' is outside the framework directory") if args.json_output: print(json.dumps({"allowed": False, "reason": "out_of_scope", "task": args.task, "phase": base})) return 1 else: if not (str(file_path).startswith(proj_str + os.sep) or str(file_path) == proj_str): print(f"OUT_OF_SCOPE: File '{file_path}' is outside project '{project_dir}'. Only framework project can modify framework files.") if args.json_output: print(json.dumps({"allowed": False, "reason": "out_of_scope", "task": args.task, "phase": base, "file": str(file_path)})) return 1 if base in ("implement", "doc_review"): import time as _time edit_mtime = _get_edit_timestamp(task_path) now = _time.time() age_minutes = (now - edit_mtime) / 60 if age_minutes > 30: print(f"DENIED: Task '{args.task}' has been in {base} phase for {age_minutes:.0f} minutes (stale). Create a new task for new work.") if args.json_output: print(json.dumps({"allowed": False, "reason": "stale_task", "stale_task": args.task, "stale_minutes": round(age_minutes)})) return 1 print(f"ALLOWED: Task '{args.task}' is in {base} phase — code edits are permitted.") if args.json_output: print(json.dumps({"allowed": True, "reason": "edit_phase", "task": args.task, "phase": base})) _touch_lastedit(task_path) return 0 print(f"DENIED: Task '{args.task}' is in {base} phase. Code edits require implement or doc_review phase.") if args.json_output: print(json.dumps({"allowed": False, "reason": "wrong_phase", "task": args.task, "phase": base, "allowed_phases": ["implement", "doc_review"]})) return 1 def cmd_touch(args): """Update .state.lastedit to reset stale-task timer without changing phase.""" task_path = _task_dir(args.task, args.project) if not task_path.exists(): print(f"ERROR: Task '{args.task}' not found") return 2 state_file = task_path / ".state" if not state_file.exists(): print(f"ERROR: Task '{args.task}' has no .state file. Run --upgrade first.") return 1 _touch_lastedit(task_path) phase = _read_state(task_path) print(f"Touched task '{args.task}' (phase: {phase}) — activity clock reset.") return 0 def cmd_scope_check(args): project_dir = _find_project_dir(args.project) file_path = Path(args.file).resolve() proj_str = str(project_dir.resolve()) if str(file_path).startswith(proj_str + os.sep) or str(file_path) == proj_str: print(f"IN_SCOPE: File '{file_path}' is within project '{project_dir}'") return 0 if project_dir != AUTOMATON_DIR: auto_str = str(AUTOMATON_DIR) if str(file_path).startswith(auto_str + os.sep) or str(file_path) == auto_str: print(f"OUT_OF_SCOPE: File '{file_path}' is in the framework directory, but current project is '{project_dir}'. Only framework project can modify framework files.") return 1 print(f"OUT_OF_SCOPE: File '{file_path}' is outside project '{project_dir}'") return 1 def cmd_same_session(args): task_path = _task_dir(args.task, args.project) if not task_path.exists(): print(f"ERROR: Task '{args.task}' not found") return 2 state_file = task_path / ".state" if not state_file.exists(): print(f"DIFFERENT_SESSION: Task '{args.task}' has no .state file") return 0 import time mtime = _get_edit_timestamp(task_path) age_minutes = (time.time() - mtime) / 60 threshold = 30 if age_minutes < threshold: print(f"SAME_SESSION: Task '{args.task}' last edit activity {age_minutes:.0f} minutes ago (threshold: {threshold} min)") return 1 print(f"DIFFERENT_SESSION: Task '{args.task}' last edit activity {age_minutes:.0f} minutes ago (threshold: {threshold} min)") return 0 def cmd_claim(args): task_path = _task_dir(args.task, args.project) if not task_path.exists(): print(f"ERROR: Task '{args.task}' not found in {task_path.parent}") return 2 config = _parse_agent_config(args.project) if config["mode"] != "multi-agent": print(f"Claimed task '{args.task}' (single-agent mode — no lock needed)") return 0 if not args.agent: print("ERROR: --agent is required in multi-agent mode") return 2 agent_phases = config["agents"].get(args.agent, {}).get("phases", []) if not agent_phases: print(f"ERROR: Agent '{args.agent}' not found in Agent Configuration") return 2 phase = _require_state(task_path, args.task) if phase is None: return 1 base = _base_phase(phase) if "*" not in agent_phases and base not in agent_phases: print(f"ERROR: Agent '{args.agent}' is not configured for phase '{base}'. Allowed phases: {', '.join(agent_phases)}") return 1 if base == "code_review": implementer_file = task_path / ".state.implementer" if implementer_file.exists(): implementer = implementer_file.read_text().strip() if implementer == args.agent: print(f"ERROR: Agent '{args.agent}' implemented this task and cannot claim the code_review phase. Reviewer must be different from implementer.") return 1 lock_file = task_path / ".state.lock" timeout_sec = _lock_timeout_seconds(args.project) if lock_file.exists(): content = lock_file.read_text().strip() lines = dict(l.split(": ", 1) for l in content.splitlines() if ": " in l) existing_agent = lines.get("agent", "unknown") expires_str = lines.get("expires", "") if expires_str: try: from datetime import datetime as dt expires = dt.fromisoformat(expires_str.replace("Z", "+00:00")) if datetime.now(timezone.utc) < expires: print(f"ERROR: Task '{args.task}' is claimed by agent '{existing_agent}' (expires: {expires_str}). Retry after expiry or release the claim.") return 1 else: print(f"WARN: Task '{args.task}' had stale lock from agent '{existing_agent}' (expired: {expires_str}). Overclaiming for agent '{args.agent}'.") except Exception: pass from datetime import datetime as dt, timedelta now = datetime.now(timezone.utc) expires = now + timedelta(seconds=timeout_sec) lock_content = f"agent: {args.agent}\nphase: {phase}\nclaimed: {now.isoformat()}\nexpires: {expires.isoformat()}\n" tmp = task_path / ".state.lock.tmp" tmp.write_text(lock_content) tmp.replace(task_path / ".state.lock") print(f"Claimed task '{args.task}' for agent '{args.agent}' — phase: {phase}") print(f"Lock expires: {expires.isoformat()}") return 0 def cmd_release(args): task_path = _task_dir(args.task, args.project) if not task_path.exists(): print(f"ERROR: Task '{args.task}' not found in {task_path.parent}") return 2 config = _parse_agent_config(args.project) if config["mode"] != "multi-agent": print(f"Released task '{args.task}' (single-agent mode — no lock to release)") return 0 if not args.agent: print("ERROR: --agent is required in multi-agent mode") return 2 lock_file = task_path / ".state.lock" if not lock_file.exists(): print(f"WARN: Task '{args.task}' has no lock. Nothing to release.") return 0 content = lock_file.read_text().strip() lines = dict(l.split(": ", 1) for l in content.splitlines() if ": " in l) existing_agent = lines.get("agent", "unknown") if existing_agent != args.agent: print(f"ERROR: Task '{args.task}' is claimed by agent '{existing_agent}', not '{args.agent}'. Only the claiming agent can release.") return 1 lock_file.unlink() print(f"Released task '{args.task}' from agent '{args.agent}'") return 0 def cmd_next_available(args): config = _parse_agent_config(args.project) if config["mode"] != "multi-agent": print("Single-agent mode — use --list to see all tasks") return 0 if not args.agent: print("ERROR: --agent is required in multi-agent mode") return 2 agent_phases = config["agents"].get(args.agent, {}).get("phases", []) if not agent_phases: print(f"ERROR: Agent '{args.agent}' not found in Agent Configuration") return 2 tasks = _all_task_dirs(args.project) candidates = [] for name, path in tasks: phase = _read_state(path) if phase is None: continue base = _base_phase(phase) if phase in ("complete", "human_intervention"): continue if "*" not in agent_phases and base not in agent_phases: continue lock_file = path / ".state.lock" if lock_file.exists(): content = lock_file.read_text().strip() lines = dict(l.split(": ", 1) for l in content.splitlines() if ": " in l) expires_str = lines.get("expires", "") if expires_str: try: from datetime import datetime as dt expires = dt.fromisoformat(expires_str.replace("Z", "+00:00")) if datetime.now(timezone.utc) < expires: continue except Exception: pass priority = PHASE_PRIORITY.get(base, 0) candidates.append((name, base, phase, priority)) if not candidates: print(f"No tasks available for agent '{args.agent}'. All tasks are claimed by other agents, in phases this agent cannot work on, or completed.") return 0 candidates.sort(key=lambda x: -x[3]) best = candidates[0] print(f"Next available task for agent '{args.agent}':") print(f"Task: {best[0]}") print(f"Phase: {best[2]}") print(f"Phase priority: {best[3]} ({'high — close to completion' if best[3] >= 7 else 'medium' if best[3] >= 4 else 'low'})") print(f"Status: unclaimed") print(f"\nTo claim: python ~/.automaton/scripts/status.py --claim --task {best[0]} --agent {args.agent}") return 0 def cmd_available(args): config = _parse_agent_config(args.project) if config["mode"] != "multi-agent": print("Single-agent mode — use --list to see all tasks") return 0 if not args.agent: print("ERROR: --agent is required in multi-agent mode") return 2 agent_phases = config["agents"].get(args.agent, {}).get("phases", []) if not agent_phases: print(f"ERROR: Agent '{args.agent}' not found in Agent Configuration") return 2 tasks = _all_task_dirs(args.project) print(f"Available tasks for agent '{args.agent}':") idx = 1 for name, path in tasks: phase = _read_state(path) if phase is None: continue base = _base_phase(phase) if phase in ("complete", "human_intervention"): continue if "*" not in agent_phases and base not in agent_phases: continue lock_file = path / ".state.lock" status = "unclaimed" if lock_file.exists(): content = lock_file.read_text().strip() lines = dict(l.split(": ", 1) for l in content.splitlines() if ": " in l) existing_agent = lines.get("agent", "unknown") expires_str = lines.get("expires", "") status = f"claimed by '{existing_agent}' (expires: {expires_str})" try: from datetime import datetime as dt expires = dt.fromisoformat(expires_str.replace("Z", "+00:00")) if datetime.now(timezone.utc) >= expires: status = "expired lock — available" except Exception: pass print(f"{idx}. Task: {name} | Phase: {phase} | Status: {status}") idx += 1 if idx == 1: print("No tasks available.") return 0 def cmd_list_states(args): """Print all valid phase names.""" print("Valid phases:") for phase in VALID_PHASES: base = _base_phase(phase) approvals = " * requires approval" if base in APPROVAL_PHASES and phase == base else "" print(f" {phase}{approvals}") return 0 # ============================================================================ # Loop management (v1 — task add-status-brakes) # # All loop-aware behavior lives inside status.py so harnesses cannot route # around it. Loops are stored per-project under /.automaton/loops/ # /. The .state.loop file is the single source of truth for runtime # state. Loops without .state.loop are UNTRACKED. See design/loops/technical.md. # ============================================================================ LOOP_STATES = ("running", "halted", "paused", "complete") LOOP_HALTS = ( "iterations_exhausted", "budget_exhausted", "verifier_failed", "drift_detected", "human_intervention", ) LOOP_STATE_SCHEMA_VERSION = 1 LOOP_TICK_LOG_NAME = ".state.log" LOOP_STATE_FILE = ".state.loop" LOOP_CONFIG_FILE = "loop.json" LOOP_WORKTREE_DIR = "worktree" LOOP_TICK_SCRIPT_SH = "automaton-loop-tick.sh" LOOP_TICK_SCRIPT_BAT = "automaton-loop-tick.bat" LOOP_CONTEXT_FLOOR_KB = 16_000 # mirrors vram_detect.LOOP_MODE_CONTEXT_FLOOR_KB CLEANUP_DEFAULT_DAYS = 7 CLEANUP_DEFAULT_INTERVAL_SEC = 86400 # daily CLEANUP_TICK_SCRIPT_SH = "automaton-cleanup.sh" CLEANUP_TICK_SCRIPT_BAT = "automaton-cleanup.bat" CLEANUP_LABEL = "cleanup" # launchd label: com.automaton.cleanup def _completed_task_age_days(task_path: Path) -> Optional[float]: """Days since the task last transitioned to its current phase, measured by the ``.state`` file mtime (the moment ``_write_state`` recorded completion). Returns None if ``.state`` is missing or unreadable.""" state_file = task_path / ".state" if not state_file.exists(): return None try: mtime = state_file.stat().st_mtime except OSError: return None return (datetime.now(timezone.utc).timestamp() - mtime) / 86400.0 def _launchagents_dir() -> Path: """LaunchAgents directory, overridable via AUTOMATON_LAUNCHAGENTS_DIR so tests can redirect plist writes to a temp dir without touching the host.""" override = os.environ.get("AUTOMATON_LAUNCHAGENTS_DIR") if override: return Path(override) return Path.home() / "Library" / "LaunchAgents" def _tasks_root_for(task_path: Path) -> Path: """Return the tasks directory that should hold the ``complete/`` archive for this task (handles subtask paths).""" parent = task_path.parent if parent.name == "complete": return parent.parent return parent def _loops_dir(project: Optional[str] = None) -> Path: project_dir = _find_project_dir(project) if project_dir == AUTOMATON_DIR: return AUTOMATON_DIR / "loops" return project_dir / ".automaton" / "loops" def _loop_dir(name: str, project: Optional[str] = None) -> Path: if not _is_kebab_case(name): raise ValueError(f"loop name '{name}' must be kebab-case") return _loops_dir(project) / name def _all_loop_dirs(project: Optional[str] = None) -> list[tuple[str, Path]]: base = _loops_dir(project) out: list[tuple[str, Path]] = [] if not base.exists(): return out for entry in sorted(base.iterdir()): if entry.is_dir() and not entry.name.startswith("."): out.append((entry.name, entry)) return out def _read_state_loop(loop_path: Path) -> Optional[dict]: state_file = loop_path / LOOP_STATE_FILE if not state_file.exists(): return None try: return json.loads(state_file.read_text()) except (OSError, json.JSONDecodeError): return None def _write_state_loop(loop_path: Path, state: dict) -> None: state_file = loop_path / LOOP_STATE_FILE tmp = loop_path / ".state.loop.tmp" tmp.write_text(json.dumps(state, indent=2, sort_keys=True) + "\n") tmp.replace(state_file) def _initial_state_loop(name: str) -> dict: return { "schema_version": LOOP_STATE_SCHEMA_VERSION, "name": name, "status": "running", "halt_reason": None, "iteration_count": 0, "resumed_count": 0, "last_tick_at": None, "last_verdict": None, "score_history": [], "current_task": None, "worktree_branch": None, "worktree_path": None, } def _read_loop_config(loop_path: Path) -> Optional[dict]: cfg = loop_path / LOOP_CONFIG_FILE if not cfg.exists(): return None try: return json.loads(cfg.read_text()) except (OSError, json.JSONDecodeError): return None def _append_tick_log(loop_path: Path, line: str) -> None: log = loop_path / LOOP_TICK_LOG_NAME ts = datetime.now(timezone.utc).isoformat() with log.open("a", encoding="utf-8") as fh: fh.write(f"[{ts}] {line}\n") def _loop_owning_task(task_name: str, project: Optional[str] = None) -> Optional[tuple[str, Path]]: """Scan all loops for one whose current_task == task_name. Returns (name, path) or None.""" for name, path in _all_loop_dirs(project): state = _read_state_loop(path) if state and state.get("current_task") == task_name: return (name, path) return None def _loop_untracked_hint(name: str) -> str: return (f"ERROR: loop '{name}' has no {LOOP_STATE_FILE} (UNTRACKED). " f"All --loop commands refuse. Run --create-loop (or future " f"--upgrade-loops) to bootstrap.") def _halt_loop(loop_path: Path, state: dict, reason: str, project: Optional[str]) -> None: state["status"] = "halted" state["halt_reason"] = reason _write_state_loop(loop_path, state) _append_tick_log(loop_path, f"HALT reason={reason}") # Best-effort schedule disable — never fatal if it fails (no cron/plist on # this platform, etc.). The .state.loop is the source of truth; the OS unit # will read the state on next wake and self-skip. try: _disable_schedule(loop_path.name, project) except Exception: pass def _claim_loop_task_impl(name: str, task_name: str, project: Optional[str]) -> int: """Implement --claim-loop-task. Scans all loops to verify no OTHER loop already owns task_name. If self owns it → exit 0 (idempotent). If other loop owns it → prints 'task_already_claimed:{other}' to stderr, exit 2. If nobody owns it → sets state['current_task'] = task_name, writes .state.loop, exit 0. If loop is untracked → prints 'loop_untracked' to stderr, exit 2. Must be called inside the claiming loop's _loop_lock to serialize writes. """ loop_path = _loop_dir(name, project) state = _read_state_loop(loop_path) if state is None: print(_loop_untracked_hint(name), file=sys.stderr) return 2 # Self-ownership check (idempotent) if state.get("current_task") == task_name: print("already_self_claimed") return 0 # Cross-loop scan: check no OTHER running/paused loop owns this task for other_name, other_path in _all_loop_dirs(project): if other_name == name: continue other_state = _read_state_loop(other_path) if other_state is None: continue other_status = other_state.get("status") if other_status not in ("running", "paused"): continue if other_state.get("current_task") == task_name: print(f"task_already_claimed:{other_name}", file=sys.stderr) return 2 # Nobody owns it → claim it state["current_task"] = task_name _write_state_loop(loop_path, state) print("OK") return 0 def cmd_claim_loop_task(args) -> int: name = args.claim_loop_task task_name = args.task if not task_name: print("ERROR: --task is required for --claim-loop-task", file=sys.stderr) return 2 if not name: print("ERROR: --claim-loop-task requires a loop name", file=sys.stderr) return 2 loop_path = _loop_dir(name, args.project) state = _read_state_loop(loop_path) if state is None: print(_loop_untracked_hint(name), file=sys.stderr) return 2 with _loop_lock(loop_path): return _claim_loop_task_impl(name, task_name, args.project) _LOOP_LOCK_ENV_BYPASS = "AUTOMATON_NO_LOOP_LOCK" @contextlib.contextmanager def _loop_lock(loop_path: Path, exclusive: bool = True): """Cross-process file lock on /.state.lock. Used to serialize read-modify-write cycles on `.state.loop` across concurrent ticks (two scheduler firings, or a tick vs `--approve` or `--pause-loop`). Blocking acquire; no timeout (operators notice a wedged tick via `--loop-list` stale `last_tick_at`). Per-loop granularity: the `.state.lock` file lives in the loop's own dir, not the framework root. NOT re-entrant across processes. When the runner invokes `status.py --check-gate` as a subprocess while holding the parent's lock, the subprocess would otherwise deadlock waiting on the same flock. To avoid this, the runner sets $AUTOMATON_NO_LOOP_LOCK=1 in the subprocess environment; `_loop_lock` sees that env var and becomes a no-op (`yield` without flock), trusting the caller's outer lock to cover the critical section. Manual CLI invocations don't set the env var, so they lock normally. NFS caveat: `flock` semantics differ on NFS-mounted loops. The loop dir is documented to be local (project root or `~/.automaton`). """ if os.environ.get(_LOOP_LOCK_ENV_BYPASS) == "1": yield return lock_file = loop_path / ".state.lock" fd = os.open(str(lock_file), os.O_RDWR | os.O_CREAT, 0o644) acquired = False try: if sys.platform == "win32": import msvcrt msvcrt.locking(fd, msvcrt.LK_LOCK if exclusive else msvcrt.LK_NBLCK, 1) else: import fcntl fcntl.flock(fd, fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH) acquired = True yield finally: if acquired: if sys.platform == "win32": import msvcrt try: msvcrt.locking(fd, msvcrt.LK_UNLCK, 1) except OSError: pass else: import fcntl fcntl.flock(fd, fcntl.LOCK_UN) os.close(fd) def _disable_schedule(name: str, project: Optional[str]) -> None: """Best-effort schedule disable. Implemented per R6 — see _install_schedule.""" loop_path = _loop_dir(name, project) system = platform.system() if system == "Darwin": plist = _launchagents_dir() / f"com.automaton.loop.{name}.plist" if plist.exists(): plist.rename(plist.with_suffix(".plist.disabled")) try: subprocess.run(["launchctl", "unload", str(plist)], capture_output=True, timeout=5, check=False) except (OSError, subprocess.SubprocessError): pass elif system == "Linux": try: res = subprocess.run(["crontab", "-l"], capture_output=True, text=True, timeout=5, check=False) lines = res.stdout.splitlines() if res.returncode == 0 else [] kept = [] inside_block = False for line in lines: if line.strip() == f"# automaton-loop:{name}": inside_block = True continue if inside_block and line.strip().startswith("# end automaton-loop:"): inside_block = False continue if not inside_block: kept.append(line) subprocess.run(["crontab", "-"], input="\n".join(kept) + "\n", capture_output=True, text=True, timeout=5, check=False) except (OSError, subprocess.SubprocessError): pass elif system == "Windows": subprocess.run(["schtasks", "/end", "/tn", f"AutomatonLoop_{name}"], capture_output=True, timeout=5, check=False) def _install_cron_block_generic(marker: str, stub_path: Path, interval: int) -> int: """Install (or re-install) an idempotent cron block delimited by ``# `` / ``# end ``. Strips any prior block with the same marker before appending the new one. Returns 0 on success, 2 on failure.""" minutes_interval = max(1, interval // 60) try: res = subprocess.run(["crontab", "-l"], capture_output=True, text=True, timeout=5, check=False) existing = res.stdout.splitlines() if res.returncode == 0 else [] except (OSError, subprocess.SubprocessError): existing = [] kept = [] inside = False for line in existing: if line.strip() == f"# {marker}": inside = True continue if inside and line.strip() == f"# end {marker}": inside = False continue if not inside: kept.append(line) kept.append(f"# {marker}") kept.append(f"*/{minutes_interval} * * * * {stub_path}") kept.append(f"# end {marker}") try: subprocess.run(["crontab", "-"], input="\n".join(kept) + "\n", capture_output=True, text=True, timeout=5, check=False) except (OSError, subprocess.SubprocessError) as exc: print(f"ERROR: failed to write crontab: {exc}") return 2 return 0 def _install_cron_block(name: str, loop_path: Path, interval: int) -> int: """Install (or re-install) the cron block for a loop. Idempotent: strips any prior block for the same loop name before appending the new one. Returns 0 on success, 2 on crontab write failure. """ stub_path = loop_path / LOOP_TICK_SCRIPT_SH return _install_cron_block_generic(f"automaton-loop:{name}", stub_path, interval) def _enable_schedule(name: str, project: Optional[str]) -> None: """Inverse of _disable_schedule. Called from --resume-loop after pause clear.""" loop_path = _loop_dir(name, project) system = platform.system() if system == "Darwin": plist = _launchagents_dir() / f"com.automaton.loop.{name}.plist" disabled = plist.with_suffix(".plist.disabled") if disabled.exists() and not plist.exists(): disabled.rename(plist) elif system == "Linux": stub = loop_path / LOOP_TICK_SCRIPT_SH if not stub.exists(): print(f"WARNING: cannot re-enable schedule for '{name}': " f"no tick stub at {stub}; run --install-schedule first.", file=sys.stderr) return cfg = _read_loop_config(loop_path) or {} raw_interval = cfg.get("schedule", {}).get("interval_seconds", 3600) try: interval = int(raw_interval) except (TypeError, ValueError): interval = 3600 _install_cron_block(name, loop_path, interval) elif system == "Windows": subprocess.run(["schtasks", "/run", "/tn", f"AutomatonLoop_{name}"], capture_output=True, timeout=5, check=False) # --------------------------------------------------------------------------- # cmd_* functions for loop commands # --------------------------------------------------------------------------- def cmd_create_loop(args) -> int: name = args.create_loop template = args.from_template if not _is_kebab_case(name): print(f"ERROR: loop name '{name}' must be kebab-case") return 2 templates_dir = AUTOMATON_DIR / "templates" / "loops" template_dir = templates_dir / template if not template_dir.exists() or not (template_dir / "loop.json").exists(): print(f"ERROR: template '{template}' not found at {template_dir}/loop.json") return 2 loop_path = _loop_dir(name, args.project) if loop_path.exists(): print(f"ERROR: loop '{name}' already exists at {loop_path}") return 2 loop_path.mkdir(parents=True) cfg_src = template_dir / "loop.json" cfg_dst = loop_path / LOOP_CONFIG_FILE cfg_dst.write_text(cfg_src.read_text()) # Patch the "name" field to match the actual loop name (templates use a placeholder). try: cfg = json.loads(cfg_dst.read_text()) if "name" not in cfg or cfg.get("name") != name: cfg["name"] = name cfg_dst.write_text(json.dumps(cfg, indent=2) + "\n") except json.JSONDecodeError: pass _write_state_loop(loop_path, _initial_state_loop(name)) (loop_path / LOOP_TICK_LOG_NAME).write_text("") print(f"Created loop '{name}' from template '{template}'.") print(f"State: running. Install schedule with: --install-schedule {name}") return 0 def cmd_install_schedule(args) -> int: name = args.install_schedule loop_path = _loop_dir(name, args.project) state = _read_state_loop(loop_path) if state is None: print(_loop_untracked_hint(name)) return 2 cfg = _read_loop_config(loop_path) if cfg is None: print(f"ERROR: loop '{name}' has no {LOOP_CONFIG_FILE}; cannot install schedule") return 2 interval = args.interval if interval is None: interval = int(cfg.get("schedule", {}).get("interval_seconds", 3600)) project_dir = _find_project_dir(args.project) framework_dir = AUTOMATON_DIR runner = framework_dir / "scripts" / "loop-runner.py" project_root_str = str(project_dir) framework_str = str(framework_dir) # Generate the OS-unit tick stub (locked by technical.md §6). if platform.system() == "Windows": stub_path = loop_path / LOOP_TICK_SCRIPT_BAT stub = ( "@echo off\r\n" f"cd /d \"{project_root_str}\"\r\n" f"python \"{runner}\" --mode tick --loop \"{name}\"\r\n" ) else: stub_path = loop_path / LOOP_TICK_SCRIPT_SH stub = ( "#!/usr/bin/env bash\n" f"cd \"{project_root_str}\"\n" f"python3 \"{runner}\" --mode tick --loop \"{name}\"\n" ) stub_path.write_text(stub) if platform.system() != "Windows": stub_path.chmod(0o755) system = platform.system() if system == "Darwin": plist_dir = _launchagents_dir() plist_dir.mkdir(parents=True, exist_ok=True) plist_path = plist_dir / f"com.automaton.loop.{name}.plist" plist = ( f"\n" f"\n" f"\n" f"\n" f" Labelcom.automaton.loop.{name}\n" f" ProgramArguments\n" f" \n" f" {stub_path}\n" f" \n" f" StartInterval{interval}\n" f" RunAtLoad\n" f"\n" f"\n" ) plist_path.write_text(plist) print(f"Installed launchd unit: {plist_path}") print(f"Interval: {interval}s. Tick stub: {stub_path}") return 0 elif system == "Linux": rc = _install_cron_block(name, loop_path, interval) if rc == 0: minutes_interval = max(1, interval // 60) print(f"Installed crontab block (every {minutes_interval} min). Tick stub: {stub_path}") return rc elif system == "Windows": minutes_interval = max(1, interval // 60) subprocess.run( ["schtasks", "/create", "/tn", f"AutomatonLoop_{name}", "/tr", str(stub_path), "/sc", "minute", "/mo", str(minutes_interval), "/f"], capture_output=True, timeout=10, check=False, ) print(f"Installed schtasks unit (every {minutes_interval} min). Tick stub: {stub_path}") return 0 else: print(f"ERROR: unsupported platform '{system}' for --install-schedule") return 2 def cmd_cleanup_done(args) -> int: """--cleanup-done [--days N] [--dry-run] [--json] Move tasks whose ``.state`` base phase is ``complete`` and whose completion age (measured by ``.state`` mtime) is >= --days into the ``tasks/complete/`` archive folder. Archived tasks are excluded from ``--list``/``--audit`` and from the dashboard. Idempotent. ``--dry-run`` lists candidates without moving. """ days = args.days if args.days is not None else CLEANUP_DEFAULT_DAYS if days < 0: print("ERROR: --days must be non-negative") return 2 tasks = _all_task_dirs(args.project) moved = [] skipped = [] candidates = [] for name, path in tasks: phase = _read_state(path) if phase is None: continue if _base_phase(phase) != "complete": continue age = _completed_task_age_days(path) if age is None: continue if age < days: continue candidates.append((name, path, age)) tasks_root = _tasks_root_for(path) complete_dir = tasks_root / "complete" dest = complete_dir / name if dest.exists(): skipped.append({"name": name, "reason": "dest_exists", "age_days": round(age, 2)}) continue if args.dry_run: moved.append({"name": name, "dest": str(dest), "age_days": round(age, 2), "dry_run": True}) continue complete_dir.mkdir(parents=True, exist_ok=True) try: path.rename(dest) except OSError as exc: skipped.append({"name": name, "reason": f"rename_failed: {exc}", "age_days": round(age, 2)}) continue moved.append({"name": name, "dest": str(dest), "age_days": round(age, 2)}) verb = "would move" if args.dry_run else "moved" for item in moved: print(f"[{verb}] {item['name']} (age {item['age_days']}d) -> {item['dest']}") for item in skipped: print(f"[skipped] {item['name']} ({item['reason']}, age {item['age_days']}d)") if args.json_output: summary = { "days": days, "dry_run": bool(args.dry_run), "moved": [m["name"] for m in moved], "skipped": skipped, "candidates": [c[0] for c in candidates], } print(json.dumps(summary)) return 0 def _cleanup_stub_path() -> Path: return AUTOMATON_DIR / "scripts" / "automaton-cleanup.sh" def cmd_install_cleanup_schedule(args) -> int: """--install-cleanup-schedule [--days N] [--interval SEC] Install an OS scheduler unit that runs ``--cleanup-done`` periodically (default daily). Mirrors ``--install-schedule`` but for the cleanup job, which is framework-wide (project = the framework dir).""" days = args.days if args.days is not None else CLEANUP_DEFAULT_DAYS interval = args.interval if args.interval is not None else CLEANUP_DEFAULT_INTERVAL_SEC project_dir = _find_project_dir(args.project) status_script = AUTOMATON_DIR / "scripts" / "status.py" project_arg = f"--project \"{project_dir}\"" if str(project_dir) != str(AUTOMATON_DIR) else "" if platform.system() == "Windows": stub_path = AUTOMATON_DIR / "scripts" / "automaton-cleanup.bat" stub = ( "@echo off\r\n" f"python \"{status_script}\" --cleanup-done --days {days} {project_arg}\r\n" ) else: stub_path = _cleanup_stub_path() stub = ( "#!/usr/bin/env bash\n" f"python3 \"{status_script}\" --cleanup-done --days {days}" + (f" {project_arg}\n" if project_arg else "\n") ) stub_path.write_text(stub) if platform.system() != "Windows": stub_path.chmod(0o755) system = platform.system() if system == "Darwin": plist_dir = _launchagents_dir() plist_dir.mkdir(parents=True, exist_ok=True) plist_path = plist_dir / f"com.automaton.cleanup.plist" plist = ( f"\n" f"\n" f"\n" f"\n" f" Labelcom.automaton.cleanup\n" f" ProgramArguments\n" f" \n" f" {stub_path}\n" f" \n" f" StartInterval{interval}\n" f" RunAtLoad\n" f"\n" f"\n" ) plist_path.write_text(plist) print(f"Installed launchd unit: {plist_path}") print(f"Interval: {interval}s, cleanup threshold: {days}d. Stub: {stub_path}") return 0 elif system == "Linux": rc = _install_cron_block_generic("automaton-cleanup", stub_path, interval) if rc == 0: print(f"Installed crontab block (every {max(1, interval // 60)} min), " f"threshold {days}d. Stub: {stub_path}") return rc elif system == "Windows": minutes_interval = max(1, interval // 60) subprocess.run( ["schtasks", "/create", "/tn", "AutomatonCleanup", "/tr", str(stub_path), "/sc", "minute", "/mo", str(minutes_interval), "/f"], capture_output=True, timeout=10, check=False, ) print(f"Installed schtasks unit (every {minutes_interval} min). Stub: {stub_path}") return 0 else: print(f"ERROR: unsupported platform '{system}' for --install-cleanup-schedule") return 2 def _parse_launchd_plist(plist_path: Path) -> dict: """Minimal plist extractor for ProgramArguments, StartInterval, Label.""" import plistlib try: with plist_path.open("rb") as fh: data = plistlib.load(fh) return data except (OSError, ValueError): return {} def _read_loop_runtime_state(name: str, project: Optional[str]) -> Optional[dict]: try: loop_path = _loop_dir(name, project) except ValueError: return None return _read_state_loop(loop_path) def _list_scheduled_jobs(project: Optional[str] = None) -> list[dict]: """Enumerate installed scheduled jobs (loops + cleanup) for the dashboard 'Background' tab. Each entry: ``{label, kind, status, next_run_seconds, stub_path, runtime_state?}``. ``status`` ∈ {enabled, disabled, misconfigured}.""" jobs: list[dict] = [] system = platform.system() def _launchd_label_kind(label: str) -> tuple[str, str]: # returns (kind, name_or_empty) if label.startswith("com.automaton.loop."): return ("loop", label[len("com.automaton.loop."):]) if label in ("com.automaton.cleanup",): return ("cleanup", "") if label.startswith("com.automaton.cleanup."): return ("cleanup", label[len("com.automaton.cleanup."):]) return ("unknown", label) if system == "Darwin" or system == "Windows": # On Windows we cannot easily enumerate schtasks entries here; on Darwin # we scan LaunchAgents. (Windows listing is best-effort left to the # dashboard surfacing 'unknown' for the tab.) if system == "Darwin": plist_dir = _launchagents_dir() if plist_dir.exists(): for entry in sorted(plist_dir.iterdir()): if not entry.name.startswith("com.automaton."): continue disabled = entry.name.endswith(".plist.disabled") real = entry.with_suffix("") if disabled else entry label = real.stem # filename without .plist data = _parse_launchd_plist(real) if real.exists() else {} interval = data.get("StartInterval") stub = (data.get("ProgramArguments") or [None])[0] kind, sub = _launchd_label_kind(label) runtime = None agent_type = None if kind == "loop" and sub: s = _read_loop_runtime_state(sub, project) runtime = s loop_dir = _loop_dir(sub, project) cfg = _read_loop_config(loop_dir) or {} ws = cfg.get("work_source") or {} agent_type = ws.get("kind", "single") status = "misconfigured" if not real.exists() else ( "disabled" if disabled else "enabled") # flag stub pointing at a missing path if stub and not Path(stub).exists(): status = "misconfigured" jobs.append({ "label": label, "kind": kind, "name": sub, "status": status, "next_run_seconds": interval, "stub_path": stub, "runtime_state": runtime, "agent_type": agent_type, }) if system == "Linux": try: res = subprocess.run(["crontab", "-l"], capture_output=True, text=True, timeout=5, check=False) lines = res.stdout.splitlines() if res.returncode == 0 else [] except (OSError, subprocess.SubprocessError): lines = [] marker = None for line in lines: s = line.strip() if s.startswith("# automaton-loop:") or s.startswith("#automaton-loop:"): marker = s.lstrip("#").strip() name = marker.split("automaton-loop:")[1] cfg = _read_loop_config(_loop_dir(name, project)) or {} ws = cfg.get("work_source") or {} agent_type = ws.get("kind", "single") jobs.append({ "label": f"automaton-loop:{name}", "kind": "loop", "name": name, "status": "enabled", "next_run_seconds": None, "stub_path": None, "runtime_state": _read_loop_runtime_state(name, project), "agent_type": agent_type, }) elif s in ("# automaton-cleanup", "#automaton-cleanup"): jobs.append({ "label": "automaton-cleanup", "kind": "cleanup", "name": "", "status": "enabled", "next_run_seconds": None, "stub_path": None, "runtime_state": None, }) return jobs def cmd_pause_loop(args) -> int: name = args.pause_loop loop_path = _loop_dir(name, args.project) state = _read_state_loop(loop_path) if state is None: print(_loop_untracked_hint(name)) return 2 # Serialize read-modify-write against concurrent ticks / approve. with _loop_lock(loop_path): state = _read_state_loop(loop_path) if state is None: print(_loop_untracked_hint(name)) return 2 if state["status"] != "running" and state["status"] != "paused": print(f"ERROR: loop '{name}' is in status '{state['status']}', " f"cannot pause. Only 'running' can be paused.") return 1 state["status"] = "paused" _write_state_loop(loop_path, state) _append_tick_log(loop_path, "PAUSED by user") _disable_schedule(name, args.project) print(f"Paused loop '{name}'. Schedule disabled. Resume with --resume-loop {name}") return 0 def cmd_resume_loop(args) -> int: name = args.resume_loop loop_path = _loop_dir(name, args.project) state = _read_state_loop(loop_path) if state is None: print(_loop_untracked_hint(name)) return 2 with _loop_lock(loop_path): state = _read_state_loop(loop_path) if state is None: print(_loop_untracked_hint(name)) return 2 if state["status"] == "halted": print(f"ERROR: loop '{name}' is HALTED (halt_reason={state['halt_reason']}). " f"--resume-loop cannot clear halts. Use: --approve --loop {name}") return 1 if state["status"] != "paused": print(f"ERROR: loop '{name}' is in status '{state['status']}', " f"cannot resume. Only 'paused' can be resumed.") return 1 state["status"] = "running" _write_state_loop(loop_path, state) _append_tick_log(loop_path, "RESUMED by user") _enable_schedule(name, args.project) print(f"Resumed loop '{name}'. Schedule re-enabled.") return 0 def cmd_approve_loop(args) -> int: """--approve --loop — the only way to clear a halt. (D4)""" name = args.loop loop_path = _loop_dir(name, args.project) state = _read_state_loop(loop_path) if state is None: print(_loop_untracked_hint(name)) return 2 with _loop_lock(loop_path): state = _read_state_loop(loop_path) if state is None: print(_loop_untracked_hint(name)) return 2 if state["status"] != "halted": print(f"ERROR: loop '{name}' is in status '{state['status']}', not 'halted'. " f"--approve --loop only clears halts.") return 1 state["status"] = "running" state["halt_reason"] = None state["resumed_count"] = int(state.get("resumed_count", 0)) + 1 _write_state_loop(loop_path, state) _append_tick_log(loop_path, f"APPROVED by user (resumed_count={state['resumed_count']})") _enable_schedule(name, args.project) print(f"Approved loop '{name}'. Halt cleared. Resumed count: {state['resumed_count']}") return 0 def _gate_loop_status(loop_path: Path, state: dict, cfg: dict) -> Optional[dict]: if state["status"] != "running": return { "ok": False, "reason": f"{state['status']}:{state.get('halt_reason') or ''}".rstrip(":"), "halt_reason": state.get("halt_reason"), "remaining_iterations": max(0, _loop_max_iterations(cfg) - int(state.get("iteration_count", 0))), "remaining_budget_usd": None, "task_phase": None, "task_in_halt_loop": state["status"] == "halted", "out_of_scope_files": [], } return None # running — no failure on this check def _loop_max_iterations(cfg: Optional[dict]) -> int: if cfg is None: return 0 return int(cfg.get("brakes", {}).get("max_iterations", 0)) def _gate_iterations(state: dict, cfg: dict) -> Optional[dict]: max_iter = _loop_max_iterations(cfg) if max_iter <= 0: return None # no cap configured remaining = max_iter - int(state.get("iteration_count", 0)) if remaining <= 0: return { "ok": False, "reason": "halted:iterations_exhausted", "halt_reason": "iterations_exhausted", "remaining_iterations": 0, "remaining_budget_usd": None, "task_phase": None, "task_in_halt_loop": True, "out_of_scope_files": [], } return None def _gate_budget(state: dict, cfg: dict, loop_path: Path) -> Optional[dict]: max_budget = cfg.get("brakes", {}).get("max_budget_usd") if max_budget is None: return None # informational only, remote-only per D3 # Best-effort: read from a cost.json the harness may have written. cost_file = loop_path / "cost.json" if cost_file.exists(): try: spent = json.loads(cost_file.read_text()).get("spent_usd", 0.0) except (OSError, json.JSONDecodeError): spent = 0.0 else: spent = 0.0 if spent >= float(max_budget): return { "ok": False, "reason": "halted:budget_exhausted", "halt_reason": "budget_exhausted", "remaining_iterations": None, "remaining_budget_usd": max(0.0, float(max_budget) - spent), "task_phase": None, "task_in_halt_loop": True, "out_of_scope_files": [], } return None def _gate_task_phase(state: dict, cfg: dict, project: Optional[str]) -> Optional[dict]: current_task = state.get("current_task") if not current_task: return None task_path = _task_dir(current_task, project) if not task_path.exists(): return { "ok": False, "reason": "halted:human_intervention", "halt_reason": "human_intervention", "remaining_iterations": None, "remaining_budget_usd": None, "task_phase": None, "task_in_halt_loop": True, "out_of_scope_files": [], } task_state = _read_state(task_path) if task_state == "human_intervention": return { "ok": False, "reason": "halted:human_intervention", "halt_reason": "human_intervention", "remaining_iterations": None, "remaining_budget_usd": None, "task_phase": task_state, "task_in_halt_loop": True, "out_of_scope_files": [], } return None def _base_branch(cfg: Optional[dict]) -> str: if cfg is None: return "main" br = (cfg.get("blast_radius") or {}).get("base_branch") if br is None: return "main" if not isinstance(br, str): print(f"WARNING: blast_radius.base_branch={br!r} is not a string; " f"coercing to {str(br)!r}", file=sys.stderr) return str(br) if br.strip() == "": print(f"WARNING: blast_radius.base_branch is empty; " f"falling back to 'main'", file=sys.stderr) return "main" return br def _gate_worktree_drift(state: dict, cfg: dict, project: Optional[str]) -> Optional[dict]: if not state.get("worktree_path"): return None # --no-worktree mode — drift check N/A worktree_path = Path(state["worktree_path"]) if not worktree_path.exists(): return None # worktree missing — runner will recreate; treat as no-drift file_scope = cfg.get("blast_radius", {}).get("file_scope", []) if not file_scope: return None base = _base_branch(cfg) try: res = subprocess.run( ["git", "diff", "--name-only", f"{base}...HEAD"], cwd=worktree_path, capture_output=True, text=True, timeout=10, check=False, ) if res.returncode != 0: print(f"WARNING: could not run git diff for drift check: {res.stderr.strip()}", file=sys.stderr) return None # no-git environment — skip with warning, not halt changed = [l.strip() for l in res.stdout.splitlines() if l.strip()] except (OSError, subprocess.SubprocessError) as exc: print(f"WARNING: drift check subprocess failed: {exc}", file=sys.stderr) return None out_of_scope: list[str] = [] for f in changed: if not any(f.startswith(scope) for scope in file_scope): out_of_scope.append(f) if out_of_scope: return { "ok": False, "reason": "halted:drift_detected", "halt_reason": "drift_detected", "remaining_iterations": None, "remaining_budget_usd": None, "task_phase": None, "task_in_halt_loop": True, "out_of_scope_files": out_of_scope, } return None def _gate_score_plateau(state: dict, cfg: dict) -> Optional[dict]: window = int(cfg.get("brakes", {}).get("score_plateau_window", 0)) if window <= 0: return None history = list(state.get("score_history", []))[-window:] if len(history) < window: return None # Flat or monotonically non-increasing across the window. if all(s <= history[0] for s in history[1:]): return { "ok": False, "reason": "halted:verifier_failed", "halt_reason": "verifier_failed", "remaining_iterations": None, "remaining_budget_usd": None, "task_phase": None, "task_in_halt_loop": True, "out_of_scope_files": [], } return None def cmd_check_gate(args) -> int: name = args.check_gate loop_path = _loop_dir(name, args.project) state = _read_state_loop(loop_path) if state is None: print(_loop_untracked_hint(name)) return 2 cfg = _read_loop_config(loop_path) or {} # Serialize read-evaluate-(maybe halt-write) against concurrent ticks / # --pause / --approve. When invoked as a subprocess by the loop runner, # the runner sets $AUTOMATON_NO_LOOP_LOCK=1 and `_loop_lock` becomes a # no-op (the runner's outer `with _loop_lock` covers the full tick). When # invoked standalone (manual CLI), env var is unset and this acquires the # lock normally. Either way the critical section below is serialized. with _loop_lock(loop_path): state = _read_state_loop(loop_path) if state is None: print(_loop_untracked_hint(name)) return 2 gates = [ _gate_loop_status(loop_path, state, cfg), _gate_iterations(state, cfg), _gate_budget(state, cfg, loop_path), _gate_task_phase(state, cfg, args.project), _gate_worktree_drift(state, cfg, args.project), _gate_score_plateau(state, cfg), ] failure = next((g for g in gates if g is not None), None) if failure is None: out = { "ok": True, "reason": "running", "halt_reason": None, "remaining_iterations": max(0, _loop_max_iterations(cfg) - int(state.get("iteration_count", 0))), "remaining_budget_usd": cfg.get("brakes", {}).get("max_budget_usd"), "task_phase": _task_phase_for_loop(state, args.project), "task_in_halt_loop": False, "out_of_scope_files": [], } verdict = out else: verdict = failure # Atomically halt the loop for halts that are non-transient (verifier # plateau, drift). iterations_exhausted / budget_exhausted already # fired via _gate_iterations/_gate_budget returning the failure dict; # mark halted consistently. if state["status"] != "halted" and failure.get("halt_reason"): _halt_loop(loop_path, state, failure["halt_reason"], args.project) # refresh state from disk after halt state = _read_state_loop(loop_path) or state if args.json_output: print(json.dumps(verdict)) else: print(f"loop: {name}") print(f"ok: {verdict['ok']}") print(f"reason: {verdict['reason']}") if verdict.get("halt_reason"): print(f"halt_reason: {verdict['halt_reason']}") print(f"remaining_iterations: {verdict.get('remaining_iterations')}") print(f"remaining_budget_usd: {verdict.get('remaining_budget_usd')}") print(f"task_phase: {verdict.get('task_phase')}") print(f"task_in_halt_loop: {verdict.get('task_in_halt_loop')}") if verdict.get("out_of_scope_files"): print(f"out_of_scope_files: {verdict['out_of_scope_files']}") return 0 if verdict["ok"] else 1 def _task_phase_for_loop(state: dict, project: Optional[str]) -> Optional[str]: current_task = state.get("current_task") if not current_task: return None task_path = _task_dir(current_task, project) return _read_state(task_path) def cmd_can_continue(args) -> int: name = args.can_continue loop_path = _loop_dir(name, args.project) state = _read_state_loop(loop_path) if state is None: print(_loop_untracked_hint(name)) return 2 out = { "ok": state["status"] == "running", "status": state["status"], "halt_reason": state.get("halt_reason"), "name": name, } if args.json_output: print(json.dumps(out)) else: print(f"loop: {name}") print(f"status: {state['status']}") print(f"ok: {out['ok']}") if state.get("halt_reason"): print(f"halt_reason: {state['halt_reason']}") return 0 if out["ok"] else 1 def cmd_loop_list(args) -> int: loops = _all_loop_dirs(args.project) if not loops: print("No loops found.") return 0 print(f"{'NAME':<30} {'STATUS':<10} {'HALT_REASON':<24} {'TICKS':<6} {'RESUMES':<8}") print("-" * 80) for name, path in loops: state = _read_state_loop(path) or {} print(f"{name:<30} {state.get('status', 'UNTRACKED'):<10} " f"{str(state.get('halt_reason') or '-'):<24} " f"{int(state.get('iteration_count', 0)):<6} " f"{int(state.get('resumed_count', 0)):<8}") return 0 def cmd_version(args) -> int: """Print framework version from config.md's ## Framework Version section. Exit 0 always (POSIX convention for --version).""" config_path = AUTOMATON_DIR / "config.md" version = None if config_path.exists(): try: text = config_path.read_text(encoding="utf-8") except OSError: text = "" in_section = False for raw in text.splitlines(): line = raw.strip() if line.startswith("## Framework Version"): in_section = True continue if in_section and line.startswith("##"): in_section = False if not in_section: continue m = re.search(r"version\**\s*[:=]\s*\**\s*([0-9][\w.\-]*)", line, re.IGNORECASE) if m: version = m.group(1) break if version is None: print("automaton (unknown version)") else: print(f"automaton {version}") return 0 def cmd_can_edit_loop(args) -> int: """--can-edit --loop [--loop-worktree] --file {path} Checks the file falls inside this loop's declared blast radius. The harness must pass the absolute path (resolved). With --loop-worktree the harness is confirming it has already cd'd into the worktree and the path is worktree-relative; we then check the path against the loop's file_scope as-is. """ name = args.loop loop_path = _loop_dir(name, args.project) state = _read_state_loop(loop_path) if state is None: print(_loop_untracked_hint(name)) return 2 cfg = _read_loop_config(loop_path) or {} file_scope = cfg.get("blast_radius", {}).get("file_scope", []) if not args.file: print(f"ERROR: --file is required for --can-edit --loop") return 2 file_path = args.file if args.loop_worktree: check_path = file_path else: check_path = str(Path(file_path).resolve()) project_dir = _find_project_dir(args.project) proj_str = str(project_dir.resolve()) auto_str = str(AUTOMATON_DIR) if not (check_path.startswith(proj_str + os.sep) or check_path == proj_str or check_path.startswith(auto_str + os.sep) or check_path == auto_str): print(f"DENIED: File '{check_path}' is outside project/framework root.") if args.json_output: print(json.dumps({"allowed": False, "reason": "out_of_root", "loop": name})) return 1 in_scope = True if file_scope: in_scope = any(check_path == s or check_path.startswith(s + os.sep) or check_path.startswith(s.rstrip("/") + "/") for s in file_scope) if not in_scope: print(f"DENIED: File '{check_path}' is outside loop '{name}' blast radius ({file_scope}).") if args.json_output: print(json.dumps({"allowed": False, "reason": "out_of_scope", "loop": name, "file_scope": file_scope, "file": check_path})) return 1 print(f"ALLOWED: File '{check_path}' is within loop '{name}' blast radius.") if args.json_output: print(json.dumps({"allowed": True, "reason": "in_scope", "loop": name, "file_scope": file_scope, "file": check_path})) return 0 def main(): parser = argparse.ArgumentParser(description="Automaton status and enforcement script") parser.add_argument("--project", help="Project root directory (defaults to CWD)") parser.add_argument("--task", help="Task name") parser.add_argument("--agent", help="Agent ID (for multi-agent commands)") parser.add_argument("--transition", metavar="PHASE", help="Transition task to a new phase") parser.add_argument("--create-task", metavar="NAME", help="Create a new task") parser.add_argument("--approve", action="store_true", help="Approve current phase (for approval-gated phases)") parser.add_argument("--validate-folder", action="store_true", help="Validate task folder for out-of-order artifacts") parser.add_argument("--list", action="store_true", help="List all tasks") parser.add_argument("--audit", action="store_true", help="Audit all tasks for violations") parser.add_argument("--claim", action="store_true", help="Claim task for agent (multi-agent)") parser.add_argument("--release", action="store_true", help="Release task claim (multi-agent)") parser.add_argument("--next-available", action="store_true", help="Find next available task for agent") parser.add_argument("--available", action="store_true", help="List all available tasks for agent") parser.add_argument("--can-edit", action="store_true", help="Check if code edits are allowed. Without --task, checks if ANY task allows edits. With --task, checks specific task. With --file, also checks file scope.") parser.add_argument("--upgrade", action="store_true", help="Bootstrap .state files for pre-v2.0 tasks (use --task for single task, or omit for all)") parser.add_argument("--scope-check", action="store_true", help="Check if a file is in project scope") parser.add_argument("--file", help="File path for scope check or can-edit file scope check") parser.add_argument("--same-session", action="store_true", help="Check if task was created in current session") parser.add_argument("--list-states", action="store_true", help="List all valid phase names") parser.add_argument("--touch", action="store_true", help="Update .state.lastedit to reset stale-task timer without changing phase") parser.add_argument("--json", action="store_true", dest="json_output", help="Output machine-readable JSON on last line (for harness integration)") # ----- Loop engineering (v1) ----- parser.add_argument("--create-loop", metavar="NAME", help="Create a new loop from a template") parser.add_argument("--from-template", metavar="TEMPLATE", default="ci-triage", help="Template name for --create-loop (default: ci-triage)") parser.add_argument("--install-schedule", metavar="NAME", help="Install OS scheduler unit for a loop") parser.add_argument("--interval", type=int, help="Tick interval in seconds (for --install-schedule)") parser.add_argument("--pause-loop", metavar="NAME", help="Pause a running loop") parser.add_argument("--resume-loop", metavar="NAME", help="Resume a paused loop") parser.add_argument("--loop", metavar="NAME", help="Loop name (used with --approve, --can-edit, --check-gate)") parser.add_argument("--loop-worktree", action="store_true", help="With --can-edit --loop, check worktree scope instead of main tree") parser.add_argument("--check-gate", metavar="NAME", help="Run all loop gate checks before a tick (runner pre-check)") parser.add_argument("--can-continue", metavar="NAME", help="Cheap status check: is the loop runnable right now?") parser.add_argument("--loop-list", action="store_true", help="List all loops and their status") parser.add_argument("--claim-loop-task", metavar="NAME", help="Claim a task for this loop (cross-loop ownership check)") parser.add_argument("--cleanup-done", action="store_true", help="Move complete tasks older than --days into tasks/complete/ (age by .state mtime)") parser.add_argument("--install-cleanup-schedule", action="store_true", help="Install an OS scheduler unit that runs --cleanup-done periodically (default daily)") parser.add_argument("--days", type=int, help="Cleanup age threshold in days (default 7, used with --cleanup-done / --install-cleanup-schedule)") parser.add_argument("--dry-run", action="store_true", help="With --cleanup-done, list candidates without moving them") parser.add_argument("--version", action="store_true", help="Print framework version and exit") args = parser.parse_args() if args.version: return cmd_version(args) if args.claim_loop_task: return cmd_claim_loop_task(args) if args.create_loop: return cmd_create_loop(args) if args.install_schedule: return cmd_install_schedule(args) if args.pause_loop: return cmd_pause_loop(args) if args.resume_loop: return cmd_resume_loop(args) if args.check_gate: return cmd_check_gate(args) if args.can_continue: return cmd_can_continue(args) if args.loop_list: return cmd_loop_list(args) if args.install_cleanup_schedule: return cmd_install_cleanup_schedule(args) if args.cleanup_done: return cmd_cleanup_done(args) if args.create_task: return cmd_create_task(args) if args.approve: if args.loop: return cmd_approve_loop(args) if not args.task: print("ERROR: --task (or --loop) is required for --approve") return 2 return cmd_approve(args) if args.transition: if not args.task: print("ERROR: --task is required for --transition") return 2 return cmd_transition(args) if args.validate_folder: if not args.task: print("ERROR: --task is required for --validate-folder") return 2 return cmd_validate_folder(args) if args.list: return cmd_list(args) if args.audit: return cmd_audit(args) if args.claim: if not args.task: print("ERROR: --task is required for --claim") return 2 return cmd_claim(args) if args.release: if not args.task: print("ERROR: --task is required for --release") return 2 return cmd_release(args) if args.next_available: return cmd_next_available(args) if args.available: return cmd_available(args) if args.can_edit: return cmd_can_edit(args) if args.upgrade: return cmd_upgrade(args) if args.scope_check: if not args.task or not args.file: print("ERROR: --task and --file are required for --scope-check") return 2 return cmd_scope_check(args) if args.same_session: if not args.task: print("ERROR: --task is required for --same-session") return 2 return cmd_same_session(args) if args.list_states: return cmd_list_states(args) if args.touch: if not args.task: print("ERROR: --task is required for --touch") return 2 return cmd_touch(args) if args.task: return cmd_show_task(args) print("ERROR: No command specified. Use --help for usage information.") return 2 if __name__ == "__main__": sys.exit(main())