278 lines
15 KiB
Python
278 lines
15 KiB
Python
#!/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)
|