128 lines
3.9 KiB
Python
128 lines
3.9 KiB
Python
"""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())
|