feat(evals): add generalization audit runner skeleton

This commit is contained in:
Shay 2026-06-23 07:23:43 -07:00
parent 3100f84e73
commit f823eb437b
5 changed files with 609 additions and 0 deletions

View file

@ -14,6 +14,13 @@ from evals.generalization.cache_verifier import (
CacheVerificationReport,
verify_local_generalization_cache,
)
from evals.generalization.item_schema import (
GENERALIZATION_AUDIT_RUNNER_POLICY_VERSION,
GeneralizationAuditItem,
GeneralizationAuditOutcome,
GeneralizationAuditReport,
)
from evals.generalization.audit_runner import run_generalization_audit
__all__ = [
"GeneralizationBenchmarkManifest",
@ -24,4 +31,9 @@ __all__ = [
"CacheVerificationRecord",
"CacheVerificationReport",
"verify_local_generalization_cache",
"GENERALIZATION_AUDIT_RUNNER_POLICY_VERSION",
"GeneralizationAuditItem",
"GeneralizationAuditOutcome",
"GeneralizationAuditReport",
"run_generalization_audit",
]

View file

@ -0,0 +1,110 @@
"""Generalization audit runner execution logic."""
from __future__ import annotations
from collections import Counter
from typing import Callable
from evals.generalization.item_schema import (
GENERALIZATION_AUDIT_RUNNER_POLICY_VERSION,
GeneralizationAuditItem,
GeneralizationAuditOutcome,
GeneralizationAuditReport,
)
def run_generalization_audit(
*,
dataset: str,
split: str,
items: tuple[GeneralizationAuditItem, ...],
evaluator: Callable[[GeneralizationAuditItem], GeneralizationAuditOutcome],
) -> GeneralizationAuditReport:
"""Executes a generalization audit over a sequence of items using the provided evaluator.
Args:
dataset: Name of the dataset under audit.
split: Name of the dataset split.
items: Tuple of audit items. Refuses empty items (raises ValueError).
evaluator: Injected evaluation function mapping an item to an outcome.
Returns:
An aggregated, deterministic GeneralizationAuditReport.
"""
if not items:
raise ValueError("Audit execution requires a non-empty sequence of items.")
correct = 0
wrong = 0
refused = 0
unsupported = 0
candidate_attempts = 0
binding_failures = 0
replay_refusals = 0
sealed_trace_counter: Counter[str] = Counter()
residual_kind_counter: Counter[str] = Counter()
reason_codes_set: set[str] = set()
for item in items:
try:
outcome = evaluator(item)
except Exception as exc:
# Evaluator exceptions fail closed to refused outcome
outcome = GeneralizationAuditOutcome(
item_id=item.item_id,
disposition="refused",
residual_kinds=(),
candidate_attempt_count=0,
binding_failure_count=0,
replay_refusal_count=0,
sealed_trace_dispositions=(),
reason_codes=("evaluator_exception", type(exc).__name__),
)
disp = outcome.disposition
if disp == "correct":
correct += 1
elif disp == "wrong":
wrong += 1
elif disp == "refused":
refused += 1
elif disp == "unsupported":
unsupported += 1
else:
refused += 1 # Fallback to refused for unrecognized dispositions
candidate_attempts += outcome.candidate_attempt_count
binding_failures += outcome.binding_failure_count
replay_refusals += outcome.replay_refusal_count
for d in outcome.sealed_trace_dispositions:
sealed_trace_counter[d] += 1
for r in outcome.residual_kinds:
residual_kind_counter[r] += 1
reason_codes_set.update(outcome.reason_codes)
# Sort histograms descending by count, then alphabetically by key for determinism
sorted_sealed_traces = tuple(
sorted(sealed_trace_counter.items(), key=lambda x: (-x[1], x[0]))
)
sorted_residual_kinds = tuple(
sorted(residual_kind_counter.items(), key=lambda x: (-x[1], x[0]))
)
return GeneralizationAuditReport(
policy_version=GENERALIZATION_AUDIT_RUNNER_POLICY_VERSION,
dataset=dataset,
split=split,
n_items=len(items),
correct=correct,
wrong=wrong,
refused=refused,
unsupported=unsupported,
candidate_attempts=candidate_attempts,
binding_failures=binding_failures,
replay_refusals=replay_refusals,
sealed_trace_dispositions=sorted_sealed_traces,
dominant_residual_kinds=sorted_residual_kinds,
reason_codes=tuple(sorted(reason_codes_set)),
)

View file

@ -0,0 +1,53 @@
"""Schemas for generalization audit items, outcomes, and reports."""
from __future__ import annotations
from dataclasses import dataclass
GENERALIZATION_AUDIT_RUNNER_POLICY_VERSION = "generalization_audit_runner.v1"
@dataclass(frozen=True, slots=True)
class GeneralizationAuditItem:
"""A normalized item representing a single benchmark problem instance for audit."""
dataset: str
split: str
item_id: str
prompt_ref: str
answer_kind: str
metadata: tuple[tuple[str, str], ...]
@dataclass(frozen=True, slots=True)
class GeneralizationAuditOutcome:
"""The result of evaluating a single generalization audit item."""
item_id: str
disposition: str # correct | wrong | refused | unsupported
residual_kinds: tuple[str, ...]
candidate_attempt_count: int
binding_failure_count: int
replay_refusal_count: int
sealed_trace_dispositions: tuple[str, ...]
reason_codes: tuple[str, ...]
@dataclass(frozen=True, slots=True)
class GeneralizationAuditReport:
"""Aggregated report summarizing outcomes across a set of audit items."""
policy_version: str
dataset: str
split: str
n_items: int
correct: int
wrong: int
refused: int
unsupported: int
candidate_attempts: int
binding_failures: int
replay_refusals: int
sealed_trace_dispositions: tuple[tuple[str, int], ...]
dominant_residual_kinds: tuple[tuple[str, int], ...]
reason_codes: tuple[str, ...]

View file

@ -0,0 +1,157 @@
#!/usr/bin/env python3
"""CLI script to run generalization audit (skeleton)."""
from __future__ import annotations
import argparse
import json
import sys
from dataclasses import asdict
from pathlib import Path
# Resolve repository root and add to sys.path to support imports
script_path = Path(__file__).resolve()
repo_root = script_path.parent.parent.parent
if str(repo_root) not in sys.path:
sys.path.insert(0, str(repo_root))
from evals.generalization.audit_runner import run_generalization_audit # noqa: E402
from evals.generalization.item_schema import ( # noqa: E402
GeneralizationAuditItem,
GeneralizationAuditOutcome,
)
def main() -> None:
parser = argparse.ArgumentParser(description="Run generalization audit.")
parser.add_argument(
"--synthetic-smoke",
action="store_true",
help="Run a synthetic smoke audit.",
)
parser.add_argument("--dataset", type=str, help="Name of the dataset to audit.")
parser.add_argument(
"--split", type=str, default="test", help="Dataset split to audit."
)
parser.add_argument(
"--json", action="store_true", help="Print report in deterministic JSON format."
)
args = parser.parse_args()
if not args.synthetic_smoke and not args.dataset:
print(
"Error: Either --synthetic-smoke or --dataset <name> must be specified.",
file=sys.stderr,
)
sys.exit(1)
if args.dataset:
# In PR-2, no real dataset adapters exist, so we fail with the required error code
print("Error: dataset_adapter_unavailable", file=sys.stderr)
sys.exit(1)
if args.synthetic_smoke:
# Generate synthetic items and run smoke audit
items = (
GeneralizationAuditItem(
dataset="SYNTHETIC_SMOKE",
split="test",
item_id="item_1",
prompt_ref="synthetic:smoke:item_1",
answer_kind="numeric_text",
metadata=(("difficulty", "easy"),),
),
GeneralizationAuditItem(
dataset="SYNTHETIC_SMOKE",
split="test",
item_id="item_2",
prompt_ref="synthetic:smoke:item_2",
answer_kind="numeric_text",
metadata=(("difficulty", "medium"),),
),
GeneralizationAuditItem(
dataset="SYNTHETIC_SMOKE",
split="test",
item_id="item_3",
prompt_ref="synthetic:smoke:item_3",
answer_kind="numeric_text",
metadata=(("difficulty", "hard"),),
),
)
def synthetic_evaluator(
item: GeneralizationAuditItem,
) -> GeneralizationAuditOutcome:
if item.item_id == "item_1":
return GeneralizationAuditOutcome(
item_id=item.item_id,
disposition="correct",
residual_kinds=("none",),
candidate_attempt_count=1,
binding_failure_count=0,
replay_refusal_count=0,
sealed_trace_dispositions=("success",),
reason_codes=(),
)
elif item.item_id == "item_2":
return GeneralizationAuditOutcome(
item_id=item.item_id,
disposition="wrong",
residual_kinds=("numeric_precision",),
candidate_attempt_count=2,
binding_failure_count=0,
replay_refusal_count=0,
sealed_trace_dispositions=("fail", "success"),
reason_codes=("wrong_value",),
)
else:
return GeneralizationAuditOutcome(
item_id=item.item_id,
disposition="refused",
residual_kinds=(),
candidate_attempt_count=1,
binding_failure_count=1,
replay_refusal_count=1,
sealed_trace_dispositions=("refused",),
reason_codes=("safety_policy",),
)
try:
report = run_generalization_audit(
dataset="SYNTHETIC_SMOKE",
split="test",
items=items,
evaluator=synthetic_evaluator,
)
except Exception as exc:
print(f"Audit Failed: {exc}", file=sys.stderr)
sys.exit(1)
if args.json:
print(json.dumps(asdict(report), indent=2, sort_keys=True))
else:
print(
f"Generalization Audit Report (Policy: {report.policy_version})"
)
print("=" * 80)
print(f"Dataset: {report.dataset}")
print(f"Split: {report.split}")
print(f"Total Items: {report.n_items}")
print(f"Correct: {report.correct}")
print(f"Wrong: {report.wrong}")
print(f"Refused: {report.refused}")
print(f"Unsupported: {report.unsupported}")
print(f"Candidate Attempts: {report.candidate_attempts}")
print(f"Binding Failures: {report.binding_failures}")
print(f"Replay Refusals: {report.replay_refusals}")
print(f"Sealed Trace Dispositions: {report.sealed_trace_dispositions}")
print(f"Dominant Residual Kinds: {report.dominant_residual_kinds}")
print(f"Reason Codes: {', '.join(report.reason_codes)}")
print("=" * 80)
sys.exit(0)
if __name__ == "__main__":
main()

View file

@ -0,0 +1,277 @@
"""Tests for the generalization audit runner skeleton."""
from __future__ import annotations
import json
import subprocess
import sys
import pytest
from evals.generalization.audit_runner import run_generalization_audit
from evals.generalization.item_schema import (
GENERALIZATION_AUDIT_RUNNER_POLICY_VERSION,
GeneralizationAuditItem,
GeneralizationAuditOutcome,
)
def test_aggregates_counts_and_metrics() -> None:
"""Ensure run_generalization_audit aggregates dispositions, counts, and reasons correctly."""
items = (
GeneralizationAuditItem(
dataset="TEST_DATA",
split="test",
item_id="id_1",
prompt_ref="test:test:id_1",
answer_kind="numeric",
metadata=(("tag", "a"),),
),
GeneralizationAuditItem(
dataset="TEST_DATA",
split="test",
item_id="id_2",
prompt_ref="test:test:id_2",
answer_kind="numeric",
metadata=(("tag", "b"),),
),
GeneralizationAuditItem(
dataset="TEST_DATA",
split="test",
item_id="id_3",
prompt_ref="test:test:id_3",
answer_kind="numeric",
metadata=(("tag", "c"),),
),
GeneralizationAuditItem(
dataset="TEST_DATA",
split="test",
item_id="id_4",
prompt_ref="test:test:id_4",
answer_kind="numeric",
metadata=(("tag", "d"),),
),
)
def mock_evaluator(
item: GeneralizationAuditItem,
) -> GeneralizationAuditOutcome:
if item.item_id == "id_1":
return GeneralizationAuditOutcome(
item_id=item.item_id,
disposition="correct",
residual_kinds=("none",),
candidate_attempt_count=1,
binding_failure_count=0,
replay_refusal_count=0,
sealed_trace_dispositions=("success",),
reason_codes=(),
)
elif item.item_id == "id_2":
return GeneralizationAuditOutcome(
item_id=item.item_id,
disposition="wrong",
residual_kinds=("precision",),
candidate_attempt_count=3,
binding_failure_count=0,
replay_refusal_count=0,
sealed_trace_dispositions=("fail", "fail", "success"),
reason_codes=("precision_error",),
)
elif item.item_id == "id_3":
return GeneralizationAuditOutcome(
item_id=item.item_id,
disposition="refused",
residual_kinds=(),
candidate_attempt_count=1,
binding_failure_count=1,
replay_refusal_count=1,
sealed_trace_dispositions=("refused",),
reason_codes=("safety_policy",),
)
else:
return GeneralizationAuditOutcome(
item_id=item.item_id,
disposition="unsupported",
residual_kinds=(),
candidate_attempt_count=0,
binding_failure_count=0,
replay_refusal_count=0,
sealed_trace_dispositions=(),
reason_codes=("unsupported_format",),
)
report = run_generalization_audit(
dataset="TEST_DATA",
split="test",
items=items,
evaluator=mock_evaluator,
)
# Core disposition counts
assert report.correct == 1
assert report.wrong == 1
assert report.refused == 1
assert report.unsupported == 1
assert report.n_items == 4
# Direct additions of metrics
assert report.candidate_attempts == 5 # 1 + 3 + 1 + 0
assert report.binding_failures == 1 # 0 + 0 + 1 + 0
assert report.replay_refusals == 1 # 0 + 0 + 1 + 0
# Histograms
# "success" -> 2 times, "fail" -> 2 times, "refused" -> 1 time
# Sorted by count descending, then by key ascending:
# (('fail', 2), ('success', 2), ('refused', 1))
assert report.sealed_trace_dispositions == (
("fail", 2),
("success", 2),
("refused", 1),
)
# residual kinds: "none" -> 1, "precision" -> 1
# Sorted by count descending, then key:
# (('none', 1), ('precision', 1))
assert report.dominant_residual_kinds == (
("none", 1),
("precision", 1),
)
# Reason codes: sorted union of all codes
assert report.reason_codes == (
"precision_error",
"safety_policy",
"unsupported_format",
)
def test_empty_item_set_refuses() -> None:
"""Ensure run_generalization_audit refuses (raises ValueError) if items tuple is empty."""
with pytest.raises(ValueError, match="requires a non-empty sequence"):
run_generalization_audit(
dataset="TEST_DATA",
split="test",
items=(),
evaluator=lambda x: None,
)
def test_evaluator_exception_fail_closed() -> None:
"""Ensure run_generalization_audit catches evaluator exceptions and fails closed."""
items = (
GeneralizationAuditItem(
dataset="TEST_DATA",
split="test",
item_id="id_1",
prompt_ref="test:test:id_1",
answer_kind="numeric",
metadata=(),
),
)
def exploding_evaluator(
item: GeneralizationAuditItem,
) -> GeneralizationAuditOutcome:
raise RuntimeError("Something exploded!")
report = run_generalization_audit(
dataset="TEST_DATA",
split="test",
items=items,
evaluator=exploding_evaluator,
)
assert report.n_items == 1
assert report.refused == 1
assert report.correct == 0
assert report.wrong == 0
assert "evaluator_exception" in report.reason_codes
assert "RuntimeError" in report.reason_codes
def test_json_output_deterministic_and_no_raw_prompts() -> None:
"""Ensure serialized report is deterministic and does not contain raw prompt/answer fields."""
items = (
GeneralizationAuditItem(
dataset="TEST_DATA",
split="test",
item_id="id_1",
prompt_ref="test:test:id_1",
answer_kind="numeric",
metadata=(),
),
)
report = run_generalization_audit(
dataset="TEST_DATA",
split="test",
items=items,
evaluator=lambda x: GeneralizationAuditOutcome(
item_id=x.item_id,
disposition="correct",
residual_kinds=(),
candidate_attempt_count=1,
binding_failure_count=0,
replay_refusal_count=0,
sealed_trace_dispositions=(),
reason_codes=(),
),
)
from dataclasses import asdict
report_dict = asdict(report)
# 1. Deterministic check
json1 = json.dumps(report_dict, indent=2, sort_keys=True)
json2 = json.dumps(report_dict, indent=2, sort_keys=True)
assert json1 == json2
# 2. Check no raw prompt/answer leakages in the report
serialized_str = json1.lower()
assert "prompt" not in report_dict
assert "answer" not in report_dict
# Check that common prompt/answer related strings are absent
assert "raw_prompt" not in serialized_str
assert "raw_answer" not in serialized_str
assert "chain_of_thought" not in serialized_str
assert "example_text" not in serialized_str
def test_cli_synthetic_smoke() -> None:
"""Verify that --synthetic-smoke runs and produces expected report structure."""
result = subprocess.run(
[
sys.executable,
"scripts/benchmarks/run_generalization_audit.py",
"--synthetic-smoke",
"--json",
],
capture_output=True,
text=True,
check=True,
)
assert result.returncode == 0
report = json.loads(result.stdout)
assert report["dataset"] == "SYNTHETIC_SMOKE"
assert report["policy_version"] == GENERALIZATION_AUDIT_RUNNER_POLICY_VERSION
assert report["n_items"] == 3
assert report["correct"] == 1
assert report["wrong"] == 1
assert report["refused"] == 1
def test_cli_real_dataset_refuses() -> None:
"""Verify that requesting a real dataset fails with dataset_adapter_unavailable."""
result = subprocess.run(
[
sys.executable,
"scripts/benchmarks/run_generalization_audit.py",
"--dataset",
"gsm1k",
],
capture_output=True,
text=True,
)
assert result.returncode != 0
assert "dataset_adapter_unavailable" in result.stderr