gitea
21f16b7da2
CI / build (push) Has been cancelled
- task.py: remove naive substring fallback from parse_verdict_status(), only parse ## Status: header; no header → ambiguous (REFEREE) - status.py: add _auto_update_verdict_on_complete() — when transitioning human_intervention→complete, auto-update VERDICT.md to PASS - status.py: add stale-task detection to --can-edit — deny edits if all edit tasks have .state mtime >30 min old (reason: stale_task) - status.py: add --touch command to reset task activity clock - guard plugin: handle stale_task reason with specific error message - Update test to match new verdict parsing behavior
1335 lines
57 KiB
Python
Executable File
1335 lines
57 KiB
Python
Executable File
#!/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
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
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",
|
|
"bug_find",
|
|
"adversarial_bug_find",
|
|
"doc_review",
|
|
"referee",
|
|
"complete",
|
|
"human_intervention",
|
|
]
|
|
|
|
BASE_PHASES = [
|
|
"new", "research", "decomposition", "design", "test_design",
|
|
"implement", "bug_find", "adversarial_bug_find", "doc_review",
|
|
"referee", "complete", "human_intervention",
|
|
]
|
|
|
|
APPROVAL_PHASES = {"research", "decomposition", "design", "test_design"}
|
|
|
|
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": ["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",
|
|
"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", "BUG_REPORT.md", "ADVERSARIAL_BUG_REPORT.md",
|
|
"DOC_REVIEW.md", "VERDICT.md"],
|
|
"research": ["DESIGN.md", "DECOMPOSITION.md", "TEST_PLAN.md",
|
|
"IMPLEMENTATION.md", "BUG_REPORT.md", "ADVERSARIAL_BUG_REPORT.md",
|
|
"DOC_REVIEW.md", "VERDICT.md"],
|
|
"decomposition": ["DESIGN.md", "TEST_PLAN.md", "IMPLEMENTATION.md",
|
|
"BUG_REPORT.md", "ADVERSARIAL_BUG_REPORT.md",
|
|
"DOC_REVIEW.md", "VERDICT.md"],
|
|
"design": ["DECOMPOSITION.md", "TEST_PLAN.md", "IMPLEMENTATION.md",
|
|
"BUG_REPORT.md", "ADVERSARIAL_BUG_REPORT.md",
|
|
"DOC_REVIEW.md", "VERDICT.md"],
|
|
"test_design": ["DECOMPOSITION.md", "IMPLEMENTATION.md",
|
|
"BUG_REPORT.md", "ADVERSARIAL_BUG_REPORT.md",
|
|
"DOC_REVIEW.md", "VERDICT.md"],
|
|
"implement": ["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",
|
|
"VRAM_CONFIG.md", "PARENT_SPEC.md", "REVIEW.md"}
|
|
|
|
PHASE_PRIORITY = {
|
|
"referee": 11, "doc_review": 10, "adversarial_bug_find": 9,
|
|
"bug_find": 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"],
|
|
"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)"],
|
|
"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": "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]
|
|
return base / task_name
|
|
|
|
|
|
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("."):
|
|
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 _infer_state_from_artifacts(task_path: Path) -> Optional[str]:
|
|
artifacts = {}
|
|
for name in ["SPEC.md", "DECOMPOSITION.md", "DESIGN.md", "TEST_PLAN.md",
|
|
"IMPLEMENTATION.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()
|
|
if "PASS" in content:
|
|
return "complete"
|
|
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 "IMPLEMENTATION.md" in artifacts:
|
|
return "bug_find"
|
|
if "TEST_PLAN.md" in artifacts:
|
|
return "implement"
|
|
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 <project-path>")
|
|
print(f" Or: python ~/.automaton/scripts/status.py --audit --project <project-path> (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
|
|
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)
|
|
_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 <project-path>' 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 <project-path>' 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 <project-path>' 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",
|
|
"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(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
|
|
is_in_task_folder = len(parts) >= 2 and parts[0] == "tasks" and parts[1] 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 cmd_audit(args):
|
|
project_dir = _find_project_dir(args.project)
|
|
tasks = _all_task_dirs(args.project)
|
|
if not tasks:
|
|
print("No tasks 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", "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 <project-path>' 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(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 cmd_can_edit(args):
|
|
project_dir = _find_project_dir(args.project)
|
|
|
|
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"):
|
|
state_file = path / ".state"
|
|
state_mtime = state_file.stat().st_mtime if state_file.exists() else 0
|
|
edit_tasks.append((name, base, path, state_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):
|
|
scope_tasks.append({"task": name, "phase": base, "state_mtime": state_mtime})
|
|
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]}))
|
|
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]}))
|
|
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):
|
|
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):
|
|
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"):
|
|
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}))
|
|
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 mtime 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
|
|
import os
|
|
os.utime(str(state_file), None)
|
|
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):
|
|
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):
|
|
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 = state_file.stat().st_mtime
|
|
age_minutes = (time.time() - mtime) / 60
|
|
threshold = 30
|
|
if age_minutes < threshold:
|
|
print(f"SAME_SESSION: Task '{args.task}' .state was modified {age_minutes:.0f} minutes ago (threshold: {threshold} min)")
|
|
return 1
|
|
print(f"DIFFERENT_SESSION: Task '{args.task}' .state was modified {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
|
|
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
|
|
|
|
|
|
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 mtime 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)")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.create_task:
|
|
return cmd_create_task(args)
|
|
if args.approve:
|
|
if not args.task:
|
|
print("ERROR: --task 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()) |