Files

94 lines
4.6 KiB
Python

#!/usr/bin/env python3
"""Executable role planner, family metadata, policy binding and projection tests."""
import glob
import importlib.util
import os
import subprocess
import sys
import yaml
ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
HOOKS = os.path.join(ROOT, ".claude", "hooks")
sys.path.insert(0, HOOKS)
from orgos.planning.intake_classifier import classify_request # noqa: E402
from orgos.planning.role_selector import resolve_family, select_minimum_sufficient_roles # noqa: E402
passed = failed = 0
def check(name, ok, detail=""):
global passed, failed
if ok:
passed += 1
print(f" PASS {name}")
else:
failed += 1
print(f" FAIL {name}{': ' + detail if detail else ''}")
print("== family metadata is not an execution identity ==")
cards = glob.glob(os.path.join(ROOT, ".claude", "agents", "*.md"))
check("75 role cards", len(cards) == 75, str(len(cards)))
check("zero fam-* cards", glob.glob(os.path.join(ROOT, ".claude", "agents", "fam-*.md")) == [])
compiled = subprocess.run([sys.executable, os.path.join(HOOKS, "compile_orgos_registry.py"), "--check"],
cwd=ROOT, capture_output=True, text=True)
check("Pack registries have no drift", compiled.returncode == 0, compiled.stderr)
print("== minimum sufficient selection ==")
architecture = resolve_family("FAM-ARCHITECTURE-TECH", ["api", "service-boundary"], "standard")
check("fan-out family does not return all seven members", len(architecture["resolved-workers"]) == 1,
str(architecture["resolved-workers"]))
check("resolved identity is concrete", not architecture["primary-worker"].startswith("FAM-"))
plan = select_minimum_sufficient_roles({
"workflow-id": "wf-role-plan",
"tier": "standard",
"workflow-stage": "design",
"candidate-families": ["FAM-ARCHITECTURE-TECH"],
"required-artifacts": ["api-contract"],
"risks": ["security"],
"independent-review-required": True,
"signals": ["api", "application", "appsec"],
})["selection-plan"]
selected = [plan["selected"]["owner"]] + plan["selected"]["contributors"] + plan["selected"]["reviewers"]
check("required coverage is complete", plan["status"] == "ready" and plan["coverage"]["missing"] == [],
str(plan["coverage"]))
check("api producer selected", any("artifact:api-contract" in item["coverage"] for item in selected))
check("security risk selected", any("risk:security" in item["coverage"] for item in selected))
producer_ids = {item["role-id"] for item in selected if "artifact:api-contract" in item["coverage"]}
reviewer_ids = {item["role-id"] for item in plan["selected"]["reviewers"]}
check("producer and reviewer identities are independent", bool(reviewer_ids) and producer_ids.isdisjoint(reviewer_ids))
check("skipped roles explain why", bool(plan["skipped"]) and all(item.get("reason") for item in plan["skipped"]))
blocked = select_minimum_sufficient_roles({
"tier": "standard", "candidate-families": ["FAM-ARCHITECTURE-TECH"],
"required-artifacts": ["api-contract"], "token-budget": 1,
})["selection-plan"]
check("insufficient token budget blocks selection", blocked["status"] == "blocked")
print("== deterministic intake avoids blanket CEO ==")
check("typo routes light without executive", not classify_request("오탈자 하나 수정")["executive-required"])
check("portfolio pricing routes strategic executive", classify_request("포트폴리오 가격 전략 결정")["executive-required"])
print("== task allowlist binds active subagent ==")
import guard_tools as GT # noqa: E402
orig_record, orig_pkg = GT._registry_record, GT._load_bound_package
try:
GT._registry_record = lambda _agent: {"report_producing": True, "agent_type": "eng-be"}
GT._load_bound_package = lambda _record: ({
"target-role-agent": "eng-be", "allowed-tools": ["Read"], "allowed-paths": [ROOT],
}, None)
check("allowed task tool passes", GT.check("Read", {"file_path": "README.md"}, {"agent_id": "a"}) == (None, None))
check("tool absent from task allowlist blocks",
GT.check("Write", {"file_path": "README.md"}, {"agent_id": "a"})[0] == "task-tool-allowlist")
finally:
GT._registry_record, GT._load_bound_package = orig_record, orig_pkg
print("== new reports opt into structured projection ==")
new_report_text = open(os.path.join(HOOKS, "new_report.py"), encoding="utf-8").read()
check("new report scaffold has projection v1", "projection-version: 1" in new_report_text)
check("new report scaffold has all projection fields",
all(name in new_report_text for name in ("decision-summary", "evidence-index", "dissent", "open-risks", "artifact-refs")))
print(f"\n{passed} passed, {failed} failed")
sys.exit(1 if failed else 0)