#!/usr/bin/env python3
"""P3-B enforcement 강제기 — standalone check(pytest 아님). exit 0=통과.

validate_method_execution(보고서 step-results 증명) + handoff gate + transition gate + debt.
합성 계약/활성화를 methods/activations 인자로 직접 주입해 policy 분기를 검증한다(디스크 미오염).
"""
import importlib.util
import os
import sys

import yaml  # noqa: F401

ROOT = os.environ.get("CLAUDE_PROJECT_DIR", os.getcwd())
HOOKS = os.path.join(ROOT, ".claude", "hooks")

passed = failed = 0


def check(name, ok):
    global passed, failed
    if ok:
        passed += 1
        print(f"  ✅ {name}")
    else:
        failed += 1
        print(f"  ❌ {name}")


def _load(mod, path):
    spec = importlib.util.spec_from_file_location(mod, os.path.join(HOOKS, path))
    m = importlib.util.module_from_spec(spec)
    if HOOKS not in sys.path:
        sys.path.insert(0, HOOKS)
    spec.loader.exec_module(m)
    return m


mc = _load("method_contracts", "method_contracts.py")

# --- 합성 계약(v2) + 활성화 --------------------------------------------------
_PROF = {
    "method-id": "m1",
    "workflow": [
        {"step-id": "s1", "required-output": "brief"},
        {"step-id": "s2"},
        {"step-id": "s3", "skippable": True, "skip-rules": [{"rule-id": "no-visual"}]},
    ],
    "alternatives-policy": {"min-alternatives": 2},
}
METHODS = {
    "TST": {"method-contract": {"version": 2}, "methods": [_PROF]},
    "V1": {"working-method": ["..."]},
}
HASH = mc.canonical_contract_hash(_PROF)
ACTS = {"TST": {"methods": {"m1": {"status": "active", "contract-sha256": HASH}}}}
ACTS_DRAFT = {"TST": {"methods": {"m1": {"status": "draft", "contract-sha256": HASH}}}}


def _me_ok():
    return {
        "role-id": "TST", "method-id": "m1", "contract-sha256": HASH,
        "step-results": [
            {"step-id": "s1", "status": "completed", "artifact-refs": [{"report-id": "r1", "sha256": "a" * 64}]},
            {"step-id": "s2", "status": "completed"},
            {"step-id": "s3", "status": "skipped", "skip-rule-id": "no-visual"},
        ],
        "decisions": [{"decision-id": "d1",
                       "alternatives": [{"option-id": "a"}, {"option-id": "b"}],
                       "selected-option-id": "a"}],
    }


def _vme(me, role="TST", tier="standard", acts=ACTS):
    rep = {"role-id": role, "tier": tier}
    if me is not None:
        rep["method-execution"] = me
    return mc.validate_method_execution(rep, activations=acts, methods=METHODS)


print("== T4.1: validate_method_execution 무강제 조건 ==")
check("v1 역할 → 무강제", _vme(None, role="V1") == [])
check("draft 계약 → 무강제(trace-only)", _vme(None, acts=ACTS_DRAFT) == [])
check("light tier → 무강제", _vme(None, tier="light") == [])

print("== T4.1: validate_method_execution 강제(active·standard) ==")
check("정상 실행 → 통과", _vme(_me_ok()) == [])
check("method-execution 누락 → 에러(자기신고 금지)",
      any("method-execution.method-id 없음" in e for e in _vme(None)))
_bad = _me_ok(); _bad["contract-sha256"] = "0" * 64
check("contract-sha256 불일치 → 에러(구버전 실행)",
      any("contract-sha256" in e for e in _vme(_bad)))
_bad = _me_ok(); _bad["step-results"] = [s for s in _bad["step-results"] if s["step-id"] != "s2"]
check("필수 step 누락(s2) → 에러",
      any("필수 step 's2'" in e for e in _vme(_bad)))
_bad = _me_ok(); _bad["step-results"][0].pop("artifact-refs")
check("completed·required-output인데 artifact-ref 없음 → 에러",
      any("artifact-ref 없음" in e for e in _vme(_bad)))
_bad = _me_ok(); _bad["step-results"][1]["status"] = "skipped"; _bad["step-results"][1]["skip-rule-id"] = "x"
check("skippable 아닌 step(s2) skip → 에러",
      any("skippable 아님" in e for e in _vme(_bad)))
_bad = _me_ok(); _bad["step-results"][2]["skip-rule-id"] = "unauthorized"
check("허용 안 된 skip-rule → 에러",
      any("skip-rule-id" in e for e in _vme(_bad)))
_bad = _me_ok(); _bad["decisions"][0]["alternatives"] = [{"option-id": "a"}]
check("대안 부족(<min) → 에러",
      any("대안" in e for e in _vme(_bad)))

print("== T4.1: validate_report 배선(통합) ==")
vr = _load("validate_report", "validate_report.py")
# validate_report 가 실제로 policy engine 을 호출하는지 — v2 active 역할 주입해 확인
vr._MC.load_role_methods = lambda: METHODS
vr._MC.load_activations = lambda: ACTS
try:
    rep = {"role-id": "TST", "tier": "standard",
           "report-header": {"bottom-line": "x"}, "method-execution": None}
    errs = vr.validate(rep)
    check("validate_report 가 method-execution 게이트 호출(누락 에러 포함)",
          any("method-execution.method-id 없음" in e for e in errs))
    rep2 = {"role-id": "V1", "tier": "standard", "report-header": {"bottom-line": "x"}}
    errs2 = vr.validate(rep2)
    check("v1 역할 보고서 → method-execution 에러 없음(무회귀)",
          not any("method-execution" in e for e in errs2))
finally:
    pass

print("== T4.2: evaluate_handoff_edge (hard/soft·debt) ==")
_edge = {"edge-id": "e1", "artifact-type": "selected-direction", "required-state": "Accepted",
         "from": {"role-id": "P", "method-id": "mp"}, "to": {"role-id": "C", "method-id": "mc"}}
_both_active = {"P": {"methods": {"mp": {"status": "active"}}},
                "C": {"methods": {"mc": {"status": "active"}}}}
_one_draft = {"P": {"methods": {"mp": {"status": "draft"}}},
              "C": {"methods": {"mc": {"status": "active"}}}}
_yes = lambda e: True   # noqa: E731
_no = lambda e: False   # noqa: E731

_r = mc.evaluate_handoff_edge(_edge, present=_yes, accepted=_yes, activations=_both_active)
check("both-active·충족 → ok·hard", _r["ok"] and _r["hard"] and _r["violations"] == [])
_r = mc.evaluate_handoff_edge(_edge, present=_no, accepted=_no, activations=_both_active)
check("both-active·아티팩트 부재 → 차단(ok=False, hard)", (not _r["ok"]) and _r["hard"] and _r["violations"])
_r = mc.evaluate_handoff_edge(_edge, present=_yes, accepted=_no, activations=_both_active)
check("both-active·미수락(Accepted 필요) → 차단", not _r["ok"])
_r = mc.evaluate_handoff_edge(_edge, present=_no, accepted=_no, activations=_one_draft)
check("한쪽 draft·위반 → 비차단(ok=True) + debt opened",
      _r["ok"] and (not _r["hard"]) and _r["debt"] and _r["debt"]["type"] == "handoff-draft-unmet")

print("== T4.2: handoff_violations (consumer required-inputs) ==")
_PROF_C = {"method-id": "mc", "required-inputs": [
    {"artifact-type": "selected-direction", "from-role": "P", "from-method": "mp", "required-state": "Accepted"},
    {"artifact-type": "optional-note", "from-role": "P", "optional": True},
]}
_M2 = {"C": {"method-contract": {"version": 2}, "methods": [_PROF_C]},
       "P": {"method-contract": {"version": 2}, "methods": [{"method-id": "mp"}]}}
_errs, _debts = mc.handoff_violations("C", "mc", present=_no, accepted=_no,
                                      activations=_both_active, methods=_M2)
check("required-input 미충족(both-active) → blocking error", _errs != [])
check("optional 입력은 게이트 제외", all("optional-note" not in e for e in _errs))
_errs2, _ = mc.handoff_violations("C", "mc", present=_yes, accepted=_yes,
                                  activations=_both_active, methods=_M2)
check("required-input 충족 → 통과", _errs2 == [])
_errs3, _ = mc.handoff_violations("V1", "x", present=_no, accepted=_no, methods=METHODS)
check("v1/미존재 profile → 무게이트", _errs3 == [])

print("== T4.2: context_package spawn 게이트 배선(무회귀) ==")
cp = _load("context_package", "context_package.py")
_sel = [e for e in cp.validate({"target-role-agent": "prod-pm", "tier": "standard",
                                "method-selection": {"method-id": "x"}, "workflow-id": "wf-x"})
        if "handoff" in e]
check("현행 v1 역할 spawn → handoff 게이트 no-op(무회귀)", _sel == [])

print("== T4.2b: design-direction independent-review handoff 경계 ==")
_review_pkg = {
    "workflow-id": "wf-review", "task-id": "review-systematizability",
    "target-role-agent": "DES-PLATFORM", "tier": "standard",
    "method-selection": {"method-id": "tokenize"},
}
check("critique의 review-* task만 independent-review로 인식",
      cp._is_independent_design_review(
          _review_pkg, {"plan": "design-direction", "stage": "design-direction-critique"}))
check("같은 review-*라도 prototype stage에서는 handoff 면제 금지",
      not cp._is_independent_design_review(
          _review_pkg, {"plan": "design-direction", "stage": "design-direction-prototype"}))
check("critique라도 review-*가 아닌 생산 task는 handoff 면제 금지",
      not cp._is_independent_design_review(
          {**_review_pkg, "task-id": "build-prototype"},
          {"plan": "design-direction", "stage": "design-direction-critique"}))
_orig_review_boundary = cp._is_independent_design_review
try:
    cp._is_independent_design_review = lambda pkg: True
    check("independent-review spawn은 생산 method required-input을 요구하지 않음",
          cp._handoff_spawn_errors(_review_pkg) == [])
finally:
    cp._is_independent_design_review = _orig_review_boundary

# F2(2026-07-16 실측): 카드 존재 체크는 case-무관이어야 한다. role-id 는 대문자(DES-DIRECTOR)이고
# 에이전트 카드 파일은 소문자(des-director.md)라, verbatim 체크가 대문자 role 을 '카드 없다'로 거부했다
# (validate_report 은 commit 13d39a2 에서 이미 case-무관화됨 — 두 강제기 정합).
_f2_upper = [e for e in cp.validate({"target-role-agent": "DES-DIRECTOR", "tier": "standard",
                                     "method-selection": {"method-id": "frame-divergence"}, "workflow-id": "wf-x"})
             if "에이전트 카드" in e]
check("F2: 대문자 role-id(DES-DIRECTOR) → 소문자 카드(des-director.md)로 인정(카드-없음 오류 없음)", _f2_upper == [])
_f2_bogus = [e for e in cp.validate({"target-role-agent": "NOT-A-REAL-ROLE-XYZ", "tier": "standard",
                                     "method-selection": {"method-id": "x"}, "workflow-id": "wf-x"})
             if "에이전트 카드" in e]
check("F2: 진짜 미등록 role 은 여전히 카드-없음으로 차단(우회 방지)", _f2_bogus != [])

print("== T4.3: state_engine transition handoff gate ==")
se = _load("state_engine", "state_engine.py")
# predicate 존재 + bool 판정
_pred = se._PREDICATES.get("method-handoff-satisfied")
check("method-handoff-satisfied predicate 등록", _pred is not None)
check("미충족 fact → predicate False(전이 차단)", _pred({"method_handoff_unmet": True})[0] is False)
check("충족 fact → predicate True(전이 허용)", _pred({"method_handoff_unmet": False})[0] is True)
check("method_handoff_unmet 은 PROTECTED(자기신고 오버라이드 불가)",
      "method_handoff_unmet" in se._PROTECTED_FACTS)
# _method_handoff_unmet: ctx 미지정 → False(무회귀), 지정 시 handoff_violations 위임
check("ctx 무지정 → 미검사(False)", se._method_handoff_unmet("wf", {}, None) is False)
check("ctx.handoff-check 지정 → 검사 위임(에러 시 True)",
      se._method_handoff_unmet("wf", {}, {"handoff-check": [{"role": "C", "method": "mc"}]}) in (True, False))
# 위임 결과가 handoff_violations 를 실제 반영하는지 — state_engine 이 lazy import 하는
# sys.modules["method_contracts"] 를 패치(같은 인스턴스)
import method_contracts as _mcmod  # noqa: E402 — sys.modules 등록본
_orig_hv = _mcmod.handoff_violations
try:
    _mcmod.handoff_violations = lambda role, mid, **k: (["blocked!"], [])
    check("consumer required-inputs 위반 → 전이 게이트 unmet=True",
          se._method_handoff_unmet("wf", {}, {"handoff-check": [{"role": "C", "method": "mc"}]}) is True)
    _mcmod.handoff_violations = lambda role, mid, **k: ([], [])
    check("위반 없음 → 전이 게이트 unmet=False",
          se._method_handoff_unmet("wf", {}, {"handoff-check": [{"role": "C", "method": "mc"}]}) is False)
finally:
    _mcmod.handoff_violations = _orig_hv

print("== T4.4: migration-debt 원장(opened/resolved fold) ==")
import tempfile as _tf  # noqa: E402
with _tf.TemporaryDirectory() as _dtd:
    _dp = os.path.join(_dtd, "method-contract-debt.jsonl")
    mc.record_debt({"debt-id": "d1", "type": "handoff-draft-unmet", "edge-id": "e1"}, path=_dp)
    mc.record_debt({"debt-id": "d2", "type": "handoff-draft-unmet", "edge-id": "e2"}, path=_dp)
    check("2개 opened → unresolved 2", len(mc.unresolved_debt(path=_dp)) == 2)
    # d1 resolved → unresolved 1(최신 status fold)
    mc.record_debt({"debt-id": "d1", "status": "resolved"}, path=_dp)
    _u = mc.unresolved_debt(path=_dp)
    check("d1 resolved 후 → unresolved 1(fold)", len(_u) == 1 and _u[0]["debt-id"] == "d2")
    # 같은 key 중복 opened → fold 로 1개(중복 무해)
    mc.record_debt({"debt-id": "d2", "type": "handoff-draft-unmet", "edge-id": "e2"}, path=_dp)
    check("중복 opened → fold 로 1개(중복 무해)", len(mc.unresolved_debt(path=_dp)) == 1)
    # edge-id/type 키(debt-id 없을 때)
    _dp2 = os.path.join(_dtd, "d2.jsonl")
    mc.record_debt({"type": "t", "edge-id": "x"}, path=_dp2)
    mc.record_debt({"type": "t", "edge-id": "x", "status": "resolved"}, path=_dp2)
    check("debt-id 없으면 type:edge-id 키로 fold", mc.unresolved_debt(path=_dp2) == [])
    # 크래시 안전: 쓸 수 없는 경로 → False(예외 없음). (실 workspace 오염 방지 — 임시경로만 사용)
    check("record_debt 쓰기불가 경로 → False(크래시 없음)",
          mc.record_debt({"debt-id": "z"}, path="/nonexistent-dir-xyz/\x00/debt.jsonl") is False)
check("debt_ledger_path 는 str 또는 None", isinstance(mc.debt_ledger_path(), (str, type(None))))

print("== T4.4: doctor migration-debt surfacing(0개→OK) ==")
import subprocess as _sp2  # noqa: E402
_d2 = _sp2.run([sys.executable, os.path.join(ROOT, ".claude/hooks/doctor.py")],
               capture_output=True, text=True,
               env={**os.environ, "CLAUDE_PROJECT_DIR": ROOT, "ORGOS_WORKSPACE": "_sandbox"})
check("doctor: migration-debt 0 → 계약 machinery OK",
      "migration-debt 0" in _d2.stdout and _d2.returncode == 0)

print(f"\n{passed} passed · {failed} failed")
sys.exit(1 if failed else 0)