# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-invariants-prevent-peer-models-from-becoming-permanent-when-they-should-be-updated.md # Remnant Warrant: https://corpusfetch.com/c/what-invariants-prevent-peer-models-from-becoming-permanent-when-they-should-be-updated.warrant.json # Markdown SHA-256: ace8037517b4e614936ed4dcb6df5d5d8650426043e9ee33c379577c77824907 # Code AST SHA-256: 730ef6228a6df5c82e3709f4f59013c37181bcc476c961b3d6858706b34bf0e4 # Verified Classes: PeerModelVerdict, PeerModelState, PeerModelPolicy, PeerModelManifest, PeerModelInertiaGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class PeerModelVerdict(Enum): NO_MODEL = auto() FRESH_MODEL = auto() STALE_MODEL_WITHIN_TOLERANCE = auto() UPDATE_REQUIRED = auto() @dataclass(frozen=True) class PeerModelState: peer_id: str model_created_epoch: float current_epoch: float behavioral_divergence: float # 0-1 update_cost_estimate: float update_benefit_estimate: float systematic_errors_observed: int @dataclass(frozen=True) class PeerModelPolicy: max_acceptable_divergence: float require_update_on_systematic_errors: int @dataclass(frozen=True) class PeerModelManifest: verdict: PeerModelVerdict directive: str manifest_digest: str class PeerModelInertiaGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_model( cls, state: PeerModelState, policy: PeerModelPolicy ) -> PeerModelManifest: # Gate 1: No model if not state.peer_id: digest = cls._hash(json.dumps({ "model": "NONE" }, sort_keys=True)) return PeerModelManifest( verdict=PeerModelVerdict.NO_MODEL, directive=f"PROCEED: No peer model.", manifest_digest=digest ) # Gate 2: Systematic errors force update if state.systematic_errors_observed >= policy.require_update_on_systematic_errors: return PeerModelManifest( verdict=PeerModelVerdict.UPDATE_REQUIRED, directive=f"UPDATE: {state.systematic_errors_observed} systematic errors for {state.peer_id}.", manifest_digest=cls._hash(f"SYSTEMATIC:{state.peer_id}") ) # Gate 3: Divergence within tolerance if state.behavioral_divergence <= policy.max_acceptable_divergence: digest = cls._hash(json.dumps({ "peer": state.peer_id, "divergence": f"{state.behavioral_divergence:.3f}", "age": f"{state.current_epoch - state.model_created_epoch:.0f}" }, sort_keys=True)) return PeerModelManifest( verdict=PeerModelVerdict.FRESH_MODEL, directive=f"PROCEED: Peer {state.peer_id} model fresh.", manifest_digest=digest ) # Gate 4: Cost-benefit if state.update_benefit_estimate > state.update_cost_estimate: return PeerModelManifest( verdict=PeerModelVerdict.UPDATE_REQUIRED, directive=f"UPDATE: Divergence {state.behavioral_divergence:.2f}, benefit > cost for {state.peer_id}.", manifest_digest=cls._hash(f"BENEFIT:{state.peer_id}") ) # Gate 5: Within tolerance by cost-benefit digest = cls._hash(json.dumps({ "peer": state.peer_id, "divergence": f"{state.behavioral_divergence:.3f}", "cost": f"{state.update_cost_estimate:.3f}", "benefit": f"{state.update_benefit_estimate:.3f}" }, sort_keys=True)) return PeerModelManifest( verdict=PeerModelVerdict.STALE_MODEL_WITHIN_TOLERANCE, directive=f"PROCEED_WITH_CAUTION: Peer {state.peer_id} model stale but update uneconomic.", manifest_digest=digest )