"""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
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 가 담당)."""
    return stage_targets.stage_lifecycle_snapshot_from_rows(
        stage_map,
        rows,
    ).handoff_eligibility()


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

    판정의 SSOT 는 final-report 의 data.json 이다 (record_verified 와 동일
    필드: header.taskType / verificationScope / finalVerdict). 통과 조건은
    `release_gate.release_handoff_allowed` 한 곳이 소유한다 — `accepted`,
    또는 모든 조건이 릴리스를 막지 않는다고 선언한 `conditional-accept`.
    whole-task run 산출물만 평면 reports/ 에 남으므로 stage-* 는 걸리지 않는다."""
    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 ""
    reports_dir = RunRef.from_task_root(root, "final-verification").reports_dir
    for dj in sorted(reports_dir.glob("final-report-*.data.json"),
                     reverse=True):
        try:
            data = load_owned_object(dj, artifact="final verification report")
        except JsonBoundaryError:
            continue
        if ((data.get("header") or {}).get("taskType") == "final-verification"
                and data.get("verificationScope") == "whole-task"
                and release_handoff_allowed(data)):
            return str(dj)
    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 _local_checkout_target(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.

    An `active` row is refused: `status == "active"` is the occupancy SSOT a
    concurrent run holds (`worktree_registry.list_active_stage_numbers` feeds
    the stage resolver), and local checkout would remove that run's worktree
    and release its reservation, letting a third `--stage auto` run claim the
    same stage."""
    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 대상 없음")
        if entry.status == "active":
            raise HandoffError(
                "task 워크트리를 실행 중인 런이 점유하고 있습니다 (status=active) — "
                "그 런이 끝나 점유가 해제된 뒤 다시 시도하세요")
        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_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}) — "
            "커밋/스태시 후 다시 시도하세요")
    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)
    consumers.append_verified(Path(plan_run_root), impl_task_key=key,
                              stage=stage, verdict=token,
                              report_path=report_path)
    return {"ok": True, "stage": stage, "report_path": report_path}


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


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

    p = argparse.ArgumentParser(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())