#!/usr/bin/env python3 """Focused local benchmark for the plain and structured Reason Guard paths.""" from __future__ import annotations import argparse import importlib.util import json import statistics import tempfile import time from pathlib import Path from typing import Any, Callable ROOT = Path(__file__).resolve().parents[1] GUARD_PATH = ROOT / ".prd_plugin" / "hooks" / "prd_reason_guard.py" LIVE_SESSION_FIXTURE = ( ROOT / "tests" / "fixtures" / "reason_guard_live_session_eval.json" ) LIVE_SESSION_ROUND5_FIXTURE = ( ROOT / "tests" / "fixtures" / "reason_guard_live_session_eval_round5.json" ) LIVE_SESSION_ROUND6_FIXTURE = ( ROOT / "tests" / "fixtures" / "reason_guard_live_session_eval_round6.json" ) MAX_STRUCTURED_P95_MS = 50.0 MAX_P95_DELTA_MS = 20.0 MAX_CLASSIFICATION_P95_MS = 1.0 SUMMARY_CASES = ( ("active-hooks", True, True, "Investigating hooks change cause."), ( "active-config", True, True, "Assessing matrix coverage and wiki drift; Tracing config skill_log change source.", ), ( "active-routing", True, True, "Investigating index job absence; Tracing repo-level tab routing flaw.", ), ( "active-repo-route", True, True, "Diagnosing repo routing and workspace tab issues.", ), ( "active-guard", True, True, "Verifying guard fallback policy before changing routing.", ), ( "active-runtime", True, True, "Checking runtime adapter capability before routing.", ), ( "active-model", True, True, "Comparing model profiles before changing routing.", ), ( "active-latency", True, True, "Measuring hook latency before changing the guard.", ), ( "active-inspection", True, True, "Inspecting config routing fallback behavior.", ), ( "active-reproduction", True, True, "Reproducing runtime guard failure before disabling the hook.", ), ( "commitment-control", True, True, "I will compare both routes before changing routing.", ), ( "alternatives-control", True, True, "Possible causes are model capability or batching; comparing both would distinguish routing behavior.", ), ("active-anomaly", True, True, "Investigating config timestamp anomaly."), ("active-failure", True, True, "Tracing fallback source after hook failure."), ("active-inconsistency", True, True, "Diagnosing model adapter inconsistency."), ("active-impact", True, True, "Assessing guard policy change impact."), ( "active-disable", True, True, "Checking runtime behavior before disabling fallback.", ), ( "active-difference", True, True, "Inspecting routing differences before config change.", ), ( "active-error", True, True, "Reproducing hook error before changing policy.", ), ( "active-performance", True, True, "Benchmarking model adapter latency before routing change.", ), ( "active-codex-markdown", True, True, "**Planning the safe replay** **Investigating routing policy change cause**", ), ("summary-only", False, False, "Summarizing docs audit and config changes."), ("explanation-only", False, False, "Explaining route source and UI behavior discrepancies."), ("completion-only", False, False, "Completed the config documentation."), ("assertion-only", False, False, "The routing policy is stable."), ("cosmetic-investigation", False, False, "Investigating button spacing in the settings UI."), ("formatting-check", False, False, "Checking formatting before changing comment wording."), ("documentation-review", False, False, "Reviewing hook documentation."), ("readme-diagnosis", False, False, "Diagnosing a typo in the README."), ("generic-assessment", False, False, "Assessing the latest screenshots."), ("plain-progress", False, False, "Continuing implementation."), ("ordinary-test", False, False, "Testing the login form validation."), ("ordinary-measure", False, False, "Measuring the image width."), ("cosmetic-config", False, False, "Inspecting config file formatting."), ("cosmetic-routing", False, False, "Tracing routing icon alignment."), ("cosmetic-profile", False, False, "Assessing model profile screenshot."), ("cosmetic-hook", False, False, "Checking hook documentation links."), ("cosmetic-runtime", False, False, "Measuring runtime label width."), ("cosmetic-guard", False, False, "Investigating guard heading capitalization."), ("cosmetic-policy", False, False, "Comparing policy page typography."), ("cosmetic-adapter", False, False, "Testing adapter documentation examples."), ("cosmetic-config-ui", False, False, "Diagnosing config UI spacing."), ("cosmetic-animation", False, False, "Reproducing hook animation."), ("cosmetic-fallback", False, False, "Verifying fallback label copy."), ("cosmetic-model-card", False, False, "Benchmarking model card rendering."), ( "cosmetic-codex-markdown", False, False, "**Planning the settings review** **Investigating button spacing**", ), ) def _load_guard(): spec = importlib.util.spec_from_file_location( "prd_reason_guard_benchmark_target", GUARD_PATH ) if spec is None or spec.loader is None: raise RuntimeError("Reason Guard benchmark target could not be loaded") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def _percentile(values: list[float], fraction: float) -> float: ordered = sorted(values) index = max(0, round(fraction * (len(ordered) - 1))) return ordered[index] def _stats(values: list[float]) -> dict[str, float]: return { "median_ms": round(statistics.median(values), 3), "p95_ms": round(_percentile(values, 0.95), 3), "max_ms": round(max(values), 3), } def _directive(index: int) -> dict[str, Any]: return { "version": 1, "kind": "evidence-check", "claim": f"Validate unitword{index} pathword{index}.", "required_evidence": [ { "type": "tool-result", "tool": "benchmark-status", "acceptance": [ {"predicate": "outcome_is", "value": "success"}, ], } ], "expected_result": "The benchmark status succeeds.", "blocking": False, } def _run_case( guard: Any, root: Path, samples: int, warmup: int, payload_for: Callable[[int], dict[str, Any]], ) -> list[float]: (root / ".prd_plugin").mkdir(parents=True) (root / ".prd_plugin" / "config.json").write_text( json.dumps({"reasoning_guard": {"mode": "report"}}), encoding="utf-8", ) measured: list[float] = [] for index in range(warmup + samples): started = time.perf_counter() guard.process_event( root, "UserPromptSubmit", "claude", payload_for(index), ) elapsed_ms = (time.perf_counter() - started) * 1000 if index >= warmup: measured.append(elapsed_ms) return measured def _classification_metrics( guard: Any, cases: list[dict[str, Any]], *, weighted: bool = False, ) -> dict[str, Any]: true_positive = false_positive = true_negative = false_negative = 0 uncertain_positive = uncertain_negative = 0 required_misses = [] false_positive_ids = [] false_negative_ids = [] timings = [] ambiguous_ids = [] evaluated_cases = 0 evaluated_weight = 0 for case in cases: case_id = case["id"] label = case["label"] text = case["text"] started = time.perf_counter() candidate = bool(guard._classify_obligation(text)) uncertain = ( not candidate and guard._has_uncertain_active_summary(text) ) timings.append((time.perf_counter() - started) * 1000) if label == "ambiguous": ambiguous_ids.append(case_id) continue expected = label == "positive" required = bool(case.get("required", expected)) weight = int(case.get("occurrences", 1)) if weighted else 1 evaluated_cases += 1 evaluated_weight += weight if candidate and expected: true_positive += weight elif candidate: false_positive += weight false_positive_ids.append(case_id) elif uncertain and expected: uncertain_positive += weight elif uncertain: uncertain_negative += weight elif expected: false_negative += weight false_negative_ids.append(case_id) if required: required_misses.append(case_id) else: true_negative += weight candidate_predictions = true_positive + false_positive candidate_precision = ( true_positive / (true_positive + false_positive) if true_positive + false_positive else None ) candidate_recall = ( true_positive / (true_positive + uncertain_positive + false_negative) if true_positive + uncertain_positive + false_negative else 1.0 ) safety_recall = ( (true_positive + uncertain_positive) / (true_positive + uncertain_positive + false_negative) if true_positive + uncertain_positive + false_negative else 1.0 ) clear_specificity = ( true_negative / (true_negative + false_positive + uncertain_negative) if true_negative + false_positive + uncertain_negative else 1.0 ) abstention_rate = ( (uncertain_positive + uncertain_negative) / evaluated_weight if evaluated_weight else 0.0 ) return { "cases": len(cases), "evaluated_cases": evaluated_cases, "evaluated_weight": evaluated_weight, "ambiguous_ids": ambiguous_ids, "true_positive": true_positive, "false_positive": false_positive, "true_negative": true_negative, "false_negative": false_negative, "uncertain_positive": uncertain_positive, "uncertain_negative": uncertain_negative, "candidate_predictions": candidate_predictions, "candidate_precision": ( round(candidate_precision, 4) if candidate_precision is not None else None ), "candidate_recall": round(candidate_recall, 4), "safety_recall": round(safety_recall, 4), "clear_specificity": round(clear_specificity, 4), "abstention_rate": round(abstention_rate, 4), "false_positive_ids": false_positive_ids, "false_negative_ids": false_negative_ids, "required_misses": required_misses, "latency": _stats(timings), } def _synthetic_cases() -> list[dict[str, Any]]: return [ { "id": case_id, "label": "positive" if expected else "negative", "required": required, "text": text, } for case_id, expected, required, text in SUMMARY_CASES ] def _load_live_session_cases(split: str) -> list[dict[str, Any]]: fixture = json.loads(LIVE_SESSION_FIXTURE.read_text(encoding="utf-8")) round5 = json.loads(LIVE_SESSION_ROUND5_FIXTURE.read_text(encoding="utf-8")) round6 = json.loads(LIVE_SESSION_ROUND6_FIXTURE.read_text(encoding="utf-8")) cases = [ *fixture.get("cases", []), *round5.get("cases", []), *round6.get("cases", []), ] if not isinstance(cases, list): raise ValueError("live-session fixture cases must be a list") allowed_labels = {"positive", "negative", "ambiguous"} current_round = int(fixture.get("sampling", {}).get("current_round", 1)) selected = [] seen_ids = set() seen_texts = set() for case in cases: if not isinstance(case, dict): raise ValueError("live-session fixture case must be an object") if case.get("id") in seen_ids or case.get("text") in seen_texts: raise ValueError("live-session fixture cases must be deduplicated") seen_ids.add(case.get("id")) seen_texts.add(case.get("text")) if case.get("label") not in allowed_labels: raise ValueError("live-session fixture label is invalid") if case.get("split") not in {"development", "holdout"}: raise ValueError("live-session fixture split is invalid") case_round = int(case.get("round", 1)) if ( split == "all" or ( split == "development" and ( case_round < current_round or ( case_round == current_round and case["split"] == "development" ) ) ) or ( split == "holdout" and case_round == current_round and case["split"] == "holdout" ) ): selected.append(case) if not selected: raise ValueError(f"live-session fixture has no {split} cases") return selected def _classification_benchmark(guard: Any, evaluation_split: str) -> dict[str, Any]: live_cases = _load_live_session_cases(evaluation_split) by_stratum = { "challenge": [case for case in live_cases if case["stratum"] != "representative-neutral"], "representative_neutral": [ case for case in live_cases if case["stratum"] == "representative-neutral" ], } return { "synthetic": _classification_metrics(guard, _synthetic_cases()), "live_session": { "split": evaluation_split, "headline": _classification_metrics(guard, live_cases), "occurrence_weighted": _classification_metrics( guard, live_cases, weighted=True ), "strata": { name: _classification_metrics(guard, cases) for name, cases in by_stratum.items() if cases }, }, } def benchmark( samples: int = 100, warmup: int = 10, evaluation_split: str = "development", ) -> dict[str, Any]: if samples < 1 or warmup < 0: raise ValueError("samples must be positive and warmup cannot be negative") if evaluation_split not in {"development", "holdout", "all"}: raise ValueError("evaluation_split must be development, holdout, or all") guard = _load_guard() with tempfile.TemporaryDirectory() as directory: base = Path(directory) plain_values = _run_case( guard, base / "plain", samples, warmup, lambda index: { "session_id": "benchmark-plain", "event_id": f"plain-{index}", "reasoning_summary": "Visible benchmark summary.", }, ) structured_values = _run_case( guard, base / "structured", samples, warmup, lambda index: { "session_id": "benchmark-structured", "event_id": f"structured-{index}", "reasoning_summary": "Visible benchmark summary.", "reasoning_evidence_check": _directive(index), }, ) plain = _stats(plain_values) structured = _stats(structured_values) classification = _classification_benchmark(guard, evaluation_split) synthetic = classification["synthetic"] live = classification["live_session"]["headline"] live_strata = classification["live_session"]["strata"].values() p95_delta = round(structured["p95_ms"] - plain["p95_ms"], 3) def candidate_precision_passes(metrics: dict[str, Any]) -> bool: precision = metrics["candidate_precision"] return ( precision >= 0.95 if precision is not None else metrics["false_positive"] == 0 ) passed = ( structured["p95_ms"] <= MAX_STRUCTURED_P95_MS and p95_delta <= MAX_P95_DELTA_MS and candidate_precision_passes(synthetic) and synthetic["safety_recall"] >= 0.95 and synthetic["clear_specificity"] >= 0.50 and not synthetic["required_misses"] and candidate_precision_passes(live) and live["safety_recall"] >= 0.95 and live["clear_specificity"] >= 0.50 and not live["required_misses"] and all( candidate_precision_passes(stratum) and stratum["safety_recall"] >= 0.95 for stratum in live_strata ) and classification["live_session"]["strata"] ["representative_neutral"]["clear_specificity"] >= 0.50 and max( synthetic["latency"]["p95_ms"], live["latency"]["p95_ms"], ) <= MAX_CLASSIFICATION_P95_MS ) return { "schema_version": "1.2", "method": ( "high-confidence candidate classification plus conservative abstention; " "same-host event path; unique events; local atomic report writes" ), "samples": samples, "warmup": warmup, "plain": plain, "structured": structured, "classification": classification, "comparison": {"p95_delta_ms": p95_delta}, "budget": { "structured_p95_ms_max": MAX_STRUCTURED_P95_MS, "p95_delta_ms_max": MAX_P95_DELTA_MS, "candidate_precision_min": 0.95, "safety_recall_min": 0.95, "clear_specificity_min": 0.50, "classification_p95_ms_max": MAX_CLASSIFICATION_P95_MS, "passed": passed, }, "dependencies": {"model": False, "network": False}, } def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--samples", type=int, default=100) parser.add_argument("--warmup", type=int, default=10) parser.add_argument( "--evaluation-split", choices=("development", "holdout", "all"), default="development", help="Use development while tuning; run holdout once after the rule is frozen.", ) parser.add_argument("--json", action="store_true") args = parser.parse_args(argv) report = benchmark(args.samples, args.warmup, args.evaluation_split) if args.json: print(json.dumps(report, indent=2, ensure_ascii=True)) else: print( "Reason Guard structured p95 " f"{report['structured']['p95_ms']} ms; delta " f"{report['comparison']['p95_delta_ms']} ms; " "live-session safety recall " f"{report['classification']['live_session']['headline']['safety_recall']}; " f"budget {'passed' if report['budget']['passed'] else 'failed'}" ) return 0 if report["budget"]["passed"] else 1 if __name__ == "__main__": raise SystemExit(main())