#!/usr/bin/env python3
"""P3-B family 정합성 검증기(standalone check) — Phase 6 wave 재사용.

계약 lint(모든 v2 역할): uses-capability section 해소·handoff.to 실존·artifact-type 통제어휘·
machine-gate 어휘 유효. + DESIGN family 정합: 역할경계 not-owns 실존·handoff↔required-input
양방향 일치·닫힌 방향 루프·대표 task 경로 walkable.
"""
import importlib.util
import os
import sys

import yaml

ROOT = os.environ.get("CLAUDE_PROJECT_DIR", os.getcwd())
HOOKS = os.path.join(ROOT, ".claude", "hooks")
REG = os.path.join(ROOT, "org-os", "00-role-registry")
passed = failed = 0


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


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")
se = _load("state_engine_family_invariant", "state_engine.py")
cp = _load("context_package_family_invariant", "context_package.py")
rm = mc.load_role_methods()
VOCAB = set(yaml.safe_load(open(os.path.join(ROOT, "org-os/06-agent-work/artifact-type-vocabulary.yaml")))["artifact-types"])
_idx = yaml.safe_load(open(os.path.join(REG, "role-working-methods", "index.yaml")))
MACHINE_VOCAB = set((_idx.get("contract-v2-schema") or {}).get("machine-check-vocabulary") or [])
ALL_ROLES = set(rm)


def profile_index():
    idx = {}
    for rid, e in rm.items():
        if (e.get("method-contract") or {}).get("version") == 2:
            for m in e.get("methods", []):
                idx[(rid, m.get("method-id"))] = m
    return idx


PROFILES = profile_index()


print("== family resolver → executable concrete card 정합 ==")
_families = yaml.safe_load(open(os.path.join(REG, "capability-families.yaml")))[
    "capability-families"]["families"]
_unexecutable = []
for _family in _families:
    _resolutions = [se.resolve_family(_family["family-id"])]
    for _route in _family.get("collapse-routes", []) or []:
        _signals = list(_route.get("when-any", []) or [])
        if _signals:
            _resolutions.append(se.resolve_family(_family["family-id"], [_signals[0]]))
    for _resolved in _resolutions:
        for _worker in (_resolved or {}).get("resolved-workers", []):
            _card_name = str(_worker).lower()
            _card = os.path.join(ROOT, ".claude", "agents", _card_name + ".md")
            _semantic = cp._semantic_errors({
                "workflow-id": "family-invariant",
                "task-id": "resolve-family-invariant",
                "target-role-agent": _card_name,
                "tier": "standard",
                "allowed-tools": ["Read"],
                "token-budget": {"max-input-tokens": 1, "max-output-tokens": 1},
            })
            if not os.path.isfile(_card) or any(
                    "에이전트 카드" in error or "family는 resolver" in error
                    for error in _semantic):
                _unexecutable.append(
                    f"{_family['family-id']}->{_worker}: card={os.path.isfile(_card)} errors={_semantic}")
check("모든 resolve-family 결과가 context-package 실행 가능한 concrete card",
      _unexecutable == [], detail=str(_unexecutable[:3]))


def lint_contract(rid, entry):
    """계약 하나의 스키마/참조 lint → 문제 리스트."""
    probs = []
    rb = entry.get("role-boundary") or {}
    for token in rb.get("not-owns", []):  # not-owns 가 '(-> ROLE-ID)' 를 언급하면 실존 역할이어야
        for r in ALL_ROLES:
            if f"-> {r}" in token or f"->{r}" in token:
                break
    for m in entry.get("methods", []):
        mid = m.get("method-id")
        tag = f"{rid}/{mid}"
        for i in m.get("required-inputs", []):
            if i.get("artifact-type") not in VOCAB:
                probs.append(f"{tag}: required-input artifact-type '{i.get('artifact-type')}' 통제어휘 밖")
            # from-role 가 v2 역할이면 from-method 필수(그래야 producer profile 식별 → 엣지 hard 판정 가능)
            fr = i.get("from-role")
            if fr and (rm.get(fr, {}).get("method-contract") or {}).get("version") == 2:
                fm = i.get("from-method")
                if not fm:
                    probs.append(f"{tag}: required-input '{i.get('artifact-type')}' 의 from-role {fr} 는 v2 인데 "
                                 f"from-method 미지정(엣지가 soft 로 강등됨)")
                elif (fr, fm) not in PROFILES:
                    probs.append(f"{tag}: required-input from {fr}/{fm} 미존재 profile")
        for a in m.get("output-artifacts", []):
            if a not in VOCAB:
                probs.append(f"{tag}: output-artifact '{a}' 통제어휘 밖")
        for s in m.get("workflow", []):
            uc = s.get("uses-capability")
            if uc and mc.resolve_capability_section(uc.get("skill-id"), uc.get("section-id")) is None:
                probs.append(f"{tag}/{s.get('step-id')}: uses-capability {uc} 미해소")
            for g in (s.get("completion-gates") or {}).get("machine", []):
                if g.get("check") not in MACHINE_VOCAB:
                    probs.append(f"{tag}/{s.get('step-id')}: machine check '{g.get('check')}' 어휘 밖")
        for h in m.get("handoff-contract", []):
            to = h.get("to") or {}
            if (to.get("role-id"), to.get("method-id")) not in PROFILES:
                probs.append(f"{tag}: handoff '{h.get('edge-id')}' to {to} 미존재 profile")
            if h.get("artifact-type") not in VOCAB:
                probs.append(f"{tag}: handoff artifact-type '{h.get('artifact-type')}' 통제어휘 밖")
    return probs


print("== 계약 lint (모든 v2 역할) ==")
_v2 = [r for r, e in rm.items() if (e.get("method-contract") or {}).get("version") == 2]
_all_probs = []
for rid in _v2:
    _all_probs += lint_contract(rid, rm[rid])
check(f"v2 계약 {len(_v2)}역할({len(PROFILES)} profile) 스키마/참조 lint 통과", _all_probs == [], detail=str(_all_probs[:3]))

# 전역 양방향 정합: 모든 v2 producer handoff 엣지의 to 가 v2 면 consumer required-input 이 일치
_g_mismatch = []
for (rid, mid), m in PROFILES.items():
    for h in m.get("handoff-contract", []):
        to = h.get("to") or {}
        cons = PROFILES.get((to.get("role-id"), to.get("method-id")))
        if cons is None:
            continue  # lint 이 별도 처리
        match = [i for i in cons.get("required-inputs", [])
                 if i.get("artifact-type") == h.get("artifact-type")
                 and i.get("from-role") == rid and i.get("from-method") == mid]
        if not match:
            _g_mismatch.append(f"{rid}/{mid} --{h.get('artifact-type')}--> {to.get('role-id')}/{to.get('method-id')}")
check("전 v2 handoff 엣지가 consumer required-input 과 양방향 일치(from-method 포함)",
      _g_mismatch == [], detail=str(_g_mismatch))

print("== DESIGN family 정합 ==")
DESIGN = ["DES-DIRECTOR", "DES-PROD", "DES-PLATFORM", "DES-VISUAL", "DES-INTERNAL"]
check("DESIGN 5역할 전부 v2 계약", all((rm[r].get("method-contract") or {}).get("version") == 2 for r in DESIGN))

# handoff ↔ required-input 양방향 정합(family 내부 엣지)
_mismatch = []
for rid in DESIGN:
    for m in rm[rid].get("methods", []):
        for h in m.get("handoff-contract", []):
            to = h.get("to") or {}
            if to.get("role-id") not in DESIGN:
                continue  # family 외부 handoff 는 제외
            consumer = PROFILES.get((to.get("role-id"), to.get("method-id"))) or {}
            match = [i for i in consumer.get("required-inputs", [])
                     if i.get("artifact-type") == h.get("artifact-type") and i.get("from-role") == rid]
            if not match:
                _mismatch.append(f"{rid}/{m['method-id']} --{h.get('artifact-type')}--> "
                                 f"{to.get('role-id')}/{to.get('method-id')}: consumer required-input 미선언")
check("family 내부 handoff 는 consumer required-input 과 양방향 일치", _mismatch == [], detail=str(_mismatch))

# 방향 결정 DAG: charter 가 생성과 비교에 각각 전달되고, 둘 다 수렴 전에 합류해야 한다.
def has_edge(frm, fmid, art, to, tmid):
    prof = PROFILES.get((frm, fmid)) or {}
    return any((h.get("to") or {}).get("role-id") == to and (h.get("to") or {}).get("method-id") == tmid
               and h.get("artifact-type") == art for h in prof.get("handoff-contract", []))


_direction_dag = [
    ("DES-PROD", "pre-direction", "direction-input-brief", "DES-DIRECTOR", "frame-divergence"),
    ("DES-DIRECTOR", "frame-divergence", "divergence-charter", "DES-VISUAL", "art-direction"),
    ("DES-DIRECTOR", "frame-divergence", "divergence-charter", "DES-VISUAL", "compare-directions"),
    ("DES-VISUAL", "art-direction", "reference-cluster", "DES-DIRECTOR", "converge-directions"),
    ("DES-VISUAL", "compare-directions", "comparative-divergence-audit", "DES-DIRECTOR", "converge-directions"),
    ("DES-DIRECTOR", "converge-directions", "selected-direction", "DES-PROD", "post-direction"),
    ("DES-PROD", "post-direction", "design-decision-record", "DES-PLATFORM", "tokenize"),
]
_broken = [e for e in _direction_dag if not has_edge(*e)]
check("방향 결정 DAG 7-edge walkable(charter→생성·비교→수렴)", _broken == [], detail=str(_broken))

# 역할 경계: DESIGN 5역할의 owns 가 서로 겹치지 않음(대표 키워드)
_owns_terms = {"방향": [], "화면": [], "토큰": [], "아트": [], "사내": []}
_boundary_ok = True
# DIRECTOR=방향, PROD=화면, PLATFORM=토큰, VISUAL=아트, INTERNAL=사내 — 각자 고유 owns 존재
_expect = {"DES-DIRECTOR": "방향", "DES-PROD": "화면", "DES-PLATFORM": "토큰",
           "DES-VISUAL": "아트", "DES-INTERNAL": "사내"}
for r, kw in _expect.items():
    owns = " ".join(rm[r].get("role-boundary", {}).get("owns", []))
    if kw not in owns:
        _boundary_ok = False
check("역할 경계 고유성(각 역할이 자기 도메인 owns 보유)", _boundary_ok)

print("== EXEC family 결정 DAG 정합 ==")
EXEC = ["EXEC-CEO", "EXEC-CTO", "EXEC-CPO", "EXEC-CFO", "EXEC-COO", "EXEC-CPTO", "EXEC-VPENG", "STR-ANALYST", "OPS-ORCH"]
check("EXEC 9역할 전부 v2 계약", all((rm[r].get("method-contract") or {}).get("version") == 2 for r in EXEC))
# 결정 DAG: source(STR)→CEO, CFO/COO→CEO, CTO/CPO→CPTO, CPTO→CEO. EXEC-CEO=sink(수렴)
_dag = [
    ("STR-ANALYST", "strategy-analysis", "grounding-evidence", "EXEC-CEO", "decide-direction"),
    ("STR-ANALYST", "strategy-analysis", "option-set", "EXEC-CEO", "decide-direction"),
    ("EXEC-CFO", "financial-judgment", "financial-assessment", "EXEC-CEO", "decide-direction"),
    ("EXEC-COO", "ops-judgment", "ops-assessment", "EXEC-CEO", "decide-direction"),
    ("EXEC-CTO", "tech-judgment", "tech-assessment", "EXEC-CPTO", "integration-judgment"),
    ("EXEC-CPO", "product-judgment", "product-assessment", "EXEC-CPTO", "integration-judgment"),
    ("EXEC-CPTO", "integration-judgment", "integration-decision", "EXEC-CEO", "decide-direction"),
]
_dag_broken = [e for e in _dag if not has_edge(*e)]
check("EXEC 결정 DAG 7-edge walkable(STR→CEO·CFO/COO→CEO·CTO/CPO→CPTO→CEO)", _dag_broken == [], detail=str(_dag_broken))
# EXEC-CEO 는 sink: decide-direction 이 5개 상류 required-input 을 모두 gate
_ceo = PROFILES.get(("EXEC-CEO", "decide-direction")) or {}
_ceo_inputs = {i.get("artifact-type") for i in _ceo.get("required-inputs", [])}
check("EXEC-CEO(sink)가 grounding+option+3평가를 required-input 으로 수렴",
      {"grounding-evidence", "option-set", "financial-assessment", "ops-assessment", "integration-decision"} <= _ceo_inputs)
check("EXEC 결정권자(CEO/CPTO/CFO...)는 alternatives-policy min≥2(옵션 발산 강제)",
      all((PROFILES.get((r, m)) or {}).get("alternatives-policy", {}).get("min-alternatives", 0) >= 2
          for r, m in [("EXEC-CEO", "decide-direction"), ("EXEC-CFO", "financial-judgment"),
                       ("STR-ANALYST", "strategy-analysis"), ("EXEC-CPTO", "integration-judgment")]))

print("== PRODUCT family discovery→PRD 체인 정합 ==")
PRODUCT = ["PROD-PM", "PROD-PO", "PROD-TPO", "PROD-PPO", "UX-RESEARCHER", "DATA-ANALYST"]
check("PRODUCT 6역할 전부 v2 계약", all((rm[r].get("method-contract") or {}).get("version") == 2 for r in PRODUCT))
_pchain = [
    ("UX-RESEARCHER", "user-research", "user-research", "PROD-PM", "product-discovery"),
    ("DATA-ANALYST", "metrics-analysis", "metrics-analysis", "PROD-PM", "product-discovery"),
    ("PROD-PM", "product-discovery", "prd", "PROD-PO", "backlog-definition"),
]
check("PRODUCT discovery→PRD→backlog 체인 walkable", [e for e in _pchain if not has_edge(*e)] == [])
# 크로스패밀리: PROD-PM 이 EXEC-CEO/decide-direction 의 product-decision 을 required-input 으로 소비
_pm = PROFILES.get(("PROD-PM", "product-discovery")) or {}
check("PROD-PM 이 product-decision(EXEC-CEO/decide-direction) 크로스패밀리 입력 선언",
      any(i.get("artifact-type") == "product-decision" and i.get("from-role") == "EXEC-CEO"
          and i.get("from-method") == "decide-direction" for i in _pm.get("required-inputs", [])))

print("== 활성화 현황 + drift 무결성(active hash == 현재 계약 hash) ==")
acts = mc.load_activations()
_active = {(r, m) for r, rec in acts.items() for m, d in (rec.get("methods") or {}).items() if d.get("status") == "active"}
# 변경하지 않은 안정 계약은 active. v3로 바뀐 발산/수렴/아트 계약은 기존 승인을
# 재사용하지 않고 draft로 무효화한다(golden 재검증 + HUMAN-001 재승인 필요).
check("DESIGN 미변경 안정 계약(PROD/pre·post·PLATFORM/tokenize·INTERNAL) active",
      {("DES-PROD", "pre-direction"), ("DES-PROD", "post-direction"),
       ("DES-PLATFORM", "tokenize"), ("DES-INTERNAL", "internal-tool-design")} <= _active)
_changed_design = {
    ("DES-DIRECTOR", "frame-divergence"),
    ("DES-DIRECTOR", "converge-directions"),
    ("DES-VISUAL", "art-direction"),
}
_draft = {(r, m) for r, rec in acts.items() for m, d in (rec.get("methods") or {}).items()
          if d.get("status") == "draft"}
check("v3 변경 DESIGN 계약은 재승인 전 draft", _changed_design <= _draft)
check("v3 신규 비교 계약은 자동 active 처리하지 않음",
      ("DES-VISUAL", "compare-directions") not in _active)
# drift 무결성: active 레코드 hash 가 현재 계약 hash 와 일치해야(변경 후 미재활성=drift, doctor 도 FAIL)
_drift = []
for r, rec in acts.items():
    for m, d in (rec.get("methods") or {}).items():
        if d.get("status") == "active":
            cur = mc.canonical_contract_hash(mc.resolve_method_profile(r, m))
            if cur != d.get("contract-sha256"):
                _drift.append(f"{r}/{m}")
check("active 계약 drift 0(변경 후 재활성 완료)", _drift == [], detail=str(_drift))

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