#!/usr/bin/env python3 """Prepare one Git Flow feature worktree for an active Milestone.""" from __future__ import annotations import argparse import fcntl import json import os from pathlib import Path import re import subprocess import sys from typing import Any, Iterable, NamedTuple CATALOG_ENV = "AGENT_TASK_EXECUTION_CATALOG" MILESTONE_PATTERN = re.compile( r"^agent-roadmap/phase/(?P[a-z0-9-]+)/milestones/(?P[a-z0-9-]+)\.md$" ) EPIC_HEADING = re.compile(r"^### Epic:\s*\[(?P[a-z0-9-]+)\]\s*(?P.+?)\s*$") TASK_LINE = re.compile(r"^- \[(?P<done>[ xX])\] \[(?P<id>[a-z0-9-]+)\]\s+(?P<body>.+)$") EPIC_RANGE = re.compile(r"^(?P<start>[1-9][0-9]*)\.\.(?P<end>[1-9][0-9]*)$") SENSITIVE_PARTS = {".env", "secret", "secrets", "credential", "credentials", "password", "passwords"} SENSITIVE_SUFFIXES = {".pem", ".key", ".p12"} class PreparationError(RuntimeError): """A fail-closed preparation error.""" class Epic(NamedTuple): epic_id: str title: str task_ids: tuple[str, ...] incomplete_ids: tuple[str, ...] def emit(event: str, **payload: Any) -> None: print(json.dumps({"event": event, **payload}, ensure_ascii=False, sort_keys=True), flush=True) def run( command: list[str], *, cwd: Path, check: bool = True, capture: bool = True, ) -> subprocess.CompletedProcess[str]: result = subprocess.run( command, cwd=cwd, text=True, stdout=subprocess.PIPE if capture else None, stderr=subprocess.PIPE if capture else None, check=False, ) if check and result.returncode != 0: detail = (result.stderr or result.stdout or "").strip() raise PreparationError(f"command failed ({result.returncode}): {' '.join(command)}: {detail}") return result def git(repo: Path, *arguments: str, check: bool = True) -> str: result = run(["git", *arguments], cwd=repo, check=check) return (result.stdout or "").strip() def resolve_repo(raw: str) -> Path: repo = Path(raw).expanduser().resolve() if not repo.is_dir(): raise PreparationError(f"repository directory not found: {repo}") top = Path(git(repo, "rev-parse", "--show-toplevel")).resolve() if top != repo: raise PreparationError(f"--repo must be the repository root: expected={top} actual={repo}") return repo def resolve_milestone(repo: Path, raw: str) -> tuple[Path, re.Match[str]]: candidate = Path(raw).expanduser() path = (repo / candidate).resolve() if not candidate.is_absolute() else candidate.resolve() try: relative = path.relative_to(repo).as_posix() except ValueError as exc: raise PreparationError(f"milestone is outside repository: {path}") from exc match = MILESTONE_PATTERN.fullmatch(relative) if match is None or not path.is_file(): raise PreparationError(f"active milestone path required: {relative}") return path, match def section(text: str, heading: str) -> str: pattern = re.compile( rf"^## {re.escape(heading)}\s*$\n(?P<body>.*?)(?=^##\s|\Z)", re.MULTILINE | re.DOTALL, ) match = pattern.search(text) return match.group("body").strip() if match else "" def first_heading(text: str, prefix: str) -> str: match = re.search(rf"^{re.escape(prefix)}\s*(.+?)\s*$", text, re.MULTILINE) if not match: raise PreparationError(f"missing heading: {prefix}") return match.group(1).strip() def milestone_contract(path: Path) -> dict[str, str]: text = path.read_text(encoding="utf-8") status_body = section(text, "상태") status_match = re.search(r"^\[(.+?)\]\s*$", status_body, re.MULTILINE) status = status_match.group(1).strip() if status_match else "" lock = section(text, "구현 잠금") lock_state = re.search(r"^- 상태:\s*(.+?)\s*$", lock, re.MULTILINE) decision = re.search(r"^- 결정 필요:\s*(.+?)\s*$", lock, re.MULTILINE) if status != "계획": raise PreparationError(f"milestone must be [계획]: actual={status or 'missing'}") if not lock_state or lock_state.group(1).strip() != "해제": raise PreparationError("milestone implementation lock is not 해제") if not decision or decision.group(1).strip() != "없음": raise PreparationError("milestone has unresolved 결정 필요") return { "title": first_heading(text, "# Milestone:"), "status": status, } def existing_milestone_contract(path: Path) -> dict[str, str]: text = path.read_text(encoding="utf-8") status_body = section(text, "상태") status_match = re.search(r"^\[(.+?)\]\s*$", status_body, re.MULTILINE) status = status_match.group(1).strip() if status_match else "" lock = section(text, "구현 잠금") lock_state = re.search(r"^- 상태:\s*(.+?)\s*$", lock, re.MULTILINE) decision = re.search(r"^- 결정 필요:\s*(.+?)\s*$", lock, re.MULTILINE) if status not in {"계획", "진행중"}: raise PreparationError( f"Milestone work requires [계획] or [진행중]: actual={status or 'missing'}" ) if not lock_state or lock_state.group(1).strip() != "해제": raise PreparationError("milestone implementation lock is not 해제") if not decision or decision.group(1).strip() != "없음": raise PreparationError("milestone has unresolved 결정 필요") return { "title": first_heading(text, "# Milestone:"), "status": status, } def parse_epics(text: str) -> list[Epic]: lines = text.splitlines() starts: list[tuple[int, re.Match[str]]] = [] for index, line in enumerate(lines): match = EPIC_HEADING.fullmatch(line) if match: starts.append((index, match)) epics: list[Epic] = [] for position, (start, match) in enumerate(starts): end = starts[position + 1][0] if position + 1 < len(starts) else len(lines) task_matches = [ value for value in (TASK_LINE.fullmatch(line) for line in lines[start + 1 : end]) if value is not None ] epics.append( Epic( epic_id=match.group("id"), title=match.group("title"), task_ids=tuple(value.group("id") for value in task_matches), incomplete_ids=tuple( value.group("id") for value in task_matches if value.group("done") == " " ), ) ) return epics def select_epics(epics: list[Epic], selector: str) -> list[Epic]: if not epics: raise PreparationError("target Milestone has no Epic") if selector == "remaining": return [epic for epic in epics if epic.incomplete_ids] if selector == "first-incomplete": selected = next((epic for epic in epics if epic.incomplete_ids), None) if selected is None: raise PreparationError("target Milestone has no incomplete Epic") return [selected] range_match = EPIC_RANGE.fullmatch(selector) if range_match: start = int(range_match.group("start")) end = int(range_match.group("end")) if start > end: raise PreparationError(f"Epic range start must not exceed end: {selector}") if end > len(epics): raise PreparationError( f"Epic range exceeds document order: requested={selector} available={len(epics)}" ) return epics[start - 1 : end] selected: list[Epic] = [] for raw in selector.split(","): value = raw.strip() if not value: raise PreparationError("--epics contains an empty selector") matches = [ epic for epic in epics if epic.epic_id == value or epic.title.casefold().strip() == value.casefold() ] if len(matches) != 1: raise PreparationError(f"Epic selector must resolve exactly once: {value}") if matches[0] in selected: raise PreparationError(f"duplicate Epic selector: {value}") selected.append(matches[0]) return [epic for epic in epics if epic in selected] def resolve_workspace(repo: Path, raw: str) -> Path: candidate = Path(raw).expanduser() return (repo / candidate).resolve() if not candidate.is_absolute() else candidate.resolve() def phase_contract(repo: Path, phase_slug: str) -> dict[str, str]: path = repo / "agent-roadmap" / "phase" / phase_slug / "PHASE.md" if not path.is_file(): raise PreparationError(f"phase document not found: {path}") text = path.read_text(encoding="utf-8") status_body = section(text, "상태") status_match = re.search(r"^\[(.+?)\]\s*$", status_body, re.MULTILINE) return { "title": first_heading(text, "# Phase:"), "status": status_match.group(1).strip() if status_match else "계획", } def git_common_dir(repo: Path) -> Path: raw = Path(git(repo, "rev-parse", "--git-common-dir")) return (repo / raw).resolve() if not raw.is_absolute() else raw.resolve() def atomic_json(path: Path, value: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(path.suffix + f".tmp.{os.getpid()}") temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") os.replace(temporary, path) def read_json(path: Path) -> dict[str, Any] | None: if not path.exists(): return None try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise PreparationError(f"invalid state file: {path}: {exc}") from exc if not isinstance(value, dict): raise PreparationError(f"state file must contain an object: {path}") return value def process_start_token(pid: int) -> str | None: stat = Path(f"/proc/{pid}/stat") try: remainder = stat.read_text(encoding="utf-8").rsplit(")", 1)[1].split() return f"proc:{remainder[19]}" except (OSError, IndexError): return None def process_is_same(pid: object, expected_token: object) -> bool: if not isinstance(pid, int) or pid <= 0: return False try: os.kill(pid, 0) except (OSError, ValueError): return False actual_token = process_start_token(pid) if expected_token is None or actual_token is None: return True return actual_token == expected_token def changed_paths(workspace: Path) -> list[str]: unmerged = git(workspace, "diff", "--name-only", "--diff-filter=U") if unmerged: raise PreparationError(f"workspace has unmerged paths: {','.join(unmerged.splitlines())}") values: set[str] = set() for arguments in ( ("diff", "--name-only", "--no-renames", "-z", "HEAD"), ("ls-files", "--others", "--exclude-standard", "-z"), ): result = subprocess.run( ["git", *arguments], cwd=workspace, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, ) if result.returncode != 0: raise PreparationError(result.stderr.decode("utf-8", "replace").strip()) values.update( part.decode("utf-8", "surrogateescape") for part in result.stdout.split(b"\0") if part ) return sorted(values) def sensitive(path: str) -> bool: candidate = Path(path) lowered = {part.casefold() for part in candidate.parts} if lowered & SENSITIVE_PARTS: return True name = candidate.name.casefold() return ( any(token in name for token in ("secret", "credential", "password")) or candidate.suffix.casefold() in SENSITIVE_SUFFIXES ) def ref_exists(repo: Path, ref: str) -> bool: return run(["git", "show-ref", "--verify", "--quiet", ref], cwd=repo, check=False).returncode == 0 def ensure_clean(repo: Path, label: str) -> None: status = git(repo, "status", "--porcelain=v1", "--untracked-files=all") if status: raise PreparationError(f"{label} has repository changes; clean checkout required") def worktrees(repo: Path) -> list[dict[str, str]]: records: list[dict[str, str]] = [] current: dict[str, str] = {} for line in git(repo, "worktree", "list", "--porcelain").splitlines() + [""]: if not line: if current: records.append(current) current = {} continue key, _, value = line.partition(" ") current[key] = value return records def probe_targets( repo: Path, execution_catalog: str, planner_target: str, review_target: str, ) -> None: runner = ( Path(__file__).resolve().parents[2] / "prepare-epic-work-items" / "scripts" / "run_agent_once.py" ) if not runner.is_file(): raise PreparationError(f"agent runner not found: {runner}") seen: set[str] = set() for target_id in (planner_target, review_target): if target_id in seen: continue seen.add(target_id) command = [ sys.executable, str(runner), "--execution-catalog", execution_catalog, "--target-id", target_id, "--workspace", str(repo), "--probe", ] result = run(command, cwd=repo, check=False, capture=False) if result.returncode != 0: raise PreparationError( f"execution target capability probe failed: target_id={target_id}" ) def render_current( *, phase_slug: str, phase: dict[str, str], milestone_slug: str, milestone: dict[str, str], ) -> str: return f"""# 현재 로드맵 컨텍스트 ## 고정 참조 - Phase를 가로지르는 다음 작업 후보는 [전역 마일스톤 실행 순서](priority-queue.md)를 먼저 확인한다. - Phase는 도메인/책임 영역이며 순차 실행 게이트가 아니다. ## 활성 Phase - [{phase['status']}] {phase['title']} - 경로: [PHASE.md](phase/{phase_slug}/PHASE.md) ## 활성 Milestone - [{milestone['status']}] {milestone['title']} - Phase: [PHASE.md](phase/{phase_slug}/PHASE.md) - 경로: [{milestone['title']}](phase/{phase_slug}/milestones/{milestone_slug}.md) ## 선택 규칙 - 이 문서는 현재 feature workspace의 활성 Phase와 Milestone 후보를 가리킨다. - 실제 진행·완료 상태는 Milestone 문서와 active task evidence로 판정한다. - `[완료]` 또는 `[폐기]` 항목은 활성 항목에 남기지 않는다. """ def epic_cycle_script(workspace: Path) -> Path: path = ( workspace / "agent-ops" / "skills" / "common" / "prepare-epic-work-items" / "scripts" / "run_epic_cycle.py" ) if not path.is_file(): raise PreparationError(f"Epic cycle script not found: {path}") return path def dispatcher_script(workspace: Path) -> Path: skills_root = workspace / "agent-ops" / "skills" project_root = skills_root / "project" / "orchestrate-agent-task-loop" project_dispatcher = project_root / "scripts" / "dispatch.py" if project_dispatcher.is_file(): private_root = skills_root / "private" / "orchestrate-agent-task-loop" private_dispatcher = private_root / "scripts" / "dispatch.py" if (private_root / "SKILL.md").is_file() and private_dispatcher.is_file(): return private_dispatcher return project_dispatcher common_dispatcher = ( skills_root / "common" / "orchestrate-agent-task-loop" / "scripts" / "dispatch.py" ) if not common_dispatcher.is_file(): raise PreparationError(f"dispatcher script not found: {common_dispatcher}") return common_dispatcher def dispatcher_command( *, workspace: Path, dispatcher: Path, task_group: str, execution_catalog: str, ) -> list[str]: command = [ sys.executable, str(dispatcher), "--workspace", str(workspace), "--task-group", task_group, ] common_dispatcher = ( workspace / "agent-ops" / "skills" / "common" / "orchestrate-agent-task-loop" / "scripts" / "dispatch.py" ) if dispatcher.resolve() == common_dispatcher.resolve(): command.extend(["--execution-catalog", execution_catalog]) return command def epic_cycle_command( *, args: argparse.Namespace, workspace: Path, milestone: Path, epic: Epic, batch_ids: list[str], validate_only: bool = False, ) -> list[str]: command = [ sys.executable, str(epic_cycle_script(workspace)), "--workspace", str(workspace), "--milestone", str(milestone), "--epic", epic.epic_id, "--execution-catalog", args.execution_catalog, "--planner-target", args.planner_target, "--batch-task-ids", ",".join(batch_ids), ] if args.review_target: command.extend(["--review-target", args.review_target]) if validate_only: command.append("--validate-only") elif args.retry: command.append("--retry") return command def validate_batch( *, args: argparse.Namespace, workspace: Path, milestone: Path, epics: list[Epic], batch_ids: list[str], ) -> None: for epic in epics: result = run( epic_cycle_command( args=args, workspace=workspace, milestone=milestone, epic=epic, batch_ids=batch_ids, validate_only=True, ), cwd=workspace, check=False, capture=False, ) if result.returncode != 0: raise PreparationError( f"selected Epic batch validation failed: epic={epic.epic_id} exit={result.returncode}" ) def publish_batch_review(workspace: Path) -> str: paths = changed_paths(workspace) if paths: unsafe = [path for path in paths if sensitive(path)] if unsafe: raise PreparationError(f"sensitive path refused: {','.join(unsafe)}") for path in paths: git(workspace, "add", "--", path) git(workspace, "diff", "--cached", "--check") git(workspace, "commit", "-m", "chore(milestone): 복수 Epic 준비 결과를 검증한다") git(workspace, "push") if changed_paths(workspace): raise PreparationError("workspace is dirty after batch review publish") return git(workspace, "rev-parse", "HEAD") def cross_epic_review( *, args: argparse.Namespace, workspace: Path, milestone: Path, milestone_slug: str, phase_slug: str, epics: list[Epic], batch_ids: list[str], common: Path, ) -> tuple[int, dict[str, Any] | None]: reviewer_target = args.review_target or args.planner_target state_root = common / "milestone-work-preparation" / milestone_slug prompt_path = state_root / "prompts" / "cross-epic-review.txt" result_path = ( common / "epic-work-preparation" / milestone_slug / "_batch" / "cross-epic-review.json" ) label = f"{milestone_slug}-cross-epic-review" selected = "\n".join( f"- [{epic.epic_id}] {epic.title}: {','.join(epic.task_ids)}" for epic in epics ) prompt = f"""You are a fresh child agent launched for one bounded cross-Epic preparation review. Work only in {workspace}. Read the repository AGENTS.md completely, then read agent-ops/skills/common/router.md and only the plan/refine skills required to correct artifacts. Do not start subagents, orchestration dispatchers, preparation scripts, or monitoring loops. Do not commit or push; the parent runtime owns the checkpoint. Target Milestone: {milestone} Selected Epics in document order: {selected} Allowed Milestone Task ids: {','.join(batch_ids)} Review the complete prepared artifact union across these Epics from a fresh context. Check scope coverage, cross-Epic assumptions and dependencies, write-set collisions, PLAN/CODE_REVIEW pairing and first-line metadata, task ids, indices, routing, and verification. Fix every material defect through the owning plan/refine procedure without adding an official code-review verdict. Do not implement active PLAN work. If a genuine product or scope decision is required, create the Milestone SDD USER_REVIEW artifact and stop without inventing a decision. Final in Korean. """ prompt_path.parent.mkdir(parents=True, exist_ok=True) prompt_path.write_text(prompt, encoding="utf-8") previous = read_json(result_path) if previous is not None: if previous.get("workspace") != str(workspace) or previous.get("label") != label: raise PreparationError(f"cross-Epic review result identity mismatch: {result_path}") if previous.get("status") == "succeeded": emit( "AGENT_RESULT_RECOVERED", stage="cross-epic-review", locator=previous.get("locator"), result="succeeded", ) elif previous.get("status") in {"running", "tracking"} and process_is_same( previous.get("agent_pid"), previous.get("agent_process_start_token") ): emit( "AGENT_TRACKING", stage="cross-epic-review", locator=previous.get("locator"), pid=previous.get("agent_pid"), ) return 3, previous elif not args.retry: emit( "AGENT_RECOVERY_REQUIRED", stage="cross-epic-review", locator=previous.get("locator"), action="inspect locator, then rerun with --retry", ) return 3, previous record = previous if record is None or record.get("status") != "succeeded": runner = ( workspace / "agent-ops" / "skills" / "common" / "prepare-epic-work-items" / "scripts" / "run_agent_once.py" ) command = [ sys.executable, str(runner), "--execution-catalog", args.execution_catalog, "--target-id", reviewer_target, "--workspace", str(workspace), "--prompt-file", str(prompt_path), "--label", label, "--result-file", str(result_path), ] starting_head = git(workspace, "rev-parse", "HEAD") result = run(command, cwd=workspace, check=False, capture=False) if git(workspace, "rev-parse", "HEAD") != starting_head: raise PreparationError("cross-Epic reviewer committed unexpectedly") record = read_json(result_path) if result.returncode != 0 or record is None or record.get("status") != "succeeded": return (3 if result.returncode == 3 else 2), record review_path = ( workspace / "agent-roadmap" / "sdd" / phase_slug / milestone_slug / "USER_REVIEW.md" ) if review_path.exists(): head = publish_batch_review(workspace) emit("USER_REVIEW", path=str(review_path), head=head) return 2, record task_group = workspace / "agent-task" / f"m-{milestone_slug}" if list(task_group.glob("USER_REVIEW.md")) + list(task_group.glob("*/USER_REVIEW.md")): raise PreparationError("cross-Epic reviewer created forbidden agent-task USER_REVIEW") validate_batch( args=args, workspace=workspace, milestone=milestone, epics=epics, batch_ids=batch_ids, ) head = publish_batch_review(workspace) emit("CROSS_EPIC_REVIEW_FINISHED", head=head, epics=[epic.epic_id for epic in epics]) return 0, record def coordinate_batch( *, args: argparse.Namespace, workspace: Path, milestone: Path, milestone_slug: str, phase_slug: str, common: Path, ) -> int: selected = select_epics(parse_epics(milestone.read_text(encoding="utf-8")), args.epics) empty = [epic.epic_id for epic in selected if not epic.task_ids] if empty: raise PreparationError(f"selected Epic has no Task ids: {','.join(empty)}") batch_ids = [task_id for epic in selected for task_id in epic.task_ids] duplicates = sorted({value for value in batch_ids if batch_ids.count(value) > 1}) if duplicates: raise PreparationError(f"selected Epic Task ids are not unique: {','.join(duplicates)}") identity = { "milestone": str(milestone), "workspace": str(workspace), "selected_epics": [epic.epic_id for epic in selected], "batch_task_ids": batch_ids, "execution_catalog": str(Path(args.execution_catalog).expanduser().resolve()), "planner_target": args.planner_target, "review_target": args.review_target, } state_path = common / "milestone-work-preparation" / milestone_slug / "batch-state.json" state = read_json(state_path) if state is None: state = { **identity, "status": "active", "epic_events": {}, "cross_epic_review_done": False, "dispatcher_dry_run_done": False, "dispatcher_live_started": False, } atomic_json(state_path, state) else: mismatched = [key for key, expected in identity.items() if state.get(key) != expected] if mismatched and state.get("status") == "completed": state = { **identity, "status": "active", "epic_events": {}, "cross_epic_review_done": False, "dispatcher_dry_run_done": False, "dispatcher_live_started": False, } atomic_json(state_path, state) elif mismatched: key = mismatched[0] raise PreparationError( f"active Milestone preparation batch identity changed: field={key} " f"state={state.get(key)} requested={identity[key]}" ) if state.get("status") == "completed": emit( "MILESTONE_PREPARATION_COMPLETED", milestone=milestone_slug, epics=identity["selected_epics"], resumed=True, head=state.get("head"), ) return 0 resume_blocked_dispatcher = state.get("status") == "dispatcher-blocked" if state.get("status") == "dispatching" and process_is_same( state.get("dispatcher_pid"), state.get("dispatcher_process_start_token") ): emit( "DISPATCHER_TRACKING", pid=state.get("dispatcher_pid"), task_group=f"m-{milestone_slug}", ) return 3 if state.get("status") in {"dispatching", "dispatcher-tracking"} and not args.retry: emit( "DISPATCHER_RECOVERY_REQUIRED", task_group=f"m-{milestone_slug}", action="inspect dispatcher state, then rerun with --retry", ) return 3 if state.get("status") == "dispatcher-blocked" and not args.retry: emit( "DISPATCHER_BLOCKED", task_group=f"m-{milestone_slug}", exit_code=state.get("dispatcher_exit_code"), resumed=True, ) return 2 epic_events = state.get("epic_events") if not isinstance(epic_events, dict): raise PreparationError(f"invalid Epic event map: {state_path}") unknown_events = sorted(set(epic_events) - set(identity["selected_epics"])) if unknown_events: raise PreparationError(f"Epic event map contains unselected ids: {','.join(unknown_events)}") for epic in selected: if epic_events.get(epic.epic_id) in {"EPIC_WORK_ITEMS_READY", "EPIC_COMPLETED"}: continue state.update(status="epic-preparing", current_epic=epic.epic_id) atomic_json(state_path, state) result = run( epic_cycle_command( args=args, workspace=workspace, milestone=milestone, epic=epic, batch_ids=batch_ids, ), cwd=workspace, check=False, capture=False, ) epic_state_path = ( common / "epic-work-preparation" / milestone_slug / epic.epic_id / "state.json" ) epic_state = read_json(epic_state_path) or {} terminal = epic_state.get("event") if result.returncode != 0 or terminal not in {"EPIC_WORK_ITEMS_READY", "EPIC_COMPLETED"}: status = epic_state.get("status") if terminal == "USER_REVIEW" or status == "user-review": state.update(status="user-review", event="USER_REVIEW", current_epic=epic.epic_id) atomic_json(state_path, state) return 2 if result.returncode == 3 or status == "tracking": state.update(status="epic-tracking", event="AGENT_TRACKING", current_epic=epic.epic_id) atomic_json(state_path, state) return 3 state.update( status="failed", event="FAILED", current_epic=epic.epic_id, reason=epic_state.get("reason", f"Epic cycle exit={result.returncode}"), ) atomic_json(state_path, state) return 2 epic_events[epic.epic_id] = terminal state.update(status="active", epic_events=epic_events) state.pop("current_epic", None) atomic_json(state_path, state) if len(selected) > 1 and not state.get("cross_epic_review_done"): state.update(status="cross-epic-review") atomic_json(state_path, state) result, record = cross_epic_review( args=args, workspace=workspace, milestone=milestone, milestone_slug=milestone_slug, phase_slug=phase_slug, epics=selected, batch_ids=batch_ids, common=common, ) if result != 0: review_path = ( workspace / "agent-roadmap" / "sdd" / phase_slug / milestone_slug / "USER_REVIEW.md" ) user_review = result == 2 and review_path.exists() state.update( status=( "user-review" if user_review else "cross-epic-review-tracking" if result == 3 else "failed" ), event=( "USER_REVIEW" if user_review else "AGENT_TRACKING" if result == 3 else "FAILED" ), review_locator=record.get("locator") if record else None, ) atomic_json(state_path, state) return result state.update( status="active", cross_epic_review_done=True, review_locator=record.get("locator") if record else None, ) atomic_json(state_path, state) validate_batch( args=args, workspace=workspace, milestone=milestone, epics=selected, batch_ids=batch_ids, ) refreshed_epics = { epic.epic_id: epic for epic in parse_epics(milestone.read_text(encoding="utf-8")) } for epic_id in identity["selected_epics"]: refreshed = refreshed_epics.get(epic_id) if refreshed is None: raise PreparationError(f"selected Epic disappeared before batch barrier: {epic_id}") if not refreshed.incomplete_ids: epic_events[epic_id] = "EPIC_COMPLETED" state.update( status="batch-ready", event="MILESTONE_WORK_ITEMS_READY", epic_events=epic_events, ) atomic_json(state_path, state) emit( "MILESTONE_WORK_ITEMS_READY", milestone=milestone_slug, epics=[epic.epic_id for epic in selected], terminals=epic_events, ) if all(event == "EPIC_COMPLETED" for event in epic_events.values()): state.update( status="completed", event="MILESTONE_PREPARATION_COMPLETED", dispatcher="skipped-no-active-plans", head=git(workspace, "rev-parse", "HEAD"), ) atomic_json(state_path, state) emit( "MILESTONE_PREPARATION_COMPLETED", milestone=milestone_slug, epics=[epic.epic_id for epic in selected], dispatcher="skipped-no-active-plans", ) return 0 dispatcher = dispatcher_script(workspace) task_group = f"m-{milestone_slug}" if not state.get("dispatcher_dry_run_done"): dry_run_command = dispatcher_command( workspace=workspace, dispatcher=dispatcher, task_group=task_group, execution_catalog=args.execution_catalog, ) dry_run_command.append("--dry-run") dry_run = run( dry_run_command, cwd=workspace, check=False, capture=False, ) if dry_run.returncode != 0: state.update(status="failed", event="FAILED", reason=f"dispatcher dry-run exit={dry_run.returncode}") atomic_json(state_path, state) return 2 state.update(dispatcher_dry_run_done=True, dispatcher=str(dispatcher)) atomic_json(state_path, state) emit("DISPATCHER_DRY_RUN_FINISHED", task_group=task_group) command = dispatcher_command( workspace=workspace, dispatcher=dispatcher, task_group=task_group, execution_catalog=args.execution_catalog, ) if resume_blocked_dispatcher and args.retry: command.append("--retry-blocked") try: process = subprocess.Popen(command, cwd=workspace, start_new_session=True) except OSError as exc: raise PreparationError(f"dispatcher launch failed: {exc}") from exc state.update( status="dispatching", event="DISPATCHER_STARTED", dispatcher_live_started=True, dispatcher_pid=process.pid, dispatcher_process_start_token=process_start_token(process.pid), dispatcher_command=command, ) atomic_json(state_path, state) emit("DISPATCHER_STARTED", task_group=task_group, pid=process.pid) try: exit_code = process.wait() except KeyboardInterrupt: state.update(status="dispatcher-tracking", event="DISPATCHER_TRACKING") atomic_json(state_path, state) emit("DISPATCHER_TRACKING", task_group=task_group, pid=process.pid) return 3 state.update(dispatcher_exit_code=exit_code) if exit_code == 3: state.update(status="dispatcher-tracking", event="DISPATCHER_TRACKING") atomic_json(state_path, state) emit("DISPATCHER_TRACKING", task_group=task_group, exit_code=exit_code) return 3 if exit_code != 0: state.update(status="dispatcher-blocked", event="DISPATCHER_BLOCKED") atomic_json(state_path, state) emit("DISPATCHER_BLOCKED", task_group=task_group, exit_code=exit_code) return 2 validate_batch( args=args, workspace=workspace, milestone=milestone, epics=selected, batch_ids=batch_ids, ) refreshed = {epic.epic_id: epic for epic in parse_epics(milestone.read_text(encoding="utf-8"))} incomplete = [ epic_id for epic_id in identity["selected_epics"] if epic_id not in refreshed or refreshed[epic_id].incomplete_ids ] if incomplete: raise PreparationError( "dispatcher exited successfully but selected Epic workstate is incomplete: " + ",".join(incomplete) ) state.update( status="completed", event="MILESTONE_PREPARATION_COMPLETED", head=git(workspace, "rev-parse", "HEAD"), ) atomic_json(state_path, state) emit( "MILESTONE_PREPARATION_COMPLETED", milestone=milestone_slug, epics=identity["selected_epics"], dispatcher_exit_code=exit_code, head=state["head"], ) return 0 def parser() -> argparse.ArgumentParser: value = argparse.ArgumentParser(description=__doc__) value.add_argument("--repo") value.add_argument("--milestone", required=True) value.add_argument("--workspace") value.add_argument( "--existing-workspace", action="store_true", help="start selected Epic work in the current prepared feature workspace", ) value.add_argument("--execution-catalog", default=os.environ.get(CATALOG_ENV)) value.add_argument( "--planner-target", default=os.environ.get("AGENT_TASK_PLANNER_TARGET") ) value.add_argument( "--review-target", default=os.environ.get("AGENT_TASK_REVIEW_TARGET") ) value.add_argument( "--epics", help="prepare and dispatch remaining/first-incomplete/one/list/range selector", ) value.add_argument( "--retry", action="store_true", help="resume a stopped batch after its recorded recovery condition was handled", ) value.add_argument("--remote", default="origin") value.add_argument( "--skip-agent-probe", action="store_true", help="tests only; requires --dry-run or AGENT_OPS_TESTING=1", ) value.add_argument("--dry-run", action="store_true") return value def apply_defaults(args: argparse.Namespace) -> argparse.Namespace: if not args.execution_catalog: raise PreparationError( f"--execution-catalog or {CATALOG_ENV} is required" ) if not args.planner_target: raise PreparationError( "--planner-target or AGENT_TASK_PLANNER_TARGET is required" ) args.execution_catalog = str(Path(args.execution_catalog).expanduser().resolve()) if args.review_target is None: args.review_target = args.planner_target return args def prepare_existing(args: argparse.Namespace) -> int: apply_defaults(args) if not args.epics: raise PreparationError("--epics is required with --existing-workspace") if args.skip_agent_probe and not ( args.dry_run or os.environ.get("AGENT_OPS_TESTING") == "1" ): raise PreparationError( "--skip-agent-probe is test-only; use AGENT_OPS_TESTING=1" ) workspace = resolve_repo(args.workspace or ".") if args.repo is not None and resolve_repo(args.repo) != workspace: raise PreparationError("--repo must match --workspace in existing-workspace mode") milestone_path, milestone_match = resolve_milestone(workspace, args.milestone) milestone = existing_milestone_contract(milestone_path) milestone_slug = milestone_match.group("slug") phase_slug = milestone_match.group("phase") selected = select_epics(parse_epics(milestone_path.read_text(encoding="utf-8")), args.epics) batch_ids = [task_id for epic in selected for task_id in epic.task_ids] identity = { "milestone": str(milestone_path), "workspace": str(workspace), "selected_epics": [epic.epic_id for epic in selected], "batch_task_ids": batch_ids, } develop = git(workspace, "config", "--get", "gitflow.branch.develop") feature_prefix = git(workspace, "config", "--get", "gitflow.prefix.feature") branch = git(workspace, "branch", "--show-current") expected_branch = f"{feature_prefix}{milestone_slug}" if feature_prefix else "" if not develop or not feature_prefix or branch != expected_branch: raise PreparationError( f"current workspace must use the target Milestone feature branch: " f"expected={expected_branch or 'missing-gitflow-config'} actual={branch or 'detached'}" ) current_path = workspace / "agent-roadmap" / "current.md" expected_current_target = f"phase/{phase_slug}/milestones/{milestone_slug}.md" if ( not current_path.is_file() or expected_current_target not in current_path.read_text(encoding="utf-8") ): raise PreparationError( f"workspace-local current does not select target Milestone: {current_path}" ) common = git_common_dir(workspace) state_root = common / "milestone-work-preparation" / milestone_slug state_root.mkdir(parents=True, exist_ok=True) state_path = state_root / "workspace-state.json" with (state_root / "workspace.lock").open("a+", encoding="utf-8") as lock: try: fcntl.flock(lock.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError as exc: raise PreparationError( f"workspace preparation already running: {state_root / 'workspace.lock'}" ) from exc batch_state = read_json(state_root / "batch-state.json") batch_matches = bool( batch_state and all(batch_state.get(key) == expected for key, expected in identity.items()) ) resuming_batch = bool( batch_matches and batch_state and batch_state.get("status") != "completed" ) if batch_state and batch_state.get("status") != "completed" and not batch_matches: raise PreparationError("another active Epic batch owns the current Milestone workspace") if not resuming_batch: ensure_clean(workspace, "feature workspace") upstream = git( workspace, "rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}", ) if not upstream.endswith(f"/{branch}"): raise PreparationError( f"feature branch upstream mismatch: branch={branch} upstream={upstream}" ) if not resuming_batch and not args.dry_run: remote_name = upstream.split("/", 1)[0] git(workspace, "fetch", remote_name, branch) head = git(workspace, "rev-parse", "HEAD") upstream_head = git(workspace, "rev-parse", "@{u}") if head != upstream_head: if not resuming_batch or run( ["git", "merge-base", "--is-ancestor", "@{u}", "HEAD"], cwd=workspace, check=False, ).returncode != 0: raise PreparationError("current feature branch is not synchronized with its upstream") emit( "PREFLIGHT_READY", branch=branch, milestone=str(milestone_path), workspace=str(workspace), existing_workspace=True, ) if args.dry_run: return 0 if selected and not args.skip_agent_probe and not resuming_batch: probe_targets( workspace, args.execution_catalog, args.planner_target, args.review_target, ) ensure_clean(workspace, "feature workspace after agent probe") state = { "status": "workspace-ready", "milestone": str(milestone_path), "milestone_slug": milestone_slug, "branch": branch, "workspace": str(workspace), "execution_catalog": args.execution_catalog, "planner_target": args.planner_target, "review_target": args.review_target, "existing_workspace": True, } atomic_json(state_path, state) emit("WORKSPACE_READY", **state) return coordinate_batch( args=args, workspace=workspace, milestone=milestone_path, milestone_slug=milestone_slug, phase_slug=phase_slug, common=common, ) def prepare(args: argparse.Namespace) -> int: apply_defaults(args) if args.repo is None or args.workspace is None: raise PreparationError("--repo and --workspace are required unless --existing-workspace is used") if args.skip_agent_probe and not ( args.dry_run or os.environ.get("AGENT_OPS_TESTING") == "1" ): raise PreparationError( "--skip-agent-probe is test-only; use --dry-run or AGENT_OPS_TESTING=1" ) repo = resolve_repo(args.repo) milestone_path, milestone_match = resolve_milestone(repo, args.milestone) milestone = milestone_contract(milestone_path) phase_slug = milestone_match.group("phase") milestone_slug = milestone_match.group("slug") phase = phase_contract(repo, phase_slug) workspace = resolve_workspace(repo, args.workspace) if workspace == repo: raise PreparationError("feature workspace must differ from the develop checkout") try: workspace.relative_to(repo) except ValueError: pass else: raise PreparationError("feature workspace must not be nested inside the develop checkout") common = git_common_dir(repo) state_root = common / "milestone-work-preparation" / milestone_slug state_path = state_root / "workspace-state.json" batch_state_path = state_root / "batch-state.json" target_milestone_path = workspace / milestone_path.relative_to(repo) state_root.mkdir(parents=True, exist_ok=True) lock_path = state_root / "workspace.lock" with lock_path.open("a+", encoding="utf-8") as lock: try: fcntl.flock(lock.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError as exc: raise PreparationError(f"workspace preparation already running: {lock_path}") from exc develop = git(repo, "config", "--get", "gitflow.branch.develop") feature_prefix = git(repo, "config", "--get", "gitflow.prefix.feature") if not develop or not feature_prefix: raise PreparationError("gitflow.branch.develop and gitflow.prefix.feature are required") current_branch = git(repo, "branch", "--show-current") if current_branch != develop: raise PreparationError(f"prepare from Git Flow develop branch: expected={develop} actual={current_branch}") ensure_clean(repo, "develop checkout") if not ref_exists(repo, f"refs/remotes/{args.remote}/{develop}") and args.dry_run: raise PreparationError(f"remote develop ref missing: {args.remote}/{develop}") if not args.dry_run: git(repo, "fetch", args.remote, develop) remote_develop = f"refs/remotes/{args.remote}/{develop}" if git(repo, "rev-parse", "HEAD") != git(repo, "rev-parse", remote_develop): raise PreparationError(f"develop checkout is not exactly synchronized with {args.remote}/{develop}") if run( ["git", "check-ignore", "--quiet", "agent-roadmap/current.md"], cwd=repo, check=False, ).returncode != 0: raise PreparationError( "agent-roadmap/current.md must be ignored for workspace-local Milestone selection" ) if not args.skip_agent_probe and not args.dry_run: probe_targets( repo, args.execution_catalog, args.planner_target, args.review_target, ) ensure_clean(repo, "develop checkout after agent probe") branch = f"{feature_prefix}{milestone_slug}" local_ref = f"refs/heads/{branch}" remote_ref = f"refs/remotes/{args.remote}/{branch}" branch_worktrees = [ item for item in worktrees(repo) if item.get("branch") == local_ref ] if branch_worktrees and Path(branch_worktrees[0]["worktree"]).resolve() != workspace: raise PreparationError( f"feature branch already belongs to another worktree: {branch_worktrees[0]['worktree']}" ) emit( "PREFLIGHT_READY", branch=branch, milestone=str(milestone_path), workspace=str(workspace), ) if args.dry_run: return 0 local_exists = ref_exists(repo, local_ref) remote_exists = ref_exists(repo, remote_ref) prior_batch = read_json(batch_state_path) if args.epics else None resuming_batch = bool( prior_batch and prior_batch.get("status") != "completed" and prior_batch.get("workspace") == str(workspace) and prior_batch.get("milestone") == str(target_milestone_path) ) if not local_exists: if remote_exists: git(repo, "branch", "--track", branch, f"{args.remote}/{branch}") else: git(repo, "branch", branch, remote_develop) if run( ["git", "merge-base", "--is-ancestor", remote_develop, local_ref], cwd=repo, check=False, ).returncode != 0: raise PreparationError(f"feature branch does not contain current {args.remote}/{develop}") if remote_exists and git(repo, "rev-parse", local_ref) != git(repo, "rev-parse", remote_ref): if not resuming_batch or run( ["git", "merge-base", "--is-ancestor", remote_ref, local_ref], cwd=repo, check=False, ).returncode != 0: raise PreparationError("existing local and remote feature branches differ") if not remote_exists: git(repo, "push", "--set-upstream", args.remote, branch) else: git(repo, "branch", "--set-upstream-to", f"{args.remote}/{branch}", branch) emit("FEATURE_BRANCH_PUSHED", branch=branch, remote=args.remote) if branch_worktrees: if not resuming_batch: ensure_clean(workspace, "feature workspace") else: if workspace.exists() and any(workspace.iterdir()): raise PreparationError(f"workspace exists and is not empty: {workspace}") workspace.parent.mkdir(parents=True, exist_ok=True) git(repo, "worktree", "add", str(workspace), branch) ensure_clean(workspace, "feature workspace") current_path = workspace / "agent-roadmap" / "current.md" current_path.parent.mkdir(parents=True, exist_ok=True) current_path.write_text( render_current( phase_slug=phase_slug, phase=phase, milestone_slug=milestone_slug, milestone=milestone, ), encoding="utf-8", ) if not resuming_batch: ensure_clean(workspace, "feature workspace") state = { "status": "workspace-ready", "milestone": str(target_milestone_path), "milestone_slug": milestone_slug, "branch": branch, "workspace": str(workspace), "execution_catalog": args.execution_catalog, "planner_target": args.planner_target, "review_target": args.review_target, } atomic_json(state_path, state) emit("WORKSPACE_READY", **state) if args.epics: return coordinate_batch( args=args, workspace=workspace, milestone=target_milestone_path, milestone_slug=milestone_slug, phase_slug=phase_slug, common=common, ) return 0 def main(argv: Iterable[str] | None = None) -> int: args = parser().parse_args(argv) try: apply_defaults(args) return prepare_existing(args) if args.existing_workspace else prepare(args) except (OSError, PreparationError) as exc: emit("FAILED", reason=str(exc)) return 2 if __name__ == "__main__": raise SystemExit(main())