127 lines
6.8 KiB
Python
127 lines
6.8 KiB
Python
#!/usr/bin/env python3
|
|
"""P4 cascade benchmark — 실행계열(probe·runner·judge orchestration). standalone check. exit 0=통과.
|
|
실제 claude CLI/model 호출은 mock — 오케스트레이션 로직만 검증(실행은 --execute 게이트)."""
|
|
import importlib.util
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
|
|
ROOT = os.environ.get("CLAUDE_PROJECT_DIR", os.getcwd())
|
|
HOOKS = os.path.join(ROOT, ".claude", "hooks")
|
|
if HOOKS not in sys.path:
|
|
sys.path.insert(0, HOOKS)
|
|
|
|
passed = failed = 0
|
|
|
|
|
|
def check(name, ok):
|
|
global passed, failed
|
|
if ok:
|
|
passed += 1
|
|
print(f" ✅ {name}")
|
|
else:
|
|
failed += 1
|
|
print(f" ❌ {name}")
|
|
|
|
|
|
from bench_cascade import probe # noqa: E402
|
|
|
|
# adapter 결정: command 본문이 순수 slash 지시면 direct-slash, 아니면 adapter prompt 구성
|
|
inv = probe.build_stage_invocation("ground", "# /ground\n사용자 문제를 접지한다.", "/ctrl/brief.md")
|
|
check("probe: stage invocation 은 prompt 에 brief 경로를 실는다", "/ctrl/brief.md" in inv["prompt"])
|
|
check("probe: mode 는 direct-slash 또는 adapter", inv["mode"] in ("direct-slash", "adapter"))
|
|
check("probe: argv 는 claude -p 형태(-p 포함)", "-p" in inv["argv"])
|
|
|
|
# 원장 재개: 새 원장에 stage 산출 anchor 가 있으면 resume 가능
|
|
check("probe: 원장에 다음 stage anchor 있으면 resume True",
|
|
probe.resume_ok({"stages": []}, {"stages": ["ground"], "accepted": ["ground-report"]}) is True)
|
|
check("probe: 원장 변화 없으면 resume False",
|
|
probe.resume_ok({"stages": ["ground"]}, {"stages": ["ground"]}) is False)
|
|
|
|
from bench_cascade import judge # noqa: E402
|
|
|
|
calls = judge.plan_calls(["A", "B", "C"], panel_size=3)
|
|
check("judge: 3-arm panel3 → 18 호출", len(calls) == 18)
|
|
check("judge: 각 pair 마다 forward+reversed", sum(1 for c in calls if c["orientation"] == "reversed") == 9)
|
|
xy = judge.assign_xy(("A", "B"), "forward", seed="s")
|
|
check("judge: forward 는 X=첫 arm", xy == {"X": "A", "Y": "B"})
|
|
xyr = judge.assign_xy(("A", "B"), "reversed", seed="s")
|
|
check("judge: reversed 는 X=둘째 arm", xyr == {"X": "B", "Y": "A"})
|
|
lv = judge.logical_vote_id("run-1", "A-vs-B", 1, "forward")
|
|
check("judge: logical-vote-id 결정론", lv == judge.logical_vote_id("run-1", "A-vs-B", 1, "forward"))
|
|
check("judge: judgment-id 는 attempt 별로 다름",
|
|
judge.judgment_id(lv, 1) != judge.judgment_id(lv, 2))
|
|
check("judge: prompt 에 injection 방어 원칙 포함", judge.INJECTION_GUARD in judge.build_prompt({}, {}, {}))
|
|
# dedup: 같은 lvid 에서 마지막 성공본만
|
|
recs = [
|
|
{"logical-vote-id": lv, "attempt": 1, "status": "malformed"},
|
|
{"logical-vote-id": lv, "attempt": 2, "status": "valid", "pairwise-judgment": {"overall": {"winner": "X"}}},
|
|
]
|
|
ded = judge.dedup(recs)
|
|
check("judge: dedup 은 lvid별 마지막 성공 1개", len(ded) == 1 and ded[0]["attempt"] == 2)
|
|
# run_panel: mock model_call 이 malformed 2회면 panel-incomplete
|
|
def _bad_call(prompt):
|
|
return "이건 YAML 아님 @@@"
|
|
res = judge.run_panel([{"pair": ("A", "B"), "orientation": "forward", "judge-index": 1}],
|
|
bundles={"A": {}, "B": {}}, rubric={}, run_id="run-1",
|
|
model_call=_bad_call, budget_path=None)
|
|
check("judge: malformed 2회 → panel-incomplete", res[0]["status"] == "panel-incomplete")
|
|
|
|
from bench_cascade import budget as _bud # noqa: E402
|
|
def _ok_call(prompt):
|
|
return "pairwise-judgment:\n overall: {winner: X}\n"
|
|
_bp = os.path.join(tempfile.mkdtemp(prefix="p4jbud_"), "r.json")
|
|
_bud.approve("pj", 1_000_000, 100.0, _bp)
|
|
judge.run_panel([{"pair": ("A", "B"), "orientation": "forward", "judge-index": 1}],
|
|
bundles={"A": {}, "B": {}}, rubric={}, run_id="run-b",
|
|
model_call=_ok_call, budget_path=_bp)
|
|
check("judge: run_panel 이 예산을 실제 charge(상한 라이브)", _bud.remaining(_bp)["tokens"] < 1_000_000)
|
|
_bp2 = os.path.join(tempfile.mkdtemp(prefix="p4jbud2_"), "r.json")
|
|
_bud.approve("pj2", 1, 100.0, _bp2) # token 1개(cost 넉넉) — require 통과 후 charge()의 token overage 가 halt(신규 charge 경로 실검증)
|
|
_raised = False
|
|
try:
|
|
judge.run_panel([{"pair": ("A", "B"), "orientation": "forward", "judge-index": 1}],
|
|
bundles={"A": {}, "B": {}}, rubric={}, run_id="run-c",
|
|
model_call=_ok_call, budget_path=_bp2)
|
|
except (ValueError, RuntimeError):
|
|
_raised = True
|
|
check("judge: 너무 작은 예산이면 run_panel 이 유료호출을 막고 halt(fail-closed)", _raised)
|
|
|
|
from bench_cascade import runner # noqa: E402
|
|
check("runner: STAGES 는 ground·decide·design-direction·design-system-dryrun 4개",
|
|
[s["name"] for s in runner.STAGES] == ["ground", "decide", "design-direction", "design-system-dryrun"])
|
|
env = runner.evidence_env("/ctrl/evidence-pack")
|
|
check("runner: evidence env 는 외부웹 차단(Blocker 1)", env["ORGOS_EXTERNAL_WEB"] == "denied")
|
|
check("runner: evidence env 는 evidence-pack 경로 주입", env["BENCHMARK_EVIDENCE_PACK"] == "/ctrl/evidence-pack")
|
|
rc = runner.human_receipt("run-1", "briefsha", ["A", "B", "C"])
|
|
check("runner: HUMAN receipt 는 전 arm 동일 scope(Blocker 2)", rc["accepted-scope"]["arm-ids"] == ["A", "B", "C"])
|
|
check("runner: HUMAN receipt forbidden 에 deployment/real-purchase",
|
|
"deployment" in rc["forbidden"] and "real-purchase" in rc["forbidden"])
|
|
# run_stage: mock exec_fn 이 exit0 + artifact 리턴
|
|
def _mock_exec(argv, cwd, env):
|
|
return {"exit-code": 0, "artifacts": ["ground-report.yaml"], "transcript": []}
|
|
_ws = tempfile.mkdtemp(prefix="p4ws_")
|
|
res = runner.run_stage(worktree="/wt", workspace=_ws, stage=runner.STAGES[0], env={}, exec_fn=_mock_exec)
|
|
check("runner: run_stage 성공 시 exit-code 0", res["exit-code"] == 0)
|
|
check("runner: run_stage 산출물 기록", "ground-report.yaml" in res["artifacts"])
|
|
# 실패 stage 는 다음 진행 억지 금지 신호
|
|
def _fail_exec(argv, cwd, env):
|
|
return {"exit-code": 1, "artifacts": [], "transcript": []}
|
|
resf = runner.run_stage(worktree="/wt", workspace=_ws, stage=runner.STAGES[0], env={}, exec_fn=_fail_exec)
|
|
check("runner: 실패 stage 는 exit-code 비0(다음 stage 차단 신호)", resf["exit-code"] != 0)
|
|
# dry-run stage 는 ORGOS_DRY_RUN env 실어보냄 + evidence_env 스레딩 확인
|
|
_cap = {}
|
|
def _cap_exec(argv, cwd, env):
|
|
_cap.update(env)
|
|
return {"exit-code": 0, "artifacts": [], "transcript": []}
|
|
_ws2 = tempfile.mkdtemp(prefix="p4ws2_")
|
|
runner.run_stage(worktree="/wt", workspace=_ws2, stage=runner.STAGES[3],
|
|
env=runner.evidence_env("/ctrl/evidence-pack"), exec_fn=_cap_exec)
|
|
check("runner: dry-run stage 는 env 에 ORGOS_DRY_RUN=true(실제 /design-system 미실행 신호)",
|
|
_cap.get("ORGOS_DRY_RUN") == "true")
|
|
check("runner: evidence_env 가 exec env 로 전달됨(외부웹 차단 Blocker1)",
|
|
_cap.get("ORGOS_EXTERNAL_WEB") == "denied" and _cap.get("BENCHMARK_EVIDENCE_PACK") == "/ctrl/evidence-pack")
|
|
|
|
print(f"\n{passed} passed · {failed} failed")
|
|
sys.exit(1 if failed else 0)
|