"""Deterministic coverage audit for an illustrative golden evaluation dataset. This local fixture validates declared records only. It does not collect user data, inspect training corpora, prove that an exclusion attestation is true, measure model quality, or certify that a dataset is representative, safe, or fit for a production release. Those claims require governed provenance, domain review, privacy controls, deployment evidence, and continuing audits. """ from __future__ import annotations from dataclasses import asdict, dataclass from hashlib import sha256 import json import re from typing import Literal CaseKind = Literal["hard_negative", "long_tail", "policy_boundary", "real_task"] Decision = Literal["PASS_COVERAGE", "FAIL_COVERAGE"] ID_PATTERN = re.compile(r"^[a-z][a-z0-9._:-]{2,95}$") SLICE_PATTERN = re.compile(r"^[a-z][a-z0-9_-]{1,63}$") DIGEST_PATTERN = re.compile(r"^[0-9a-f]{64}$") ALLOWED_KINDS = ("hard_negative", "long_tail", "policy_boundary", "real_task") 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 _slice_name(name: str, value: object) -> str: text = _exact_text(name, value, maximum=64) if not SLICE_PATTERN.fullmatch(text): raise ValueError(f"{name} must be a canonical slice name") return text def _digest(name: str, value: object) -> str: text = _exact_text(name, value, maximum=64) if not DIGEST_PATTERN.fullmatch(text): raise ValueError(f"{name} must be lowercase SHA-256") 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 _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 _checked_kinds(name: str, value: object) -> tuple[CaseKind, ...]: try: kinds = tuple(value) # type: ignore[arg-type] except TypeError as error: raise ValueError(f"{name} must be iterable") from error if not kinds: raise ValueError(f"{name} must not be empty") for kind in kinds: _exact_text(f"{name} entry", kind, maximum=32) if kind not in ALLOWED_KINDS: raise ValueError(f"{name} contains an unsupported case kind") if len(kinds) != len(set(kinds)): raise ValueError(f"{name} must be unique") if tuple(sorted(kinds)) != kinds: raise ValueError(f"{name} must be sorted") return kinds # type: ignore[return-value] @dataclass(frozen=True) class CoverageRequirement: slice_name: str minimum_cases: int required_kinds: tuple[CaseKind, ...] def __post_init__(self) -> None: _slice_name("requirement.slice_name", self.slice_name) _exact_int("requirement.minimum_cases", self.minimum_cases, minimum=1, maximum=10_000) object.__setattr__( self, "required_kinds", _checked_kinds("requirement.required_kinds", self.required_kinds), ) @dataclass(frozen=True) class GoldenCase: case_id: str slice_name: str kind: CaseKind input_digest: str expected_behavior_digest: str provenance_ref: str reviewer_ids: tuple[str, ...] adjudicated: bool training_exclusion_attested: bool def __post_init__(self) -> None: _canonical_id("case.case_id", self.case_id) _slice_name("case.slice_name", self.slice_name) _exact_text("case.kind", self.kind, maximum=32) if self.kind not in ALLOWED_KINDS: raise ValueError("case.kind is unsupported") _digest("case.input_digest", self.input_digest) _digest("case.expected_behavior_digest", self.expected_behavior_digest) _canonical_id("case.provenance_ref", self.provenance_ref) try: reviewers = tuple(self.reviewer_ids) except TypeError as error: raise ValueError("case.reviewer_ids must be iterable") from error object.__setattr__(self, "reviewer_ids", reviewers) if not reviewers: raise ValueError("case.reviewer_ids must not be empty") for reviewer in reviewers: _canonical_id("case.reviewer_id", reviewer) if len(reviewers) != len(set(reviewers)): raise ValueError("case.reviewer_ids must be unique") if tuple(sorted(reviewers)) != reviewers: raise ValueError("case.reviewer_ids must be sorted") _exact_bool("case.adjudicated", self.adjudicated) _exact_bool( "case.training_exclusion_attested", self.training_exclusion_attested, ) @dataclass(frozen=True) class GoldenDatasetContract: version: str scope: str requirements: tuple[CoverageRequirement, ...] required_kinds: tuple[CaseKind, ...] minimum_total_cases: int max_slice_share_bps: int minimum_reviewers: int def __post_init__(self) -> None: _canonical_id("contract.version", self.version) _canonical_id("contract.scope", self.scope) try: requirements = tuple(self.requirements) except TypeError as error: raise ValueError("contract.requirements must be iterable") from error object.__setattr__(self, "requirements", requirements) if not requirements or any(type(item) is not CoverageRequirement for item in requirements): raise ValueError("requirements must contain exact CoverageRequirement records") names = [item.slice_name for item in requirements] if len(names) != len(set(names)): raise ValueError("requirement slice names must be unique") if tuple(sorted(names)) != tuple(names): raise ValueError("requirements must be sorted by slice_name") object.__setattr__( self, "required_kinds", _checked_kinds("contract.required_kinds", self.required_kinds), ) _exact_int( "contract.minimum_total_cases", self.minimum_total_cases, minimum=1, maximum=1_000_000, ) _exact_int( "contract.max_slice_share_bps", self.max_slice_share_bps, minimum=1, maximum=10_000, ) _exact_int( "contract.minimum_reviewers", self.minimum_reviewers, minimum=1, maximum=100, ) @property def content_id(self) -> str: return "golden-dataset@sha256:" + _stable_digest(asdict(self)) @dataclass(frozen=True) class GoldenDatasetSnapshot: scope: str contract_content_id: str dataset_version: str cases: tuple[GoldenCase, ...] 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"golden-dataset@sha256:[0-9a-f]{64}", content_id): raise ValueError("snapshot.contract_content_id must be a golden dataset digest") _canonical_id("snapshot.dataset_version", self.dataset_version) try: cases = tuple(self.cases) except TypeError as error: raise ValueError("snapshot.cases must be iterable") from error object.__setattr__(self, "cases", cases) if not cases or any(type(item) is not GoldenCase for item in cases): raise ValueError("cases must contain exact GoldenCase records") case_ids = [item.case_id for item in cases] input_digests = [item.input_digest for item in cases] if len(case_ids) != len(set(case_ids)): raise ValueError("case IDs must be unique") if len(input_digests) != len(set(input_digests)): raise ValueError("case input digests must be unique") @dataclass(frozen=True) class CoverageReport: contract: GoldenDatasetContract snapshot: GoldenDatasetSnapshot decision: Decision covered_kinds: tuple[CaseKind, ...] missing_requirements: tuple[str, ...] slice_counts: tuple[tuple[str, int], ...] def _rebuild_requirement(value: object) -> CoverageRequirement: if type(value) is not CoverageRequirement: raise ValueError("requirements must contain exact CoverageRequirement records") return CoverageRequirement(value.slice_name, value.minimum_cases, value.required_kinds) def _rebuild_case(value: object) -> GoldenCase: if type(value) is not GoldenCase: raise ValueError("cases must contain exact GoldenCase records") return GoldenCase( value.case_id, value.slice_name, value.kind, value.input_digest, value.expected_behavior_digest, value.provenance_ref, value.reviewer_ids, value.adjudicated, value.training_exclusion_attested, ) def _rebuild_contract(value: object) -> GoldenDatasetContract: if type(value) is not GoldenDatasetContract: raise ValueError("contract must be exact GoldenDatasetContract") try: requirements = tuple(_rebuild_requirement(item) for item in value.requirements) return GoldenDatasetContract( value.version, value.scope, requirements, value.required_kinds, value.minimum_total_cases, value.max_slice_share_bps, value.minimum_reviewers, ) except (AttributeError, TypeError) as error: raise ValueError("contract failed reconstruction") from error def _rebuild_snapshot(value: object) -> GoldenDatasetSnapshot: if type(value) is not GoldenDatasetSnapshot: raise ValueError("snapshot must be exact GoldenDatasetSnapshot") try: cases = tuple(_rebuild_case(item) for item in value.cases) return GoldenDatasetSnapshot( value.scope, value.contract_content_id, value.dataset_version, cases, ) except (AttributeError, TypeError) as error: raise ValueError("snapshot failed reconstruction") from error def audit_coverage(contract: object, snapshot: object) -> CoverageReport: """Audit declared coverage without running a model or changing the dataset.""" 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") counts: dict[str, int] = { requirement.slice_name: 0 for requirement in checked_contract.requirements } kinds_by_slice: dict[str, set[CaseKind]] = { requirement.slice_name: set() for requirement in checked_contract.requirements } missing: list[str] = [] for case in checked_snapshot.cases: if case.slice_name not in counts: raise ValueError(f"case {case.case_id} uses an undeclared slice") counts[case.slice_name] += 1 kinds_by_slice[case.slice_name].add(case.kind) if len(case.reviewer_ids) < checked_contract.minimum_reviewers: missing.append(f"reviewers:{case.case_id}") if not case.adjudicated: missing.append(f"adjudication:{case.case_id}") if not case.training_exclusion_attested: missing.append(f"training-exclusion:{case.case_id}") for requirement in checked_contract.requirements: if counts[requirement.slice_name] < requirement.minimum_cases: missing.append(f"minimum:{requirement.slice_name}") absent = set(requirement.required_kinds) - kinds_by_slice[requirement.slice_name] for kind in sorted(absent): missing.append(f"kind:{requirement.slice_name}:{kind}") covered_kinds = tuple(sorted({case.kind for case in checked_snapshot.cases})) for kind in checked_contract.required_kinds: if kind not in covered_kinds: missing.append(f"global-kind:{kind}") total = len(checked_snapshot.cases) if total < checked_contract.minimum_total_cases: missing.append("minimum:total") largest_slice = max(counts.values()) if largest_slice * 10_000 > total * checked_contract.max_slice_share_bps: missing.append("concentration:slice") missing_requirements = tuple(sorted(set(missing))) decision: Decision = "FAIL_COVERAGE" if missing_requirements else "PASS_COVERAGE" return CoverageReport( checked_contract, checked_snapshot, decision, covered_kinds, missing_requirements, tuple(sorted(counts.items())), ) def _hex(label: str) -> str: return sha256(label.encode("utf-8")).hexdigest() ILLUSTRATIVE_CONTRACT = GoldenDatasetContract( version="golden-dataset-v1", scope="support-assistant-eval", requirements=( CoverageRequirement("billing", 2, ("hard_negative", "real_task")), CoverageRequirement("policy", 2, ("policy_boundary", "real_task")), CoverageRequirement("returns", 2, ("long_tail", "real_task")), CoverageRequirement("technical", 2, ("hard_negative", "long_tail")), ), required_kinds=ALLOWED_KINDS, minimum_total_cases=8, max_slice_share_bps=2500, minimum_reviewers=2, ) def _case(case_id: str, slice_name: str, kind: CaseKind, reviewers: tuple[str, ...]) -> GoldenCase: return GoldenCase( case_id=case_id, slice_name=slice_name, kind=kind, input_digest=_hex(case_id + ":input"), expected_behavior_digest=_hex(case_id + ":expected"), provenance_ref="ticket:" + case_id, reviewer_ids=reviewers, adjudicated=True, training_exclusion_attested=True, ) ILLUSTRATIVE_CASES = ( _case("case-billing-hard", "billing", "hard_negative", ("reviewer:ops", "reviewer:support")), _case("case-billing-real", "billing", "real_task", ("reviewer:ops", "reviewer:support")), _case("case-policy-boundary", "policy", "policy_boundary", ("reviewer:legal", "reviewer:safety")), _case("case-policy-real", "policy", "real_task", ("reviewer:legal", "reviewer:safety")), _case("case-returns-tail", "returns", "long_tail", ("reviewer:ops", "reviewer:support")), _case("case-returns-real", "returns", "real_task", ("reviewer:ops", "reviewer:support")), _case("case-technical-hard", "technical", "hard_negative", ("reviewer:engineering", "reviewer:support")), _case("case-technical-tail", "technical", "long_tail", ("reviewer:engineering", "reviewer:support")), ) ILLUSTRATIVE_SNAPSHOT = GoldenDatasetSnapshot( scope="support-assistant-eval", contract_content_id=ILLUSTRATIVE_CONTRACT.content_id, dataset_version="support-golden-2026-09-22", cases=ILLUSTRATIVE_CASES, ) def format_example() -> str: report = audit_coverage(ILLUSTRATIVE_CONTRACT, ILLUSTRATIVE_SNAPSHOT) return "\n".join( ( "example=illustrative_only", f"dataset_id={report.contract.content_id}", f"cases={len(report.snapshot.cases)};slices={len(report.slice_counts)}", "coverage=" + ",".join(report.covered_kinds), f"decision={report.decision}", "claim=LOCAL_COVERAGE_AUDIT_NOT_DATASET_QUALITY_CERTIFICATION", ) ) def main() -> None: print(format_example()) if __name__ == "__main__": main()