"""anu_v3.branch_ref_allocator — task-2553+9 collision-safe branch name
allocator (회장 §4-1~4-4 / 9-R.3 / 9-R.4 / 9-R.6).

근본원인: live 메인 워크스페이스가 ``task/task-2553p1-f1-clean-replacement``
에 checkout 돼 있고, activation runner 의 ``git checkout -B <branch> <base>``
가 그 **live-checked-out branch ref 를 reset** → task-2553+8 DEFENSIVE_HOLD.

본 모듈 = 충돌하지 않는 **고유 신규 replacement branch name 을 할당**.
충돌은 강제 reset 이 아니라 **회피(새 name)** 로만 해소한다 (회장 §3).

설계 불변식:
  - **read-only**: ``git worktree list`` / ``git show-ref`` / ``git ls-remote``
    만 호출. branch 생성·reset·checkout·push·prune **부재**(정적·런타임).
  - **loop-until-clean (9-R.3)**: 단발 uuid 확률 방어 금지. candidate 가
    (worktree checked-out ∪ local refs ∪ remote refs) 3-source 전부에서
    absent 될 때까지 반복. bounded ``max_attempts``(기본 8) 초과 → HOLD.
  - **강제 reset 0 (회장 §3)**: 어떤 경우에도 기존 ref 를 reset/삭제 않음.
  - **9-R.6 audit trail**: 결과 provenance 는 ``schemas/
    branch_allocation_provenance.schema.json`` 와 1:1 (base_sha single
    authority — base_name 미기록).
"""

from __future__ import annotations

import os
import subprocess
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Callable, Final

ALLOCATOR_MODULE: Final[str] = "anu_v3.branch_ref_allocator"
ALLOCATOR_VERSION: Final[str] = "1.0.0"
PROVENANCE_SCHEMA: Final[str] = "anu_v3.branch_allocation_provenance.v1"

STATUS_ALLOCATED: Final[str] = "ALLOCATED"
STATUS_HOLD: Final[str] = "HOLD_FOR_CHAIR"

DEFAULT_MAX_ATTEMPTS: Final[int] = 8

#: read-only git verbs — 이 셋만 호출 (정적 부재 검증 대상 = 그 외 mutating verb).
READ_ONLY_GIT_VERBS: Final[tuple[str, ...]] = (
    "worktree",  # worktree list --porcelain (read-only sub-cmd 만)
    "show-ref",  # --verify --quiet refs/heads/<name>
    "ls-remote",  # --heads origin <name>
    "rev-parse",  # --abbrev-ref HEAD (live checkout branch 식별)
)


class BranchAllocatorError(Exception):
    """allocation 실패. fail-closed: branch mutation 절대 부재."""


def _now_utc() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def _sanitized_env() -> dict[str, str]:
    """9-R.2 동일 패턴 — GIT_* 오염 변수 제거 (isolation 강제)."""
    e = os.environ.copy()
    for k in (
        "GIT_DIR",
        "GIT_WORK_TREE",
        "GIT_INDEX_FILE",
        "GIT_OBJECT_DIRECTORY",
        "GIT_COMMON_DIR",
    ):
        e.pop(k, None)
    return e


def _git_ro(target: str | Path, *args: str, timeout: int = 60) -> tuple[int, str]:
    """absolute ``git -C <target>`` **read-only** 전용. (rc, stdout) 반환.

    호출 verb 가 READ_ONLY_GIT_VERBS 화이트리스트 밖이면 즉시 거부
    (런타임 fail-closed — mutating git 호출 경로 부재 강제).
    """
    if not args or args[0] not in READ_ONLY_GIT_VERBS:
        raise BranchAllocatorError(
            f"read-only git verb 화이트리스트 위반: {args[:1]!r} "
            f"(allowed={READ_ONLY_GIT_VERBS})"
        )
    abspath = str(Path(target).resolve())
    proc = subprocess.run(  # noqa: S603 — git read-only, sanitized env, no shell
        ["git", "-C", abspath, *args],
        capture_output=True,
        text=True,
        timeout=timeout,
        check=False,
        env=_sanitized_env(),
    )
    return proc.returncode, proc.stdout.strip()


# ─────────────────────────────────────────────────────────────────────────────
# detection (전부 read-only)
# ─────────────────────────────────────────────────────────────────────────────
def list_checked_out_branches(repo_path: str | Path) -> dict[str, str]:
    """``git worktree list --porcelain`` → {branch_name: worktree_path}.

    모든 worktree(live 포함) 의 checked-out branch ref 를 파싱한다.
    """
    rc, out = _git_ro(repo_path, "worktree", "list", "--porcelain")
    if rc != 0:
        raise BranchAllocatorError(f"worktree list rc={rc}")
    mapping: dict[str, str] = {}
    cur_wt: str | None = None
    for line in out.splitlines():
        if line.startswith("worktree "):
            cur_wt = line[len("worktree "):].strip()
        elif line.startswith("branch ") and cur_wt is not None:
            ref = line[len("branch "):].strip()
            if ref.startswith("refs/heads/"):
                mapping[ref[len("refs/heads/"):]] = cur_wt
    return mapping


def live_checkout_branch(repo_path: str | Path) -> str | None:
    """live(메인) workspace 가 현재 checkout 한 branch (detached → None).

    이 branch ref 는 회장 §3 에 따라 **절대 reset 하지 않는다** (식별 전용).
    """
    rc, out = _git_ro(repo_path, "rev-parse", "--abbrev-ref", "HEAD")
    if rc != 0 or not out or out == "HEAD":
        return None
    return out


def local_ref_exists(repo_path: str | Path, name: str) -> bool:
    """``git show-ref --verify --quiet refs/heads/<name>`` (read-only)."""
    rc, _ = _git_ro(
        repo_path, "show-ref", "--verify", "--quiet", f"refs/heads/{name}"
    )
    return rc == 0


def remote_ref_exists(
    repo_path: str | Path, name: str, *, remote: str = "origin"
) -> bool:
    """``git ls-remote --heads origin <name>`` 비어있지 않으면 True."""
    rc, out = _git_ro(repo_path, "ls-remote", "--heads", remote, name)
    if rc != 0:
        # 원격 조회 실패 = 충돌 단정 불가 → 보수적으로 '존재함' 취급
        # (collision-safe: 의심되면 회피하여 새 name 할당).
        return True
    return bool(out.strip())


def _new_runid() -> str:
    """uuid4 hex 12 (9-R.3 suffix)."""
    return uuid.uuid4().hex[:12]


# ─────────────────────────────────────────────────────────────────────────────
# loop-until-clean allocation (9-R.3) — 강제 reset 0, 회피만
# ─────────────────────────────────────────────────────────────────────────────
def _probe(
    repo_path: str | Path,
    candidate: str,
    checked_out: dict[str, str],
) -> dict[str, bool]:
    """candidate 의 3-source 점유 여부 (read-only). True = 점유(충돌)."""
    return {
        "worktree": candidate in checked_out,
        "local": local_ref_exists(repo_path, candidate),
        "remote": remote_ref_exists(repo_path, candidate),
    }


def allocate_branch(
    repo_path: str | Path,
    base_name: str,
    *,
    base_sha: str,
    max_attempts: int = DEFAULT_MAX_ATTEMPTS,
    runid_factory: Callable[[], str] = _new_runid,
) -> dict:
    """collision-safe branch name 할당 (read-only, loop-until-clean).

    1. candidate = ``base_name`` → 3-source(worktree/local/remote) 전부
       absent 면 ``chosen_strategy=base`` 로 즉시 확정.
    2. 하나라도 present → ``<base_name>-<runid>`` 재생성·재검사 반복.
    3. ``max_attempts`` 회 내 clean candidate 미발견 → ``STATUS_HOLD``
       ("clean unique branch allocation 실패", 회장 §9 / 9-R.3).

    강제 reset/삭제/생성 **0** — 충돌은 새 name 회피로만 해소(회장 §3).
    반환 = branch_allocation_provenance.schema.json 와 1:1 dict.
    """
    checked_out = list_checked_out_branches(repo_path)
    live_branch = live_checkout_branch(repo_path)

    base_probe = _probe(repo_path, base_name, checked_out)
    collision_detected = any(base_probe.values())

    # 9-R.6 audit: sources_checked = **base name** 점유 출처 (왜 충돌했나).
    # 최종 allocated name 은 loop 보장상 항상 clean → 그 probe 는 무의미.
    runid: str | None = None
    attempts = 0
    if not collision_detected:
        allocated = base_name
        strategy = "base"
        attempts = 1
    else:
        allocated = None
        strategy = "suffixed"
        while attempts < max_attempts:
            attempts += 1
            runid = runid_factory()
            candidate = f"{base_name}-{runid}"
            probe = _probe(repo_path, candidate, checked_out)
            if not any(probe.values()):
                allocated = candidate
                break

    base_record = {
        "schema": PROVENANCE_SCHEMA,
        "allocator_module": ALLOCATOR_MODULE,
        "allocator_version": ALLOCATOR_VERSION,
        "ts_utc": _now_utc(),
        "base_sha": base_sha,
        "collision_detected": collision_detected,
        "sources_checked": {
            "worktree": bool(base_probe["worktree"]),
            "local": bool(base_probe["local"]),
            "remote": bool(base_probe["remote"]),
        },
        "runid": runid,
        "live_checkout_branch": live_branch,
        "chosen_strategy": strategy,
        "attempts": attempts,
        "read_only": True,
    }

    if allocated is None:
        base_record["allocated_branch_name"] = ""
        base_record["status"] = STATUS_HOLD
        base_record["hold_reasons"] = [
            "clean unique branch allocation 실패 "
            f"(loop-until-clean {max_attempts}회 초과, 회장 §9 / 9-R.3). "
            "강제 reset 0 — 회장 보고."
        ]
        return base_record

    base_record["allocated_branch_name"] = allocated
    base_record["status"] = STATUS_ALLOCATED
    return base_record


__all__ = [
    "ALLOCATOR_MODULE",
    "ALLOCATOR_VERSION",
    "PROVENANCE_SCHEMA",
    "STATUS_ALLOCATED",
    "STATUS_HOLD",
    "DEFAULT_MAX_ATTEMPTS",
    "BranchAllocatorError",
    "list_checked_out_branches",
    "live_checkout_branch",
    "local_ref_exists",
    "remote_ref_exists",
    "allocate_branch",
]