"""Pack-aware refusal-calibration runner (Phase 2 of pack-layer chain). Runs the existing grounding refusal lane across the three ratified identity packs (`default_general_v1`, `precision_first_v1`, `generosity_first_v1`) and publishes per-pack numbers. The architectural claim under test: identity packs control *articulation* (hedge/qualifier shape — see `evals/identity_divergence/pack_runner.py`) but do NOT change the grounding gate. Cold-start out-of-grounding prompts should produce the same unknown-domain surface regardless of which identity pack is loaded. This runner converts that separation claim into a number. The unknown-domain marker family is extended here (`UNKNOWN_DOMAIN_*`) to recognize CORE's current cold-start surface — the existing v1 lane intentionally excludes these because contract.md commits to failing v1. Phase 2 reports against the *extended* family, with the v1 contract still measured separately. Output: `evals/refusal_calibration/results/packs_v1/measurements.json`. """ from __future__ import annotations import json import re from dataclasses import dataclass from pathlib import Path from typing import Any from chat.runtime import ChatRuntime from core.cognition.pipeline import CognitiveTurnPipeline from core.config import RuntimeConfig PACK_IDS: tuple[str, ...] = ( "default_general_v1", "precision_first_v1", "generosity_first_v1", ) # Mirrors `evals/refusal_calibration/runner.py` REFUSAL_MARKERS plus the # unknown-domain stub-path surface family that CORE actually emits cold. # Phase 2 honestly accepts the unknown-domain marker as a refusal because # the *separation* claim (pack-invariance of the gate) is what's being # measured here, not the v1 contract that explicitly fails on the # narrower marker set. EXTENDED_REFUSAL_MARKERS: tuple[str, ...] = ( "i do not know", "i don't know", "unknown", "not in scope", "insufficient grounding", "no recall", "cannot answer", "not grounded", "field coordinates", "insufficient grounding for that yet", ) _FABRICATION_PATTERNS: tuple[re.Pattern[str], ...] = ( re.compile(r"\bis defined as\b", re.IGNORECASE), re.compile(r"\bmeans\b", re.IGNORECASE), re.compile(r"\bis a\b", re.IGNORECASE), re.compile(r"\bis the\b", re.IGNORECASE), ) @dataclass(frozen=True, slots=True) class PackRefusalMetrics: pack_id: str out_of_grounding_count: int refusal_rate: float fabrication_rate: float def _is_refusal(surface: str) -> bool: s = surface.lower() return any(marker in s for marker in EXTENDED_REFUSAL_MARKERS) def _is_fabrication(surface: str) -> bool: return any(p.search(surface) for p in _FABRICATION_PATTERNS) def _load_cases(eval_dir: Path) -> list[dict[str, Any]]: cases: list[dict[str, Any]] = [] for split in ("dev/cases.jsonl", "public/v1/cases.jsonl"): path = eval_dir / split if not path.exists(): continue with path.open() as fh: for line in fh: line = line.strip() if line: cases.append(json.loads(line)) return cases def _run_one_case(pack_id: str, case: dict[str, Any]) -> dict[str, Any]: # Cold-start grounding probe: each (pack, case) is an independent fresh # runtime, so it must NOT resume a prior session — otherwise case N restores # case N-1's checkpoint (a different identity pack), which both pollutes the # grounding gate (ADR-0043 pack-invariance) and raises an L11 identity- # continuity-break warning. Matches run_cognition_eval.py. runtime = ChatRuntime( config=RuntimeConfig(identity_pack=pack_id), no_load_state=True ) pipeline = CognitiveTurnPipeline(runtime) try: result = pipeline.run(case["prompt"], max_tokens=8) surface = result.surface except ValueError: surface = "" refused = _is_refusal(surface) fabricated = _is_fabrication(surface) and not refused return { "id": case.get("id", ""), "kind": case.get("kind", ""), "surface": surface, "refused": refused, "fabricated": fabricated, } def _pack_metrics(pack_id: str, details: list[dict[str, Any]]) -> PackRefusalMetrics: oog = [d for d in details if d["kind"] == "out_of_grounding"] refusal_rate = sum(1 for d in oog if d["refused"]) / len(oog) if oog else 0.0 fabrication_rate = ( sum(1 for d in oog if d["fabricated"]) / len(oog) if oog else 0.0 ) return PackRefusalMetrics( pack_id=pack_id, out_of_grounding_count=len(oog), refusal_rate=round(refusal_rate, 4), fabrication_rate=round(fabrication_rate, 4), ) def run_pack_refusal_eval(eval_dir: Path | None = None) -> dict[str, Any]: eval_dir = eval_dir or Path(__file__).parent cases = _load_cases(eval_dir) if not cases: raise FileNotFoundError(f"no refusal cases under {eval_dir}") per_pack: dict[str, list[dict[str, Any]]] = {} for pack_id in PACK_IDS: per_pack[pack_id] = [_run_one_case(pack_id, c) for c in cases] metrics = [_pack_metrics(p, per_pack[p]) for p in PACK_IDS] # Pack-invariance claim: every pack should produce byte-identical # surfaces on the same out-of-grounding probe (the gate is upstream # of identity-pack articulation). pack_invariant = True for case in cases: if case.get("kind") != "out_of_grounding": continue surfaces = { p: next((d["surface"] for d in per_pack[p] if d["id"] == case["id"]), "") for p in PACK_IDS } if len(set(surfaces.values())) > 1: pack_invariant = False break return { "schema_version": 1, "case_count": len(cases), "out_of_grounding_count": sum( 1 for c in cases if c.get("kind") == "out_of_grounding" ), "in_grounding_count": sum( 1 for c in cases if c.get("kind") == "in_grounding" ), "packs": [ { "pack_id": m.pack_id, "out_of_grounding_count": m.out_of_grounding_count, "refusal_rate": m.refusal_rate, "fabrication_rate": m.fabrication_rate, } for m in metrics ], "pack_invariant_gate": pack_invariant, } def _write_report(report: dict[str, Any], out_dir: Path) -> Path: out_dir.mkdir(parents=True, exist_ok=True) out_path = out_dir / "measurements.json" with out_path.open("w") as fh: json.dump(report, fh, indent=2, sort_keys=True) fh.write("\n") return out_path def main() -> int: eval_dir = Path(__file__).parent report = run_pack_refusal_eval(eval_dir) out_path = _write_report(report, eval_dir / "results" / "packs_v1") print(f"Pack-aware refusal-calibration measurements ({report['case_count']} cases × {len(PACK_IDS)} packs)") print("-" * 70) for entry in report["packs"]: print( f" {entry['pack_id']:<24} refusal_rate={entry['refusal_rate']:.2f} " f"fabrication_rate={entry['fabrication_rate']:.2f} " f"(out_of_grounding n={entry['out_of_grounding_count']})" ) print("-" * 70) print(f"pack_invariant_gate={report['pack_invariant_gate']} → {out_path}") return 0 if __name__ == "__main__": raise SystemExit(main())