261 lines
11 KiB
Python
261 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""P1-D 강제 테스트 (#10 audit Write + tools 정본, #9 primary-artifacts 분리).
|
|
|
|
test_enforcement.py를 건드리지 않는 독립 테스트(공유 규칙). pytest 불필요 — exit 0 = 전부 통과.
|
|
|
|
검증:
|
|
#10 - 모든 생성 agent의 tools 프론트매터가 tool-permission-matrix.yaml(정본)에서 파생된다.
|
|
- audit-capable family(및 워커)는 Write를 갖는다(Bash redirection으로 불변 guard 우회 방지).
|
|
- GTM/OPS(FAM-GTM-GROWTH/SALES, FAM-OPS-DELIVERY)는 불필요한 Edit/Bash를 잃는다.
|
|
- primary-artifacts envelope 계약이 agent 본문에 embed된다.
|
|
#9 - design/spec/build/completion 유형 report는 primary-artifacts[] 없으면 차단.
|
|
- primary-artifacts.path가 실존하지 않으면 차단.
|
|
- 실존 path면 통과(다른 검사 통과 전제). sha가 receipt와 불일치하면 차단.
|
|
- report-type 없는 report는 primary-artifacts 없이도 통과(하위호환).
|
|
P0 회귀 없음 - receipt 없는 자기신고 E5는 여전히 차단(C6 유지).
|
|
"""
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
|
|
import yaml
|
|
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
HOOKS = os.path.join(ROOT, ".claude", "hooks")
|
|
AGENTS = os.path.join(ROOT, ".claude", "agents")
|
|
REG = os.path.join(ROOT, "org-os", "00-role-registry")
|
|
sys.path.insert(0, HOOKS)
|
|
import validate_report as vr # noqa: E402
|
|
|
|
passed, failed = 0, 0
|
|
|
|
|
|
def check(name, ok):
|
|
global passed, failed
|
|
if ok:
|
|
passed += 1
|
|
print(f" PASS {name}")
|
|
else:
|
|
failed += 1
|
|
print(f" FAIL {name}")
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 공용: 매트릭스 정본 파싱 + agent 프론트매터
|
|
# --------------------------------------------------------------------------- #
|
|
_MX = yaml.safe_load(open(os.path.join(REG, "tool-permission-matrix.yaml")))[
|
|
"tool-permission-matrix"]["agent-tools"]
|
|
_PROFILES = _MX["profiles"]
|
|
_FAM_PROFILES = _MX["family-profiles"]
|
|
_DEFAULT_PROFILE = _MX["default-profile"]
|
|
_FAMS = yaml.safe_load(open(os.path.join(REG, "capability-families.yaml")))[
|
|
"capability-families"]["families"]
|
|
_BYID = {f["family-id"]: f for f in _FAMS}
|
|
|
|
|
|
def expected_tools(fid):
|
|
prof = _FAM_PROFILES.get(fid, _DEFAULT_PROFILE)
|
|
return ", ".join(_PROFILES[prof])
|
|
|
|
|
|
def frontmatter(path):
|
|
txt = open(path).read()
|
|
return yaml.safe_load(txt.split("---\n")[1])
|
|
|
|
|
|
def family_of(name, fm):
|
|
# 모든 concrete worker/lead는 family 필드를 보유한다. family agent card는 없다.
|
|
return fm.get("family") or name.upper()
|
|
|
|
|
|
print("== #10 tools derive from tool-permission-matrix (single source) ==")
|
|
_agent_files = [f for f in os.listdir(AGENTS) if f.endswith(".md")]
|
|
check("75 concrete agent files present", len(_agent_files) == 75)
|
|
_mismatch = []
|
|
for f in sorted(_agent_files):
|
|
name = f[:-3]
|
|
fm = frontmatter(os.path.join(AGENTS, f))
|
|
fid = family_of(name, fm)
|
|
exp = expected_tools(fid)
|
|
if str(fm.get("tools", "")).strip() != exp:
|
|
_mismatch.append((name, fid, fm.get("tools"), exp))
|
|
check("every agent's tools == matrix-derived profile for its family", not _mismatch)
|
|
if _mismatch:
|
|
for m in _mismatch[:8]:
|
|
print(f" mismatch {m}")
|
|
|
|
print("== #10 audit-capable families gained Write (no Edit; immutable-report safe) ==")
|
|
_audit_fams = [f["family-id"] for f in _FAMS if f.get("audit-capable")]
|
|
for fid in _audit_fams:
|
|
toks = expected_tools(fid)
|
|
check(f"{fid} derived tools include Write", "Write" in toks)
|
|
# 대표 audit agent 파일에도 실제 Write가 박혀 있어야 함(생성물 확인)
|
|
for a in ("sec-engineer", "qa", "gtm-legal", "exec-vpeng", "arch-swat", "sec-appsec", "arch-app"):
|
|
p = os.path.join(AGENTS, a + ".md")
|
|
toks = str(frontmatter(p).get("tools", ""))
|
|
check(f"audit agent {a} has Write", "Write" in toks)
|
|
check(f"audit agent {a} has NO Edit (report immutable)", "Edit" not in toks)
|
|
|
|
print("== #10 GTM/OPS lost unneeded Edit/Bash (moved ENG -> ADVISORY) ==")
|
|
for fid in ("FAM-GTM-GROWTH", "FAM-GTM-SALES", "FAM-OPS-DELIVERY"):
|
|
toks = expected_tools(fid)
|
|
check(f"{fid} has NO Edit", "Edit" not in toks)
|
|
check(f"{fid} has NO Bash", "Bash" not in toks)
|
|
check(f"{fid} keeps Write", "Write" in toks)
|
|
# ENG 계열은 여전히 Edit+Bash
|
|
for fid in ("FAM-ENG-BACKEND", "FAM-DATA", "FAM-DESIGN", "FAM-PLATFORM-INFRA"):
|
|
toks = expected_tools(fid)
|
|
check(f"{fid} keeps Edit+Bash (code/pipeline)", "Edit" in toks and "Bash" in toks)
|
|
|
|
print("== #9 primary-artifacts envelope contract embedded in agent bodies ==")
|
|
for a in ("arch-app", "eng-be", "sec-engineer", "prod-pm", "des-prod"):
|
|
body = open(os.path.join(AGENTS, a + ".md")).read()
|
|
check(f"{a} body carries primary-artifacts envelope contract",
|
|
"primary-artifacts" in body and "envelope" in body)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# #9 validate_report: primary-artifacts 강제
|
|
# --------------------------------------------------------------------------- #
|
|
print("== #9 validate_report requires primary-artifacts for artifact report-types ==")
|
|
TMP = tempfile.mkdtemp(prefix="p1d-")
|
|
try:
|
|
# 실존 아티팩트 파일(절대경로)
|
|
art = os.path.join(TMP, "service.py")
|
|
with open(art, "w") as fh:
|
|
fh.write("def handler():\n return 1\n")
|
|
|
|
def hdr(evidence_uri):
|
|
return {
|
|
"bottom-line": "구현 완료.",
|
|
"decision-needed": {"needed": False},
|
|
"confidence": {"value": "Med"},
|
|
"risks": [],
|
|
"evidence": [{"source-uri": evidence_uri, "grade": "E3"}],
|
|
}
|
|
|
|
# P0-3/P0-5: 유효 report 는 report-id/workflow-id/role-id 필수. 통과해야 하는 fixture 에 부여.
|
|
# P3-B cutover: 이 파일은 primary-artifact/receipt 바인딩만 검증(계약 강제 대상 아님). 일부 체크는
|
|
# report_path(워크스페이스 경로)를 넘기므로 TST-* 는 production 가드에 막힌다 → 실역할(EXEC-CTO)을
|
|
# 유지하되 tier: light 로 두어 method-execution 강제를 면제(paperwork∝risk). 계약 강제는
|
|
# test_p3b_* 가 standard-tier 로 검증. **IDN 이 report 최상위로 펼쳐져 tier 가 실린다.
|
|
IDN = {"report-id": "r-art", "workflow-id": "wfB", "role-id": "EXEC-CTO", "tier": "light"}
|
|
|
|
# (1) completion + primary-artifacts(실존) + verification-performed -> 통과
|
|
good_completion = {
|
|
"report-type": "completion", **IDN,
|
|
"verification-performed": "pytest -q (12 passed)",
|
|
"primary-artifacts": [
|
|
{"path": art, "kind": "code", "verification": "pytest -q"}],
|
|
"report-header": hdr(art),
|
|
}
|
|
check("completion WITH existing primary-artifact -> pass (no violation)",
|
|
vr.validate(good_completion) == [])
|
|
|
|
# (2) completion 인데 primary-artifacts 없음 -> 차단
|
|
no_pa = {
|
|
"report-type": "completion",
|
|
"verification-performed": "pytest",
|
|
"report-header": hdr(art),
|
|
}
|
|
_v = vr.validate(no_pa)
|
|
check("completion WITHOUT primary-artifacts -> blocked",
|
|
any("primary-artifacts" in e for e in _v))
|
|
|
|
# (3) design 인데 primary-artifacts.path 실존하지 않음 -> 차단
|
|
bad_path = {
|
|
"report-type": "design",
|
|
"decisions": ["ADR-1: 이벤트 소싱 채택"],
|
|
"primary-artifacts": [
|
|
{"path": os.path.join(TMP, "does-not-exist.md"),
|
|
"kind": "adr", "verification": "review"}],
|
|
"report-header": hdr(art),
|
|
}
|
|
_v = vr.validate(bad_path)
|
|
check("design WITH non-existent artifact path -> blocked",
|
|
any("실존하지 않음" in e for e in _v))
|
|
|
|
# (4) build 인데 primary-artifacts 비어있음 -> 차단
|
|
build_empty = {
|
|
"report-type": "build",
|
|
"verification-performed": "build ok",
|
|
"primary-artifacts": [],
|
|
"report-header": hdr(art),
|
|
}
|
|
_v = vr.validate(build_empty)
|
|
check("build WITH empty primary-artifacts[] -> blocked",
|
|
any("primary-artifacts" in e for e in _v))
|
|
|
|
# (5) P0-5: report-type 은 이제 필수 — 없으면 차단(예전의 back-compat 통과는 폐기).
|
|
plain = {"report-header": hdr(art)}
|
|
_v = vr.validate(plain)
|
|
check("report with NO report-type -> BLOCKED (P0-5 report-type 필수)",
|
|
any("report-type" in e for e in _v))
|
|
|
|
# (6) sha 교차검증: ledger receipt와 불일치하는 sha -> 차단
|
|
# ledger를 completion-records 형제 evidence/ledger.jsonl에 시드한다.
|
|
ws = os.path.join(TMP, "ws")
|
|
recdir = os.path.join(ws, "completion-records", "wfB")
|
|
os.makedirs(recdir, exist_ok=True)
|
|
os.makedirs(os.path.join(ws, "evidence"), exist_ok=True)
|
|
art2 = os.path.join(ws, "model.sql")
|
|
with open(art2, "w") as fh:
|
|
fh.write("create table t(id int);\n")
|
|
import hashlib as _hashlib
|
|
art2_sha = _hashlib.sha256(open(art2, "rb").read()).hexdigest()
|
|
import json as _json
|
|
with open(os.path.join(ws, "evidence", "ledger.jsonl"), "w") as fh:
|
|
fh.write(_json.dumps({
|
|
"tool_use_id": "w1", "tool_name": "Write", "ts": "2026-07-10T00:00:00Z",
|
|
"cwd": ws, "artifact_path": art2, "artifact_sha256": art2_sha,
|
|
"workflow_id": "wfB", "session_id": "fixture-session",
|
|
"agent_id": "fixture-agent"}) + "\n")
|
|
rep_path = os.path.join(recdir, "role-x-20260101T000000Z.report.yaml")
|
|
sha_bad = {
|
|
"report-type": "build", **IDN,
|
|
"verification-performed": "ok",
|
|
"primary-artifacts": [
|
|
{"path": art2, "kind": "data-model", "sha": "b" * 64, "verification": "psql"}],
|
|
"report-header": hdr(art2),
|
|
}
|
|
with open(rep_path, "w") as fh:
|
|
yaml.safe_dump(sha_bad, fh)
|
|
_v = vr.validate(sha_bad, report_path=rep_path)
|
|
check("build primary-artifact sha != ledger receipt -> blocked",
|
|
any("receipt와 불일치" in e for e in _v))
|
|
|
|
# (7) 같은 것을 올바른 sha로 선언하면 sha 검사 통과(다른 검사도 통과)
|
|
sha_ok = dict(sha_bad)
|
|
sha_ok["primary-artifacts"] = [
|
|
{"path": art2, "kind": "data-model", "sha": art2_sha, "verification": "psql"}]
|
|
with open(rep_path.replace("role-x", "role-y"), "w") as fh:
|
|
yaml.safe_dump(sha_ok, fh)
|
|
_v = vr.validate(sha_ok, report_path=rep_path.replace("role-x", "role-y"))
|
|
check("build primary-artifact sha == ledger receipt -> pass",
|
|
_v == [])
|
|
|
|
# (8) P0 회귀 없음: receipt 없는 자기신고 E5는 여전히 차단(C6 유지)
|
|
e5_dir = os.path.join(TMP, "e5ws", "completion-records", "wfE")
|
|
os.makedirs(e5_dir, exist_ok=True)
|
|
e5_report = {
|
|
"report-header": {
|
|
"bottom-line": "테스트 통과, 배포 가능.",
|
|
"decision-needed": {"needed": False},
|
|
"confidence": {"value": "High"},
|
|
"risks": [],
|
|
"evidence": [{"command": "pytest -q", "exit-code": 0, "grade": "E5"}],
|
|
}
|
|
}
|
|
e5_path = os.path.join(e5_dir, "role-z-20260101T000000Z.report.yaml")
|
|
with open(e5_path, "w") as fh:
|
|
yaml.safe_dump(e5_report, fh)
|
|
_v = vr.validate(e5_report, report_path=e5_path)
|
|
check("P0 intact: self-reported E5 (no ledger receipt) -> still blocked",
|
|
any("receipt" in e for e in _v))
|
|
finally:
|
|
shutil.rmtree(TMP, ignore_errors=True)
|
|
|
|
print(f"\n{passed} passed, {failed} failed")
|
|
sys.exit(1 if failed else 0)
|