# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-sampling-determinism-invariants-must-survive-cross-region-inference-routing.md # Remnant Warrant: https://corpusfetch.com/c/what-sampling-determinism-invariants-must-survive-cross-region-inference-routing.warrant.json # Markdown SHA-256: e0d4c5d1f1923318effe4e14855db3ea9a0a514fb5bcd0c3b46fc5c66f7a2963 # Code AST SHA-256: d13909c9e2b72ebb1d12ef261bd1a18c648b1002b8aa8ce77077aba1428bb1fb # Verified Classes: SamplingVerdict, SamplingState, SamplingPolicy, SamplingManifest, SamplingDeterminismGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class SamplingVerdict(Enum): NON_DETERMINISTIC = auto() DETERMINISM_VERIFIED = auto() DIVERGENCE_DETECTED = auto() TEST_STALE = auto() @dataclass(frozen=True) class SamplingState: requires_determinism: bool determinism_test_epoch: float current_epoch: float divergent_regions: tuple[str, ...] all_regions_checked: bool @dataclass(frozen=True) class SamplingPolicy: max_test_staleness_sec: float @dataclass(frozen=True) class SamplingManifest: verdict: SamplingVerdict directive: str manifest_digest: str class SamplingDeterminismGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_determinism( cls, state: SamplingState, policy: SamplingPolicy ) -> SamplingManifest: # Gate 1: No determinism required if not state.requires_determinism: digest = cls._hash(json.dumps({ "determinism": "NOT_REQUIRED" }, sort_keys=True)) return SamplingManifest( verdict=SamplingVerdict.NON_DETERMINISTIC, directive="PROCEED: Sampling is probabilistic by policy.", manifest_digest=digest ) # Gate 2: Test freshness age = state.current_epoch - state.determinism_test_epoch if age > policy.max_test_staleness_sec: return SamplingManifest( verdict=SamplingVerdict.TEST_STALE, directive=f"HALT: Determinism test {age:.0f}s stale > {policy.max_test_staleness_sec:.0f}s. Re-test.", manifest_digest=cls._hash(f"STALE:{age:.0f}") ) # Gate 3: All regions checked if not state.all_regions_checked: return SamplingManifest( verdict=SamplingVerdict.TEST_STALE, directive="HALT: Determinism test did not cover all regions.", manifest_digest=cls._hash("PARTIAL_TEST") ) # Gate 4: Divergence if state.divergent_regions: return SamplingManifest( verdict=SamplingVerdict.DIVERGENCE_DETECTED, directive=f"HALT: Divergence in regions {sorted(state.divergent_regions)}.", manifest_digest=cls._hash(f"DIVERGENT:{len(state.divergent_regions)}") ) # Gate 5: Verified digest = cls._hash(json.dumps({ "test_epoch": f"{state.determinism_test_epoch:.0f}", "age": f"{age:.0f}", "regions_checked": "ALL" }, sort_keys=True)) return SamplingManifest( verdict=SamplingVerdict.DETERMINISM_VERIFIED, directive=f"PROCEED: Cross-region determinism verified {age:.0f}s ago.", manifest_digest=digest )