diff --git a/scripts/tier2/run_track.py b/scripts/tier2/run_track.py new file mode 100644 index 00000000..6b37e58b --- /dev/null +++ b/scripts/tier2/run_track.py @@ -0,0 +1,127 @@ +"""CLI entry point for the Tier 2 autonomous track execution. + +Duplicates the /tier-2-auto-execute slash command's protocol so the +smoke e2e test can run without an OpenCode session. The slash command +itself is a thin wrapper that calls this CLI. +""" +from __future__ import annotations + +import argparse +import subprocess +import sys +from datetime import datetime, timezone +from pathlib import Path + +from scripts.tier2.failcount import ( + FailcountState, + load_state, + save_state, +) +from scripts.tier2.write_report import ( + TaskResult, + write_failure_report, +) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Run a Tier 2 track autonomously.") + parser.add_argument("track_name", help="The track name (e.g., result_migration_review_pass)") + parser.add_argument("--resume", action="store_true", help="Continue from a previous run") + parser.add_argument("--toast", action="store_true", help="Windows toast on give-up") + parser.add_argument("--repo-path", default=".", help="Path to the Tier 2 clone") + parser.add_argument("--mode", default="init", choices=["init", "commit", "report", "status"], help="Action to perform") + return parser.parse_args() + + +def _git_fetch(repo_path: Path, remote: str = "origin", branch: str = "main") -> tuple[int, str, str]: + r = subprocess.run( + ["git", "fetch", remote, branch], + cwd=repo_path, capture_output=True, text=True, + ) + return r.returncode, r.stdout, r.stderr + + +def _git_switch_create(repo_path: Path, branch_name: str, start_point: str) -> tuple[int, str, str]: + r = subprocess.run( + ["git", "switch", "-c", branch_name, start_point], + cwd=repo_path, capture_output=True, text=True, + ) + return r.returncode, r.stdout, r.stderr + + +def _git_current_branch(repo_path: Path) -> str: + r = subprocess.run( + ["git", "rev-parse", "--abbrev-ref", "HEAD"], + cwd=repo_path, capture_output=True, text=True, + ) + return r.stdout.strip() if r.returncode == 0 else "" + + +def run_init(args: argparse.Namespace) -> int: + repo_path = Path(args.repo_path) + branch_name = f"tier2/{args.track_name}" + print(f"[tier2] starting track: {args.track_name}") + print(f"[tier2] repo: {repo_path}") + print(f"[tier2] branch: {branch_name}") + + rc, _, err = _git_fetch(repo_path) + if rc != 0: + print(f"[tier2] ERROR: git fetch failed: {err}", file=sys.stderr) + return 1 + + rc, _, err = _git_switch_create(repo_path, branch_name, "origin/main") + if rc != 0: + print(f"[tier2] ERROR: git switch -c failed: {err}", file=sys.stderr) + return 1 + + state = load_state(args.track_name) if args.resume else FailcountState() + save_state(args.track_name, state) + started_at = datetime.now(timezone.utc) + print(f"[tier2] track initialized; started_at={started_at.isoformat()}") + print(f"[tier2] ready for plan execution") + return 0 + + +def run_status(args: argparse.Namespace) -> int: + state = load_state(args.track_name) + print(f"[tier2] status: track={args.track_name}") + print(f"[tier2] red_phase_failures={state.red_phase_failures}") + print(f"[tier2] green_phase_failures={state.green_phase_failures}") + print(f"[tier2] no_progress_started_at={state.no_progress_started_at.isoformat() if state.no_progress_started_at else 'None'}") + return 0 + + +def run_report(args: argparse.Namespace) -> int: + repo_path = Path(args.repo_path) + branch_name = _git_current_branch(repo_path) or f"tier2/{args.track_name}" + started_at = datetime.now(timezone.utc) + state = load_state(args.track_name) + path = write_failure_report( + track_name=args.track_name, + branch_name=branch_name, + started_at=started_at, + stopped_at=started_at, + give_up_signal="manual report requested", + completed_tasks=[], + current_task=None, + last_failures=[], + state=state, + repo_path=repo_path, + ) + print(f"[tier2] report written to: {path}") + return 0 + + +def main() -> int: + args = parse_args() + if args.mode == "init": + return run_init(args) + if args.mode == "status": + return run_status(args) + if args.mode == "report": + return run_report(args) + return run_init(args) + + +if __name__ == "__main__": + sys.exit(main())