"""Explicit structured-head capabilities, independent of model dependencies. Reference availability, training coverage and prediction availability are separate facts. A capability records which trained heads may be used; it does not certify accuracy. Missing checkpoint metadata must not be interpreted as all-enabled. """ from __future__ import annotations import math from collections.abc import Iterable, Mapping from dataclasses import dataclass from typing import Any @dataclass(frozen=True) class SafetyCapabilities: risk: bool = True presence: tuple[bool, bool, bool] = (True, True, True) links: tuple[bool, bool, bool] = (True, True, True) def __post_init__(self) -> None: if not isinstance(self.risk, bool): raise ValueError("risk capability must be boolean") for name in ("presence", "links"): values = getattr(self, name) if ( not isinstance(values, tuple) or len(values) != 3 or any(not isinstance(value, bool) for value in values) ): raise ValueError(f"{name} capabilities must be a tuple of three booleans") @property def token_indices(self) -> tuple[int, ...]: flags = (self.risk, *self.presence, *self.links) return tuple(index for index, enabled in enumerate(flags) if enabled) def as_dict(self) -> dict[str, Any]: return {"risk": self.risk, "presence": list(self.presence), "links": list(self.links)} @classmethod def unavailable(cls) -> SafetyCapabilities: return cls(False, (False, False, False), (False, False, False)) @classmethod def from_dict(cls, value: Mapping[str, Any]) -> SafetyCapabilities: if not isinstance(value, Mapping) or set(value) != {"risk", "presence", "links"}: raise ValueError("safety capabilities require exactly risk, presence, and links") if not isinstance(value["presence"], list | tuple) or not isinstance( value["links"], list | tuple ): raise ValueError("presence and links capabilities must be arrays") return cls(value["risk"], tuple(value["presence"]), tuple(value["links"])) def restrict(self, allowed: SafetyCapabilities) -> SafetyCapabilities: """Ablations can disable capabilities, never enable untrained heads.""" return SafetyCapabilities( self.risk and allowed.risk, tuple(a and b for a, b in zip(self.presence, allowed.presence, strict=True)), tuple(a and b for a, b in zip(self.links, allowed.links, strict=True)), ) ALL_SAFETY_HEADS = SafetyCapabilities() SSA_ABLATIONS = ("none", "no_ssa", "no_presence", "no_links", "no_risk") def apply_ablation(capabilities: SafetyCapabilities, name: str) -> SafetyCapabilities: """Disable named heads before token construction; never invent capability.""" if name not in SSA_ABLATIONS: raise ValueError(f"unknown SSA ablation {name!r}") mask = { "none": ALL_SAFETY_HEADS, "no_ssa": SafetyCapabilities.unavailable(), "no_presence": SafetyCapabilities(True, (False, False, False), (True, True, True)), "no_links": SafetyCapabilities(True, (True, True, True), (False, False, False)), "no_risk": SafetyCapabilities(False, (True, True, True), (True, True, True)), }[name] return capabilities.restrict(mask) def supervision_coverage(rows: Iterable[Mapping[str, Any]]) -> dict[str, Any]: """Count authoritative labels per image, without substituting missing values. Callers must provide unique training images and validated provenance. This function validates numeric domains, not the authority of the source itself. """ coverage = {"images": 0, "risk": 0, "presence": [0, 0, 0], "links": [0, 0, 0]} def observed(value: Any, *, binary: bool = False) -> bool: if value is None: return False if isinstance(value, bool) or not isinstance(value, int | float): raise ValueError("structured labels must be numeric or null") if not math.isfinite(value) or not 0 <= value <= 1: raise ValueError("structured labels must be finite and in [0, 1]") if binary and value not in (0, 1): raise ValueError("presence labels must be binary") return True for row in rows: if not isinstance(row, Mapping): raise ValueError("structured targets must be objects") coverage["images"] += 1 coverage["risk"] += int(observed(row.get("risk"))) for name in ("presence", "links"): values = row.get(name, [None, None, None]) if not isinstance(values, list | tuple) or len(values) != 3: raise ValueError(f"{name} targets must have three entries") for index, value in enumerate(values): coverage[name][index] += int(observed(value, binary=name == "presence")) return coverage def capabilities_from_coverage(coverage: Mapping[str, Any]) -> SafetyCapabilities: """An observed label enables training, not a claim of calibrated predictions.""" for name in ("images", "risk"): value = coverage.get(name) if isinstance(value, bool) or not isinstance(value, int) or value < 0: raise ValueError(f"{name} coverage must be a nonnegative integer") counts = [] for name in ("presence", "links"): values = coverage.get(name) if ( not isinstance(values, list | tuple) or len(values) != 3 or any( isinstance(value, bool) or not isinstance(value, int) or value < 0 for value in values ) ): raise ValueError(f"{name} coverage must contain three nonnegative integers") counts.extend(values) if any(value > coverage["images"] for value in [coverage["risk"], *counts]): raise ValueError("label coverage cannot exceed training image count") return SafetyCapabilities( coverage["risk"] > 0, tuple(value > 0 for value in coverage["presence"]), tuple(value > 0 for value in coverage["links"]), )