#!/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<phase>[a-z0-9-]+)/milestones/(?P<slug>[a-z0-9-]+)\.md$"
)
EPIC_HEADING = re.compile(r"^### Epic:\s*\[(?P<id>[a-z0-9-]+)\]\s*(?P<title>.+?)\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:
    common_dispatcher = (
        workspace
        / "agent-ops"
        / "skills"
        / "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,
    ]
    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="prepare or reuse the target feature branch in the current workspace and start selected Epic work",
    )
    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:
        raise PreparationError("gitflow.branch.develop and gitflow.prefix.feature are required")
    if branch not in {develop, expected_branch}:
        raise PreparationError(
            "current workspace must use the Git Flow develop branch or target Milestone "
            f"feature branch: expected={develop}|{expected_branch} actual={branch or 'detached'}"
        )
    start_from_develop = branch == develop
    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 start_from_develop and resuming_batch:
            raise PreparationError("an active Epic batch must resume on its target feature branch")
        if not resuming_batch:
            ensure_clean(workspace, "current workspace")
        upstream = git(
            workspace,
            "rev-parse",
            "--abbrev-ref",
            "--symbolic-full-name",
            "@{u}",
        )
        if not upstream.endswith(f"/{branch}"):
            raise PreparationError(
                f"workspace 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 workspace branch is not synchronized with its upstream")
        if start_from_develop and not args.dry_run:
            git(workspace, "fetch", args.remote)

        local_ref = f"refs/heads/{expected_branch}"
        remote_ref = f"refs/remotes/{args.remote}/{expected_branch}"
        local_exists = ref_exists(workspace, local_ref)
        remote_exists = ref_exists(workspace, remote_ref)
        if start_from_develop:
            if upstream != f"{args.remote}/{develop}":
                raise PreparationError(
                    f"develop branch must track {args.remote}/{develop}: actual={upstream}"
                )
            remote_develop = f"refs/remotes/{args.remote}/{develop}"
            if not ref_exists(workspace, remote_develop):
                raise PreparationError(f"remote develop ref missing: {args.remote}/{develop}")
            branch_worktrees = [
                item for item in worktrees(workspace) if item.get("branch") == local_ref
            ]
            if branch_worktrees:
                raise PreparationError(
                    "target feature branch already belongs to another worktree: "
                    f"{branch_worktrees[0]['worktree']}"
                )
            if local_exists and run(
                ["git", "merge-base", "--is-ancestor", remote_develop, local_ref],
                cwd=workspace,
                check=False,
            ).returncode != 0:
                raise PreparationError(
                    f"feature branch does not contain current {args.remote}/{develop}"
                )
            if not local_exists and remote_exists and run(
                ["git", "merge-base", "--is-ancestor", remote_develop, remote_ref],
                cwd=workspace,
                check=False,
            ).returncode != 0:
                raise PreparationError(
                    f"remote feature branch does not contain current {args.remote}/{develop}"
                )
            if remote_exists and local_exists and git(
                workspace, "rev-parse", local_ref
            ) != git(workspace, "rev-parse", remote_ref):
                raise PreparationError("existing local and remote feature branches differ")
        emit(
            "PREFLIGHT_READY",
            branch=expected_branch,
            source_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, "current workspace after agent probe")
        if start_from_develop:
            if not local_exists:
                if remote_exists:
                    git(
                        workspace,
                        "branch",
                        "--track",
                        expected_branch,
                        f"{args.remote}/{expected_branch}",
                    )
                else:
                    git(workspace, "branch", expected_branch, f"{args.remote}/{develop}")
            git(workspace, "switch", expected_branch)
            if not remote_exists:
                git(workspace, "push", "--set-upstream", args.remote, expected_branch)
            else:
                git(
                    workspace,
                    "branch",
                    "--set-upstream-to",
                    f"{args.remote}/{expected_branch}",
                    expected_branch,
                )
            branch = expected_branch
            emit("FEATURE_BRANCH_PUSHED", branch=branch, remote=args.remote)
            ensure_clean(workspace, "feature workspace")
        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())
