"""test_isolated_worktree_evidence_2553plus8.py — task-2553+8 regression.

ISOLATED-WORKTREE EVIDENCE SOURCE REFINEMENT — §8 필수 15 + 9-R.3 추가 9
(총 24) + fixture 3종 시나리오.

전부 부작용 0: isolated worktree 는 self-cleaning(git worktree remove 만,
live workspace·타 worktree 무영향 — 정적 grep + 런타임 전후 불변 검증).
실 GitHub write/PR open 0.
"""

from __future__ import annotations

import ast
import copy
import json
import subprocess
import sys
from pathlib import Path
from typing import Any

import pytest

WORKSPACE = Path(__file__).resolve().parents[2]
if str(WORKSPACE) not in sys.path:
    sys.path.insert(0, str(WORKSPACE))

from anu_v3.isolated_worktree_evidence_source import (  # noqa: E402
    SOURCE_WORKSPACE_TYPE_ISOLATED,
    SOURCE_WORKSPACE_TYPE_LIVE,
    build_isolated_evidence_bundle,
    _sanitized_env,
)
from anu_v3.pre_authorized_contract_deriver import (  # noqa: E402
    derive,
)
from anu_v3.pre_authorized_action_gate import (  # noqa: E402
    DECISION_ALLOW,
    evaluate,
)
from anu_v3.pre_authorized_executor_binding import (  # noqa: E402
    STATUS_DRY_RUN_PLAN,
    bind,
)
from anu_v3.pre_authorized_activation_runner import (  # noqa: E402
    STATUS_ACTIVATED_PR_OPEN,
    STATUS_DRY_RUN_VERIFIED,
    STATUS_HOLD,
    run_activation,
)

FX = WORKSPACE / "memory" / "fixtures"
ISO_CLEAN = FX / "task-2553+1.f1solo.isolated-clean-worktree.contract.json"
LIVE_CONTAM = FX / "task-2553+1.live-workspace-contaminated.contract.json"
ISO_7TH = FX / "task-2553+1.isolated-7th-file-contamination.contract.json"
MODULE_SRC = WORKSPACE / "anu_v3" / "isolated_worktree_evidence_source.py"
RUNNER_SRC = WORKSPACE / "anu_v3" / "pre_authorized_activation_runner.py"


def _j(p: Path) -> dict[str, Any]:
    return json.loads(p.read_text(encoding="utf-8"))


def _live_head() -> str:
    return subprocess.run(
        ["git", "-C", str(WORKSPACE), "rev-parse", "HEAD"],
        capture_output=True, text=True, check=True, env=_sanitized_env(),
    ).stdout.strip()


@pytest.fixture(scope="module")
def iso_clean() -> dict[str, Any]:
    return _j(ISO_CLEAN)


# ── fixture 3종 존재 ──────────────────────────────────────────────────────────
def test_fixtures_three_exist():
    assert ISO_CLEAN.exists() and LIVE_CONTAM.exists() and ISO_7TH.exists()


# ── §8 #1 — live workspace contaminated evidence → HOLD ──────────────────────
def test_01_live_workspace_contaminated_hold():
    d = derive(_j(LIVE_CONTAM))
    assert d["status"] == STATUS_HOLD


# ── §8 #2 — isolated clean + exact 6-file delta → ALLOW ──────────────────────
def test_02_isolated_clean_six_file_allow(iso_clean):
    d = derive(iso_clean)
    assert d["status"] == "DERIVED", d.get("hold_reasons")
    dec = evaluate(d["contract"])
    assert dec["decision"] == DECISION_ALLOW
    iso = iso_clean["_isolated_worktree_evidence"]
    assert len(iso["isolated_effective_files"]) == 6
    assert (
        iso_clean["_provenance"]["source_workspace_type"]
        == SOURCE_WORKSPACE_TYPE_ISOLATED
    )


# ── §8 #3 — isolated + 7th file contamination → HOLD ─────────────────────────
def test_03_isolated_seventh_file_hold():
    assert derive(_j(ISO_7TH))["status"] == STATUS_HOLD


# ── §8 #4 — source_workspace_type missing → HOLD (PR-open) ───────────────────
def test_04_source_workspace_type_missing_hold_pr_open(iso_clean):
    b = copy.deepcopy(iso_clean)
    b["_provenance"].pop("source_workspace_type", None)
    r = run_activation(
        repo_path=str(WORKSPACE), evidence_bundle=b, activate=True
    )
    assert r["status"] == STATUS_HOLD
    assert r.get("github_mutation_performed", False) is False


# ── §8 #5 — source_workspace_type=live_workspace → HOLD (PR-open) ────────────
def test_05_source_workspace_type_live_hold_pr_open(iso_clean):
    b = copy.deepcopy(iso_clean)
    b["_provenance"]["source_workspace_type"] = SOURCE_WORKSPACE_TYPE_LIVE
    r = run_activation(
        repo_path=str(WORKSPACE), evidence_bundle=b, activate=True
    )
    assert r["status"] == STATUS_HOLD
    assert r.get("real_write_performed", False) is False


# ── §8 #6 — provenance mismatch → HOLD ───────────────────────────────────────
def test_06_provenance_mismatch_hold(iso_clean):
    # 위조 contract _provenance (실 evidence 재해시와 불일치) → binding 차단.
    b = copy.deepcopy(iso_clean)
    c = derive(b)["contract"]
    dec = evaluate(c)
    forged = copy.deepcopy(c)
    forged["_provenance"]["evidence_bundle_sha256"] = "0" * 64
    pf = bind(decision=dec, contract=forged, evidence_bundle=b, dry_run=True)
    assert pf["status"] != STATUS_DRY_RUN_PLAN


# ── §8 #7 — expected_files mismatch → HOLD ───────────────────────────────────
def test_07_expected_files_mismatch_hold(iso_clean):
    b = copy.deepcopy(
        {k: v for k, v in iso_clean.items() if k != "_provenance"}
    )
    b["evidence"]["expected_files"]["observed_value"]["files"] = [
        "anu_v2/owner_trigger_pat.py"
    ]
    assert derive(b)["status"] == STATUS_HOLD


# ── §8 #8 — F2 changed → HOLD ────────────────────────────────────────────────
def test_08_f2_changed_hold(iso_clean):
    b = copy.deepcopy(
        {k: v for k, v in iso_clean.items() if k != "_provenance"}
    )
    ev = b["evidence"]["git_effective_diff"]["observed_value"]
    ev["files"] = sorted(set(ev["files"]) | {"anu_v2/owner_trigger_pat_f2.py"})
    assert derive(b)["status"] == STATUS_HOLD


# ── §8 #9 — phase3/mqe changed → HOLD ────────────────────────────────────────
def test_09_phase3_mqe_changed_hold(iso_clean):
    b = copy.deepcopy(
        {k: v for k, v in iso_clean.items() if k != "_provenance"}
    )
    ev = b["evidence"]["git_effective_diff"]["observed_value"]
    ev["files"] = sorted(set(ev["files"]) | {"anu_v2/merge_queue_executor.py"})
    assert derive(b)["status"] == STATUS_HOLD


# ── §8 #10 — same_branch_push=true → HOLD ────────────────────────────────────
def test_10_same_branch_push_true_hold(iso_clean):
    b = copy.deepcopy(
        {k: v for k, v in iso_clean.items() if k != "_provenance"}
    )
    b["evidence"]["same_branch_push_zero"]["observed_value"][
        "source_branch_push_count"
    ] = 3
    assert derive(b)["status"] == STATUS_HOLD


# ── §8 #11 — source PR preserved=false → HOLD ────────────────────────────────
def test_11_source_pr_not_preserved_hold(iso_clean):
    b = copy.deepcopy(
        {k: v for k, v in iso_clean.items() if k != "_provenance"}
    )
    b["evidence"]["source_pr_preservation"]["observed_value"][
        "recomputed_head_sha"
    ] = "f" * 40
    assert derive(b)["status"] == STATUS_HOLD


# ── §8 #12 — merge_required=true → HOLD ──────────────────────────────────────
def test_12_merge_required_true_hold(iso_clean):
    b = copy.deepcopy(
        {k: v for k, v in iso_clean.items() if k != "_provenance"}
    )
    b["evidence"]["scope_declaration"]["observed_value"][
        "requires_merge"
    ] = True
    d = derive(b)
    if d["status"] == "DERIVED":
        assert evaluate(d["contract"])["decision"] != DECISION_ALLOW
    else:
        assert d["status"] == STATUS_HOLD


# ── §8 #13 — callback missing → HOLD ─────────────────────────────────────────
def test_13_callback_missing_hold(iso_clean):
    b = copy.deepcopy(
        {k: v for k, v in iso_clean.items() if k != "_provenance"}
    )
    b["evidence"]["callback_policy_marker"]["observed_value"][
        "authority"
    ] = "broad"
    assert derive(b)["status"] == STATUS_HOLD


# ── §8 #14 — PR-open path only after isolated evidence ALLOW ──────────────────
def test_14_pr_open_only_after_isolated_allow(iso_clean):
    # isolated ALLOW + activate → reaches real path (but structural guard
    # / live-branch conflict yields HOLD, NEVER silent PR open).
    b = copy.deepcopy(iso_clean)
    b["_provenance"]["source_workspace_type"] = "live_workspace"
    r = run_activation(
        repo_path=str(WORKSPACE), evidence_bundle=b, activate=True
    )
    assert r["status"] == STATUS_HOLD
    assert r["status"] != STATUS_ACTIVATED_PR_OPEN


# ── §8 #15 — merge path unreachable (static) ─────────────────────────────────
def test_15_merge_path_unreachable_static():
    src = MODULE_SRC.read_text(encoding="utf-8")
    tree = ast.parse(src)
    bad = []
    for node in ast.walk(tree):
        if isinstance(node, ast.Call):
            seg = (ast.get_source_segment(src, node) or "").lower().replace(
                " ", ""
            )
            if "ghpr" in seg and "merge" in seg:
                bad.append(seg)
            if "pr_merge" in seg or "--auto" in seg:
                bad.append(seg)
    assert not bad


# ── 9-R.3 #16 — evidence_bundle=None + activate → 구조적 HOLD (seal ①) ───────
def test_16_none_evidence_activate_structural_hold():
    r = run_activation(
        repo_path=str(WORKSPACE), evidence_bundle=None, activate=True
    )
    assert r["status"] == STATUS_HOLD
    assert r.get("real_write_performed", False) is False
    assert any(
        "SEAL" in x or "live build 봉인" in x or "live_fallback" in str(
            r.get("stage", "")
        )
        for x in r.get("hold_reasons", [])
    )


# ── 9-R.3 #17 — swt missing/=live_workspace → HOLD ───────────────────────────
def test_17_swt_missing_or_live_hold(iso_clean):
    for mutate in (
        lambda b: b["_provenance"].pop("source_workspace_type", None),
        lambda b: b["_provenance"].__setitem__(
            "source_workspace_type", "live_workspace"
        ),
    ):
        b = copy.deepcopy(iso_clean)
        mutate(b)
        r = run_activation(
            repo_path=str(WORKSPACE), evidence_bundle=b, activate=True
        )
        assert r["status"] == STATUS_HOLD


# ── 9-R.3 #18 — GIT_DIR/GIT_WORK_TREE 오염 주입 → isolation 유지 ─────────────
def test_18_git_env_pollution_isolation_preserved(monkeypatch):
    before = _live_head()
    monkeypatch.setenv("GIT_DIR", "/tmp/__poison_gitdir__")
    monkeypatch.setenv("GIT_WORK_TREE", "/tmp/__poison_worktree__")
    monkeypatch.setenv("GIT_INDEX_FILE", "/tmp/__poison_index__")
    se = _sanitized_env()
    assert "GIT_DIR" not in se and "GIT_WORK_TREE" not in se
    assert "GIT_INDEX_FILE" not in se and "GIT_COMMON_DIR" not in se
    b = build_isolated_evidence_bundle(repo_path=str(WORKSPACE))
    assert b.get("status") != "HOLD_FOR_CHAIR", b
    assert (
        b["_provenance"]["source_workspace_type"]
        == SOURCE_WORKSPACE_TYPE_ISOLATED
    )
    assert _live_head() == before  # isolation: live 불변


# ── 9-R.3 #19 — live-대상 파괴적 op 정적 부재 ────────────────────────────────
def test_19_no_destructive_ops_on_live_static():
    src = MODULE_SRC.read_text(encoding="utf-8")
    tree = ast.parse(src)
    forbidden = ("rmtree", "unlink", "os.remove", "shutil.rmtree")
    git_destructive = ("reset", "clean", "stash")
    for node in ast.walk(tree):
        if isinstance(node, ast.Call):
            seg = (ast.get_source_segment(src, node) or "")
            low = seg.lower()
            for f in forbidden:
                assert f not in low, f"파괴적 호출 발견: {seg[:80]}"
            if "git" in low and ('"-c"' in seg or "git_destructive" in low):
                pass
            for g in git_destructive:
                # git reset/clean/stash 인자 부재 (worktree remove/prune 만 허용)
                assert (
                    f'"{g}"' not in seg and f"'{g}'" not in seg
                ), f"git {g} 발견: {seg[:80]}"


# ── 9-R.3 #20 — 7th-file 오염 deriver→gate 체인 HOLD 전파 ────────────────────
def test_20_seventh_file_propagates_hold_through_chain():
    b = _j(ISO_7TH)
    d = derive(b)
    assert d["status"] == STATUS_HOLD
    # gate 도 (혹시 DERIVED 되어도) ALLOW 아님 — 체인 전파 보장
    if d.get("contract"):
        assert evaluate(d["contract"])["decision"] != DECISION_ALLOW


# ── 9-R.3 #21 — replay 중 live mtime/path 미참조 ─────────────────────────────
def test_21_no_live_path_or_mtime_reference_static():
    src = MODULE_SRC.read_text(encoding="utf-8")
    assert "getmtime" not in src and "st_mtime" not in src
    assert "/home/jay/workspace/anu_v2" not in src
    # 모든 git 호출은 git -C <target> + _sanitized_env (cwd 인자 부재)
    assert "cwd=" not in src
    assert "_sanitized_env()" in src
    iso = build_isolated_evidence_bundle(repo_path=str(WORKSPACE))
    assert iso["_isolated_worktree_evidence"]["live_workspace_referenced"] is False


# ── 9-R.3 #22 — 중복 PR idempotency guard ────────────────────────────────────
def test_22_duplicate_idempotency_no_side_effect():
    before = _live_head()
    b1 = build_isolated_evidence_bundle(repo_path=str(WORKSPACE))
    b2 = build_isolated_evidence_bundle(repo_path=str(WORKSPACE))
    # 반복 호출 무부작용 (live 불변, evidence 결정적 일관)
    assert _live_head() == before
    assert (
        b1["_provenance"]["source_workspace_type"]
        == b2["_provenance"]["source_workspace_type"]
        == SOURCE_WORKSPACE_TYPE_ISOLATED
    )
    assert sorted(
        b1["_isolated_worktree_evidence"]["isolated_effective_files"]
    ) == sorted(b2["_isolated_worktree_evidence"]["isolated_effective_files"])


# ── 9-R.3 #23 — worktree 제거 실패 시에도 live 무결 ──────────────────────────
def test_23_live_integrity_even_if_worktree_remove_fails():
    before = _live_head()
    # cleanup=False → worktree 남김 (제거 미수행) — live 는 여전히 불변
    b = build_isolated_evidence_bundle(
        repo_path=str(WORKSPACE), cleanup=False
    )
    try:
        assert _live_head() == before
        assert b.get("status") != "HOLD_FOR_CHAIR"
    finally:
        # 테스트 정리: git worktree remove 만 (live·타 worktree 무영향)
        from anu_v3.isolated_worktree_evidence_source import (
            remove_isolated_worktree,
        )

        iso = b.get("_isolated_worktree_evidence", {})
        wt_path = iso.get("isolated_worktree_path")
        if wt_path:
            remove_isolated_worktree(WORKSPACE, wt_path)
        assert _live_head() == before


# ── 9-R.3 #24 — isolated clean ALLOW → 실 PR open path 도달, 그 외 HOLD ──────
def test_24_isolated_clean_allow_reaches_real_path_else_hold(iso_clean):
    # dry-run: isolated clean → 전 체인 통과 (DRY_RUN_VERIFIED, 실 write 0)
    r_dry = run_activation(
        repo_path=str(WORKSPACE), evidence_bundle=iso_clean, activate=False
    )
    assert r_dry["status"] == STATUS_DRY_RUN_VERIFIED
    assert r_dry["gate_decision"] == DECISION_ALLOW
    assert r_dry["binding_preflight_status"] == STATUS_DRY_RUN_PLAN
    assert r_dry["real_write_performed"] is False
    # 비isolated provenance → 실 PR open path 차단 (그 외 전부 HOLD)
    b = copy.deepcopy(iso_clean)
    b["_provenance"]["source_workspace_type"] = None
    r_blk = run_activation(
        repo_path=str(WORKSPACE), evidence_bundle=b, activate=True
    )
    assert r_blk["status"] == STATUS_HOLD
    assert r_blk.get("github_mutation_performed", False) is False


# ── runner seal ② 정적 증명 ──────────────────────────────────────────────────
def test_runner_seal_static_present():
    src = RUNNER_SRC.read_text(encoding="utf-8")
    assert "SEAL ①" in src and "SEAL ②" in src
    assert "isolated_clean_worktree" in src
    assert "source_workspace_type_precondition" in src