#!/usr/bin/env python3 """Nightly advance plan builder. Reads agentctl registry + CURRENT.md + cleanup report and produces an ordered task list for the nightly advance run. Dependency ordering rules: 1. Integrate-candidate tasks (integrate-* worktrees) whose branch head is NOT an ancestor of dev come FIRST — they unblock the dev tip their feature tasks were built against, and their merge must land before feature tasks that rebase onto the new tip. 2. Feature/fix tasks follow, ordered by status: - in_progress / active first (work to advance) - cleanup=pending done tasks after (closeout + verification + removal) 3. Tasks whose CURRENT status is done and whose worktree verdict is ready land last (removal only). Output: JSON plan on stdout (or --out file) with entries: {task_id, worktree, branch, owner, status, phase, reason} Phases: integrate_closeout | feature_advance | cleanup_verify Exit codes: 0 ok, 2 usage/preflight failure, 3 unparsable upstream data. """ from __future__ import annotations import argparse import json import re import subprocess import sys from dataclasses import dataclass, field from pathlib import Path DEFAULT_REPO = "/home/weichen/projects/Agent-System" AGENTCTL = "scripts/agentctl.py" def _agentctl(repo: Path, *args: str) -> dict: """Run agentctl (notices on stderr) and parse its stdout JSON.""" proc = subprocess.run( [sys.executable, "-B", str(repo / AGENTCTL), *args], capture_output=True, text=True, cwd=repo, ) if proc.returncode != 0: raise RuntimeError( f"agentctl {' '.join(args)} rc={proc.returncode}: {proc.stderr[-500:]}" ) # Some subcommands emit [notice] lines on stdout too; keep last JSON blob. out = "\n".join( ln for ln in proc.stdout.splitlines() if not ln.startswith("[notice]") ).strip() try: return json.loads(out) except json.JSONDecodeError as exc: raise RuntimeError(f"unparsable agentctl output for {args}: {out[:300]}") from exc def _git(repo: Path, *args: str) -> str: proc = subprocess.run( ["git", "-C", str(repo), *args], capture_output=True, text=True ) if proc.returncode != 0: raise RuntimeError(f"git {' '.join(args)}: {proc.stderr.strip()[:300]}") return proc.stdout.strip() def parse_current_rows(repo: Path) -> dict[str, dict]: """Parse CURRENT.md 并行认领 table rows keyed by task id (col 2).""" rows: dict[str, dict] = {} text = (repo / ".doc/ops/CURRENT.md").read_text(encoding="utf-8") for line in text.splitlines(): if not line.strip().startswith("|"): continue cells = [c.strip() for c in line.strip().strip("|").split("|")] if len(cells) < 6 or cells[0] in {"owner", "---"} or set(cells[0]) <= {"-", " "}: continue task_id, status = cells[1], cells[5] if not re.match(r"^[A-Za-z0-9._-]+-\d{4}-\d{2}-\d{2}", task_id): continue rows[task_id] = { "owner": cells[0], "task": task_id, "branch": cells[2], "worktree": cells[3], "mutex": cells[4], "status": status, } return rows def registry_tasks(repo: Path) -> dict[str, dict]: """task registry state; empty when unavailable (plan falls back to CURRENT).""" try: status = _agentctl(repo, "status") except RuntimeError: return {} if not isinstance(status, dict): return {} out = status.get("tasks") or {} if isinstance(out, dict): return out return {} @dataclass class PlanEntry: task_id: str worktree: str branch: str owner: str status: str phase: str reason: str extra: dict = field(default_factory=dict) def as_dict(self) -> dict: d = { "task_id": self.task_id, "worktree": self.worktree, "phase": self.phase, "reason": self.reason, } d.update(self.extra) return d def build_plan(repo: Path) -> list[PlanEntry]: try: cleanup = _agentctl(repo, "cleanup", "report") if not isinstance(cleanup, dict): cleanup = {} except RuntimeError: # Preflight (advance.py) already gates on daemon health; a failing # cleanup probe here means no worktree info — plan nothing rather # than crash the whole nightly run. cleanup = {} current = parse_current_rows(repo) reg = registry_tasks(repo) entries: list[PlanEntry] = [] EXCLUDE = {"nightly-advance-automation-2026-09-28"} for wt in cleanup.get("worktrees", []): path = wt.get("path", "") if not path: continue wt_name = Path(path).name if wt_name in EXCLUDE: continue # never plan the nightly automation task itself branch = wt.get("branch", "") tasks = wt.get("tasks") or [] blockers = wt.get("blockers") or [] notes = wt.get("notes") or [] ancestor = "ancestor_of_dev" in notes reg_states = [t.get("state", "") for t in tasks] # Pair integrate worktrees with their base feature task id. m = re.match(r"^integrate-(.+?)-\d{4}-\d{2}-\d{2}(-r\d+)?$", wt_name) base_feature = m.group(1) if m else None cur = None for t in tasks: if t.get("id") in current: cur = current[t["id"]] break if cur is None and base_feature and base_feature in current: cur = current[base_feature] status = (cur or {}).get("status") or ( reg_states[0] if reg_states else "unknown" ) task_id = next((t["id"] for t in tasks if t.get("id")), wt_name) owner = (cur or {}).get("owner") or ( tasks[0].get("owner", "unknown") if tasks else "unknown" ) if base_feature and not ancestor: entries.append( PlanEntry( task_id=task_id, worktree=path, branch=branch, owner=owner, status=status, phase="integrate_closeout", reason="integrate candidate not yet on dev; land first so " "dependent feature tasks rebase onto the new tip", extra={"blockers": blockers, "base_feature": base_feature}, ) ) elif "TASK_ACTIVE" in " ".join(blockers) or status in {"in_progress", "active"}: entries.append( PlanEntry( task_id=task_id, worktree=path, branch=branch, owner=owner, status=status, phase="feature_advance", reason="active task; advance along its task doc to acceptance", extra={"blockers": blockers}, ) ) elif wt.get("verdict") == "ready" or ( status in {"done", "cancelled"} and not blockers ): entries.append( PlanEntry( task_id=task_id, worktree=path, branch=branch, owner=owner, status=status, phase="cleanup_verify", reason="finished and cleanup-ready; verify no-content-loss " "then remove worktree+branch", extra={"blockers": blockers}, ) ) # done-but-blocked worktrees stay out of the plan; they are reported # by the nightly report as pending-closeout items. order = {"integrate_closeout": 0, "feature_advance": 1, "cleanup_verify": 2} entries.sort(key=lambda e: (order[e.phase], e.task_id)) return entries def main() -> int: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--repo", default=DEFAULT_REPO) ap.add_argument("--out", help="write plan JSON here (stdout otherwise)") ap.add_argument( "--max-tasks", type=int, default=0, help="cap entries (0 = no cap)" ) args = ap.parse_args() repo = Path(args.repo).resolve() plan = [e.as_dict() for e in build_plan(repo)] if args.max_tasks > 0: plan = plan[: args.max_tasks] payload = { "generated_at": __import__("datetime").datetime.now().isoformat(timespec="seconds"), "repo": str(repo), "count": len(plan), "entries": plan, } text = json.dumps(payload, ensure_ascii=False, indent=2) if args.out: Path(args.out).write_text(text, encoding="utf-8") else: print(text) return 0 if __name__ == "__main__": sys.exit(main())