Files

258 lines
14 KiB
Python

#!/usr/bin/env python3
"""P3-B 계약 policy engine 강제기 — standalone check(pytest 아님). exit 0=통과.
method_contracts.py(공용 policy engine, Contract v2 해석 단일 지점)를 검증한다:
- load_role_methods 병합(파일분리 SoT)
- resolve_method_profile(v1 역할 → None, v2 profile 조회)
- canonical_contract_hash 결정성·민감도
- load_activations / resolve_activation(기본 draft)
- validate_method_selection(standard/heavy 필수·light 유일후보·미지 method-id)
이후 phase(enforcement·handoff·debt)가 이 파일에 append된다.
"""
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")
print("== T3.1: load_role_methods (파일분리 병합) ==")
_rm = mc.load_role_methods()
check("load_role_methods 병합 75역할", len(_rm) == 75)
_v2roles = [r for r, e in _rm.items() if (e.get("method-contract") or {}).get("version") == 2]
_v1roles = [r for r, e in _rm.items() if (e.get("method-contract") or {}).get("version") != 2]
check("v2 계약 역할은 methods[] 를 가진다(구조 정합)",
all(isinstance(_rm[r].get("methods"), list) and _rm[r]["methods"] for r in _v2roles))
check("v1 역할은 working-method 를 가진다(회귀 없음)",
all(_rm[r].get("working-method") for r in _v1roles))
print("== T3.1: resolve_method_profile ==")
# v1(또는 미존재) 역할 → None(계약 강제 대상 아님). 전 역할 v2면 미존재 role-id 로 동일 의미 검증.
_v1sample = _v1roles[0] if _v1roles else "NONEXISTENT-ROLE-V1"
check("v1/미존재 역할 profile 조회 → None(강제 미대상)",
mc.resolve_method_profile(_v1sample, "any-method") is None)
check("v2 역할 profile 조회 → dict(실존 method)",
_v2roles == [] or isinstance(mc.resolve_method_profile(_v2roles[0], _rm[_v2roles[0]]["methods"][0]["method-id"]), dict))
print("== T3.1: canonical_contract_hash 결정성·민감도 ==")
_c1 = {"method-id": "m1", "workflow": [{"step-id": "s1"}, {"step-id": "s2"}]}
_c1b = {"workflow": [{"step-id": "s1"}, {"step-id": "s2"}], "method-id": "m1"} # 키 순서만 다름
_c2 = {"method-id": "m1", "workflow": [{"step-id": "s1"}]}
check("같은 내용(키 순서 무관) → 같은 hash",
mc.canonical_contract_hash(_c1) == mc.canonical_contract_hash(_c1b))
check("내용 다르면 → 다른 hash",
mc.canonical_contract_hash(_c1) != mc.canonical_contract_hash(_c2))
check("hash 는 64-hex sha256",
len(mc.canonical_contract_hash(_c1)) == 64)
print("== T3.1: activation registry(기본 draft) ==")
_acts = mc.load_activations()
check("load_activations dict 반환(현행 빈 골격 → {} 또는 roles)", isinstance(_acts, dict))
check("resolve_activation 미등록 → status draft",
mc.resolve_activation("NO-SUCH-ROLE", "no-method").get("status") == "draft")
print("== T3.1: validate_method_selection (v2 branch, 합성 계약) ==")
# 합성 v2 역할로 policy 분기 검증(디스크 미오염 — load_role_methods 주입)
_orig = mc.load_role_methods
mc.load_role_methods = lambda: {
"TST-MULTI": {"method-contract": {"version": 2},
"methods": [{"method-id": "mA"}, {"method-id": "mB"}]},
"TST-SINGLE": {"method-contract": {"version": 2},
"methods": [{"method-id": "only"}]},
"TST-V1": {"working-method": ["..."]},
}
# 활성화 주입 — method-selection 은 active method 가 있을 때만 hard(draft 회귀 방지)
_ACTS = {"TST-MULTI": {"methods": {"mA": {"status": "active"}, "mB": {"status": "active"}}},
"TST-SINGLE": {"methods": {"only": {"status": "active"}}}}
def _vms(cp):
return mc.validate_method_selection(cp, activations=_ACTS)
try:
check("standard·selection 없음 → 에러(auto-infer 금지)",
_vms({"role-id": "TST-MULTI", "tier": "standard"}) != [])
check("heavy·selection 없음 → 에러",
_vms({"role-id": "TST-MULTI", "tier": "heavy"}) != [])
check("light·복수 profile·selection 없음 → 에러(선택 필요)",
_vms({"role-id": "TST-MULTI", "tier": "light"}) != [])
check("light·유일 profile·selection 없음 → 통과",
_vms({"role-id": "TST-SINGLE", "tier": "light"}) == [])
check("standard·유효 method-id 선택 → 통과",
_vms({"role-id": "TST-MULTI", "tier": "standard", "method-selection": {"method-id": "mA"}}) == [])
check("standard·미지 method-id → 에러",
_vms({"role-id": "TST-MULTI", "tier": "standard", "method-selection": {"method-id": "ghost"}}) != [])
check("v1 역할 → 미적용(빈 리스트)",
_vms({"role-id": "TST-V1", "tier": "standard"}) == [])
check("draft-only 계약(active 없음) → 미강제(회귀 방지)",
mc.validate_method_selection({"role-id": "TST-MULTI", "tier": "standard"}, activations={}) == [])
finally:
mc.load_role_methods = _orig
print("== T3.2: activate_method_contract verify (4단 게이트) ==")
import hashlib # noqa: E402
import tempfile # noqa: E402
amc = _load("activate_method_contract", "activate_method_contract.py")
_prof = {"method-id": "converge", "workflow": [{"step-id": "frame"}, {"step-id": "synthesize"}]}
_phash = mc.canonical_contract_hash(_prof)
_yes = lambda wf, stage: True # noqa: E731 — signoff 존재 주입
_no = lambda wf, stage: False # noqa: E731
with tempfile.TemporaryDirectory() as _td:
_rep = os.path.join(_td, "golden.report.yaml")
open(_rep, "w").write("bottom-line: golden ok\n")
_rhash = hashlib.sha256(open(_rep, "rb").read()).hexdigest()
# (a) profile 미존재(v1/미정의) → 거부
ok, errs = amc.verify("X-ROLE", "m", _phash, _rep, _rhash, "wf-1", profile=None, has_signoff=_yes)
check("profile 미존재 → 거부", not ok and any("profile" in e for e in errs))
# (b) hash 불일치 → 거부(리뷰된 계약과 다름)
ok, errs = amc.verify("R", "converge", "deadbeef", _rep, _rhash, "wf-1", profile=_prof, has_signoff=_yes)
check("contract-sha256 불일치 → 거부", not ok and any("sha256" in e for e in errs))
# (c) validation-report 부재 → 거부
ok, errs = amc.verify("R", "converge", _phash, os.path.join(_td, "nope.yaml"), _rhash, "wf-1",
profile=_prof, has_signoff=_yes)
check("golden report 부재 → 거부", not ok and any("validation-report" in e for e in errs))
# (d) report sha 불일치 → 거부
ok, errs = amc.verify("R", "converge", _phash, _rep, "0" * 64, "wf-1", profile=_prof, has_signoff=_yes)
check("golden report sha256 불일치 → 거부", not ok)
# (e) HUMAN signoff 없음 → 거부(핵심: OPS-ORCH 단독 활성화 불가)
ok, errs = amc.verify("R", "converge", _phash, _rep, _rhash, "wf-1", profile=_prof, has_signoff=_no)
check("HUMAN signoff 없음 → 거부(사람 게이트)", not ok and any("signoff" in e for e in errs))
# (f) 4단 전부 충족 → 통과
ok, errs = amc.verify("R", "converge", _phash, _rep, _rhash, "wf-1", profile=_prof, has_signoff=_yes)
check("4단 게이트 전부 충족 → 통과", ok and errs == [])
# signoff stage 토큰이 contract hash 에 바인딩(무관 signoff 재사용 차단)
check("signoff stage 가 계약 hash 12자에 바인딩",
amc.signoff_stage("R", "converge", _phash) == f"method-contract:R:converge:{_phash[:12]}")
# apply_activation → 원자 write + previous-status 보존
_regp = os.path.join(_td, "acts.yaml")
amc.apply_activation("R", "converge", {"status": "active", "contract-sha256": _phash}, registry_path=_regp)
_doc = yaml.safe_load(open(_regp))["method-contract-activations"]["roles"]
check("apply_activation 레코드 write(status active)",
_doc["R"]["methods"]["converge"]["status"] == "active"
and _doc["R"]["methods"]["converge"]["previous-status"] == "draft")
print("== T3.2: guard_tools 활성화 레지스트리 차단(3벡터) + 정상 CLI 미차단 ==")
gt = _load("guard_tools", "guard_tools.py")
_ACT = "org-os/00-role-registry/method-contract-activations.yaml"
_c1, _ = gt.check("Write", {"file_path": _ACT})
check("Write 직접 → 차단", _c1 == "activation-registry-boundary")
_c2, _ = gt.check("Bash", {"command": f"echo x > {_ACT}"})
check("Bash redirection → 차단", _c2 == "activation-registry-boundary")
_c3, _ = gt.check("Bash", {"command": f"python3 -c \"import os; os.replace('t','{_ACT}')\""})
check("python -c os.replace → 차단", _c3 == "activation-registry-boundary")
_c4, _ = gt.check("Bash", {"command":
"python3 .claude/hooks/activate_method_contract.py activate --role R --method m "
"--contract-sha256 h --validation-report g --validation-report-sha256 s "
"--acceptance-workflow wf"})
check("정상 CLI 호출 → 미차단(4단 게이트가 방어)", _c4 is None)
print("== T3.3: context_package method-selection spawn 게이트 ==")
cp = _load("context_package", "context_package.py")
# 합성 v2 계약 주입 — context_package 가 실제 참조하는 _MC(별도 모듈 인스턴스)를 패치
_orig2 = cp._MC.load_role_methods
_orig2a = cp._MC.load_activations
cp._MC.load_role_methods = lambda: {
"TST-MULTI": {"method-contract": {"version": 2}, "methods": [{"method-id": "mA"}, {"method-id": "mB"}]},
"TST-V1": {"working-method": ["..."]},
}
cp._MC.load_activations = lambda: {"TST-MULTI": {"methods": {"mA": {"status": "active"}, "mB": {"status": "active"}}}}
try:
def _sel_errs(pkg):
return [e for e in cp.validate(pkg) if "method-selection" in e or "method-id" in e]
check("v2 worker·standard·selection 없음 → spawn 게이트 에러",
_sel_errs({"target-role-agent": "tst-multi", "tier": "standard"}) != [])
check("v2 worker·standard·유효 method-id → 게이트 통과",
_sel_errs({"target-role-agent": "tst-multi", "tier": "standard",
"method-selection": {"method-id": "mA"}}) == [])
check("v1 worker → 게이트 미적용",
_sel_errs({"target-role-agent": "tst-v1", "tier": "standard"}) == [])
check("family metadata(fam-*) → method 게이트 미적용(실행은 semantic gate가 거부)",
_sel_errs({"target-role-agent": "fam-design", "tier": "standard"}) == [])
# draft-only(active 없음) → spawn 게이트 미적용(회귀 방지)
cp._MC.load_activations = lambda: {}
check("draft-only v2 worker → spawn 게이트 미적용(회귀 방지)",
_sel_errs({"target-role-agent": "tst-multi", "tier": "standard"}) == [])
finally:
cp._MC.load_role_methods = _orig2
cp._MC.load_activations = _orig2a
print("== T3.4: capability-sections manifest (design-craft) ==")
_secs = mc.load_capability_sections()
check("manifest 에 design-craft skill 등록", "design-craft" in _secs)
_dc = mc.resolve_capability_section("design-craft", "reference-cluster")
check("reference-cluster 해소 + section-sha256 계산",
_dc is not None and len(_dc["section-sha256"]) == 64 and "References" in _dc["text"])
_all_sec = list((_secs.get("design-craft", {}).get("sections") or {}).keys())
check("모든 선언 section 해소(헤딩 실존)",
all(mc.resolve_capability_section("design-craft", s) is not None for s in _all_sec) and len(_all_sec) == 5)
check("미정의 section → None",
mc.resolve_capability_section("design-craft", "ghost-section") is None)
check("미정의 skill → None",
mc.resolve_capability_section("no-such-skill", "brief") is None)
# section-sha256 안정성: 같은 절 두 번 → 같은 hash
check("section-sha256 결정적(같은 절 → 같은 hash)",
mc.resolve_capability_section("design-craft", "brief")["section-sha256"]
== mc.resolve_capability_section("design-craft", "brief")["section-sha256"])
print("== T3.4b: design-direction judgment floor ==")
_light_design_pkg = {
"workflow-id": "wf-design", "task-id": "direction-a", "mode": "divergent", "tier": "light",
"target-role-agent": "des-visual", "objective": "produce one direction", "output-format": "report",
"allowed-tools": ["Read"], "task-boundaries": "one direction", "must-read": ["README.md"],
"inherited-decisions": [], "expected-output": {"report-header": {"bottom-line": "x"}},
"token-budget": {"max-input-tokens": 1000, "max-output-tokens": 1000},
"workspace": "_sandbox", "target-repo": "_sandbox", "acceptance-tests": ["render"],
"non-goals": ["siblings"], "evidence-plan": ["preview receipt"],
"method-selection": {"method-id": "art-direction"},
}
check("DES-VISUAL direction task tier=light -> rejected(minimum standard)",
any("tier=light 금지" in e for e in cp.validate(_light_design_pkg)))
print("== T3.4: doctor method-contract machinery 섹션 OK ==")
import subprocess as _sp2 # noqa: E402
_d = _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: 계약 machinery OK", "계약 machinery OK" in _d.stdout and _d.returncode == 0)
print(f"\n{passed} passed · {failed} failed")
sys.exit(1 if failed else 0)