#!/usr/bin/env python3
"""P4 cascade benchmark — ruler pure-logic. standalone check. exit 0=통과."""
import importlib.util
import os
import sys

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 paths  # noqa: E402

check("paths: controller_dir 은 benchmark/cascade", paths.controller_dir().endswith(os.path.join("benchmark", "cascade")))
check("paths: run_id 는 결정론적(같은 seed→같은 id)", paths.run_id("s1") == paths.run_id("s1"))
check("paths: run_id 는 seed 별로 다름", paths.run_id("s1") != paths.run_id("s2"))
check("paths: run_id 형식 run-<hex>", paths.run_id("s1").startswith("run-") and len(paths.run_id("s1")) == 16)
check("paths: exec_root 는 /tmp 하위(worktree 격리)", paths.exec_root("run-x").startswith("/tmp/"))
check("paths: worktree 와 workspace 는 분리 경로",
      paths.worktree_dir("run-x", "A") != paths.workspace_dir("run-x", "A"))
check("paths: arm_run_dir 은 controller runs 하위(worktree 밖)",
      "benchmark" in paths.arm_run_dir("run-x", "A") and "/tmp/" not in paths.arm_run_dir("run-x", "A"))

import tempfile  # noqa: E402
from bench_cascade import inputs  # noqa: E402

_d = tempfile.mkdtemp(prefix="p4in_")
open(os.path.join(_d, "a.md"), "w").write("hello")
open(os.path.join(_d, "b.md"), "w").write("world")
h1 = inputs.sha256_tree(_d)
check("inputs: sha256_tree 결정론적(같은 내용→같은 hash)", h1 == inputs.sha256_tree(_d))
open(os.path.join(_d, "b.md"), "w").write("WORLD")
check("inputs: 내용 바뀌면 tree hash 변경", h1 != inputs.sha256_tree(_d))
_f = os.path.join(_d, "a.md")
check("inputs: sha256_file 은 64hex", len(inputs.sha256_file(_f)) == 64)
rec = inputs.benchmark_input(_f, _f, _d, _d)
check("inputs: benchmark_input 4-키", set(rec) == {"brief-sha256", "rubric-sha256", "fixture-set-sha256", "evidence-pack-sha256"})

from bench_cascade import manifest  # noqa: E402

man = manifest.load()
check("manifest: arms A/B/C 정의", set(man["arms"]) == {"A", "B", "C"})
check("manifest: commit 은 full 40hex", all(len(man["arms"][a]["commit"]) == 40 for a in "ABC"))
# 실측: arm A/B commit 은 P3-B active 0, arm C 는 DES-* active 보유
amA = manifest.active_methods_at(man["arms"]["A"]["commit"])
check("manifest: arm A 는 active 계약 0(P3 이전)", sum(len(v) for v in amA.values()) == 0)
amC = manifest.active_methods_at(man["arms"]["C"]["commit"])
check("manifest: arm C 는 DES-DIRECTOR active 보유", "converge-directions" in amC.get("DES-DIRECTOR", []))
# pre-flight 는 실제 3 commit 로 통과해야 한다
viol = manifest.preflight(man)
check("manifest: pre-flight 통과(위반 0)", viol == [], )
# drift: resolved 가 manifest 와 다르면 위반
bad_resolved = [{"stage": "x", "role-id": "DES-DIRECTOR", "method-id": "WRONG"}]
check("manifest: resolved-method-plan drift 검출", manifest.drift(man, bad_resolved) != [])
# arm C draft-fallback 시뮬: manifest 가 없는 role 을 요구하면 pre-flight 실패(가짜 manifest)
fake = {"arms": man["arms"], "pilot-invoked-methods": [{"role": "DES-DIRECTOR", "methods": ["NONEXISTENT"]}]}
check("manifest: arm C 가 요구 method 를 active 로 없으면 pre-flight 실패",
      any("arm C" in v or "active" in v for v in manifest.preflight(fake)))

from bench_cascade import budget  # noqa: E402

_bp = os.path.join(tempfile.mkdtemp(prefix="p4bud_"), "receipt.json")
budget.approve("plan-1", 1000, 5.0, _bp)
check("budget: approve 생성", budget.load(_bp)["max-tokens"] == 1000)
budget.charge(_bp, 400, 1.0)
check("budget: charge 후 잔여 토큰 600", budget.remaining(_bp)["tokens"] == 600)
_raised = False
try:
    budget.charge(_bp, 700, 0.0)  # 600 잔여에 700 요구 → 초과
except ValueError:
    _raised = True
check("budget: 초과 charge 는 ValueError", _raised)
_req = False
try:
    budget.require(os.path.join(os.path.dirname(_bp), "nope.json"))
except (RuntimeError, SystemExit):
    _req = True
check("budget: receipt 없으면 require 거부(Blocker 4)", _req)
_bp2 = os.path.join(tempfile.mkdtemp(prefix="p4bud2_"), "receipt.json")
budget.approve("plan-2", 100000, 5.0, _bp2)   # tokens plenty, cost small
budget.charge(_bp2, 10, 5.0)                   # cost fully spent, tokens barely used
_costreq = False
try:
    budget.require(_bp2)
except RuntimeError:
    _costreq = True
check("budget: cost 소진(토큰 여유)도 require 거부(Blocker 4 대칭)", _costreq)

from bench_cascade import meter  # noqa: E402

_FX = os.path.join(ROOT, ".claude", "tests", "fixtures", "p4")
m = meter.derive(os.path.join(_FX, "transcript-sample.jsonl"), os.path.join(_FX, "stage-ledger-sample.yaml"))
check("meter: input-tokens 합산 1500", m["input-tokens"] == 1500)
check("meter: output-tokens 합산 950", m["output-tokens"] == 950)
check("meter: turns 2", m["turns"] == 2)
check("meter: subagent-spawns 2", m["subagent-spawns"] == 2)
check("meter: hook-blocks 1", m["hook-blocks"] == 1)
check("meter: stage-retries 합산 1", m["stage-retries"] == 1)
check("meter: critique-revisions 2", m["critique-revisions"] == 2)
check("meter: artifacts-produced 6", m["artifacts-produced"] == 6)
check("meter: wall-seconds 215", m["wall-seconds"] == 215)
check("meter: human pre-authorized 1·interactive 0",
      m["human-interventions"] == {"interactive": 0, "pre-authorized-receipts": 1})
check("meter: execution-failures 0(전 stage exit 0)", m["execution-failures"] == 0)

_bad_tx = os.path.join(tempfile.mkdtemp(prefix="p4tx_"), "t.jsonl")
with open(_bad_tx, "w") as _f:
    _f.write('not valid json\n')
    _f.write('42\n')
    _f.write('{"type":"usage","input_tokens":5,"output_tokens":3}\n')
    _f.write('{"type":"human_intervention","kind":"interactive"}\n')
_m2 = meter.derive(_bad_tx, os.path.join(_FX, "stage-ledger-sample.yaml"))
check("meter: 손상/비-dict 라인은 skip(crash 없이 유효분만 집계)",
      _m2["input-tokens"] == 5 and _m2["human-interventions"]["interactive"] == 1)

from bench_cascade import sanitize  # noqa: E402
_ART = os.path.join(_FX, "arm-artifacts-C")
_emap = {
    "selected-direction": {"file": "approved-direction.yaml", "path": "direction.summary"},
    "selection-rationale": {"file": "approved-direction.yaml", "path": "direction.rationale"},
    "rejected-directions": {"file": "approved-direction.yaml", "path": "direction.rejected"},
    "locked-invariants": {"file": "approved-direction.yaml", "path": "locked-invariants"},
}
pkg = sanitize.project(_ART, _emap)
check("sanitize: selected-direction 투영", "정보밀도" in pkg["candidate-package"]["selected-direction"]["value"])
check("sanitize: 공통 구조 빈 필드 유지(problem-framing 존재)", "problem-framing" in pkg["candidate-package"])
check("sanitize: provenance(source-artifacts) 유지",
      pkg["candidate-package"]["selected-direction"]["source-artifacts"][0]["artifact-sha256"])
import yaml as _y  # noqa: E402
_txt = _y.safe_dump(pkg)
check("sanitize: 투영 결과에 role-id 누설 없음", sanitize.leak_scan(_txt) == [])
check("sanitize: 원본 role-id/method-execution 는 leak_scan 이 잡는다",
      sanitize.leak_scan("role-id: DES-DIRECTOR\nmethod-execution: {}") != [])
# omission: 실질 필드가 비면 검출(selected-direction 없는 map)
pkg2 = sanitize.project(_ART, {"locked-invariants": {"file": "approved-direction.yaml", "path": "locked-invariants"}})
check("sanitize: 실질필드 누락 검출(omission)", sanitize.check_omission(pkg2["candidate-package"]) != [])

# 렌더 번들 + not-evaluable 테스트
_bdir = tempfile.mkdtemp(prefix="p4bundle_")
# 렌더 없는 경우: design not-evaluable
b0 = sanitize.build_bundle("run-x", "cand-A", pkg, prototype_dir=None, render=False)
check("sanitize: prototype 없으면 design not-evaluable", sanitize.design_status(b0) == "not-evaluable")
check("sanitize: 번들에 candidate.yaml 기록", os.path.exists(os.path.join(b0["bundle-dir"], "candidate.yaml")))
# 가짜 렌더 png 를 심으면 evaluable
_pdir = tempfile.mkdtemp(prefix="p4proto_")
open(os.path.join(_pdir, "prototype-desktop.png"), "wb").write(b"\x89PNG\r\n")
open(os.path.join(_pdir, "prototype-mobile.png"), "wb").write(b"\x89PNG\r\n")
b1 = sanitize.build_bundle("run-x", "cand-B", pkg, prototype_dir=_pdir, render=True)
check("sanitize: 렌더 png 있으면 design evaluable", sanitize.design_status(b1) == "evaluable")
check("sanitize: 번들이 렌더 2장 포함", len(b1["renders"]) == 2)

_b3 = sanitize.build_bundle("run-i3", "cand-clean", pkg, prototype_dir=None, render=False)
import yaml as _y3  # noqa: E402
_cy = _y3.safe_load(open(os.path.join(_b3["bundle-dir"], "candidate.yaml")))
check("sanitize: candidate.yaml 는 candidate-package 만(프로세스 메타 미포함)",
      "projection-metrics" not in _cy and "sanitizer-version" not in _cy)
_leakpkg = {"candidate-package": {"problem-framing": {"value": "role-id: DES-DIRECTOR 누설", "source-artifacts": []}}}
_raised3 = False
try:
    sanitize.build_bundle("run-i3", "cand-leak", _leakpkg, prototype_dir=None, render=False)
except ValueError:
    _raised3 = True
check("sanitize: 누설 토큰 있으면 build_bundle 이 fail-loud(채점 금지)", _raised3)

from bench_cascade import aggregate as agg  # noqa: E402
# forward: X 승 → pair 첫 arm(=A) 승; reversed: Y 승 → 첫 arm(A) 승
check("agg: forward X→first", agg.normalize("forward", "X") == "first")
check("agg: reversed Y→first", agg.normalize("reversed", "Y") == "first")
check("agg: reversed X→second", agg.normalize("reversed", "X") == "second")
check("agg: 두 orientation 같은 실질승자면 stable", agg.stable("first", "first") is True)
check("agg: 다르면 unstable", agg.stable("first", "second") is False)
check("agg: preference-score (2승1무/3) = 0.833", abs(agg.preference_score(2, 1, 3) - 0.8333) < 1e-3)
check("agg: panel-agreement 2/3", abs(agg.panel_agreement(["first", "first", "second"]) - 0.6667) < 1e-3)
_paired = [{"fwd": "first", "rev": "first"}, {"fwd": "second", "rev": "second"}, {"fwd": "first", "rev": "second"}]
check("agg: flip-consistency 2/3(3번째 불일치)", abs(agg.flip_consistency(_paired) - 0.6667) < 1e-3)
check("agg: 최빈 2표 이상이면 그 verdict 채택", agg.panel_verdict(["first", "first", "second"]) == "first")
check("agg: stable<2 면 unstable", agg.panel_verdict(["first"]) == "unstable")
check("agg: 최빈<2면 unstable", agg.panel_verdict(["first", "second"]) == "unstable")

from bench_cascade import calibrate  # noqa: E402
check("calib: gold>bad 통과(pref .7·verdict gold·flip .7)",
      calibrate.gold_vs_bad_pass(0.7, "gold", 0.7) is True)
check("calib: gold pref<0.67 실패", calibrate.gold_vs_bad_pass(0.6, "gold", 0.9) is False)
check("calib: verdict!=gold 실패", calibrate.gold_vs_bad_pass(0.9, "bad", 0.9) is False)
th = {"target-min-drop": 1.0, "non-target-max-drop": 0.5, "target-margin-over-next": 0.5, "pairwise-target-goldwin-min": 0.67}
check("calib: 단일결함 통과(target 1.2·next 0.3·pairwise .7)",
      calibrate.single_defect_pass(1.2, 0.3, 0.3, 0.7, th) is True)
check("calib: target-drop<1.0 실패", calibrate.single_defect_pass(0.8, 0.1, 0.1, 0.9, th) is False)
check("calib: non-allowed drop>0.5 실패", calibrate.single_defect_pass(1.2, 0.6, 0.6, 0.9, th) is False)
check("calib: margin<0.5 실패(target 1.0·next 0.7)", calibrate.single_defect_pass(1.0, 0.7, 0.4, 0.9, th) is False)
check("calib: 집합 aggregate 통과(agreement .8·flip .85)",
      calibrate.aggregate_pass([{"agreement": 0.7, "flip": 0.7}], 0.8, 0.85) is True)
check("calib: 집합 flip<0.80 실패", calibrate.aggregate_pass([{"agreement": 0.9, "flip": 0.9}], 0.9, 0.7) is False)
v = calibrate.verdict({"gold-vs-bad": False, "single-defects": {}, "aggregate": True})
check("calib: 하나라도 FAIL 이면 전체 FAIL + 사유", v["pass"] is False and v["reasons"])

from bench_cascade import compare  # noqa: E402
md = compare.render_markdown({}, {}, {}, {"status": "held"}, calibrated=False)
check("compare: 강제 disclaimer 포함", "통계적 우월성" in md and compare.DISCLAIMER in md)
check("compare: uncalibrated 스탬프", "UNCALIBRATED" in md)
# 실패 arm 있으면 순위 held
rk = compare.ranking({"A-vs-B": {}}, failed_arms=["C"])
check("compare: 실패 arm 있으면 순위 held", rk["status"] == "held")
rk2 = compare.ranking({"A-vs-B": {"preference-first": 0.8}}, failed_arms=[])
check("compare: 실패 없으면 decided", rk2["status"] == "decided")
st = compare.stability_axis({"A": {"execution-failures": 0}, "C": {"execution-failures": 1}})
check("compare: stability success-rate 0.5", abs(st["execution-success-rate"] - 0.5) < 1e-9)

from bench_cascade import planner  # noqa: E402

# calibration: fixture N개 × pair조합 × panel × 2 orientation + 파일럿 18 + retry
n = planner.estimate_judge_calls(n_calibration_fixtures=8, panel_size=3, arm_ids=["A", "B", "C"], retry_factor=2)
check("planner: 총 judge 호출은 파일럿 18 초과(calibration 포함)", n > 18)
check("planner: retry_factor 반영(2배 상한)", planner.estimate_judge_calls(1, 3, ["A", "B", "C"], 2)
      > planner.estimate_judge_calls(1, 3, ["A", "B", "C"], 1))
s = planner.summary()
check("planner: summary 에 commit full hash", all(len(s["arms"][a]["commit"]) == 40 for a in "ABC"))
check("planner: summary 에 예상 judge 호출", "estimated-judge-calls" in s)
check("planner: summary 에 파일럿 pairwise=18", s["pilot-pairwise-calls"] == 18)

import subprocess as _sp  # noqa: E402
_cli = os.path.join(HOOKS, "benchmark_cascade.py")
_env = dict(os.environ); _env["CLAUDE_PROJECT_DIR"] = ROOT
r = _sp.run([sys.executable, _cli, "plan"], capture_output=True, text=True, env=_env)
check("cli: plan 은 exit 0", r.returncode == 0)
check("cli: plan 출력에 예상 judge 호출", "judge" in (r.stdout + r.stderr).lower())
_iso = tempfile.mkdtemp(prefix="p4cli_iso_")
_env_iso = dict(os.environ)
_env_iso["CLAUDE_PROJECT_DIR"] = _iso
r2 = _sp.run([sys.executable, _cli, "judge", "--execute"], capture_output=True, text=True, env=_env_iso)
check("cli: judge --execute 는 예산 receipt 없으면 거부(비0, 격리 tempdir로 결정론)", r2.returncode != 0)
r3 = _sp.run([sys.executable, _cli, "nonsense"], capture_output=True, text=True, env=_env)
check("cli: 알 수 없는 subcommand 는 비0", r3.returncode != 0)
_r4 = _sp.run([sys.executable, _cli, "compare"], capture_output=True, text=True, env=_env)
check("cli: 미배선 subcommand(compare)는 정직하게 비0(미실행 표시)", _r4.returncode != 0)

import yaml as _yy  # noqa: E402
_CD = os.path.join(ROOT, "benchmark", "cascade")
check("content: brief.md 존재·비어있지 않음", os.path.getsize(os.path.join(_CD, "brief.md")) > 200)
_pol = _yy.safe_load(open(os.path.join(_CD, "benchmark-policy.yaml")))
check("content: policy external-web-access denied", _pol["benchmark-policy"]["external-web-access"] == "denied")
_rub = _yy.safe_load(open(os.path.join(_CD, "rubric.yaml")))
from bench_cascade import JUDGE_CRITERIA  # noqa: E402
check("content: rubric 이 8 criteria 전부 정의", set(_rub["criteria"]) == set(JUDGE_CRITERIA))
_ep = os.path.join(_CD, "evidence-pack")
check("content: evidence-pack 4파일", all(os.path.exists(os.path.join(_ep, f)) for f in
      ["market-context.md", "competitor-snapshot.md", "user-observations.md", "sources.yaml"]))
_de = _yy.safe_load(open(os.path.join(_CD, "fixtures", "defect-evidence-grounding", "meta.yaml")))
check("content: 단일결함 fixture thresholds 4키",
      set(_de["fixture"]["thresholds"]) == {"target-min-drop", "non-target-max-drop", "target-margin-over-next", "pairwise-target-goldwin-min"})

print(f"\n{passed} passed · {failed} failed")
sys.exit(1 if failed else 0)