"""Stale git SHA 감지·화해·보정 (3단 방어)의 단일 reference point.

저장된 SHA(anchor / done.head_commit)가 okstra 밖의 히스토리 재작성으로
stale 해졌을 때 — patch-id 로 내용 동일성이 증명되면 자동 화해(auto),
내용이 바뀌었으면 사용자 확인 보정(confirm). 설계상 소비자는 prepare
경로(run.py), `okstra git-reconcile` subcommand, okstra-run 스킬 —
셋 모두 이 모듈 하나를 통해야 한다. 설계:
docs/superpowers/specs/2026-06-10-git-reconcile-stale-sha-recovery-design.md
"""
from __future__ import annotations

import subprocess
from dataclasses import dataclass
from pathlib import Path
from typing import Dict, List, Optional, Tuple

from .fixed_text import line
from .worktree import _resolve_commit_sha


def _git(cwd: Path, *args: str) -> subprocess.CompletedProcess:
    return subprocess.run(["git", "-C", str(cwd), *args],
                          capture_output=True, text=True)


@dataclass
class MatchResult:
    status: str            # "ancestor" | "patch-equivalent" | "not-merged" | "unresolvable"
    matched_commit: str = ""


def _patch_ids_of_range(cwd: Path, base: str, head: str) -> List[Tuple[str, str]]:
    """base..head 각 커밋의 (patch-id, sha). diff 없는 커밋은 제외된다."""
    log = _git(cwd, "log", "-p", "--no-merges", f"{base}..{head}")
    if log.returncode != 0:
        return []
    pid = subprocess.run(["git", "-C", str(cwd), "patch-id", "--stable"],
                         input=log.stdout, capture_output=True, text=True)
    out = []
    for line in pid.stdout.splitlines():
        parts = line.split()
        if len(parts) == 2:
            out.append((parts[0], parts[1]))
    return out


def _patch_id_of_diff(cwd: Path, base: str, head: str) -> str:
    """base→head 전체를 한 diff 로 본 patch-id (squash 동등성용)."""
    diff = _git(cwd, "diff", base, head)
    if diff.returncode != 0 or not diff.stdout.strip():
        return ""
    pid = subprocess.run(["git", "-C", str(cwd), "patch-id", "--stable"],
                         input=diff.stdout + "\n", capture_output=True, text=True)
    parts = pid.stdout.split()
    return parts[0] if parts else ""


def content_merged(project_root: Path, commit: str, candidate: str,
                   base: str = "") -> MatchResult:
    """commit 의 내용이 candidate 히스토리에 포함되는가.

    1) ancestor 면 그대로 통과. 2) patch-id fallback 두 granularity:
    커밋 단위(rebase/cherry-pick — base..commit 의 *모든* 커밋이 매칭돼야 함)
    + 범위 합산(squash — base..commit 전체 diff 가 한 커밋과 매칭).
    증명 실패는 not-merged — 자동 진행 금지는 호출자 계약이다."""
    resolved = _resolve_commit_sha(project_root, commit)
    cand = _resolve_commit_sha(project_root, candidate)
    if not resolved or not cand:
        return MatchResult("unresolvable")
    if _git(project_root, "merge-base", "--is-ancestor", resolved, cand).returncode == 0:
        return MatchResult("ancestor", matched_commit=resolved)
    mb = _git(project_root, "merge-base", resolved, cand).stdout.strip()
    if not mb:
        return MatchResult("not-merged")
    range_base = _resolve_commit_sha(project_root, base) or mb
    # git log 는 최신 커밋부터 출력 → patch-id 가 후보에 중복(revert·재적용,
    # cross-branch cherry-pick)되면 가장 최신(tip 쪽) SHA 를 재기록 대상으로
    # 골라야 한다. dict() 는 나중 항목이 이기므로 오래된 SHA 가 남는다 → 역순 적재.
    cand_ids: Dict[str, str] = {
        pid: sha
        for pid, sha in reversed(_patch_ids_of_range(project_root, mb, cand))
    }
    stage_ids = _patch_ids_of_range(project_root, range_base, resolved)
    if stage_ids and all(pid in cand_ids for pid, _ in stage_ids):
        # git log 는 최신 커밋부터 출력 → stage_ids[0] 이 tip. matched_commit
        # 은 재기록 대상이므로 tip 의 매칭 SHA 여야 한다.
        return MatchResult("patch-equivalent",
                           matched_commit=cand_ids[stage_ids[0][0]])
    whole = _patch_id_of_diff(project_root, range_base, resolved)
    if whole and whole in cand_ids:
        return MatchResult("patch-equivalent", matched_commit=cand_ids[whole])
    return MatchResult("not-merged")


@dataclass
class StaleItem:
    kind: str                  # "anchor" | "done"
    stage: Optional[int]
    recorded: str
    classification: str        # "ok" | "auto" | "confirm"
    reason: str
    suggested_commit: str = "" # auto 일 때 재기록 대상
    impl_task_key: str = ""


def _classify_done_row(project_root: Path, stage: int, row: dict,
                       branch: str, stage_base: str) -> StaleItem:
    recorded = row.get("head_commit", "")
    item = StaleItem(kind="done", stage=stage, recorded=recorded,
                     classification="ok", reason="",
                     impl_task_key=row.get("impl_task_key", ""))
    if not _resolve_commit_sha(project_root, recorded):
        item.classification, item.reason = "confirm", "recorded SHA unresolvable"
        return item
    tip = _resolve_commit_sha(project_root, branch)
    if not tip or tip == recorded:
        return item  # branch 없음(히스토리 intact) 또는 일치
    if _git(project_root, "merge-base", "--is-ancestor",
            recorded, tip).returncode == 0:
        return item  # 커밋이 단순히 더 쌓임 — stale 아님
    match = content_merged(project_root, recorded, tip, base=stage_base)
    if match.status in ("ancestor", "patch-equivalent"):
        item.classification = "auto"
        item.reason = f"branch {branch} rewritten, patch-equivalent"
        item.suggested_commit = tip
    else:
        item.classification = "confirm"
        item.reason = f"branch {branch} rewritten with content changes"
    return item


def classify_task(*, project_root: Path, plan_run_root: Path,
                  project_id: str, task_group: str, task_id: str,
                  work_category: str) -> List[StaleItem]:
    """task 의 anchor + 최신 done row 들을 ok/auto/confirm 으로 분류한다.
    다중 의존 gate(spec §3.2 표 5-6행)는 candidate 가 필요한 prepare 경로에서
    content_merged 로 직접 평가된다 — 여기서 중복 평가하지 않는다."""
    from . import worktree_registry as _reg
    from .consumers import read_consumers, latest_done_by_stage
    from .worktree import compute_branch_name

    items: List[StaleItem] = []
    anchor = _reg.get_implementation_base(project_id, task_group, task_id) or ""
    if anchor:
        ok = bool(_resolve_commit_sha(project_root, anchor))
        items.append(StaleItem(
            kind="anchor", stage=None, recorded=anchor,
            classification="ok" if ok else "confirm",
            reason="" if ok else "anchor SHA unresolvable",
        ))
    latest = latest_done_by_stage(read_consumers(plan_run_root))
    for stage in sorted(latest):
        row = latest[stage]
        branch = compute_branch_name(
            work_category=work_category, task_id_segment=task_id,
            stage_number=stage,
        )
        srow = _reg.get_stage_row(project_id, task_group, task_id, stage) or {}
        items.append(_classify_done_row(
            project_root, stage, row, branch, srow.get("base_ref", "")))
    return items


class ReconcileError(Exception):
    pass


def _record_reconciled(plan_run_root: Path, *, impl_task_key: str, stage: int,
                       new_commit: str, replaced: str, reason: str) -> None:
    from .consumers import append_consumer
    append_consumer(
        plan_run_root, impl_task_key=impl_task_key, stage=stage,
        status="done", force_reappend=True, head_commit=new_commit,
        reconciled=True, reconcile_reason=reason, replaced_commit=replaced,
    )


def auto_reconcile(*, project_root: Path, plan_run_root: Path,
                   project_id: str, task_group: str, task_id: str,
                   work_category: str) -> List[StaleItem]:
    """classify 의 auto 항목만 보정 row 로 재기록한다 — patch-id 증명이
    있으므로 확인 불필요(설계 §2 결정). confirm 은 건드리지 않는다."""
    applied = []
    items = classify_task(
        project_root=project_root, plan_run_root=plan_run_root,
        project_id=project_id, task_group=task_group, task_id=task_id,
        work_category=work_category)
    for item in items:
        if item.classification != "auto" or item.kind != "done":
            continue
        _record_reconciled(plan_run_root, impl_task_key=item.impl_task_key,
                           stage=item.stage, new_commit=item.suggested_commit,
                           replaced=item.recorded, reason="auto-patch-id")
        applied.append(item)
    return applied


def guidance(*, plan_run_root: Path, project_id: str, task_group: str,
             task_id: str, work_category: str) -> str:
    """PrepareError 에 첨부하는 회복 안내 (명령 예시 포함)."""
    base = (f"okstra git-reconcile --plan-run-root {plan_run_root} "
            f"--project-id {project_id} --task-group {task_group} "
            f"--task-id {task_id} --work-category {work_category}")
    return ("Recorded stage SHAs no longer match the git history "
            "(external rebase/squash/amend?). Inspect and reconcile:\n"
            f"  {base} --check --text\n"
            f"  {base} --apply --stage <N> --use-ref <branch|sha>")


def _apply_user_ref(project_root: Path, plan_run_root: Path,
                    latest: Dict[int, dict], stage: Optional[int],
                    use_ref: str) -> dict:
    if stage is None:
        raise ReconcileError("--use-ref requires --stage")
    sha = _resolve_commit_sha(project_root, use_ref)
    if not sha:
        raise ReconcileError(f"could not resolve ref `{use_ref}`")
    row = latest.get(stage)
    if not row:
        raise ReconcileError(f"stage {stage} has no done row to reconcile")
    _record_reconciled(plan_run_root, impl_task_key=row.get("impl_task_key", ""),
                       stage=stage, new_commit=sha,
                       replaced=row.get("head_commit", ""), reason="user-ref")
    return {"stage": stage, "new_commit": sha}


def apply_reconcile(*, project_root: Path, plan_run_root: Path,
                    project_id: str, task_group: str, task_id: str,
                    work_category: str, stage: Optional[int] = None,
                    use_ref: str = "", reset_anchor: str = "") -> dict:
    """auto 항목 일괄 보정 + (옵션) confirm 항목 1건 보정 + (옵션) anchor 재고정.

    enforcement(spec §3.6): confirm 항목은 `use_ref` 가 명시된 그 stage 만
    보정된다 — 어떤 경로로도 무확인 자동 보정되지 않는다.

    classify 와 보정 append 사이는 원자적이지 않다 — 동시 run 이 활발한
    시점이 아니라 사용자 보정(수동) 시점에 호출되는 것을 전제한다."""
    from . import worktree_registry as _reg
    from .consumers import read_consumers, latest_done_by_stage

    applied: List[dict] = []
    if reset_anchor:
        sha = _resolve_commit_sha(project_root, reset_anchor)
        if not sha:
            raise ReconcileError(f"could not resolve ref `{reset_anchor}`")
        _reg.reset_implementation_base(project_id, task_group, task_id, sha)
        applied.append({"anchor": sha})
    if use_ref:
        latest = latest_done_by_stage(read_consumers(plan_run_root))
        applied.append(_apply_user_ref(
            project_root, plan_run_root, latest, stage, use_ref))
    items = classify_task(
        project_root=project_root, plan_run_root=plan_run_root,
        project_id=project_id, task_group=task_group, task_id=task_id,
        work_category=work_category)
    for item in items:
        if item.classification != "auto":
            continue
        if use_ref and item.stage == stage:
            continue  # 사용자가 명시 보정한 stage 는 auto 가 덮지 않는다
        _record_reconciled(plan_run_root, impl_task_key=item.impl_task_key,
                           stage=item.stage, new_commit=item.suggested_commit,
                           replaced=item.recorded, reason="auto-patch-id")
        applied.append({"stage": item.stage, "new_commit": item.suggested_commit})
    remaining = [i for i in items if i.classification == "confirm"
                 and not (use_ref and i.stage == stage)]
    return {"applied": applied, "remaining_confirm": remaining}


def _items_as_json(items: List[StaleItem]) -> list:
    return [{"kind": i.kind, "stage": i.stage, "recorded": i.recorded,
             "classification": i.classification, "reason": i.reason,
             "suggested_commit": i.suggested_commit} for i in items]


def _parse_args(argv: Optional[list]):
    import argparse

    p = argparse.ArgumentParser(
        prog="okstra git-reconcile",
        epilog="exit codes: 0 = clean/applied, 2 = confirm items remain, 1 = error")
    p.add_argument("--project-root", default=".")
    p.add_argument("--plan-run-root", required=True,
                   help="consumers.jsonl 이 있는 runs/implementation-planning/ 경로")
    p.add_argument("--project-id", required=True)
    p.add_argument("--task-group", required=True)
    p.add_argument("--task-id", required=True)
    p.add_argument("--work-category", required=True)
    p.add_argument("--check", action="store_true",
                   help="stale 검사만 (기본 동작과 동일 — 명시용)")
    p.add_argument("--apply", action="store_true",
                   help="auto 항목 일괄 보정 (+ --stage/--use-ref, --reset-anchor)")
    p.add_argument("--stage", type=int)
    p.add_argument("--use-ref", default="")
    p.add_argument("--reset-anchor", default="")
    p.add_argument("--json", action="store_true")
    p.add_argument("--text", action="store_true")
    a = p.parse_args(argv)
    if a.check and a.apply:
        p.error("--check and --apply are mutually exclusive")
    if a.apply and a.stage is not None and not a.use_ref:
        p.error("--stage requires --use-ref (confirm 항목은 명시 ref 로만 보정된다)")
    return a


def _emit_result(payload: dict, *, as_json: bool, as_text: bool) -> None:
    import json as _json

    print(
        render_git_reconcile_text(payload)
        if as_text
        else _json.dumps(payload, ensure_ascii=False, indent=None if as_json else 2)
    )


def render_git_reconcile_text(payload: dict) -> str:
    """reconcile 결과의 승인된 분류만 고정 순서로 투영한다."""
    rows = ["Okstra git reconcile\n"]
    rows.append(line("Status", "ready" if payload.get("ok", True) else "error"))
    items = payload.get("items") if isinstance(payload.get("items"), list) else []
    rows.append(line("Item count", len(items)))
    for index, item in enumerate(items, 1):
        if not isinstance(item, dict):
            continue
        rows.append(line(f"Item {index} kind", item.get("kind")))
        rows.append(line(f"Item {index} classification", item.get("classification")))
        rows.append(line(f"Item {index} reason", item.get("reason")))
    if "error" in payload:
        rows.append(line("Failure reason", payload.get("error")))
    return "".join(rows)


def main(argv: Optional[list] = None) -> int:
    a = _parse_args(argv)

    kw = dict(project_root=Path(a.project_root).resolve(),
              plan_run_root=Path(a.plan_run_root).resolve(),
              project_id=a.project_id, task_group=a.task_group,
              task_id=a.task_id, work_category=a.work_category)
    try:
        if a.apply:
            result = apply_reconcile(**kw, stage=a.stage, use_ref=a.use_ref,
                                     reset_anchor=a.reset_anchor)
            out = {"applied": result["applied"],
                   "remaining_confirm": _items_as_json(result["remaining_confirm"])}
            _emit_result(out, as_json=a.json, as_text=a.text)
            return 2 if result["remaining_confirm"] else 0
        items = classify_task(**kw)
        stale = [i for i in items if i.classification != "ok"]
        out = {"items": _items_as_json(stale)}
        _emit_result(out, as_json=a.json, as_text=a.text)
        return 2 if any(i.classification == "confirm" for i in stale) else 0
    except ReconcileError as exc:
        out = {"ok": False, "error": str(exc)}
        _emit_result(out, as_json=a.json, as_text=a.text)
        return 1


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