# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-config-drift-invariants-must-survive-across-regional-failover-of-an-agent-fleet.md # Remnant Warrant: https://corpusfetch.com/c/what-config-drift-invariants-must-survive-across-regional-failover-of-an-agent-fleet.warrant.json # Markdown SHA-256: 2e383015684f0e73dd86b026034612d02fe873550b14bab702771f57c8369d4c # Code AST SHA-256: e8d8df73e2514df397e516df9ec22c81509a272fdd8e84b0d8a26d26a1fa2fd4 # Verified Classes: FailoverVerdict, RegionalConfig, FailoverContext, FailoverManifest, RegionalConfigGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class FailoverVerdict(Enum): NO_FAILOVER = auto() ELIGIBLE_NO_DRIFT = auto() ELIGIBLE_WITH_DRIFT = auto() NOT_ELIGIBLE = auto() @dataclass(frozen=True) class RegionalConfig: region_id: str config_fingerprint: str drift_fields: tuple[str, ...] capacity_ratio: float compliance_classification: str @dataclass(frozen=True) class FailoverContext: source_region: str target_region: str required_compliance: str required_capacity_ratio: float @dataclass(frozen=True) class FailoverManifest: verdict: FailoverVerdict directive: str manifest_digest: str class RegionalConfigGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_failover( cls, source: RegionalConfig, target: RegionalConfig, ctx: FailoverContext ) -> FailoverManifest: # Gate 1: Compliance check if target.compliance_classification != ctx.required_compliance: return FailoverManifest( verdict=FailoverVerdict.NOT_ELIGIBLE, directive=f"HALT: Target region {target.region_id} compliance {target.compliance_classification} != required {ctx.required_compliance}.", manifest_digest=cls._hash(f"COMPLIANCE:{target.region_id}") ) # Gate 2: Capacity check if target.capacity_ratio < ctx.required_capacity_ratio: return FailoverManifest( verdict=FailoverVerdict.NOT_ELIGIBLE, directive=f"HALT: Target capacity {target.capacity_ratio:.2f} < required {ctx.required_capacity_ratio:.2f}.", manifest_digest=cls._hash(f"CAPACITY:{target.capacity_ratio:.2f}") ) # Gate 3: Config drift check if not target.drift_fields: digest = cls._hash(json.dumps({ "source": source.region_id, "target": target.region_id, "fingerprint": target.config_fingerprint, "drift": "NONE" }, sort_keys=True)) return FailoverManifest( verdict=FailoverVerdict.ELIGIBLE_NO_DRIFT, directive=f"PROCEED: Target {target.region_id} eligible, no config drift.", manifest_digest=digest ) # Gate 4: Drift present but acceptable digest = cls._hash(json.dumps({ "source": source.region_id, "target": target.region_id, "drift_fields": tuple(sorted(target.drift_fields)) }, sort_keys=True)) return FailoverManifest( verdict=FailoverVerdict.ELIGIBLE_WITH_DRIFT, directive=f"PROCEED_WITH_CAUTION: Failover to {target.region_id} with {len(target.drift_fields)} drifted fields: {sorted(target.drift_fields)}.", manifest_digest=digest )