"""Deterministic calibration audit for declared LLM-judge observations. The fixture runs no model and validates no provider. It measures only the supplied labels and observations. A passing result is not evidence that a judge is unbiased, equivalent to a human, stable on another task distribution, or safe to use without review. Production use requires governed human labels, blinded sampling, repeated live measurements, privacy controls, and escalation. """ from __future__ import annotations from dataclasses import asdict, dataclass from hashlib import sha256 import json import re from typing import Literal Candidate = Literal["a", "b", "tie"] Order = Literal["ab", "ba"] Decision = Literal["PASS_CALIBRATION", "FAIL_CALIBRATION"] ID_PATTERN = re.compile(r"^[a-z][a-z0-9._:-]{2,95}$") DIGEST_PATTERN = re.compile(r"^[0-9a-f]{64}$") def _exact_text(name: str, value: object, *, maximum: int = 512) -> str: if type(value) is not str or not value.strip(): raise ValueError(f"{name} must be non-empty exact str") if value != value.strip(): raise ValueError(f"{name} must not have surrounding whitespace") if len(value) > maximum or any(ord(character) < 32 for character in value): raise ValueError(f"{name} is outside the bounded text contract") return value def _canonical_id(name: str, value: object) -> str: text = _exact_text(name, value, maximum=96) if not ID_PATTERN.fullmatch(text): raise ValueError(f"{name} must be a canonical identifier") return text def _exact_bool(name: str, value: object) -> bool: if type(value) is not bool: raise ValueError(f"{name} must be exact bool") return value def _exact_int(name: str, value: object, *, minimum: int, maximum: int) -> int: if type(value) is not int: raise ValueError(f"{name} must be exact int") if not minimum <= value <= maximum: raise ValueError(f"{name} must be between {minimum} and {maximum}") return value def _choice(name: str, value: object) -> Candidate: text = _exact_text(name, value, maximum=8) if text not in {"a", "b", "tie"}: raise ValueError(f"{name} must be a, b, or tie") return text # type: ignore[return-value] def _stable_digest(value: object) -> str: encoded = json.dumps( value, allow_nan=False, ensure_ascii=True, separators=(",", ":"), sort_keys=True, ).encode("utf-8") return sha256(encoded).hexdigest() def _rate_bps(numerator: int, denominator: int) -> int: if denominator == 0: return 0 return (numerator * 10_000 + denominator // 2) // denominator @dataclass(frozen=True) class HumanReference: case_id: str preferred_candidate: Candidate reviewer_ids: tuple[str, ...] adjudicated: bool def __post_init__(self) -> None: _canonical_id("reference.case_id", self.case_id) object.__setattr__( self, "preferred_candidate", _choice("reference.preferred_candidate", self.preferred_candidate), ) try: reviewers = tuple(self.reviewer_ids) except TypeError as error: raise ValueError("reference.reviewer_ids must be iterable") from error object.__setattr__(self, "reviewer_ids", reviewers) if not reviewers: raise ValueError("reference.reviewer_ids must not be empty") for reviewer in reviewers: _canonical_id("reference.reviewer_id", reviewer) if len(reviewers) != len(set(reviewers)): raise ValueError("reference.reviewer_ids must be unique") if tuple(sorted(reviewers)) != reviewers: raise ValueError("reference.reviewer_ids must be sorted") _exact_bool("reference.adjudicated", self.adjudicated) @dataclass(frozen=True) class JudgeObservation: case_id: str order: Order repetition: int selected_candidate: Candidate candidate_a_tokens: int candidate_b_tokens: int candidate_a_generator: str candidate_b_generator: str judge_generator: str def __post_init__(self) -> None: _canonical_id("observation.case_id", self.case_id) order = _exact_text("observation.order", self.order, maximum=2) if order not in {"ab", "ba"}: raise ValueError("observation.order must be ab or ba") _exact_int("observation.repetition", self.repetition, minimum=1, maximum=100) object.__setattr__( self, "selected_candidate", _choice("observation.selected_candidate", self.selected_candidate), ) _exact_int( "observation.candidate_a_tokens", self.candidate_a_tokens, minimum=1, maximum=1_000_000, ) _exact_int( "observation.candidate_b_tokens", self.candidate_b_tokens, minimum=1, maximum=1_000_000, ) _canonical_id("observation.candidate_a_generator", self.candidate_a_generator) _canonical_id("observation.candidate_b_generator", self.candidate_b_generator) _canonical_id("observation.judge_generator", self.judge_generator) @dataclass(frozen=True) class JudgeCalibrationContract: version: str scope: str judge_generator: str repetitions_per_order: int minimum_cases: int minimum_reviewers: int minimum_human_agreement_bps: int maximum_position_flip_bps: int maximum_repeat_variance_bps: int maximum_verbosity_bias_bps: int maximum_self_preference_bps: int def __post_init__(self) -> None: _canonical_id("contract.version", self.version) _canonical_id("contract.scope", self.scope) _canonical_id("contract.judge_generator", self.judge_generator) _exact_int( "contract.repetitions_per_order", self.repetitions_per_order, minimum=2, maximum=20, ) _exact_int("contract.minimum_cases", self.minimum_cases, minimum=1, maximum=100_000) _exact_int( "contract.minimum_reviewers", self.minimum_reviewers, minimum=1, maximum=100, ) for name in ( "minimum_human_agreement_bps", "maximum_position_flip_bps", "maximum_repeat_variance_bps", "maximum_verbosity_bias_bps", "maximum_self_preference_bps", ): _exact_int(f"contract.{name}", getattr(self, name), minimum=0, maximum=10_000) @property def content_id(self) -> str: return "judge-calibration@sha256:" + _stable_digest(asdict(self)) @dataclass(frozen=True) class JudgeCalibrationSnapshot: scope: str contract_content_id: str rubric_digest: str references: tuple[HumanReference, ...] observations: tuple[JudgeObservation, ...] def __post_init__(self) -> None: _canonical_id("snapshot.scope", self.scope) content_id = _exact_text( "snapshot.contract_content_id", self.contract_content_id, maximum=96, ) if not re.fullmatch(r"judge-calibration@sha256:[0-9a-f]{64}", content_id): raise ValueError("snapshot.contract_content_id must be a judge calibration digest") digest = _exact_text("snapshot.rubric_digest", self.rubric_digest, maximum=64) if not DIGEST_PATTERN.fullmatch(digest): raise ValueError("snapshot.rubric_digest must be lowercase SHA-256") try: references = tuple(self.references) observations = tuple(self.observations) except TypeError as error: raise ValueError("snapshot collections must be iterable") from error object.__setattr__(self, "references", references) object.__setattr__(self, "observations", observations) if not references or any(type(item) is not HumanReference for item in references): raise ValueError("references must contain exact HumanReference records") if not observations or any(type(item) is not JudgeObservation for item in observations): raise ValueError("observations must contain exact JudgeObservation records") case_ids = [item.case_id for item in references] if len(case_ids) != len(set(case_ids)): raise ValueError("reference case IDs must be unique") keys = [(item.case_id, item.order, item.repetition) for item in observations] if len(keys) != len(set(keys)): raise ValueError("observation keys must be unique") @dataclass(frozen=True) class CalibrationReport: contract: JudgeCalibrationContract snapshot: JudgeCalibrationSnapshot decision: Decision human_agreement_bps: int position_flip_bps: int repeat_variance_bps: int verbosity_bias_bps: int self_preference_bps: int reference_issues: tuple[str, ...] def _rebuild_reference(value: object) -> HumanReference: if type(value) is not HumanReference: raise ValueError("references must contain exact HumanReference records") return HumanReference( value.case_id, value.preferred_candidate, value.reviewer_ids, value.adjudicated, ) def _rebuild_observation(value: object) -> JudgeObservation: if type(value) is not JudgeObservation: raise ValueError("observations must contain exact JudgeObservation records") return JudgeObservation( value.case_id, value.order, value.repetition, value.selected_candidate, value.candidate_a_tokens, value.candidate_b_tokens, value.candidate_a_generator, value.candidate_b_generator, value.judge_generator, ) def _rebuild_contract(value: object) -> JudgeCalibrationContract: if type(value) is not JudgeCalibrationContract: raise ValueError("contract must be exact JudgeCalibrationContract") try: return JudgeCalibrationContract( value.version, value.scope, value.judge_generator, value.repetitions_per_order, value.minimum_cases, value.minimum_reviewers, value.minimum_human_agreement_bps, value.maximum_position_flip_bps, value.maximum_repeat_variance_bps, value.maximum_verbosity_bias_bps, value.maximum_self_preference_bps, ) except (AttributeError, TypeError) as error: raise ValueError("contract failed reconstruction") from error def _rebuild_snapshot(value: object) -> JudgeCalibrationSnapshot: if type(value) is not JudgeCalibrationSnapshot: raise ValueError("snapshot must be exact JudgeCalibrationSnapshot") try: references = tuple(_rebuild_reference(item) for item in value.references) observations = tuple(_rebuild_observation(item) for item in value.observations) return JudgeCalibrationSnapshot( value.scope, value.contract_content_id, value.rubric_digest, references, observations, ) except (AttributeError, TypeError) as error: raise ValueError("snapshot failed reconstruction") from error def audit_calibration(contract: object, snapshot: object) -> CalibrationReport: """Measure declared judge behavior without calling a judge or mutating inputs.""" checked_contract = _rebuild_contract(contract) checked_snapshot = _rebuild_snapshot(snapshot) if checked_snapshot.scope != checked_contract.scope: raise ValueError("snapshot belongs to another scope") if checked_snapshot.contract_content_id != checked_contract.content_id: raise ValueError("snapshot is not bound to this contract content") references = {item.case_id: item for item in checked_snapshot.references} if len(references) < checked_contract.minimum_cases: raise ValueError("snapshot has fewer cases than the calibration contract") grouped: dict[tuple[str, Order], list[JudgeObservation]] = {} metadata: dict[str, tuple[int, int, str, str]] = {} for observation in checked_snapshot.observations: if observation.case_id not in references: raise ValueError("observation has no human reference") if observation.judge_generator != checked_contract.judge_generator: raise ValueError("observation uses another judge generator") grouped.setdefault((observation.case_id, observation.order), []).append(observation) current = ( observation.candidate_a_tokens, observation.candidate_b_tokens, observation.candidate_a_generator, observation.candidate_b_generator, ) previous = metadata.setdefault(observation.case_id, current) if previous != current: raise ValueError("candidate metadata changes across observations") expected_repetitions = set(range(1, checked_contract.repetitions_per_order + 1)) for case_id in references: for order in ("ab", "ba"): observations = grouped.get((case_id, order), []) if {item.repetition for item in observations} != expected_repetitions: raise ValueError("each case requires every declared order and repetition") reference_issues: list[str] = [] for reference in checked_snapshot.references: if len(reference.reviewer_ids) < checked_contract.minimum_reviewers: reference_issues.append(f"reviewers:{reference.case_id}") if not reference.adjudicated: reference_issues.append(f"adjudication:{reference.case_id}") agreement = sum( observation.selected_candidate == references[observation.case_id].preferred_candidate for observation in checked_snapshot.observations ) human_agreement_bps = _rate_bps(agreement, len(checked_snapshot.observations)) repeat_variance = 0 order_verdicts: dict[tuple[str, Order], Candidate | None] = {} for key, observations in grouped.items(): verdicts = {item.selected_candidate for item in observations} if len(verdicts) > 1: repeat_variance += 1 order_verdicts[key] = None else: order_verdicts[key] = next(iter(verdicts)) repeat_variance_bps = _rate_bps(repeat_variance, len(grouped)) position_flips = 0 position_opportunities = 0 for case_id in references: ab = order_verdicts[(case_id, "ab")] ba = order_verdicts[(case_id, "ba")] if ab is not None and ba is not None: position_opportunities += 1 position_flips += ab != ba position_flip_bps = _rate_bps(position_flips, position_opportunities) verbosity_errors = 0 verbosity_opportunities = 0 self_errors = 0 self_opportunities = 0 for observation in checked_snapshot.observations: preferred = references[observation.case_id].preferred_candidate a_tokens, b_tokens, a_generator, b_generator = metadata[observation.case_id] if preferred == "a" and a_tokens < b_tokens: verbosity_opportunities += 1 verbosity_errors += observation.selected_candidate == "b" elif preferred == "b" and b_tokens < a_tokens: verbosity_opportunities += 1 verbosity_errors += observation.selected_candidate == "a" self_candidate: Candidate | None = None if a_generator == checked_contract.judge_generator and b_generator != a_generator: self_candidate = "a" elif b_generator == checked_contract.judge_generator and a_generator != b_generator: self_candidate = "b" if self_candidate is not None and preferred not in {self_candidate, "tie"}: self_opportunities += 1 self_errors += observation.selected_candidate == self_candidate verbosity_bias_bps = _rate_bps(verbosity_errors, verbosity_opportunities) self_preference_bps = _rate_bps(self_errors, self_opportunities) failed = bool(reference_issues) or ( human_agreement_bps < checked_contract.minimum_human_agreement_bps or position_flip_bps > checked_contract.maximum_position_flip_bps or repeat_variance_bps > checked_contract.maximum_repeat_variance_bps or verbosity_bias_bps > checked_contract.maximum_verbosity_bias_bps or self_preference_bps > checked_contract.maximum_self_preference_bps ) decision: Decision = "FAIL_CALIBRATION" if failed else "PASS_CALIBRATION" return CalibrationReport( checked_contract, checked_snapshot, decision, human_agreement_bps, position_flip_bps, repeat_variance_bps, verbosity_bias_bps, self_preference_bps, tuple(sorted(reference_issues)), ) def _observations( case_id: str, preferred: Candidate, a_tokens: int, b_tokens: int, a_generator: str, b_generator: str, ) -> tuple[JudgeObservation, ...]: return tuple( JudgeObservation( case_id, order, # type: ignore[arg-type] repetition, preferred, a_tokens, b_tokens, a_generator, b_generator, "judge-v1", ) for order in ("ab", "ba") for repetition in (1, 2) ) ILLUSTRATIVE_CONTRACT = JudgeCalibrationContract( version="judge-calibration-v1", scope="support-answer-comparison", judge_generator="judge-v1", repetitions_per_order=2, minimum_cases=4, minimum_reviewers=2, minimum_human_agreement_bps=9000, maximum_position_flip_bps=500, maximum_repeat_variance_bps=500, maximum_verbosity_bias_bps=500, maximum_self_preference_bps=500, ) ILLUSTRATIVE_REFERENCES = ( HumanReference("case-accuracy", "a", ("reviewer:domain", "reviewer:quality"), True), HumanReference("case-refusal", "b", ("reviewer:policy", "reviewer:quality"), True), HumanReference("case-self", "a", ("reviewer:domain", "reviewer:quality"), True), HumanReference("case-tie", "tie", ("reviewer:domain", "reviewer:quality"), True), ) ILLUSTRATIVE_OBSERVATIONS = ( *_observations("case-accuracy", "a", 80, 160, "candidate-a", "candidate-b"), *_observations("case-refusal", "b", 150, 70, "candidate-a", "candidate-b"), *_observations("case-self", "a", 90, 110, "candidate-a", "judge-v1"), *_observations("case-tie", "tie", 100, 100, "candidate-a", "candidate-b"), ) ILLUSTRATIVE_SNAPSHOT = JudgeCalibrationSnapshot( scope="support-answer-comparison", contract_content_id=ILLUSTRATIVE_CONTRACT.content_id, rubric_digest=sha256(b"illustrative-support-rubric-v1").hexdigest(), references=ILLUSTRATIVE_REFERENCES, observations=ILLUSTRATIVE_OBSERVATIONS, ) def format_example() -> str: report = audit_calibration(ILLUSTRATIVE_CONTRACT, ILLUSTRATIVE_SNAPSHOT) return "\n".join( ( "example=illustrative_only", f"calibration_id={report.contract.content_id}", f"cases={len(report.snapshot.references)};observations={len(report.snapshot.observations)}", f"human_agreement_bps={report.human_agreement_bps}", f"position_flip_bps={report.position_flip_bps};repeat_variance_bps={report.repeat_variance_bps}", f"verbosity_bias_bps={report.verbosity_bias_bps};self_preference_bps={report.self_preference_bps}", f"decision={report.decision}", "claim=LOCAL_DECLARED_LABEL_AUDIT_NOT_JUDGE_VALIDATION", ) ) def main() -> None: print(format_example()) if __name__ == "__main__": main()