# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-discovery-invariants-prevent-rate-limited-resources-from-appearing-healthy.md # Remnant Warrant: https://corpusfetch.com/c/what-discovery-invariants-prevent-rate-limited-resources-from-appearing-healthy.warrant.json # Markdown SHA-256: b061cb4c6a0be2a74436dcff59dd086e9cdd5d6100e0e207307296367c80c23a # Code AST SHA-256: df5d890bae4d1de47a7daee00a65f35ad4e1235d31e04b663cfd425f53df8586 # Verified Classes: RateLimitVerdict, DiscoveredResource, CapacityPolicy, RateLimitManifest, RateLimitDiscoveryGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class RateLimitVerdict(Enum): CAPACITY_VERIFIED = auto() LIVENESS_ONLY = auto() INSUFFICIENT_HEADROOM = auto() @dataclass(frozen=True) class DiscoveredResource: resource_id: str health_check_passed: bool capacity_aware_check: bool rate_limit_headroom_pct: float | None current_traffic_estimate_pct: float @dataclass(frozen=True) class CapacityPolicy: min_headroom_pct: float @dataclass(frozen=True) class RateLimitManifest: verdict: RateLimitVerdict directive: str manifest_digest: str class RateLimitDiscoveryGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_resource( cls, resource: DiscoveredResource, policy: CapacityPolicy ) -> RateLimitManifest: # Gate 1: Health check failed if not resource.health_check_passed: digest = cls._hash(json.dumps({ "resource": resource.resource_id, "status": "UNHEALTHY" }, sort_keys=True)) return RateLimitManifest( verdict=RateLimitVerdict.INSUFFICIENT_HEADROOM, directive=f"HALT: Resource {resource.resource_id} failed health check.", manifest_digest=digest ) # Gate 2: Capacity-aware check if resource.capacity_aware_check and resource.rate_limit_headroom_pct is not None: if resource.rate_limit_headroom_pct >= policy.min_headroom_pct: digest = cls._hash(json.dumps({ "resource": resource.resource_id, "headroom": f"{resource.rate_limit_headroom_pct:.1f}", "threshold": f"{policy.min_headroom_pct:.1f}" }, sort_keys=True)) return RateLimitManifest( verdict=RateLimitVerdict.CAPACITY_VERIFIED, directive=f"PROCEED: {resource.rate_limit_headroom_pct:.1f}% headroom.", manifest_digest=digest ) return RateLimitManifest( verdict=RateLimitVerdict.INSUFFICIENT_HEADROOM, directive=f"THROTTLE: {resource.rate_limit_headroom_pct:.1f}% headroom < {policy.min_headroom_pct:.1f}%.", manifest_digest=cls._hash(f"LOW_HEADROOM:{resource.rate_limit_headroom_pct:.1f}") ) # Gate 3: Liveness only return RateLimitManifest( verdict=RateLimitVerdict.LIVENESS_ONLY, directive=f"WARN: Discovery only checked liveness for {resource.resource_id}. Rate limit unknown.", manifest_digest=cls._hash(f"LIVENESS_ONLY:{resource.resource_id}") )