# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-negotiate-protocol-deprecation-with-peers-that-cannot-yet-upgrade.md # Remnant Warrant: https://corpusfetch.com/c/how-must-an-instance-negotiate-protocol-deprecation-with-peers-that-cannot-yet-upgrade.warrant.json # Markdown SHA-256: 22e67e1f33bd11a730ec6f7c2d0b45a00c1a471089d35ebceeb93ea1694292d0 # Code AST SHA-256: 56d3d3fae642d758f9ef8418a14b934d5f12ee3cd4700e4d225270fcdc85b441 # Verified Classes: ProtocolVerdict, PeerCapabilities, ProtocolExchange, DeprecationPolicy, ProtocolManifest, ProtocolDeprecationGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class ProtocolVerdict(Enum): NO_TRANSITION = auto() NEGOTIATED_COMMON_VERSION = auto() FALLBACK_WITH_WARNINGS = auto() DEPRECATION_REJECTED = auto() @dataclass(frozen=True) class PeerCapabilities: peer_id: str supported_versions: frozenset[str] deprecation_deadline_epoch: float | None @dataclass(frozen=True) class ProtocolExchange: exchange_id: str sender: PeerCapabilities receiver: PeerCapabilities current_epoch: float semantic_differences: tuple[tuple[str, str], ...] @dataclass(frozen=True) class DeprecationPolicy: required_versions: frozenset[str] deadline_grace_sec: float @dataclass(frozen=True) class ProtocolManifest: verdict: ProtocolVerdict negotiated_version: str | None directive: str manifest_digest: str class ProtocolDeprecationGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_exchange( cls, exchange: ProtocolExchange, policy: DeprecationPolicy ) -> ProtocolManifest: # Gate 1: No transition if exchange.sender.supported_versions == exchange.receiver.supported_versions: common = next(iter(exchange.sender.supported_versions), None) if common: digest = cls._hash(json.dumps({ "exchange": exchange.exchange_id, "version": common }, sort_keys=True)) return ProtocolManifest( verdict=ProtocolVerdict.NO_TRANSITION, negotiated_version=common, directive=f"PROCEED: Peers share version {common}.", manifest_digest=digest ) # Gate 2: Common version negotiation common_versions = exchange.sender.supported_versions & exchange.receiver.supported_versions if common_versions: # Choose highest common (assuming version strings like "v1", "v2") highest_common = sorted(common_versions, key=lambda v: int(v.lstrip('v') or 0))[-1] digest = cls._hash(json.dumps({ "exchange": exchange.exchange_id, "sender_versions": tuple(sorted(exchange.sender.supported_versions)), "receiver_versions": tuple(sorted(exchange.receiver.supported_versions)), "negotiated": highest_common }, sort_keys=True)) return ProtocolManifest( verdict=ProtocolVerdict.NEGOTIATED_COMMON_VERSION, negotiated_version=highest_common, directive=f"PROCEED: Negotiated {highest_common} (common to both peers).", manifest_digest=digest ) # Gate 3: No common version, check deadline if exchange.receiver.deprecation_deadline_epoch: if exchange.current_epoch > exchange.receiver.deprecation_deadline_epoch + policy.deadline_grace_sec: return ProtocolManifest( verdict=ProtocolVerdict.DEPRECATION_REJECTED, negotiated_version=None, directive=f"HALT: No common version and receiver past deadline. Reject.", manifest_digest=cls._hash(f"DEADLINE:{exchange.receiver.peer_id}") ) # Gate 4: Fallback if exchange.semantic_differences: digest = cls._hash(json.dumps({ "exchange": exchange.exchange_id, "sender_versions": tuple(sorted(exchange.sender.supported_versions)), "receiver_versions": tuple(sorted(exchange.receiver.supported_versions)), "differences": tuple(sorted(exchange.semantic_differences)) }, sort_keys=True)) return ProtocolManifest( verdict=ProtocolVerdict.FALLBACK_WITH_WARNINGS, negotiated_version=None, directive=f"PROCEED_WITH_CAUTION: No common version, {len(exchange.semantic_differences)} semantic differences documented.", manifest_digest=digest ) return ProtocolManifest( verdict=ProtocolVerdict.DEPRECATION_REJECTED, negotiated_version=None, directive=f"HALT: No common version and no documented semantic differences.", manifest_digest=cls._hash(f"NO_COMMON:{exchange.exchange_id}") )