267 lines
14 KiB
Python
267 lines
14 KiB
Python
#!/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)
|