"""Fail-closed audit for an illustrative affine neuron decision surface.""" from __future__ import annotations from dataclasses import dataclass import hashlib import json import math import sys from typing import NoReturn MAX_FEATURES = 64 MAX_TEXT_LENGTH = 240 def _fail(message: str) -> NoReturn: raise ValueError(message) def _text(value: object, field: str) -> str: if type(value) is not str or not value.strip() or len(value) > MAX_TEXT_LENGTH: _fail(f"{field} must be non-empty text of at most {MAX_TEXT_LENGTH} characters") return value def _number(value: object, field: str) -> float: if type(value) not in (int, float): _fail(f"{field} must be a concrete number") result = float(value) if not math.isfinite(result): _fail(f"{field} must be finite") if result != 0.0 and abs(result) < sys.float_info.min: _fail(f"{field} is below the supported numeric resolution") return result def _digest(payload: object) -> str: encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True) return "sha256:" + hashlib.sha256(encoded.encode("utf-8")).hexdigest() @dataclass(frozen=True) class FeatureSpec: feature_id: str unit: str semantics: str def __post_init__(self) -> None: _text(self.feature_id, "feature_id") _text(self.unit, "unit") _text(self.semantics, "semantics") @dataclass(frozen=True) class NeuronContract: contract_version: str scope_id: str model_revision: str feature_schema_version: str owner: str features: tuple[FeatureSpec, ...] weights: tuple[float, ...] bias: float activation: str action_threshold: float decision_convention: str negative_action: str positive_action: str def __post_init__(self) -> None: object.__setattr__(self, "features", tuple(self.features)) object.__setattr__(self, "weights", tuple(self.weights)) for field in ("contract_version", "scope_id", "model_revision", "feature_schema_version", "owner"): _text(getattr(self, field), field) if not 1 <= len(self.features) <= MAX_FEATURES: _fail(f"features must contain between 1 and {MAX_FEATURES} entries") if any(type(item) is not FeatureSpec for item in self.features): _fail("features must contain concrete FeatureSpec records") for item in self.features: item.__post_init__() ids = [item.feature_id for item in self.features] if len(ids) != len(set(ids)): _fail("feature IDs must be unique") if len(self.weights) != len(self.features): _fail("weights must align exactly with feature order") for index, weight in enumerate(self.weights): _number(weight, f"weights[{index}]") _number(self.bias, "bias") _text(self.activation, "activation") if self.activation not in {"sigmoid", "relu"}: _fail("activation must be sigmoid or relu") threshold = _number(self.action_threshold, "action_threshold") if self.activation == "sigmoid" and not 0.0 < threshold < 1.0: _fail("a sigmoid threshold must be strictly between zero and one") if self.activation == "relu" and threshold < 0.0: _fail("a ReLU threshold must be non-negative") if self.decision_convention != "response-gte-threshold-v1": _fail("decision_convention must bind inclusive response comparison") _text(self.negative_action, "negative_action") _text(self.positive_action, "positive_action") if self.negative_action == self.positive_action: _fail("action labels must be distinct") @dataclass(frozen=True) class InputEvidence: evidence_id: str scope_id: str model_revision: str schema_version: str source_version: str observed_at: str owner: str feature_order: tuple[str, ...] values: tuple[float, ...] def __post_init__(self) -> None: object.__setattr__(self, "feature_order", tuple(self.feature_order)) object.__setattr__(self, "values", tuple(self.values)) for field in ( "evidence_id", "scope_id", "model_revision", "schema_version", "source_version", "observed_at", "owner", ): _text(getattr(self, field), field) if not 1 <= len(self.values) <= MAX_FEATURES: _fail(f"values must contain between 1 and {MAX_FEATURES} entries") if len(self.feature_order) != len(self.values): _fail("feature_order and values must have equal length") if any(type(item) is not str for item in self.feature_order): _fail("feature_order must contain concrete strings") if len(self.feature_order) != len(set(self.feature_order)): _fail("feature_order must not contain duplicates") for index, value in enumerate(self.values): _number(value, f"values[{index}]") @dataclass(frozen=True) class NeuronAudit: evidence_content_id: str margin: float activation: float action: str decision: str def _sigmoid(value: float) -> float: if value >= 0.0: term = math.exp(-value) return 1.0 / (1.0 + term) term = math.exp(value) return term / (1.0 + term) def _checked_product(left: float, right: float, field: str) -> float: product = left * right if not math.isfinite(product): _fail(f"{field} overflowed the supported numeric range") if left != 0.0 and right != 0.0 and product == 0.0: _fail(f"{field} underflowed the supported numeric range") if product != 0.0 and abs(product) < sys.float_info.min: _fail(f"{field} is below the supported numeric resolution") return product def audit_neuron(contract: NeuronContract, evidence: InputEvidence) -> NeuronAudit: if type(contract) is not NeuronContract or type(evidence) is not InputEvidence: _fail("audit requires concrete NeuronContract and InputEvidence records") contract.__post_init__() evidence.__post_init__() if evidence.scope_id != contract.scope_id: _fail("evidence scope does not match the neuron contract") if evidence.model_revision != contract.model_revision: _fail("evidence model revision does not match the neuron contract") if evidence.schema_version != contract.feature_schema_version: _fail("evidence schema version does not match the neuron contract") if evidence.owner != contract.owner: _fail("evidence owner does not match the neuron contract") expected_order = tuple(feature.feature_id for feature in contract.features) if evidence.feature_order != expected_order: _fail("evidence feature order does not match the bound schema") products = [ _checked_product(_number(weight, "weight"), _number(value, "value"), "weighted feature") for weight, value in zip(contract.weights, evidence.values) ] margin = math.fsum([_number(contract.bias, "bias"), *products]) if not math.isfinite(margin): _fail("affine margin overflowed the supported numeric range") if margin != 0.0 and abs(margin) < sys.float_info.min: _fail("affine margin is below the supported numeric resolution") activation = _sigmoid(margin) if contract.activation == "sigmoid" else max(0.0, margin) if contract.activation == "sigmoid": threshold_margin = math.log(contract.action_threshold) - math.log1p(-contract.action_threshold) action_is_positive = margin >= threshold_margin else: action_is_positive = margin >= contract.action_threshold action = contract.positive_action if action_is_positive else contract.negative_action payload = { "contract": { "version": contract.contract_version, "scope": contract.scope_id, "model": contract.model_revision, "schema": contract.feature_schema_version, "owner": contract.owner, "features": [(f.feature_id, f.unit, f.semantics) for f in contract.features], "weights": list(contract.weights), "bias": contract.bias, "activation": contract.activation, "threshold": contract.action_threshold, "decision_convention": contract.decision_convention, "actions": [contract.negative_action, contract.positive_action], }, "evidence": { "id": evidence.evidence_id, "scope": evidence.scope_id, "model": evidence.model_revision, "schema": evidence.schema_version, "source": evidence.source_version, "observed_at": evidence.observed_at, "owner": evidence.owner, "order": list(evidence.feature_order), "values": list(evidence.values), }, } return NeuronAudit(_digest(payload), margin, activation, action, "PASS") ILLUSTRATIVE_CONTRACT = NeuronContract( contract_version="neuron-audit-v1", scope_id="illustrative-renewal-review", model_revision="renewal-neuron-r7", feature_schema_version="renewal-features-v3", owner="retention-ml", features=( FeatureSpec("usage_change_30d", "fraction", "relative product-usage change over 30 complete days"), FeatureSpec("open_incidents", "count", "unresolved support incidents at decision time"), ), weights=(2.0, -0.75), bias=-0.2, activation="sigmoid", action_threshold=0.6, decision_convention="response-gte-threshold-v1", negative_action="standard-review", positive_action="priority-retention-review", ) ILLUSTRATIVE_EVIDENCE = InputEvidence( evidence_id="renewal-input-0042", scope_id="illustrative-renewal-review", model_revision="renewal-neuron-r7", schema_version="renewal-features-v3", source_version="feature-snapshot-2026-08-25", observed_at="2026-08-25T09:00:00Z", owner="retention-ml", feature_order=("usage_change_30d", "open_incidents"), values=(0.8, 0.0), ) def format_example() -> str: audit = audit_neuron(ILLUSTRATIVE_CONTRACT, ILLUSTRATIVE_EVIDENCE) return "\n".join(( "example=illustrative_only", f"contract_version={ILLUSTRATIVE_CONTRACT.contract_version}", f"evidence_id={audit.evidence_content_id}", f"margin={audit.margin:.3f}", f"activation={audit.activation:.3f}", f"action={audit.action}", f"decision={audit.decision}", )) if __name__ == "__main__": print(format_example())