#!/usr/bin/env python3
"""Run the event-driven preparation cycle for exactly one Milestone Epic."""

from __future__ import annotations

import argparse
from dataclasses import dataclass
import fcntl
import json
import os
from pathlib import Path
import re
import subprocess
import sys
from typing import Any, Iterable


STAGES = ("materialize", "initial-review", "refine", "final-review")
PLAN_PATTERN = "PLAN-*-G??.md"
REVIEW_PATTERN = "CODE_REVIEW-*-G??.md"
HEADER = re.compile(r"^<!--\s+(?P<body>.*?)\s+-->$")
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>.+)$")
SENSITIVE_PARTS = {".env", "secret", "secrets", "credential", "credentials", "password", "passwords"}
SENSITIVE_SUFFIXES = {".pem", ".key", ".p12"}


class CycleError(RuntimeError):
    """A fail-closed Epic cycle error."""


class TrackingRequired(CycleError):
    """A one-shot agent is still owned by an existing execution handle."""

    def __init__(self, stage: str, result: dict[str, Any]) -> None:
        self.stage = stage
        self.result = result
        super().__init__(f"agent execution is still running: stage={stage}")


class TrackingRecoveryRequired(TrackingRequired):
    """A detached execution ended and its artifacts need explicit adoption."""


@dataclass(frozen=True)
class Epic:
    epic_id: str
    title: str
    task_ids: tuple[str, ...]
    incomplete_ids: tuple[str, ...]
    body: 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 CycleError(f"command failed ({result.returncode}): {' '.join(command)}: {detail}")
    return result


def git(workspace: Path, *arguments: str, check: bool = True) -> str:
    result = run(["git", *arguments], cwd=workspace, check=check)
    return (result.stdout or "").strip()


def resolve_workspace(raw: str) -> Path:
    workspace = Path(raw).expanduser().resolve()
    if not workspace.is_dir():
        raise CycleError(f"workspace not found: {workspace}")
    top = Path(git(workspace, "rev-parse", "--show-toplevel")).resolve()
    if top != workspace:
        raise CycleError(f"workspace must be git root: expected={top} actual={workspace}")
    branch = git(workspace, "branch", "--show-current")
    if not branch:
        raise CycleError("detached HEAD is not supported")
    develop = git(workspace, "config", "--get", "gitflow.branch.develop")
    feature_prefix = git(workspace, "config", "--get", "gitflow.prefix.feature")
    if not develop or not feature_prefix or not branch.startswith(feature_prefix):
        raise CycleError(f"Epic preparation requires a Git Flow feature branch: actual={branch}")
    return workspace


def resolve_milestone(workspace: Path, raw: str) -> tuple[Path, re.Match[str]]:
    candidate = Path(raw).expanduser()
    path = (workspace / candidate).resolve() if not candidate.is_absolute() else candidate.resolve()
    try:
        relative = path.relative_to(workspace).as_posix()
    except ValueError as exc:
        raise CycleError(f"milestone outside workspace: {path}") from exc
    match = MILESTONE_PATTERN.fullmatch(relative)
    if match is None or not path.is_file():
        raise CycleError(f"active milestone path required: {relative}")
    return path, match


def section(text: str, heading: str) -> str:
    match = re.search(
        rf"^## {re.escape(heading)}\s*$\n(?P<body>.*?)(?=^##\s|\Z)",
        text,
        re.MULTILINE | re.DOTALL,
    )
    return match.group("body").strip() if match else ""


def verify_milestone_gate(text: str) -> None:
    status_body = section(text, "상태")
    status_match = re.search(r"^\[(.+?)\]\s*$", status_body, re.MULTILINE)
    status = status_match.group(1).strip() if status_match else ""
    if status not in {"계획", "진행중"}:
        raise CycleError(f"milestone must be [계획] or [진행중]: actual={status or 'missing'}")
    lock = section(text, "구현 잠금")
    if not re.search(r"^- 상태:\s*해제\s*$", lock, re.MULTILINE):
        raise CycleError("milestone implementation lock is not 해제")
    if not re.search(r"^- 결정 필요:\s*없음\s*$", lock, re.MULTILINE):
        raise CycleError("milestone has unresolved 결정 필요")


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)
        body_lines = lines[start + 1 : end]
        tasks = [TASK_LINE.fullmatch(line) for line in body_lines]
        task_matches = [value for value in tasks 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") == " "
                ),
                body="\n".join(body_lines).strip(),
            )
        )
    return epics


def select_epic(epics: list[Epic], selector: str) -> Epic:
    exact_id = [epic for epic in epics if epic.epic_id == selector]
    if len(exact_id) == 1:
        return exact_id[0]
    normalized = selector.casefold().strip()
    exact_title = [epic for epic in epics if epic.title.casefold().strip() == normalized]
    if len(exact_title) == 1:
        return exact_title[0]
    raise CycleError(f"target Epic must resolve exactly once: selector={selector}")


def git_common_dir(workspace: Path) -> Path:
    raw = Path(git(workspace, "rev-parse", "--git-common-dir"))
    return (workspace / 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_state(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 CycleError(f"invalid cycle state: {path}: {exc}") from exc
    if not isinstance(value, dict):
        raise CycleError(f"cycle state must be 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 stage_result(path: Path, *, workspace: Path, label: str) -> dict[str, Any] | None:
    value = read_state(path)
    if value is None:
        return None
    if value.get("workspace") != str(workspace) or value.get("label") != label:
        raise CycleError(f"agent result identity mismatch: {path}")
    locator = value.get("locator")
    if not isinstance(locator, str) or not Path(locator).is_file():
        raise CycleError(f"agent result locator is missing: {path}")
    return value


def changed_paths(workspace: Path) -> list[str]:
    unmerged = git(workspace, "diff", "--name-only", "--diff-filter=U")
    if unmerged:
        raise CycleError(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 CycleError(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 parse_header(path: Path) -> dict[str, str]:
    try:
        first = path.read_text(encoding="utf-8").splitlines()[0]
    except (OSError, IndexError) as exc:
        raise CycleError(f"missing first-line metadata: {path}") from exc
    match = HEADER.fullmatch(first)
    if not match:
        raise CycleError(f"invalid first-line metadata: {path}")
    values: dict[str, str] = {}
    for token in match.group("body").split():
        key, separator, value = token.partition("=")
        if separator:
            values[key] = value
    required = {"task", "plan", "tag", "milestone-task"}
    missing = sorted(required - values.keys())
    if missing:
        raise CycleError(f"metadata fields missing in {path}: {','.join(missing)}")
    return values


def active_pairs(workspace: Path, task_group: str) -> list[tuple[Path, Path, dict[str, str]]]:
    root = workspace / "agent-task" / task_group
    if not root.exists():
        return []
    directories = [root, *sorted(path for path in root.iterdir() if path.is_dir())]
    pairs: list[tuple[Path, Path, dict[str, str]]] = []
    for directory in directories:
        plans = sorted(directory.glob(PLAN_PATTERN))
        reviews = sorted(directory.glob(REVIEW_PATTERN))
        if not plans and not reviews:
            continue
        if len(plans) != 1 or len(reviews) != 1:
            raise CycleError(f"active PLAN/CODE_REVIEW pair required: {directory}")
        plan_header = parse_header(plans[0])
        review_header = parse_header(reviews[0])
        if plan_header != review_header:
            raise CycleError(f"PLAN/CODE_REVIEW metadata mismatch: {directory}")
        if plan_header["task"].split("/", 1)[0] != task_group:
            raise CycleError(f"task group mismatch: {plans[0]}")
        pairs.append((plans[0], reviews[0], plan_header))
    return pairs


def validate_pairs(
    workspace: Path,
    task_group: str,
    epic_task_ids: set[str],
    allowed_task_ids: set[str] | None = None,
) -> tuple[list[tuple[Path, Path, dict[str, str]]], set[str]]:
    all_pairs = active_pairs(workspace, task_group)
    allowed = set(epic_task_ids) if allowed_task_ids is None else set(allowed_task_ids)
    if not epic_task_ids <= allowed:
        raise CycleError("target Epic Task ids must be inside the selected batch")
    pairs: list[tuple[Path, Path, dict[str, str]]] = []
    union: set[str] = set()
    dispatcher_root = (
        workspace / "agent-ops" / "skills" / "common" / "orchestrate-agent-task-loop"
    )
    dispatcher = dispatcher_root / "scripts" / "dispatch.py"
    for plan, review, header in all_pairs:
        ids = header["milestone-task"].split(",")
        if not ids or any(not value for value in ids) or len(ids) != len(set(ids)):
            raise CycleError(f"invalid milestone-task list: {plan}")
        pair_ids = set(ids)
        outside = sorted(pair_ids - allowed)
        if outside:
            raise CycleError(
                f"plan includes Task ids outside selected Epic batch: {plan}: {','.join(outside)}"
            )
        target_ids = pair_ids & epic_task_ids
        if target_ids and target_ids != pair_ids:
            raise CycleError(f"plan crosses target Epic boundary: {plan}")
        if target_ids:
            pairs.append((plan, review, header))
            union.update(target_ids)
        contents = plan.read_text(encoding="utf-8") + "\n" + review.read_text(encoding="utf-8")
        if "[TODO" in contents or "<task_group>" in contents or "<milestone-slug>" in contents:
            raise CycleError(f"unresolved template token: {plan.parent}")
        if dispatcher.is_file():
            run(
                [sys.executable, str(dispatcher), "--workspace", str(workspace), "--validate-plan", str(plan)],
                cwd=workspace,
            )
    git(workspace, "diff", "--check")
    return pairs, union


def batch_task_ids(raw: str | None, epic_task_ids: tuple[str, ...]) -> set[str]:
    target = set(epic_task_ids)
    if raw is None:
        return target
    values = raw.split(",")
    if not values or any(not value for value in values) or len(values) != len(set(values)):
        raise CycleError("--batch-task-ids must be a unique comma-separated Task id list")
    allowed = set(values)
    if not target <= allowed:
        raise CycleError("--batch-task-ids does not include every target Epic Task id")
    return allowed


def active_task_user_reviews(workspace: Path, task_group: str) -> list[Path]:
    root = workspace / "agent-task" / task_group
    if not root.exists():
        return []
    return sorted(root.glob("USER_REVIEW.md")) + sorted(root.glob("*/USER_REVIEW.md"))


def sdd_user_review(workspace: Path, phase_slug: str, milestone_slug: str) -> Path:
    return workspace / "agent-roadmap" / "sdd" / phase_slug / milestone_slug / "USER_REVIEW.md"


def stage_prompt(
    *,
    stage: str,
    workspace: Path,
    milestone: Path,
    epic: Epic,
    task_group: str,
    base_head: str,
    checkpoint_head: str | None,
) -> str:
    common = f"""You are a fresh child agent launched for one bounded Epic preparation stage, not the caller or monitor.
Work only in {workspace}.
Read the repository AGENTS.md completely, then read agent-ops/skills/common/router.md and only the skills required for this stage.
Do not start subagents, orchestration dispatchers, prepare-milestone-workspace, prepare-epic-work-items, or any monitoring loop.
Do not commit or push; the parent runtime owns Git checkpoints.
Target Milestone: {milestone}
Target Epic: [{epic.epic_id}] {epic.title}
Allowed Milestone Task ids: {','.join(epic.task_ids)}
Active task group: agent-task/{task_group}
Keep every change inside this Epic and preserve user changes. Final in Korean.
"""
    if stage == "materialize":
        return common + f"""
Materialize this Epic once. Read current source, tests, SDD, matching spec and contracts required by AGENTS.md.
Classify cohesive slices as direct-small only when each is one bounded change, has explicit verification, changes no API/wire/schema/migration/external side effect/responsibility boundary, needs no user decision, and does not collide with planned work. Treat every uncertain slice as large.
Implement and verify all direct-small slices first. Then, against that updated source, use agent-ops/skills/common/plan/SKILL.md in write mode to create valid PLAN/CODE_REVIEW pairs for every remaining large slice. Preserve exact milestone-task ids and let plan perform final routing. Do not use official code-review on unimplemented stubs.
If a genuine product/scope decision is required, use roadmap-sdd review-ready for this Milestone and stop without inventing a decision.
Starting HEAD: {base_head}
"""
    if stage == "initial-review":
        return common + f"""
Review everything produced for this Epic since {base_head}, including direct code/test/document changes and every active PLAN/CODE_REVIEW stub. This is the explicit self-review request: review the work and fix every material omission you find.
Re-run appropriate verification for direct-small work. For semantic plan defects, use the plan skill's explicit write/replan path so routing and paired files remain valid. Do not append an official code-review verdict.
Run sync-milestone-workstate mode=sync only for exact Task ids whose direct work is fully implemented and evidenced; never complete an id that still has pending plan scope.
If no material defect exists, leave correct artifacts unchanged.
"""
    if stage == "refine":
        return common + """
Read agent-ops/skills/common/refine-plans/SKILL.md and apply it once to every eligible unstarted active pair in the target task group whose milestone-task ids belong to this Epic. Preserve original scope and do not re-read source/tests or run verification. A justified no-change decision is valid. Do not recursively split a child created in this pass.
"""
    if stage == "final-review":
        return common + f"""
Review the refined active pair set for this Epic from a fresh context. Compare the child scope union, milestone-task union, write sets, verification, dependencies, indices, PLAN/CODE_REVIEW metadata, and routing against the Milestone, SDD, current source, and the pre-refine intent at checkpoint {checkpoint_head or base_head}. Fix every material defect using the owning plan/refine procedure; do not append an official code-review verdict. If a semantic replan replaces a pair, apply refine-plans once to that replacement when it remains eligible. Finish only with valid unstarted pairs or no pairs when all Epic work was direct-small.
"""
    raise CycleError(f"unsupported stage: {stage}")


def run_agent_stage(
    *,
    workspace: Path,
    state_root: Path,
    identity: str,
    stage: str,
    prompt: str,
    execution_catalog: str,
    target_id: str,
    prior_cycle_status: str,
    retry: bool,
) -> Path:
    runner = Path(__file__).resolve().with_name("run_agent_once.py")
    prompt_path = state_root / "prompts" / f"{stage}.txt"
    result_path = state_root / "attempts" / f"{stage}.json"
    prompt_path.parent.mkdir(parents=True, exist_ok=True)
    prompt_path.write_text(prompt, encoding="utf-8")
    label = f"{identity}-{stage}"
    previous = stage_result(result_path, workspace=workspace, label=label)
    if previous is not None:
        previous_status = previous.get("status")
        is_live = previous_status in {"running", "tracking"} and process_is_same(
            previous.get("agent_pid"), previous.get("agent_process_start_token")
        )
        if is_live:
            raise TrackingRequired(stage, previous)
        if previous_status == "succeeded" and not (
            retry and prior_cycle_status == "failed"
        ):
            emit(
                "AGENT_RESULT_RECOVERED",
                stage=stage,
                locator=previous["locator"],
                result="succeeded",
            )
            return result_path
        if previous_status in {"running", "tracking"} and prior_cycle_status in {
            "running",
            "tracking",
        }:
            if not retry:
                raise TrackingRecoveryRequired(stage, previous)
            emit(
                "AGENT_RESULT_RECOVERED",
                stage=stage,
                locator=previous["locator"],
                result="detached-artifacts",
            )
            return result_path
        if not retry:
            raise CycleError(
                f"prior agent result requires --retry: stage={stage} status={previous_status}"
            )
    command = [
        sys.executable,
        str(runner),
        "--execution-catalog",
        execution_catalog,
        "--target-id",
        target_id,
        "--workspace",
        str(workspace),
        "--prompt-file",
        str(prompt_path),
        "--label",
        label,
        "--result-file",
        str(result_path),
    ]
    result = run(command, cwd=workspace, check=False, capture=False)
    if result.returncode != 0:
        if result.returncode == 3:
            tracked = stage_result(result_path, workspace=workspace, label=label)
            if tracked is None:
                raise CycleError(f"agent tracking result missing: stage={stage}")
            raise TrackingRequired(stage, tracked)
        raise CycleError(f"agent stage failed: stage={stage} exit={result.returncode}")
    final = stage_result(result_path, workspace=workspace, label=label)
    if final is None or final.get("status") != "succeeded":
        raise CycleError(f"agent stage returned without succeeded result: stage={stage}")
    return result_path


def publish(workspace: Path, epic: Epic, phase: str) -> str:
    paths = changed_paths(workspace)
    if paths:
        unsafe = [path for path in paths if sensitive(path)]
        if unsafe:
            raise CycleError(f"sensitive path refused: {','.join(unsafe)}")
        for path in paths:
            git(workspace, "add", "--", path)
        git(workspace, "diff", "--cached", "--check")
        message = (
            f"feat(epic): {epic.epic_id} 작업을 준비한다"
            if phase == "initial"
            else f"chore(epic): {epic.epic_id} 준비 결과를 검증한다"
        )
        git(workspace, "commit", "-m", message)
    branch = git(workspace, "branch", "--show-current")
    upstream = git(workspace, "rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}")
    if not upstream:
        raise CycleError(f"feature branch has no upstream: {branch}")
    if not upstream.endswith(f"/{branch}"):
        raise CycleError(f"feature branch upstream mismatch: branch={branch} upstream={upstream}")
    git(workspace, "push")
    if changed_paths(workspace):
        raise CycleError("workspace is dirty after publish")
    return git(workspace, "rev-parse", "HEAD")


def parser() -> argparse.ArgumentParser:
    value = argparse.ArgumentParser(description=__doc__)
    value.add_argument("--workspace", required=True)
    value.add_argument("--milestone", required=True)
    value.add_argument("--epic", required=True)
    value.add_argument(
        "--execution-catalog",
        default=os.environ.get("AGENT_TASK_EXECUTION_CATALOG"),
    )
    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(
        "--batch-task-ids",
        help="internal selected-Epic Task id union; permits earlier Epic pairs in the same batch",
    )
    value.add_argument(
        "--validate-only",
        action="store_true",
        help="validate the selected Epic against the current batch without running an agent",
    )
    value.add_argument("--retry", action="store_true")
    return value


def apply_defaults(args: argparse.Namespace) -> argparse.Namespace:
    if not args.execution_catalog:
        raise CycleError(
            "--execution-catalog or AGENT_TASK_EXECUTION_CATALOG is required"
        )
    if not args.planner_target:
        raise CycleError(
            "--planner-target or AGENT_TASK_PLANNER_TARGET is required"
        )
    if args.review_target is None:
        args.review_target = args.planner_target
    return args


def cycle(args: argparse.Namespace) -> int:
    apply_defaults(args)
    workspace = resolve_workspace(args.workspace)
    milestone_path, milestone_match = resolve_milestone(workspace, args.milestone)
    milestone_text = milestone_path.read_text(encoding="utf-8")
    verify_milestone_gate(milestone_text)
    epic = select_epic(parse_epics(milestone_text), args.epic)
    milestone_slug = milestone_match.group("slug")
    phase_slug = milestone_match.group("phase")
    task_group = f"m-{milestone_slug}"
    identity = f"{milestone_slug}:{epic.epic_id}"
    if not epic.task_ids:
        raise CycleError(f"target Epic has no Task ids: {epic.epic_id}")
    allowed_task_ids = batch_task_ids(args.batch_task_ids, epic.task_ids)
    feature_prefix = git(workspace, "config", "--get", "gitflow.prefix.feature")
    branch = git(workspace, "branch", "--show-current")
    expected_branch = f"{feature_prefix}{milestone_slug}"
    if branch != expected_branch:
        raise CycleError(f"workspace branch does not match Milestone slug: expected={expected_branch} actual={branch}")
    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 CycleError(f"workspace-local current does not select target Milestone: {current_path}")
    if args.validate_only:
        pairs, task_union = validate_pairs(
            workspace,
            task_group,
            set(epic.task_ids),
            allowed_task_ids,
        )
        completed_with_plan = sorted(task_union - set(epic.incomplete_ids))
        if completed_with_plan:
            raise CycleError(
                "completed Task ids still have active plans: " + ",".join(completed_with_plan)
            )
        remaining_without_plan = sorted(set(epic.incomplete_ids) - task_union)
        if remaining_without_plan:
            raise CycleError(
                "incomplete Epic Task ids have neither completion sync nor active plans: "
                + ",".join(remaining_without_plan)
            )
        emit(
            "EPIC_BATCH_VALIDATED",
            identity=identity,
            event="EPIC_COMPLETED" if not epic.incomplete_ids else "EPIC_WORK_ITEMS_READY",
            plans=len(pairs),
        )
        return 0
    state_root = git_common_dir(workspace) / "epic-work-preparation" / milestone_slug / epic.epic_id
    state_path = state_root / "state.json"
    state_root.mkdir(parents=True, exist_ok=True)
    with (state_root / "cycle.lock").open("a+", encoding="utf-8") as lock:
        try:
            fcntl.flock(lock.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError as exc:
            raise CycleError(f"Epic cycle already running: {identity}") from exc

        state = read_state(state_path)
        current_head = git(workspace, "rev-parse", "HEAD")
        if state and state.get("identity") != identity:
            raise CycleError(f"cycle state identity mismatch: {state_path}")
        if (
            state
            and state.get("task_ids") is not None
            and state.get("task_ids") != list(epic.task_ids)
        ):
            raise CycleError("target Epic Task ids changed after cycle scope was fixed")
        if (
            state
            and state.get("status") != "completed"
            and state.get("batch_task_ids") is not None
            and state.get("batch_task_ids") != sorted(allowed_task_ids)
        ):
            raise CycleError("selected Epic batch Task ids changed after cycle scope was fixed")
        if state and state.get("status") == "completed":
            if changed_paths(workspace):
                raise CycleError("completed cycle requires a clean feature workspace")
            refreshed = select_epic(
                parse_epics(milestone_path.read_text(encoding="utf-8")), epic.epic_id
            )
            pairs, task_union = validate_pairs(
                workspace,
                task_group,
                set(refreshed.task_ids),
                allowed_task_ids,
            )
            completed_with_plan = sorted(task_union - set(refreshed.incomplete_ids))
            if completed_with_plan:
                raise CycleError(
                    "completed Task ids still have active plans: " + ",".join(completed_with_plan)
                )
            remaining_without_plan = sorted(set(refreshed.incomplete_ids) - task_union)
            if remaining_without_plan:
                raise CycleError(
                    "completed cycle no longer has evidence for incomplete Task ids; run workstate sync or recover plans: "
                    + ",".join(remaining_without_plan)
                )
            terminal = "EPIC_COMPLETED" if not refreshed.incomplete_ids else "EPIC_WORK_ITEMS_READY"
            state.update(event=terminal, head=current_head)
            atomic_json(state_path, state)
            emit(terminal, identity=identity, resumed=True, head=current_head, plans=len(pairs))
            return 0
        review_path = sdd_user_review(workspace, phase_slug, milestone_slug)
        if state and state.get("status") == "user-review" and review_path.exists():
            emit("USER_REVIEW", identity=identity, path=str(review_path), resumed=True)
            return 2
        prior_cycle_status = str(state.get("status")) if state else "new"
        if state and state.get("status") == "failed" and not args.retry:
            raise CycleError(f"prior terminal failure requires --retry: {state.get('reason', 'unknown')}")
        if state is None:
            if changed_paths(workspace):
                raise CycleError("clean feature workspace required before a new Epic cycle")
            if review_path.exists():
                emit("USER_REVIEW", identity=identity, path=str(review_path))
                return 2
            if active_task_user_reviews(workspace, task_group):
                raise CycleError("preparation cannot resume from agent-task USER_REVIEW")
            if not epic.incomplete_ids:
                existing, _ = validate_pairs(
                    workspace,
                    task_group,
                    set(epic.task_ids),
                    allowed_task_ids,
                )
                if existing:
                    raise CycleError(
                        "completed Epic still has active PLAN/CODE_REVIEW pairs; reconcile them before completion"
                    )
                state = {
                    "identity": identity,
                    "task_ids": list(epic.task_ids),
                    "batch_task_ids": sorted(allowed_task_ids),
                    "status": "completed",
                    "event": "EPIC_COMPLETED",
                    "head": current_head,
                }
                atomic_json(state_path, state)
                emit("EPIC_COMPLETED", identity=identity)
                return 0
            existing, _ = validate_pairs(
                workspace,
                task_group,
                set(epic.task_ids),
                allowed_task_ids,
            )
            if existing:
                raise CycleError("active pair already exists before new Epic cycle; select recovery explicitly")
            state = {
                "identity": identity,
                "task_ids": list(epic.task_ids),
                "batch_task_ids": sorted(allowed_task_ids),
                "status": "active",
                "next_stage": STAGES[0],
                "base_head": current_head,
                "checkpoint_head": None,
                "pre_refine_ids": [],
            }
            atomic_json(state_path, state)
            emit(
                "EPIC_SCOPE_RESOLVED",
                identity=identity,
                task_ids=list(epic.task_ids),
                incomplete_ids=list(epic.incomplete_ids),
            )
        elif changed_paths(workspace) and not args.retry:
            raise CycleError("dirty recovery state requires explicit --retry")

        reviewer_target = args.review_target or args.planner_target
        start_index = STAGES.index(str(state.get("next_stage", STAGES[0])))
        for stage in STAGES[start_index:]:
            stage_head = git(workspace, "rev-parse", "HEAD")
            event_prefix = stage.upper().replace("-", "_")
            emit(f"{event_prefix}_STARTED", identity=identity)
            target_id = (
                args.planner_target
                if stage in {"materialize", "refine"}
                else reviewer_target
            )
            prompt = stage_prompt(
                stage=stage,
                workspace=workspace,
                milestone=milestone_path,
                epic=epic,
                task_group=task_group,
                base_head=str(state["base_head"]),
                checkpoint_head=state.get("checkpoint_head"),
            )
            result_path = state_root / "attempts" / f"{stage}.json"
            state.update(
                status="running",
                current_stage=stage,
                active_result=str(result_path),
            )
            atomic_json(state_path, state)
            try:
                run_agent_stage(
                    workspace=workspace,
                    state_root=state_root,
                    identity=identity.replace(":", "-"),
                    stage=stage,
                    prompt=prompt,
                    execution_catalog=args.execution_catalog,
                    target_id=target_id,
                    prior_cycle_status=prior_cycle_status,
                    retry=args.retry,
                )
            except TrackingRequired as exc:
                state.update(
                    status="tracking",
                    current_stage=stage,
                    active_result=str(result_path),
                    locator=exc.result.get("locator"),
                )
                atomic_json(state_path, state)
                if isinstance(exc, TrackingRecoveryRequired):
                    state["recovery_required"] = True
                    atomic_json(state_path, state)
                    emit(
                        "AGENT_RECOVERY_REQUIRED",
                        identity=identity,
                        stage=stage,
                        locator=exc.result.get("locator"),
                        action="inspect locator, then rerun with --retry to adopt artifacts",
                    )
                else:
                    state.pop("recovery_required", None)
                    atomic_json(state_path, state)
                    emit(
                        "AGENT_TRACKING",
                        identity=identity,
                        stage=stage,
                        locator=exc.result.get("locator"),
                        pid=exc.result.get("agent_pid"),
                    )
                return 3
            if git(workspace, "rev-parse", "HEAD") != stage_head:
                raise CycleError(f"child agent committed unexpectedly: stage={stage}")
            refreshed_scope = select_epic(
                parse_epics(milestone_path.read_text(encoding="utf-8")), epic.epic_id
            )
            if refreshed_scope.task_ids != epic.task_ids:
                raise CycleError(f"target Epic Task ids changed unexpectedly: stage={stage}")
            pairs, task_union = validate_pairs(
                workspace,
                task_group,
                set(epic.task_ids),
                allowed_task_ids,
            )
            if active_task_user_reviews(workspace, task_group):
                raise CycleError("preparation agent created forbidden agent-task USER_REVIEW")
            if review_path.exists():
                head = publish(workspace, epic, "user-review")
                state.update(status="user-review", event="USER_REVIEW", head=head, next_stage=stage)
                atomic_json(state_path, state)
                emit("USER_REVIEW", identity=identity, path=str(review_path), head=head)
                return 2
            if stage == "materialize" and not changed_paths(workspace) and not pairs:
                raise CycleError("materialize produced neither direct work nor PLAN pairs")
            if stage == "refine":
                expected_union = set(state.get("pre_refine_ids", []))
                if task_union != expected_union:
                    raise CycleError(
                        f"refine changed milestone-task union: before={sorted(expected_union)} after={sorted(task_union)}"
                    )
            if stage == "final-review":
                expected_union = set(state.get("pre_refine_ids", []))
                if task_union != expected_union:
                    raise CycleError(
                        f"final review changed milestone-task union: before={sorted(expected_union)} after={sorted(task_union)}"
                    )
            emit(f"{event_prefix}_FINISHED", identity=identity, plans=len(pairs))
            state.pop("current_stage", None)
            state.pop("active_result", None)
            state.pop("locator", None)
            state.pop("recovery_required", None)

            if stage == "initial-review":
                state["pre_refine_ids"] = sorted(task_union)
                checkpoint = publish(workspace, epic, "initial")
                state["checkpoint_head"] = checkpoint
                emit("INITIAL_CHECKPOINT_PUSHED", identity=identity, head=checkpoint)
            elif stage == "final-review":
                final_head = publish(workspace, epic, "final")
                emit("FINAL_ARTIFACTS_PUSHED", identity=identity, head=final_head)
                refreshed = select_epic(
                    parse_epics(milestone_path.read_text(encoding="utf-8")), epic.epic_id
                )
                completed_with_plan = sorted(task_union - set(refreshed.incomplete_ids))
                if completed_with_plan:
                    raise CycleError(
                        "completed Task ids still have active plans: "
                        + ",".join(completed_with_plan)
                    )
                remaining_without_plan = sorted(set(refreshed.incomplete_ids) - task_union)
                if remaining_without_plan:
                    raise CycleError(
                        "incomplete Epic Task ids have neither completion sync nor active plans: "
                        + ",".join(remaining_without_plan)
                    )
                terminal = "EPIC_COMPLETED" if not refreshed.incomplete_ids else "EPIC_WORK_ITEMS_READY"
                state.update(status="completed", event=terminal, head=final_head, next_stage=None)
                atomic_json(state_path, state)
                emit(terminal, identity=identity, head=final_head, plans=len(pairs))
                return 0

            next_index = STAGES.index(stage) + 1
            state["next_stage"] = STAGES[next_index]
            state["status"] = "active"
            atomic_json(state_path, state)
            prior_cycle_status = "active"
        raise CycleError("cycle ended without terminal state")


def main(argv: Iterable[str] | None = None) -> int:
    args = parser().parse_args(argv)
    state_path: Path | None = None
    identity = "unknown"
    try:
        apply_defaults(args)
        return cycle(args)
    except (CycleError, OSError, ValueError) as exc:
        try:
            if not args.validate_only:
                workspace = Path(args.workspace).expanduser().resolve()
                milestone = Path(args.milestone)
                if not milestone.is_absolute():
                    milestone = (workspace / milestone).resolve()
                match = MILESTONE_PATTERN.fullmatch(milestone.relative_to(workspace).as_posix())
                if match:
                    selected = select_epic(
                        parse_epics(milestone.read_text(encoding="utf-8")), args.epic
                    )
                    identity = f"{match.group('slug')}:{selected.epic_id}"
                    state_path = (
                        git_common_dir(workspace)
                        / "epic-work-preparation"
                        / match.group("slug")
                        / selected.epic_id
                        / "state.json"
                    )
                    prior = read_state(state_path) or {"identity": identity}
                    prior.update(status="failed", event="FAILED", reason=str(exc))
                    atomic_json(state_path, prior)
        except Exception:
            pass
        emit("FAILED", identity=identity, reason=str(exc), state=str(state_path) if state_path else None)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())
