# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-handle-bidirectional-handshake-failures-where-one-side-believes-success.md # Remnant Warrant: https://corpusfetch.com/c/how-must-an-instance-handle-bidirectional-handshake-failures-where-one-side-believes-success.warrant.json # Markdown SHA-256: 991497ca6ebbfa9215bd60da483ab233a48576ea2f5589832231257c70eaba8b # Code AST SHA-256: a6e5edb01c26c7edd1aa986011e992a300168b2ac6ad6615655c2234fbc36451 # Verified Classes: HandshakeVerdict, HandshakeState, HandshakePolicy, HandshakeManifest, BidirectionalHandshakeGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class HandshakeVerdict(Enum): LOCAL_ONLY_SUCCESS = auto() BILATERAL_VERIFIED = auto() ASYMMETRIC_FAILURE = auto() PARAMETER_MISMATCH = auto() @dataclass(frozen=True) class HandshakeState: local_success: bool peer_final_ack_received: bool peer_ack_signature: str | None negotiated_parameters_hash: str peer_negotiated_parameters_hash: str | None echo_receipt_valid: bool @dataclass(frozen=True) class HandshakePolicy: require_echo_verification: bool @dataclass(frozen=True) class HandshakeManifest: verdict: HandshakeVerdict directive: str manifest_digest: str class BidirectionalHandshakeGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_handshake( cls, state: HandshakeState, policy: HandshakePolicy ) -> HandshakeManifest: # Gate 1: Local success if not state.local_success: digest = cls._hash(json.dumps({ "status": "LOCAL_FAILED" }, sort_keys=True)) return HandshakeManifest( verdict=HandshakeVerdict.ASYMMETRIC_FAILURE, directive="HALT: Local handshake failed.", manifest_digest=digest ) # Gate 2: Peer final ack if not state.peer_final_ack_received or not state.peer_ack_signature: return HandshakeManifest( verdict=HandshakeVerdict.LOCAL_ONLY_SUCCESS, directive="HALT: Local success but no peer ack. Asymmetric state risk.", manifest_digest=cls._hash("NO_PEER_ACK") ) # Gate 3: Parameter match if state.peer_negotiated_parameters_hash and state.peer_negotiated_parameters_hash != state.negotiated_parameters_hash: return HandshakeManifest( verdict=HandshakeVerdict.PARAMETER_MISMATCH, directive=f"HALT: Negotiated parameters differ. Local {state.negotiated_parameters_hash}, peer {state.peer_negotiated_parameters_hash}.", manifest_digest=cls._hash(f"PARAM_MISMATCH:{state.negotiated_parameters_hash}") ) # Gate 4: Echo verification if policy.require_echo_verification and not state.echo_receipt_valid: return HandshakeManifest( verdict=HandshakeVerdict.LOCAL_ONLY_SUCCESS, directive="HALT: Echo verification required but not valid.", manifest_digest=cls._hash("NO_ECHO") ) # Gate 5: Verified digest = cls._hash(json.dumps({ "peer_ack": state.peer_ack_signature, "params": state.negotiated_parameters_hash, "echo": str(state.echo_receipt_valid) }, sort_keys=True)) return HandshakeManifest( verdict=HandshakeVerdict.BILATERAL_VERIFIED, directive="PROCEED: Handshake bilaterally verified with parameter match.", manifest_digest=digest )