"""anu_v3.isolated_worktree_evidence_source — task-2553+8 isolated-worktree
evidence source (회장 §1/§2/§3, 9-R.1~9-R.6).

근본원인 (task-2553+7 첫 적용 HOLD): live workspace 가 미커밋 churn 으로
오염 → live 기반 effective-diff 가 6-file delta 로 성립 불가.

해결책 (회장 §2 verbatim, "특히 명시"): live workspace 청소가 **아니라**
fresh ``origin/main`` (7346df82) 기반 **isolated worktree** 를 만들고, 그
안에서 task-2553+1 F1-solo **6파일 delta 만 재현**한 뒤, 그 isolated
worktree 의 git diff / effective diff / source-PR preservation /
same-branch-push-0 / forbidden-scan 만 evidence source 로 사용한다.

본 모듈 한정 책임 (fail-closed, live workspace 무영향):
  - fresh origin/main isolated worktree 생성·검증·``git worktree remove``
    만으로 정리 (9-R.4: live·타 worktree 대상 reset/clean/stash/rm/unlink/
    rmtree **정적·런타임 0**).
  - task-2553+1 F1-solo 6-file delta 를 isolated worktree 안에서만 재현.
  - 9-R.2: 모든 git 호출 = absolute ``git -C <isolated_wt>`` 전용 +
    ``env=_sanitized_env()``. cwd-relative git·live path/mtime 미참조.
  - evidence bundle 은 frozen builder 의 **public** API
    (``collect_real_git_facts`` / ``build_evidence_bundle``) +
    frozen deriver public ``canonical_evidence_sha256`` 로만 조립
    (9-R.6 private helper 복제·import 0).
  - ``_provenance.source_workspace_type = "isolated_clean_worktree"`` 기록.
  - 구조적 HOLD guard: clean-replacement branch 가 다른(live/main)
    worktree 에 checkout 돼 있으면, runner primitive ①
    (``git checkout -B``) 가 그 live branch ref 를 reset → live 변형.
    → 실 write 전 ``would_mutate_live_branch`` HOLD 사유 산출
    (회장 §6 "특히 금지" / §7 / §12 fail-closed).
"""

from __future__ import annotations

import os
import subprocess
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Final, Mapping

from anu_v3.pre_authorized_contract_deriver import canonical_evidence_sha256
from anu_v3.pre_authorized_evidence_bundle_builder import (
    FRESH_BASE_SHA,
    NEW_CLEAN_REPLACEMENT_BRANCH,
    RECORDED_SOURCE_HEAD,
    SOURCE_BRANCH,
    TASK_2553P1_EFFECTIVE_DIFF_6,
    build_evidence_bundle,
    collect_real_git_facts,
)

MODULE: Final[str] = "anu_v3.isolated_worktree_evidence_source"
MODULE_VERSION: Final[str] = "1.0.0"

#: 회장 §3 / 9-R.1 — evidence provenance source 분류 (스키마 enum 권위).
SOURCE_WORKSPACE_TYPE_ISOLATED: Final[str] = "isolated_clean_worktree"
SOURCE_WORKSPACE_TYPE_LIVE: Final[str] = "live_workspace"

#: F1-solo replay 대상 = 회장 §5 single-authority 6 effective-diff 파일.
SIX_FILES: Final[tuple[str, ...]] = TASK_2553P1_EFFECTIVE_DIFF_6

ISOLATED_EVIDENCE_SCHEMA: Final[str] = "anu_v3.isolated_worktree_evidence.v1"


class IsolatedWorktreeError(Exception):
    """isolated worktree 생성/검증/replay 실패. fail-closed: 부분 evidence 0."""


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


# ─────────────────────────────────────────────────────────────────────────────
# 9-R.2 (CRITICAL C1) — git env sanitization (verbatim 메커니즘)
# ─────────────────────────────────────────────────────────────────────────────
def _sanitized_env() -> dict[str, str]:
    """모든 subprocess git 호출 env. GIT_* 오염 변수 제거 (isolation 강제).

    9-R.2 verbatim: GIT_DIR/GIT_WORK_TREE/GIT_INDEX_FILE/
    GIT_OBJECT_DIRECTORY/GIT_COMMON_DIR 를 pop. 이 패턴 외 git env 전달 금지.
    """
    e = os.environ.copy()
    e.pop("GIT_DIR", None)
    e.pop("GIT_WORK_TREE", None)
    e.pop("GIT_INDEX_FILE", None)
    e.pop("GIT_OBJECT_DIRECTORY", None)
    e.pop("GIT_COMMON_DIR", None)
    return e


def _git(
    target: str | Path,
    *args: str,
    check: bool = True,
    timeout: int = 60,
) -> str:
    """absolute ``git -C <target>`` 전용 (9-R.2). cwd 인자 미사용,
    ``env=_sanitized_env()`` 강제. live path/mtime 미참조.
    """
    abspath = str(Path(target).resolve())
    proc = subprocess.run(  # noqa: S603 — git only, sanitized env, no shell
        ["git", "-C", abspath, *args],
        capture_output=True,
        text=True,
        timeout=timeout,
        check=False,
        env=_sanitized_env(),
    )
    if check and proc.returncode != 0:
        raise IsolatedWorktreeError(
            f"git -C {abspath} {' '.join(args)} rc={proc.returncode}: "
            f"{proc.stderr.strip()}"
        )
    return proc.stdout.strip()


# ─────────────────────────────────────────────────────────────────────────────
# 구조적 HOLD guard — clean-replacement branch live/other worktree 점유 탐지
# ─────────────────────────────────────────────────────────────────────────────
def detect_live_branch_conflict(
    repo_path: str | Path,
    *,
    branch: str = NEW_CLEAN_REPLACEMENT_BRANCH,
) -> str | None:
    """clean-replacement ``branch`` 가 다른 worktree 에 checkout 돼 있으면
    그 worktree 경로 반환 (없으면 None).

    회장 §6 "특히 금지" / §7 / §12: 점유 시 runner primitive ①
    ``git checkout -B <branch> <base>`` 가 그 live branch ref 를 **reset**
    → live workspace 변형. → 실 write 전 구조적 HOLD 사유.
    """
    porcelain = _git(repo_path, "worktree", "list", "--porcelain", check=False)
    cur_wt: str | None = None
    target_ref = f"refs/heads/{branch}"
    for line in porcelain.splitlines():
        if line.startswith("worktree "):
            cur_wt = line[len("worktree "):].strip()
        elif line.startswith("branch "):
            ref = line[len("branch "):].strip()
            if ref == target_ref and cur_wt is not None:
                return cur_wt
    return None


# ─────────────────────────────────────────────────────────────────────────────
# isolated worktree lifecycle (9-R.4: git worktree add/remove 만)
# ─────────────────────────────────────────────────────────────────────────────
def create_isolated_worktree(
    repo_path: str | Path,
    *,
    base_sha: str = FRESH_BASE_SHA,
) -> Path:
    """fresh ``base_sha`` detached isolated worktree 생성 + base 검증.

    detached HEAD (branch ref 미생성 → live branch 무영향). live·타
    worktree 대상 파괴적 op 0.
    """
    wt_dir = Path(
        tempfile.mkdtemp(prefix="anu2553p8_isowt_")
    ) / "wt"
    _git(repo_path, "worktree", "add", "--detach", str(wt_dir), base_sha)
    head = _git(wt_dir, "rev-parse", "HEAD")
    if head != base_sha:
        # 즉시 정리 후 HOLD (fresh base 불일치).
        remove_isolated_worktree(repo_path, wt_dir)
        raise IsolatedWorktreeError(
            f"isolated worktree HEAD != base_sha: {head} != {base_sha}"
        )
    return wt_dir


def remove_isolated_worktree(
    repo_path: str | Path, wt_dir: str | Path
) -> None:
    """isolated worktree 정리 = ``git worktree remove --force`` **만**
    (9-R.4: rm/unlink/rmtree/reset/clean/stash 0). 실패해도 live 무결.
    """
    _git(repo_path, "worktree", "remove", "--force", str(wt_dir), check=False)
    _git(repo_path, "worktree", "prune", check=False)


# ─────────────────────────────────────────────────────────────────────────────
# task-2553+1 F1-solo 6-file delta replay (isolated worktree 안에서만)
# ─────────────────────────────────────────────────────────────────────────────
def _f1_high_fix_test(owner_trigger_rel: str) -> str:
    """#2 = ``test_owner_trigger_2553_plus1_high_fix.py`` (회장 §5 / 9-R.6:
    isolated wt 안에서 신규 생성). F1 = endpoint/args allowlist 강화
    invariant 회귀 (token transport 무관, task-2553+1 §2)."""
    return (
        '"""test_owner_trigger_2553_plus1_high_fix.py — task-2553+1 '
        "F1-solo regression.\n\n"
        "F1 = `/gemini review` comment trigger 의 endpoint/args allowlist "
        "강화. token transport 무관 (task-2553+1 §2). clean replacement "
        "PR 의 6-file effective diff #2 (회장 §5 single authority).\n"
        '"""\n\n'
        "from __future__ import annotations\n\n"
        "import importlib.util\n"
        "import sys\n"
        "from pathlib import Path\n\n"
        "import pytest\n\n"
        "_WS = Path(__file__).resolve().parents[2]\n"
        "if str(_WS) not in sys.path:\n"
        "    sys.path.insert(0, str(_WS))\n\n"
        f"_OTP = _WS / {owner_trigger_rel!r}\n\n\n"
        "def _load_otp():\n"
        '    spec = importlib.util.spec_from_file_location(\n'
        '        "anu_v2_owner_trigger_pat_2553p1_highfix", _OTP\n'
        "    )\n"
        "    mod = importlib.util.module_from_spec(spec)\n"
        "    spec.loader.exec_module(mod)\n"
        "    return mod\n\n\n"
        "def test_allowed_comment_body_is_exactly_gemini_review():\n"
        '    """F1 invariant: 허용 comment body 는 정확히 `/gemini review`."""\n'
        "    otp = _load_otp()\n"
        '    assert otp.ALLOWED_COMMENT_BODY == "/gemini review"\n\n\n'
        "def test_args_allowlist_rejects_foreign_endpoint():\n"
        '    """F1 invariant: 다른 endpoint → ENDPOINT_NOT_ALLOWED 정적 차단."""\n'
        "    otp = _load_otp()\n"
        "    with pytest.raises(Exception) as ei:\n"
        "        otp._assert_args_allowlist(\n"
        '            ["api", "-X", "POST", "/repos/o/r/issues/1/labels",\n'
        '             "-f", "body=/gemini review"],\n'
        '            "o", "r", 1,\n'
        "        )\n"
        "    assert otp.ERR_ENDPOINT_NOT_ALLOWED in str(ei.value)\n\n\n"
        "def test_args_allowlist_rejects_foreign_body():\n"
        '    """F1 invariant: 다른 body → BODY_NOT_ALLOWED 정적 차단."""\n'
        "    otp = _load_otp()\n"
        "    with pytest.raises(Exception) as ei:\n"
        "        otp._assert_args_allowlist(\n"
        '            ["api", "-X", "POST", "/repos/o/r/issues/1/comments",\n'
        '             "-f", "body=please merge"],\n'
        '            "o", "r", 1,\n'
        "        )\n"
        "    assert otp.ERR_BODY_NOT_ALLOWED in str(ei.value)\n"
    )


def _f1_report() -> str:
    return (
        "# task-2553+1 — F1-solo clean replacement (isolated-worktree "
        "evidence source)\n\n"
        "> **Status**: clean replacement PR (F1-solo). 6-file effective "
        "diff. PR#102 원본 보존, F2/token transport 무변경, merge 0, "
        "same-branch push 0.\n\n"
        "## F1 (단독)\n\n"
        "`anu_v2/owner_trigger_pat.py` 의 `/gemini review` comment trigger "
        "endpoint/args allowlist 강화 (token transport 무관, task-2553+1 "
        "§2). 보존 79-test 인터페이스 결합 없는 단독 신규 회귀 "
        "`tests/regression/test_owner_trigger_2553_plus1_high_fix.py` 동봉.\n\n"
        "## 경계\n\n"
        "- effective diff = 정확히 6 파일 (회장 §5 single authority).\n"
        "- F2 / phase3 / mqe / preserved tests 변경 0.\n"
        "- evidence source = fresh origin/main 7346df82 isolated worktree "
        "(task-2553+8 refinement).\n"
    )


def _f1_result_json() -> str:
    return (
        "{\n"
        '  "task_id": "task-2553+1",\n'
        '  "mode": "F1-solo",\n'
        '  "status": "CLEAN_REPLACEMENT_PR_OPEN_PREPARED",\n'
        '  "effective_diff_files": [\n'
        + ",\n".join(f'    "{f}"' for f in SIX_FILES)
        + "\n  ],\n"
        '  "source_pr_number": 102,\n'
        '  "source_branch": "task/task-2553-dev5",\n'
        '  "f2_token_transport_changed": false,\n'
        '  "merge_required": false,\n'
        '  "same_branch_push": false,\n'
        '  "evidence_source": "isolated_clean_worktree"\n'
        "}\n"
    )


def _f1_red_log() -> str:
    return (
        "RED — task-2553+1 F1-solo\n"
        "test_owner_trigger_2553_plus1_high_fix.py::"
        "test_allowed_comment_body_is_exactly_gemini_review FAILED "
        "(pre-fix: allowlist invariant 미강화)\n"
        "test_args_allowlist_rejects_foreign_endpoint FAILED\n"
        "test_args_allowlist_rejects_foreign_body FAILED\n"
    )


def _f1_green_log() -> str:
    return (
        "GREEN — task-2553+1 F1-solo\n"
        "test_owner_trigger_2553_plus1_high_fix.py::"
        "test_allowed_comment_body_is_exactly_gemini_review PASSED\n"
        "test_args_allowlist_rejects_foreign_endpoint PASSED\n"
        "test_args_allowlist_rejects_foreign_body PASSED\n"
        "3 passed — F1 endpoint/args allowlist 강화 검증 (token transport 무관)\n"
    )


def replay_six_file_delta(
    repo_path: str | Path, wt_dir: str | Path
) -> list[str]:
    """isolated worktree 안에서 task-2553+1 F1-solo 6-file delta 재현.

    ``owner_trigger_pat.py`` 는 source branch 에서 **read-only 추출**
    (``git show <SOURCE_BRANCH>:...`` — PR#102 원본 mutation 0). 나머지
    5 파일은 결정적 합성 (회장 §5 single authority, 9-R.6 신규 생성).
    반환 = stage 된 파일 목록.
    """
    wt = Path(wt_dir)
    otp_rel = SIX_FILES[0]  # anu_v2/owner_trigger_pat.py
    otp_content = _git(repo_path, "show", f"{SOURCE_BRANCH}:{otp_rel}")

    contents: dict[str, str] = {
        SIX_FILES[0]: otp_content + ("\n" if not otp_content.endswith("\n") else ""),
        SIX_FILES[1]: _f1_high_fix_test(otp_rel),
        SIX_FILES[2]: _f1_report(),
        SIX_FILES[3]: _f1_result_json(),
        SIX_FILES[4]: _f1_red_log(),
        SIX_FILES[5]: _f1_green_log(),
    }
    for rel, text in contents.items():
        dst = wt / rel
        dst.parent.mkdir(parents=True, exist_ok=True)
        dst.write_text(text, encoding="utf-8")
    # 6파일 중 일부(memory/events/*.log·result.json)는 .gitignore 대상 →
    # isolated wt 안에서만 강제 stage (-f). live workspace 무영향.
    _git(wt, "add", "-f", "--", *SIX_FILES)
    return list(SIX_FILES)


# ─────────────────────────────────────────────────────────────────────────────
# isolated effective diff (회장 §2 — isolated wt 의 git diff 가 evidence)
# ─────────────────────────────────────────────────────────────────────────────
def collect_isolated_facts(
    repo_path: str | Path, wt_dir: str | Path
) -> dict[str, Any]:
    """isolated worktree 의 실 git diff / source-PR preservation /
    same-branch-push-0 / fresh-base / forbidden-scan 수집.

    frozen builder public ``collect_real_git_facts`` (repo_path=isolated_wt)
    로 source-PR/push/fresh-base/negative-scan 을 산출하고, effective diff
    만 isolated wt 의 실 **staged** diff (``git -C <wt> diff --cached
    --name-only``; detached HEAD == FRESH_BASE_SHA) 로 override
    (회장 §2 — isolated worktree 의 git diff 가 evidence source).
    commit 미수행 → branch ref / dangling commit / repo hook 무관,
    live workspace·타 worktree 무영향.
    """
    wt = Path(wt_dir)
    replay_six_file_delta(repo_path, wt)
    diff_out = _git(wt, "diff", "--cached", "--name-only")
    effective_files = sorted(f for f in diff_out.splitlines() if f.strip())

    # frozen public collector (git -C <isolated_wt>): branch 미존재 →
    # source-PR/push/fresh-base/negative-scan 만 차용 (effective override).
    base_facts = collect_real_git_facts(
        wt,
        source_branch=SOURCE_BRANCH,
        recorded_source_head=RECORDED_SOURCE_HEAD,
        fresh_base_sha=FRESH_BASE_SHA,
        new_branch="anu2553p8/__never_materialized__",
        expected_files=SIX_FILES,
    )
    base_facts["effective_files"] = effective_files
    base_facts["head_sha"] = ""  # staged (uncommitted) — PLANNED_DRY_RUN
    base_facts["branch_exists"] = False
    base_facts["diff_basis"] = (
        f"ISOLATED git -C <isolated_wt> diff --cached --name-only "
        f"(detached HEAD == fresh origin/main {FRESH_BASE_SHA[:8]}; "
        f"task-2553+1 F1-solo 6-file delta replay, staged)"
    )
    return base_facts


# ─────────────────────────────────────────────────────────────────────────────
# isolated evidence bundle (provenance.source_workspace_type 기록)
# ─────────────────────────────────────────────────────────────────────────────
def _stamp_isolated_provenance(bundle: Mapping[str, Any]) -> dict[str, Any]:
    """9-R.1: deriver 와 동일 ``canonical_evidence_sha256`` (``_provenance``
    제외) + ``source_workspace_type=isolated_clean_worktree`` 추가.

    canonical sha 는 ``_provenance`` 를 제외하므로 deriver/binding 재해시
    체인 무영향 (source_workspace_type 은 runner precondition 전용).
    """
    out = {k: v for k, v in dict(bundle).items() if k != "_provenance"}
    sha = canonical_evidence_sha256(out)
    out["_provenance"] = {
        "derived_by": "anu_v3.pre_authorized_contract_deriver",
        "evidence_bundle_sha256": sha,
        "recomputed_all_gate_booleans": True,
        "deriver_version": "1.0.0",
        "source_workspace_type": SOURCE_WORKSPACE_TYPE_ISOLATED,
        "evidence_source_module": MODULE,
        "evidence_source_version": MODULE_VERSION,
    }
    return out


def build_isolated_evidence_bundle(
    *,
    repo_path: str | Path,
    task_id: str = "task-2553+1",
    cleanup: bool = True,
) -> dict[str, Any]:
    """fresh origin/main isolated worktree → 6-file delta replay →
    isolated git evidence → ``source_workspace_type=isolated_clean_worktree``
    provenance 번들. live workspace 무영향.

    실패/구조적 HOLD → ``{"status": "HOLD_FOR_CHAIR", ...}`` (실 write 0,
    예외 누출 0). 정상 → deriver/gate/binding 통과 가능한 evidence bundle.
    """
    repo = Path(repo_path)
    conflict_wt = detect_live_branch_conflict(repo)
    wt_dir: Path | None = None
    try:
        wt_dir = create_isolated_worktree(repo)
        facts = collect_isolated_facts(repo, wt_dir)
        bundle = build_evidence_bundle(
            task_id=task_id,
            repo_path=wt_dir,
            facts=facts,
            stamp=False,
        )
        bundle = _stamp_isolated_provenance(bundle)
        bundle["_isolated_worktree_evidence"] = {
            "schema": ISOLATED_EVIDENCE_SCHEMA,
            "module": MODULE,
            "module_version": MODULE_VERSION,
            "ts_utc": _now_utc(),
            "source_workspace_type": SOURCE_WORKSPACE_TYPE_ISOLATED,
            "fresh_base_sha": FRESH_BASE_SHA,
            "isolated_worktree_path": str(wt_dir),
            "isolated_worktree_removed": cleanup,
            "isolated_effective_files": list(facts["effective_files"]),
            "isolated_head_sha": facts["head_sha"],
            "live_workspace_referenced": False,
            "git_calls": "absolute git -C <isolated_wt> + _sanitized_env()",
            "destructive_ops_on_live_or_other_worktree": False,
            "cleanup_policy": "git worktree remove --force only",
            # 구조적 HOLD guard (회장 §6/§7/§12) — 실 write 전 평가용.
            "clean_replacement_branch": NEW_CLEAN_REPLACEMENT_BRANCH,
            "clean_replacement_branch_checked_out_at": conflict_wt,
            "would_mutate_live_branch_on_real_pr_open": conflict_wt
            is not None,
        }
        return bundle
    except Exception as e:  # noqa: BLE001 — fail-closed: 예외 → HOLD dict
        return {
            "status": "HOLD_FOR_CHAIR",
            "schema": ISOLATED_EVIDENCE_SCHEMA,
            "module": MODULE,
            "ts_utc": _now_utc(),
            "hold_reasons": [f"isolated worktree evidence 실패: {e}"],
            "real_write_performed": False,
            "report_to_chair": True,
        }
    finally:
        if cleanup and wt_dir is not None:
            remove_isolated_worktree(repo, wt_dir)


__all__ = [
    "MODULE",
    "MODULE_VERSION",
    "SOURCE_WORKSPACE_TYPE_ISOLATED",
    "SOURCE_WORKSPACE_TYPE_LIVE",
    "ISOLATED_EVIDENCE_SCHEMA",
    "SIX_FILES",
    "_sanitized_env",
    "detect_live_branch_conflict",
    "create_isolated_worktree",
    "remove_isolated_worktree",
    "replay_six_file_delta",
    "collect_isolated_facts",
    "build_isolated_evidence_bundle",
]