"""release-handoff stage-group 모드의 강제 지점.

자격 판정(eligible) · 수집 브랜치 생성+머지(assemble) · verified/pr 행 기록을
단일 모듈로 강제한다. lead 는 `okstra handoff <sub>` 로 호출만 한다.
설계: docs/superpowers/specs/2026-06-10-stage-group-handoff-design.md
"""

from __future__ import annotations

import json
import subprocess
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Tuple

from . import consumers, stage_targets, worktree_registry
from .final_report_paths import final_report_markdown_path, final_report_data_path
from .handoff_verification import (
    VerificationEvidenceError, load_latest_verification,
    require_finished_verification, verification_row_is_current,
)
from .paths import RunRef
from .json_boundary import JsonBoundaryError, load_owned_object
from .release_gate import (
    blocking_condition_ids,
    release_handoff_allowed,
    verdict_token,
)
from .stage_map import StageMapError, parse_stage_map_file, stage_map_records
from .worktree import (compute_branch_name, compute_worktree_path,
                       main_worktree_path, is_dirty_excluding_okstra,
                       nested_worktree_excludes, is_ancestor, merge_branch,
                       remove_worktree_force, MergeError, _git)


class HandoffError(Exception):
    """자격/전제 위반 — exit 1, actionable 메시지."""


class HandoffConflict(Exception):
    """stage 간 merge 충돌 — exit 2, 충돌 경로 동봉."""

    def __init__(self, stage: int, branch: str, paths: List[str]):
        self.stage = stage
        self.branch = branch
        self.paths = paths
        super().__init__(
            f"merge conflict while merging stage {stage} ({branch}): "
            f"{', '.join(paths)}")


def group_id_for(stages: List[int]) -> str:
    return "g" + "-".join(str(n) for n in sorted(stages))


def compute_eligibility(stage_map: List[Dict[str, Any]],
                        rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """stage 별 PR 가능 여부와 차단 사유. 의존 폐포는 선택 집합의 속성이라
    여기서는 판정하지 않는다 (assemble 의 check_dependency_closure 가 담당)."""
    eligibility = stage_targets.stage_lifecycle_snapshot_from_rows(
        stage_map,
        rows,
    ).handoff_eligibility()
    last = {row["stage"]: row for row in rows if row.get("status") == "verified"}
    for item in eligibility:
        if item["eligible"] and not verification_row_is_current(last.get(item["stage"], {})):
            item["eligible"] = False
            item["reasons"].append("verification-stale-or-unproven: re-run final-verification")
    return eligibility


def latest_whole_task_fv_release_ready(project_root, project_id: str,
                                       task_group: str, task_id: str) -> str:
    """release-handoff 로 넘어가도 되는 whole-task 검증 보고서 경로. 없으면 ''.

    최신 실행을 먼저 고르고 대상·보고서·완료 판정을 확인한다.
    실패한 최신 실행을 건너뛰어 과거 통과 보고서를 사용하지 않는다."""
    from okstra_project.state import find_task_root
    root = find_task_root(Path(project_root),
                          f"{project_id}:{task_group}:{task_id}")
    if root is None:
        return ""
    try:
        report, manifest, data = load_latest_verification(
            RunRef.from_task_root(root, "final-verification").run_dir)
        require_finished_verification(manifest, data)
        if manifest.get("taskKey") != f"{project_id}:{task_group}:{task_id}":
            return ""
        return str(report)
    except VerificationEvidenceError:
        # expected-miss: 최신 검증 미완료·차단은 전체 인계 선택지가 없는 정상 분기다.
        return ""


def check_dependency_closure(
    selected: List[int],
    stage_map: List[Dict[str, Any]],
    done_by_stage: Dict[int, Dict[str, Any]],
    is_merged_to_base: Callable[[str], bool],
) -> List[Tuple[int, int]]:
    """선택 집합의 의존 폐포 위반 목록 [(stage, 미충족 선행)].
    선행이 같은 그룹에 선택됐거나 이미 base 에 머지된 경우만 허용."""
    sel = set(selected)
    by_n = {s["stage_number"]: s for s in stage_map}
    violations = []
    for n in sorted(selected):
        for d in by_n[n]["depends_on"]:
            if d in sel:
                continue
            head = (done_by_stage.get(d) or {}).get("head_commit", "")
            if not head or not is_merged_to_base(head):
                violations.append((n, d))
    return violations


def _run_git(args: List[str], cwd, check: bool = True) -> subprocess.CompletedProcess:
    """worktree._git(공용 러너) 위에 check=True 시 HandoffError 변환만 얹는다."""
    r = _git(Path(cwd), *args)
    if check and r.returncode != 0:
        raise HandoffError(f"git {' '.join(args)} failed: {r.stderr.strip()}")
    return r


def _require_eligible(stage_map, rows, stages) -> Dict[int, Dict[str, Any]]:
    elig = {e["stage"]: e for e in compute_eligibility(stage_map, rows)}
    unknown = [n for n in stages if n not in elig]
    if unknown:
        raise HandoffError(f"stages not in Stage Map: {unknown}")
    bad = {n: elig[n]["reasons"] for n in stages if elig[n]["reasons"]}
    if bad:
        raise HandoffError(f"stages not eligible: {bad}")
    return elig


def _require_closure(stages, stage_map, done, project_root, base_branch) -> None:
    def merged(commit: str) -> bool:
        return is_ancestor(project_root, commit, f"origin/{base_branch}")
    violations = check_dependency_closure(stages, stage_map, done, merged)
    if violations:
        lines = [
            f"stage {n} depends on stage {d} which is neither selected nor "
            f"merged into origin/{base_branch} — include stage {d} in the "
            "group or PR-merge it first"
            for n, d in violations
        ]
        raise HandoffError("dependency closure violated:\n" + "\n".join(lines))


def _reuse_existing(entry, stages, done, project_root) -> Dict[str, Any]:
    head = _run_git(["rev-parse", entry.branch], project_root).stdout.strip()
    for n in stages:
        h = (done.get(n) or {}).get("head_commit", "")
        if not is_ancestor(project_root, h, head):
            raise HandoffError(
                f"collector branch {entry.branch} exists but stage {n} head "
                f"{h} is not merged into it — remove the branch/worktree and "
                "the registry group key, then retry")
    return {"ok": True, "reused": True, "group_id": group_id_for(stages),
            "branch": entry.branch, "worktree_path": entry.worktree_path,
            "head": head, "stages": sorted(stages), "merge_commits": []}


def _cleanup_group(project_root, wt_path, branch, project_id, task_group,
                   task_id, gid) -> None:
    remove_worktree_force(project_root, wt_path)
    _run_git(["branch", "-D", branch], project_root, check=False)
    worktree_registry.release(project_id, task_group, task_id, group_id=gid)


def _merge_stages(wt_path, stages, work_category, task_id, project_root,
                  project_id, task_group, gid, branch, done) -> List[str]:
    """선택 stage 들을 수집 브랜치에 위상순 --no-ff 머지한다.

    머지 대상은 `-sN` 브랜치 tip 이 아니라 기록된 done.head_commit 이다 —
    consumers 의 done.head_commit 은 git-reconcile 로 브랜치 tip 과 다른 commit
    일 수 있어, 브랜치 tip 을 머지하면 done 시점 이후의 갈라진 변경을 끌어들인다.
    stage_branch 는 충돌 식별용으로만 산출한다(머지 대상 아님)."""
    merge_commits = []
    for n in sorted(stages):
        stage_branch = compute_branch_name(
            work_category=work_category, task_id_segment=task_id,
            stage_number=n)
        done_commit = (done.get(n) or {}).get("head_commit", "")
        try:
            conflicts = merge_branch(wt_path, done_commit, no_ff=False)
        except MergeError as exc:
            # cleanup 은 assemble 의 except Exception 이 수행한다(worktree-add
            # 실패 경로와 동일 처리) — 여기서 직접 정리하지 않는다.
            raise HandoffError(
                f"stage {n} ({stage_branch}) 머지 실패(내용 충돌 아님): {exc}"
            ) from exc
        if conflicts is not None:
            _cleanup_group(project_root, wt_path, branch, project_id,
                           task_group, task_id, gid)
            raise HandoffConflict(stage=n, branch=stage_branch, paths=conflicts)
        merge_commits.append(
            _run_git(["rev-parse", "HEAD"], wt_path).stdout.strip())
    return merge_commits


def assemble(*, project_root, plan_run_root, stage_map, stages, base_branch,
             work_category, project_id, task_group, task_id) -> Dict[str, Any]:
    """수집 브랜치를 만들고 선택 stage 브랜치들을 머지한다. 멱등."""
    stages = sorted(set(stages))
    rows = consumers.read_consumers(Path(plan_run_root))
    _require_eligible(stage_map, rows, stages)
    _run_git(["fetch", "origin", base_branch], project_root)
    done = consumers.latest_done_by_stage(rows)
    _require_closure(stages, stage_map, done, project_root, base_branch)

    base_commit = worktree_registry.get_implementation_base(
        project_id, task_group, task_id)
    if not base_commit:
        raise HandoffError(
            "implementation_base_commit not recorded in worktree registry — "
            "run at least one implementation stage first")

    gid = group_id_for(stages)
    existing = worktree_registry.lookup(
        project_id, task_group, task_id, group_id=gid)
    if existing and existing.status == "active":
        return _reuse_existing(existing, stages, done, project_root)

    branch = compute_branch_name(work_category=work_category,
                                 task_id_segment=task_id, group_id=gid)
    wt_path = compute_worktree_path(
        project_id=project_id, task_group_segment=task_group,
        task_id_segment=task_id, group_id=gid)
    # reserve 충돌(이미 등록된 task-key/branch)은 RuntimeError 로 오는데, 이를
    # HandoffError 로 감싸 main 의 envelope(JSON+exit1)로 흘려보낸다. 감싸지
    # 않으면 raw traceback 으로 빠진다. cleanup 을 거는 worktree-add try 와
    # 분리해 두어, 예약 생성 실패 시 자신이 만들지도 않은 예약을 되돌리지 않는다.
    try:
        worktree_registry.reserve(
            project_id=project_id, task_group=task_group, task_id=task_id,
            worktree_path=str(wt_path), branch=branch, base_ref=base_commit,
            phase="release-handoff", group_id=gid, stages=stages)
    except RuntimeError as exc:
        raise HandoffError(str(exc))
    try:
        _run_git(["worktree", "add", "-b", branch, str(wt_path), base_commit],
                 project_root)
        merge_commits = _merge_stages(wt_path, stages, work_category, task_id,
                                      project_root, project_id, task_group, gid,
                                      branch, done)
    except HandoffConflict:
        raise  # _merge_stages 가 이미 _cleanup_group 으로 예약·worktree 를 되돌림
    except Exception:
        # worktree add 실패 등 비-충돌 경로 — 예약이 orphan 으로 남지 않도록 역전
        _cleanup_group(project_root, wt_path, branch, project_id, task_group,
                       task_id, gid)
        raise
    head = merge_commits[-1] if merge_commits else base_commit
    return {"ok": True, "reused": False, "group_id": gid, "branch": branch,
            "worktree_path": str(wt_path), "head": head, "stages": stages,
            "merge_commits": merge_commits}


def _remove_task_worktree(main_wt, target: str) -> Optional[str]:
    """Remove an okstra worktree — task-key or stage — already confirmed clean
    by the caller. No-op when it is already gone. Returns a human-readable
    detail string on failure, None on success or absence.
    Refuses while a nested worktree remains inside the target — a --force
    removal would drop only the physical directory and leave git's and the
    registry's stage-key admin entries orphaned."""
    if not Path(target).exists():
        return None  # 이미 teardown 됨 — checkout 으로 진행
    nested = nested_worktree_excludes(target)
    if nested:
        return f"nested 워크트리가 남아있습니다: {', '.join(nested)}"
    rm = remove_worktree_force(main_wt, target)
    if rm.returncode != 0:
        return rm.stderr.strip()
    return None


def _runs_in_progress(project_root, task_group, task_id, *,
                      exclude_task_type: str = "release-handoff") -> List[str]:
    """이 task 에서 지금 실행 중인 run — run 매니페스트 `status == "in-progress"`.

    `exclude_task_type` 의 run 은 세지 않는다: local checkout 을 부르는 것은
    release-handoff run 의 리드 자신이므로 그 run 은 점유가 아니다. 읽을 수 없는
    매니페스트는 판정을 열어 두지 않고 그대로 오류로 올린다 — 점유 여부를 모르는
    채 워크트리를 지우는 것보다 한 번 더 묻는 편이 낫다.
    """
    from .paths import task_runs_dir

    runs = task_runs_dir(Path(project_root), task_group, task_id)
    if not runs.is_dir():
        return []
    live: List[str] = []
    for manifest in sorted(runs.rglob("run-manifest-*.json")):
        if manifest.parent.name != "manifests":
            continue
        try:
            data = load_owned_object(manifest, artifact="run manifest")
        except JsonBoundaryError as exc:
            raise HandoffError(
                f"run 매니페스트를 읽을 수 없어 점유 여부를 판정할 수 없습니다 "
                f"({manifest}): {exc}") from exc
        if not isinstance(data, dict) or data.get("status") != "in-progress":
            continue
        if data.get("taskType") == exclude_task_type:
            continue
        live.append(f"{data.get('taskType')} seq {manifest.stem.rsplit('-', 1)[-1]}")
    return live


def _local_checkout_target(project_root, project_id, task_group, task_id,
                           stage: Optional[int]) -> Tuple[str, str]:
    """Resolve (branch, worktree_path) for a local checkout. `stage` selects
    the stage-key row; None selects the task-key row. The returned path is ""
    only when the row carries no `worktree_path`; teardown does not clear it,
    so a torn-down stage yields a stale path to a directory that no longer
    exists. The caller must probe the path before acting on it.

    Occupancy is judged differently for the two row kinds. A stage row's
    `status == "active"` is a live reservation: done-time release flips it
    (`consumers._release_stage_occupancy_keeping_branch`,
    `stage_integrate._teardown_stage`), and `list_active_stage_numbers` feeds the
    stage resolver, so an active stage row means a run holds it. The task-key
    row has no such lifecycle — `reserve` writes `active` when the worktree is
    created and nothing but this checkout ever releases it (observed 2026-09-07,
    fontsninja-v3-site dev-10626: the row stayed `active` from 09-02 through five
    completed phases, so whole-task local checkout could never be reached and
    its "retry after the run ends" message could never come true). For the
    task-key row the occupancy question is therefore asked of the runs
    themselves: another run of this task still `in-progress` refuses the
    checkout; the release-handoff run that is asking does not count."""
    if stage is None:
        entry = worktree_registry.lookup(project_id, task_group, task_id)
        if not entry or not entry.branch or not entry.worktree_path:
            raise HandoffError(
                "task-key worktree registry 엔트리가 없습니다 — local checkout 대상 없음")
        live = _runs_in_progress(project_root, task_group, task_id)
        if live:
            raise HandoffError(
                "task 워크트리를 다른 run 이 사용 중입니다 (in-progress: "
                f"{', '.join(live)}) — 그 run 이 끝난 뒤 다시 시도하세요")
        return entry.branch, entry.worktree_path
    row = worktree_registry.get_stage_row(project_id, task_group, task_id, stage)
    if not row or not row.get("branch"):
        raise HandoffError(
            f"stage {stage} 의 worktree registry 엔트리가 없습니다 — "
            "local checkout 대상 없음")
    if row.get("status") == "active":
        raise HandoffError(
            f"stage {stage} 를 실행 중인 런이 점유하고 있습니다 (status=active) — "
            "그 런이 끝나 점유가 해제된 뒤 다시 시도하세요")
    return row["branch"], row.get("worktree_path") or ""


def local_checkout(*, project_root, project_id, task_group, task_id,
                   stage: Optional[int] = None) -> Dict[str, Any]:
    """okstra 워크트리를 제거하고 대상 브랜치를 메인 워크트리에 checkout 한다
    (브랜치 보존). stage 를 주면 그 stage 의 스택 브랜치가, 주지 않으면 whole-task
    브랜치가 대상이다. 메인·대상 워크트리가 dirty 면 거부. base 브랜치는
    건드리지 않는다."""
    branch, wt_path = _local_checkout_target(
        project_root, project_id, task_group, task_id, stage)
    main_wt = main_worktree_path(Path(project_root))
    if is_dirty_excluding_okstra(main_wt):
        raise HandoffError(
            f"메인 워크트리에 미커밋 변경이 있습니다 ({main_wt}) — "
            "커밋/스태시 후 다시 시도하세요")
    # 브랜치 부재는 어떤 변경 앞에서 잡는다. 종전에는 워크트리 제거·registry
    # 해제 뒤에 `git checkout` 이 실패해(스택 브랜치를 사용자가 이미 지운 경우)
    # 절반만 바뀐 상태가 남았다(2026-09-07 dev-10626 stage-1 실측).
    probe = _git(Path(main_wt), "rev-parse", "--verify", "--quiet",
                 f"refs/heads/{branch}")
    if probe.returncode != 0:
        raise HandoffError(
            f"브랜치 {branch} 가 저장소에 없습니다(삭제됨) — checkout 대상이 없어 "
            "워크트리와 registry 는 건드리지 않았습니다")
    removed = ""
    # The removal gate must inspect the okstra worktree being destroyed, not main.
    # A torn-down stage leaves a stale path in its row, so probe the directory.
    if wt_path and Path(wt_path).exists():
        if is_dirty_excluding_okstra(wt_path):
            raise HandoffError(
                f"okstra 워크트리에 미커밋 변경이 있습니다 ({wt_path}) — "
                "해당 워크트리에서 커밋/스태시 후 다시 시도하세요")
        detail = _remove_task_worktree(main_wt, wt_path)
        if detail is not None:
            raise HandoffError(
                f"워크트리 제거 실패 ({wt_path}): {detail} — "
                "위 사유를 해소한 뒤 다시 시도하세요")
        removed = wt_path
    worktree_registry.release_status(project_id, task_group, task_id,
                                     stage_number=stage)
    co = _git(Path(main_wt), "checkout", branch)
    if co.returncode != 0:
        removal = f"워크트리는 제거됨({removed})" if removed else "제거된 워크트리 없음"
        raise HandoffError(
            f"git checkout {branch} 실패: {co.stderr.strip()} — {removal}; "
            f"메인({main_wt})에서 수동 checkout 으로 복구하세요")
    return {"ok": True, "branch": branch, "mainWorktreePath": str(main_wt),
            "removedWorktree": removed, "status": "checked-out", "stage": stage}


def _impl_task_key_for(rows: List[Dict[str, Any]], stage: int) -> str:
    done = consumers.latest_done_by_stage(rows)
    row = done.get(stage)
    if not row:
        raise HandoffError(
            f"stage {stage} has no done row in consumers.jsonl — "
            "finish the implementation stage first")
    return row.get("impl_task_key", "")


def _impl_task_key_for_any(rows: List[Dict[str, Any]], stages: List[int]) -> str:
    """task 를 식별하는 키 — stages 중 done 행이 있는 아무 stage 에서나 승계한다.
    PR 기록은 특정 stage 의 done 에 묶일 이유가 없으므로 최저 stage 가 비어도
    다른 stage 의 done 으로 충족한다."""
    done = consumers.latest_done_by_stage(rows)
    for n in sorted(stages):
        row = done.get(n)
        if row:
            return row.get("impl_task_key", "")
    raise HandoffError(
        f"none of stages {sorted(stages)} have a done row in consumers.jsonl — "
        "finish at least one implementation stage first")


def record_verified(*, plan_run_root, stage: int, report_path: str,
                    data_json) -> Dict[str, Any]:
    """릴리스로 넘어갈 수 있는 단독-stage 판정만 기록. data.json 의
    taskType/scope/verdict 를 검증해 lead 가 임의 보고서를 verified 로 올리는 것을
    막는다. 통과 조건은 whole-task 와 같고(`release_gate`), 기록되는 verdict 는
    보고서가 실제로 실은 토큰이다."""
    try:
        data = load_owned_object(Path(data_json), artifact="final verification report")
    except JsonBoundaryError as exc:
        raise HandoffError(f"cannot read final-report data.json: {exc}")
    if (data.get("header") or {}).get("taskType") != "final-verification":
        raise HandoffError("data.json is not a final-verification report")
    scope = data.get("verificationScope")
    if scope != "single-stage":
        raise HandoffError(
            f"record-verified requires verificationScope single-stage, "
            f"got {scope!r}")
    token = verdict_token(data)
    if not release_handoff_allowed(data):
        blocking = blocking_condition_ids(data)
        detail = (
            f"condition(s) {blocking} declare `blocksReleaseHandoff: true`"
            if blocking else f"got {token!r}"
        )
        raise HandoffError(
            "verdict must be `accepted`, or `conditional-accept` whose every "
            f"condition declares `blocksReleaseHandoff: false` — {detail}")
    rows = consumers.read_consumers(Path(plan_run_root))
    key = _impl_task_key_for(rows, stage)
    _record_verified_target(Path(plan_run_root), stage, key, report_path, Path(data_json), data)
    return {"ok": True, "stage": stage, "report_path": report_path}


def _record_verified_target(plan_run_root: Path, stage: int, key: str,
                            report_path: str, data_json: Path, data: dict) -> None:
    """검증한 실행과 커밋을 입증한 뒤 같은 원장 잠금 안에서 기록한다."""
    try:
        ref = RunRef.from_run_dir(Path(plan_run_root).resolve()).sibling("final-verification")
        stage_ref = RunRef.from_task_root(ref.task_root, ref.task_type, stage=stage)
        latest, manifest, verified_data = load_latest_verification(stage_ref.run_dir)
        supplied = final_report_data_path(Path(data_json).resolve())
        report_arg = Path(report_path)
        if not report_arg.is_absolute():
            report_arg = Path(str(manifest.get("projectRoot") or "")) / report_arg
        if (latest != supplied or final_report_data_path(report_arg.resolve()) != latest
                or manifest.get("taskKey") != key or verified_data != data):
            raise VerificationEvidenceError("report does not match latest verification of this task/stage")
        consumers.append_verified(
            Path(plan_run_root), impl_task_key=key, stage=stage, verdict=verdict_token(data),
            report_path=str(latest),
            head_commit=data["finalVerification"]["sourceImplementationReport"]["capturedHeadSha"],
            final_verdict=data["finalVerdict"],
        )
    except ValueError as exc:
        raise HandoffError(str(exc)) from exc


def record_pr(*, plan_run_root, stages: List[int], branch: str,
              url: str) -> Dict[str, Any]:
    if not stages:
        raise HandoffError("record-pr requires at least one stage")
    rows = consumers.read_consumers(Path(plan_run_root))
    key = _impl_task_key_for_any(rows, stages)
    consumers.append_pr(Path(plan_run_root), impl_task_key=key,
                        stages=stages, branch=branch, url=url)
    return {"ok": True, "stages": sorted(stages), "branch": branch, "url": url}


def _parse_stages_csv(raw: str) -> List[int]:
    try:
        out = sorted({int(x) for x in raw.split(",") if x.strip()})
    except ValueError:
        raise HandoffError(f"--stages must be a comma-separated int list, got {raw!r}")
    if not out:
        raise HandoffError("--stages must select at least one stage")
    return out


_CLI_EPILOG = r"""JSON 출력.

Usage:
  okstra handoff eligible --plan-run-root <dir> --approved-plan <md>
  okstra handoff assemble --plan-run-root <dir> --approved-plan <md> \
    --project-root <dir> --project-id <id> --task-group <g> --task-id <t> \
    --work-category <c> --stages 2,3 --base <branch>
  okstra handoff record-verified --plan-run-root <dir> --stage <N> \
    --report-path <md> --data-json <json>
  okstra handoff record-pr --plan-run-root <dir> --stages 2,3 \
    --branch <b> --url <u>
  okstra handoff local-checkout --project-root <dir> --project-id <id> \
    --task-group <g> --task-id <t> [--stage <N>]

Exit codes: 0 ok / 1 자격·전제 위반 / 2 stage 간 merge 충돌(conflicts 동봉)
"""
_CLI_DESCRIPTION = "Release-handoff stage-group \ubcf4\uc870 (\uc790\uaca9/\uc218\uc9d1/\uae30\ub85d)."


def main(argv: Optional[list] = None) -> int:
    import argparse

    p = argparse.ArgumentParser(
        description=_CLI_DESCRIPTION,
        epilog=_CLI_EPILOG,
        formatter_class=argparse.RawDescriptionHelpFormatter,
        prog="okstra handoff")
    sub = p.add_subparsers(dest="cmd", required=True)

    def common(sp, *, plan=True):
        if plan:
            sp.add_argument("--plan-run-root", required=True)

    sp = sub.add_parser("eligible")
    common(sp)
    sp.add_argument("--approved-plan", required=True)

    sp = sub.add_parser("assemble")
    common(sp)
    sp.add_argument("--approved-plan", required=True)
    sp.add_argument("--project-root", default=".")
    sp.add_argument("--project-id", required=True)
    sp.add_argument("--task-group", required=True)
    sp.add_argument("--task-id", required=True)
    sp.add_argument("--work-category", required=True)
    sp.add_argument("--stages", required=True)
    sp.add_argument("--base", required=True)

    sp = sub.add_parser("record-verified")
    common(sp)
    sp.add_argument("--stage", type=int, required=True)
    sp.add_argument("--report-path", required=True)
    sp.add_argument("--data-json", required=True)

    sp = sub.add_parser("record-pr")
    common(sp)
    sp.add_argument("--stages", required=True)
    sp.add_argument("--branch", required=True)
    sp.add_argument("--url", required=True)

    sp = sub.add_parser("local-checkout")
    common(sp, plan=False)
    sp.add_argument("--project-root", default=".")
    sp.add_argument("--project-id", required=True)
    sp.add_argument("--task-group", required=True)
    sp.add_argument("--task-id", required=True)
    sp.add_argument("--stage", type=int, default=None)

    a = p.parse_args(argv)
    try:
        if a.cmd == "eligible":
            stage_map = stage_map_records(parse_stage_map_file(Path(a.approved_plan)))
            rows = consumers.read_consumers(Path(a.plan_run_root))
            out = {"stages": compute_eligibility(stage_map, rows)}
        elif a.cmd == "assemble":
            out = assemble(
                project_root=Path(a.project_root).resolve(),
                plan_run_root=Path(a.plan_run_root),
                stage_map=stage_map_records(parse_stage_map_file(Path(a.approved_plan))),
                stages=_parse_stages_csv(a.stages), base_branch=a.base,
                work_category=a.work_category, project_id=a.project_id,
                task_group=a.task_group, task_id=a.task_id)
        elif a.cmd == "record-verified":
            out = record_verified(plan_run_root=Path(a.plan_run_root),
                                  stage=a.stage, report_path=a.report_path,
                                  data_json=a.data_json)
        elif a.cmd == "local-checkout":
            out = local_checkout(
                project_root=Path(a.project_root).resolve(),
                project_id=a.project_id, task_group=a.task_group,
                task_id=a.task_id, stage=a.stage)
        else:
            out = record_pr(plan_run_root=Path(a.plan_run_root),
                            stages=_parse_stages_csv(a.stages),
                            branch=a.branch, url=a.url)
    except HandoffConflict as exc:
        print(json.dumps({"error": str(exc), "stage": exc.stage,
                          "branch": exc.branch, "conflicts": exc.paths},
                         ensure_ascii=False))
        return 2
    except HandoffError as exc:
        print(json.dumps({"error": str(exc)}, ensure_ascii=False))
        return 1
    except StageMapError as exc:
        print(json.dumps({"error": str(exc)}, ensure_ascii=False))
        return 1
    print(json.dumps(out, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    import sys as _sys

    _sys.exit(main())