"""A strict, versioned inference boundary extracted from exploratory code.""" from __future__ import annotations from dataclasses import asdict, dataclass from hashlib import sha256 import json from math import isfinite import re from typing import Callable Predictor = Callable[[tuple[float, ...]], float] SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$") def _required_text(name: str, value: object) -> str: if not isinstance(value, str) or not value.strip(): raise ValueError(f"{name} must be non-empty text") return value def _strict_number(name: str, value: object) -> float: if isinstance(value, bool) or type(value) not in {int, float}: raise ValueError(f"{name} must be a non-boolean number") try: numeric = float(value) except OverflowError as error: raise ValueError(f"{name} must be finite") from error if not isfinite(numeric): raise ValueError(f"{name} must be finite") return numeric def _require_digest(name: str, value: object) -> str: if not isinstance(value, str) or not SHA256_PATTERN.fullmatch(value): raise ValueError(f"{name} must be a lowercase SHA-256 digest") return value def _stable_digest(payload: object) -> str: encoded = json.dumps( payload, allow_nan=False, ensure_ascii=True, separators=(",", ":"), sort_keys=True, ).encode("utf-8") return sha256(encoded).hexdigest() @dataclass(frozen=True) class FeatureSpec: name: str semantic: str unit: str minimum: float maximum: float def __post_init__(self) -> None: for field in ("name", "semantic", "unit"): _required_text(field, getattr(self, field)) minimum = _strict_number(f"{self.name}.minimum", self.minimum) maximum = _strict_number(f"{self.name}.maximum", self.maximum) if minimum > maximum: raise ValueError(f"{self.name}.minimum must not exceed maximum") @dataclass(frozen=True) class FeatureValue: name: str value: object semantic: str unit: str @dataclass(frozen=True) class InferenceConfig: release_version: str contract_version: str request_schema_version: str response_schema_version: str model_version: str model_digest: str preprocessing_version: str preprocessing_digest: str feature_schema: tuple[FeatureSpec, ...] output_score_semantic: str output_score_unit: str output_minimum: float output_maximum: float decision_threshold: float decision_at_or_above: str decision_below: str max_request_bytes: int max_request_id_characters: int def __post_init__(self) -> None: try: feature_schema = tuple(self.feature_schema) except TypeError as error: raise ValueError( "feature_schema must be an iterable of FeatureSpec records" ) from error if any(not isinstance(feature, FeatureSpec) for feature in feature_schema): raise ValueError("feature_schema must contain only FeatureSpec records") # frozen=True does not copy caller-owned containers. Normalize before the # release digest can observe later list mutation. object.__setattr__(self, "feature_schema", feature_schema) for field in ( "release_version", "contract_version", "request_schema_version", "response_schema_version", "model_version", "preprocessing_version", "output_score_semantic", "output_score_unit", "decision_at_or_above", "decision_below", ): _required_text(field, getattr(self, field)) _require_digest("model_digest", self.model_digest) _require_digest("preprocessing_digest", self.preprocessing_digest) names = [feature.name for feature in self.feature_schema] if not names: raise ValueError("feature_schema must not be empty") if len(names) != len(set(names)): raise ValueError("feature names must be unique") output_minimum = _strict_number("output_minimum", self.output_minimum) output_maximum = _strict_number("output_maximum", self.output_maximum) threshold = _strict_number("decision_threshold", self.decision_threshold) if output_minimum > output_maximum: raise ValueError("output_minimum must not exceed output_maximum") if not output_minimum <= threshold <= output_maximum: raise ValueError("decision_threshold must be within the output range") if type(self.max_request_bytes) is not int or self.max_request_bytes <= 0: raise ValueError("max_request_bytes must be a positive integer") if ( type(self.max_request_id_characters) is not int or self.max_request_id_characters <= 0 ): raise ValueError( "max_request_id_characters must be a positive integer" ) @property def release_digest(self) -> str: return _stable_digest( { "contract_version": self.contract_version, "decision_at_or_above": self.decision_at_or_above, "decision_below": self.decision_below, "decision_threshold": self.decision_threshold, "feature_schema": [asdict(feature) for feature in self.feature_schema], "max_request_bytes": self.max_request_bytes, "max_request_id_characters": self.max_request_id_characters, "model_digest": self.model_digest, "model_version": self.model_version, "output_maximum": self.output_maximum, "output_minimum": self.output_minimum, "output_score_semantic": self.output_score_semantic, "output_score_unit": self.output_score_unit, "preprocessing_digest": self.preprocessing_digest, "preprocessing_version": self.preprocessing_version, "release_version": self.release_version, "request_schema_version": self.request_schema_version, "response_schema_version": self.response_schema_version, } ) @property def release_id(self) -> str: return f"{self.release_version}@sha256:{self.release_digest}" @dataclass(frozen=True) class InferenceRequest: request_id: str schema_version: str features: tuple[FeatureValue, ...] @dataclass(frozen=True) class InferenceResponse: request_id: str release_id: str release_version: str contract_version: str response_schema_version: str model_version: str preprocessing_version: str score: float score_semantic: str score_unit: str decision: str class InferenceService: """Validates the complete boundary while the predictor owns model math.""" def __init__(self, config: InferenceConfig, predictor: Predictor): self._config = config self._predictor = predictor def _validated_features(self, request: InferenceRequest) -> tuple[float, ...]: if not isinstance(request.features, tuple) or any( not isinstance(feature, FeatureValue) for feature in request.features ): raise ValueError("features must be a tuple of FeatureValue records") by_name: dict[str, FeatureValue] = {} for feature in request.features: _required_text("feature.name", feature.name) if feature.name in by_name: raise ValueError(f"duplicate feature: {feature.name}") by_name[feature.name] = feature expected = {spec.name for spec in self._config.feature_schema} received = set(by_name) missing = expected - received unexpected = received - expected if missing or unexpected: details = [] if missing: details.append("missing=" + ",".join(sorted(missing))) if unexpected: details.append("unexpected=" + ",".join(sorted(unexpected))) raise ValueError("feature schema mismatch: " + "; ".join(details)) ordered = [] for spec in self._config.feature_schema: feature = by_name[spec.name] if feature.semantic != spec.semantic: raise ValueError(f"{spec.name}.semantic mismatch") if feature.unit != spec.unit: raise ValueError(f"{spec.name}.unit mismatch") numeric = _strict_number(f"{spec.name}.value", feature.value) if not spec.minimum <= numeric <= spec.maximum: raise ValueError( f"{spec.name}.value must be within " f"[{spec.minimum}, {spec.maximum}] {spec.unit}" ) ordered.append(numeric) return tuple(ordered) def _request_size_bytes(self, request: InferenceRequest) -> int: payload = { "features": [asdict(feature) for feature in request.features], "request_id": request.request_id, "schema_version": request.schema_version, } return len( json.dumps( payload, allow_nan=False, ensure_ascii=False, separators=(",", ":"), sort_keys=True, ).encode("utf-8") ) def infer(self, request: InferenceRequest) -> InferenceResponse: _required_text("request_id", request.request_id) if len(request.request_id) > self._config.max_request_id_characters: raise ValueError("request_id exceeds max_request_id_characters") if request.schema_version != self._config.request_schema_version: raise ValueError( "request schema mismatch: " f"expected {self._config.request_schema_version}, " f"got {request.schema_version}" ) ordered_features = self._validated_features(request) if self._request_size_bytes(request) > self._config.max_request_bytes: raise ValueError("request exceeds max_request_bytes") score = _strict_number("predictor score", self._predictor(ordered_features)) if not self._config.output_minimum <= score <= self._config.output_maximum: raise ValueError( "predictor score must be within " f"[{self._config.output_minimum}, {self._config.output_maximum}] " f"{self._config.output_score_unit}" ) return InferenceResponse( request_id=request.request_id, release_id=self._config.release_id, release_version=self._config.release_version, contract_version=self._config.contract_version, response_schema_version=self._config.response_schema_version, model_version=self._config.model_version, preprocessing_version=self._config.preprocessing_version, score=score, score_semantic=self._config.output_score_semantic, score_unit=self._config.output_score_unit, decision=( self._config.decision_at_or_above if score >= self._config.decision_threshold else self._config.decision_below ), ) def illustrative_predictor(features: tuple[float, ...]) -> float: """Deterministic fixture only; this is not a trained model or benchmark.""" return sum(features) / len(features) ILLUSTRATIVE_CONFIG = InferenceConfig( release_version="illustrative-release-v1", contract_version="inference-contract-v1", request_schema_version="illustrative-request-v1", response_schema_version="illustrative-response-v1", model_version="illustrative-model-v1", model_digest=sha256(b"illustrative-model-artifact-v1").hexdigest(), preprocessing_version="illustrative-preprocessing-v1", preprocessing_digest=sha256(b"illustrative-preprocessing-v1").hexdigest(), feature_schema=( FeatureSpec( "signal_a", "illustrative normalized signal A", "ratio", 0.0, 1.0, ), FeatureSpec( "signal_b", "illustrative normalized signal B", "ratio", 0.0, 1.0, ), ), output_score_semantic="illustrative review score", output_score_unit="ratio", output_minimum=0.0, output_maximum=1.0, decision_threshold=0.6, decision_at_or_above="accept", decision_below="review", max_request_bytes=512, max_request_id_characters=64, ) ILLUSTRATIVE_REQUEST = InferenceRequest( request_id="req-001", schema_version="illustrative-request-v1", features=( FeatureValue( "signal_a", 0.2, "illustrative normalized signal A", "ratio", ), FeatureValue( "signal_b", 0.8, "illustrative normalized signal B", "ratio", ), ), ) def format_example() -> str: service = InferenceService(ILLUSTRATIVE_CONFIG, illustrative_predictor) response = service.infer(ILLUSTRATIVE_REQUEST) return json.dumps(asdict(response), sort_keys=True) if __name__ == "__main__": print(format_example())