"""Illustrative release-evidence gate; declarations are not operational attestations.""" from dataclasses import asdict, dataclass, fields import hashlib import json import math import re import sys def identity(value): if type(value) is not str or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.:/@-]{0,127}", value): raise ValueError("invalid identity") def number(value, low, high): if type(value) not in (int, float) or not low <= value <= high or not math.isfinite(value): raise ValueError("finite non-boolean number required") if value != 0 and abs(value) < sys.float_info.min: raise ValueError("subnormal number") if not low <= value <= high: raise ValueError("numeric range") def count(value, low, high): if type(value) is not int or not low <= value <= high: raise ValueError("integer count range") def sequence(value, low, high): if type(value) not in (tuple, list): raise ValueError("bounded sequence required") count(len(value), low, high) return tuple(value) def digest(value): return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode()).hexdigest() def seal(record): if type(record.content_id) is not str: raise ValueError("content digest requires exact string") data = asdict(record) data.pop("content_id") return digest(data) def validate_record(record, cls): if type(record) is not cls: raise ValueError("concrete frozen record required") rebuilt = cls(**{f.name: getattr(record, f.name, None) for f in fields(cls)}) if record != rebuilt or record.content_id != rebuilt.content_id: raise ValueError("missing content digest or noncanonical record") def scope_tuple(value): result = sequence(value, 12, 12) for item in result: identity(item) return result @dataclass(frozen=True) class ReleaseContract: scope: tuple = ("release-v1", "model-v1", "eval-v1", "cohort-v1", "policy-v1", "data-v1", "ingestion-v1", "decode-v1", "privacy-v1", "license-v1", "moderation-v1", "source-v1") max_bytes: int = 1024 max_pixels: int = 1000000 max_audio_seconds: float = 30.0 max_latency_ms: float = 1000.0 max_cost: float = 0.1 max_retention_hours: int = 24 min_slice_support: int = 5 max_error_rate: float = 0.2 content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) count(self.max_bytes, 1, 65536) count(self.max_pixels, 1, 16000000) number(self.max_audio_seconds, 0.1, 3600) number(self.max_latency_ms, 1, 60000) number(self.max_cost, 0.000001, 100) count(self.max_retention_hours, 0, 720) count(self.min_slice_support, 2, 10000) number(self.max_error_rate, 0, 0.5) expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("contract digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class MediaEvidence: scope: tuple media_id: str revision: str modality: str payload_hex: str pixels: int audio_seconds: float decode: str = "verified" privacy: str = "verified" license: str = "verified" moderation: str = "verified" retention_hours: int = 24 deletion: str = "verified" content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) identity(self.media_id) identity(self.revision) if type(self.modality) is not str or self.modality not in ("image", "audio"): raise ValueError("unsupported modality") if type(self.payload_hex) is not str or not re.fullmatch("(?:[a-f0-9]{2}){1,65536}", self.payload_hex): raise ValueError("corrupt or excessive encoded payload") count(self.pixels, 0, 16000000) number(self.audio_seconds, 0, 3600) if self.modality == "image" and (self.pixels == 0 or self.audio_seconds != 0): raise ValueError("image requires pixels and no audio duration") if self.modality == "audio" and (self.audio_seconds == 0 or self.pixels != 0): raise ValueError("audio requires duration and no pixels") for value in (self.decode, self.privacy, self.license, self.moderation, self.deletion): if type(value) is not str or value not in ("verified", "failed", "unknown"): raise ValueError("explicit evidence state required") count(self.retention_hours, 0, 720) expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("media digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class SliceEvidence: scope: tuple slice_id: str media_content_ids: tuple support: int errors: int content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) if type(self.slice_id) is not str or self.slice_id not in ("image:ocr", "image:spatial", "audio:speech", "audio:noise"): raise ValueError("modality-specific slice required") values = sequence(self.media_content_ids, 1, 8) if any(type(value) is not str or not re.fullmatch("[a-f0-9]{64}", value) for value in values) or len(set(values)) != len(values): raise ValueError("unique media content digests required") object.__setattr__(self, "media_content_ids", values) count(self.support, 0, 10000) count(self.errors, 0, self.support) expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("slice digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class ReleaseEvidence: scope: tuple evidence_id: str media: tuple slices: tuple latency_ms: float cost: float fallback: str = "abstain" content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) identity(self.evidence_id) media = sequence(self.media, 1, 8) slices = sequence(self.slices, 0, 4) for item in media: validate_record(item, MediaEvidence) for item in slices: validate_record(item, SliceEvidence) object.__setattr__(self, "media", media) object.__setattr__(self, "slices", slices) number(self.latency_ms, 0, 60000) number(self.cost, 0, 100) if type(self.fallback) is not str or self.fallback != "abstain": raise ValueError("unverified fallback; only abstention is authorized") expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("release evidence digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class ReleaseReport: status: str first_failure: str action: str evidence_id: str claim: str = "LOCAL_EVIDENCE_GATE_NOT_PRODUCTION_CERTIFICATION" def gate_release(contract: ReleaseContract, evidence: ReleaseEvidence) -> ReleaseReport: """Fail closed in a fixed order; never release from aggregate-only evidence.""" validate_record(contract, ReleaseContract) validate_record(evidence, ReleaseEvidence) if any(item.scope != contract.scope for item in (evidence, *evidence.media, *evidence.slices)): raise ValueError("ingestion/model/evaluation/cohort/policy scope mismatch") if len({m.media_id for m in evidence.media}) != len(evidence.media) or len({(m.modality, m.payload_hex) for m in evidence.media}) != len(evidence.media): raise ValueError("duplicate media identity or payload") if len({s.slice_id for s in evidence.slices}) != len(evidence.slices): raise ValueError("duplicate slice") required = {"image:ocr", "image:spatial", "audio:speech", "audio:noise"} failure = "none" if {m.modality for m in evidence.media} != {"image", "audio"}: failure = "missing-modality" elif sum(len(m.payload_hex) // 2 for m in evidence.media) > contract.max_bytes or sum(m.pixels for m in evidence.media) > contract.max_pixels or math.fsum(m.audio_seconds for m in evidence.media) > contract.max_audio_seconds: failure = "media-budget" else: for boundary in ("decode", "privacy", "license", "moderation"): if any(getattr(m, boundary) != "verified" for m in evidence.media): failure = boundary break if failure == "none" and any(m.retention_hours > contract.max_retention_hours or m.deletion != "verified" for m in evidence.media): failure = "retention-deletion" if failure == "none": if {s.slice_id for s in evidence.slices} != required: failure = "slice-coverage" else: for item in evidence.slices: expected_media = tuple(m.content_id for m in evidence.media if m.modality == item.slice_id.split(":")[0]) if item.media_content_ids != expected_media: failure = "slice-content-binding" break if item.support < contract.min_slice_support or item.errors / item.support > contract.max_error_rate: failure = "slice-quality" break if failure == "none" and evidence.latency_ms > contract.max_latency_ms: failure = "latency" if failure == "none" and evidence.cost > contract.max_cost: failure = "cost" return ReleaseReport("ELIGIBLE_FOR_REVIEW" if failure == "none" else "BLOCKED", failure, "review" if failure == "none" else "abstain", digest((contract.content_id, evidence.content_id))) def illustrative_fixture(): contract = ReleaseContract() media = (MediaEvidence(contract.scope, "image-1", "rev-1", "image", "0102", 16, 0.0), MediaEvidence(contract.scope, "audio-1", "rev-1", "audio", "0304", 0, 2.0)) slices = tuple(SliceEvidence(contract.scope, label, (media[index].content_id,), 10, 1) for index, label in ((0, "image:ocr"), (0, "image:spatial"), (1, "audio:speech"), (1, "audio:noise"))) return contract, ReleaseEvidence(contract.scope, "release-eval-1", media, slices, 300.0, 0.02) def main(): report = gate_release(*illustrative_fixture()) print("example=illustrative_only") print(f"status={report.status}") print(f"first_failure={report.first_failure};action={report.action}") print(f"claim={report.claim}") if __name__ == "__main__": main()