# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-must-agents-communicate-partial-settlements-when-the-blockchain-is-congested.md # Remnant Warrant: https://corpusfetch.com/c/how-must-agents-communicate-partial-settlements-when-the-blockchain-is-congested.warrant.json # Markdown SHA-256: 9424fe6bb9893de80a39f0a6b362fcee36768dcbc04a30e343c47e2c10816eb9 # Code AST SHA-256: 780ceb2e4dbc27445749bd2b44997999edf6ffa188fcbaef016978ebccea9cfa # Verified Classes: SettlementVerdict, ChannelState, CongestionContext, SettlementManifest, OffChainSettlementGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class SettlementVerdict(Enum): ON_CHAIN_ACCEPTABLE = auto() OFF_CHAIN_ACTIVE = auto() OFF_CHAIN_UNAUTHORIZED = auto() DISPUTE_WINDOW_EXPIRED = auto() @dataclass(frozen=True) class ChannelState: channel_id: str off_chain_balance_a: int off_chain_balance_b: int signed_by_a: bool signed_by_b: bool dispute_window_epoch: float current_epoch: float @dataclass(frozen=True) class CongestionContext: gas_price_gwei: float congestion_threshold_gwei: float trade_value_wei: int @dataclass(frozen=True) class SettlementManifest: verdict: SettlementVerdict directive: str manifest_digest: str class OffChainSettlementGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_settlement( cls, channel: ChannelState, ctx: CongestionContext ) -> SettlementManifest: # Gate 1: No congestion if ctx.gas_price_gwei <= ctx.congestion_threshold_gwei: digest = cls._hash(json.dumps({ "channel": channel.channel_id, "gas": f"{ctx.gas_price_gwei:.2f}", "mode": "ON_CHAIN" }, sort_keys=True)) return SettlementManifest( verdict=SettlementVerdict.ON_CHAIN_ACCEPTABLE, directive=f"PROCEED: Gas {ctx.gas_price_gwei:.2f} below threshold {ctx.congestion_threshold_gwei:.2f}. On-chain ok.", manifest_digest=digest ) # Gate 2: Mutual signing if not (channel.signed_by_a and channel.signed_by_b): return SettlementManifest( verdict=SettlementVerdict.OFF_CHAIN_UNAUTHORIZED, directive=f"HALT: Channel {channel.channel_id} not mutually signed. Cannot use off-chain.", manifest_digest=cls._hash(f"UNSIGNED:{channel.channel_id}") ) # Gate 3: Dispute window if channel.current_epoch > channel.dispute_window_epoch: return SettlementManifest( verdict=SettlementVerdict.DISPUTE_WINDOW_EXPIRED, directive=f"HALT: Dispute window expired at {channel.dispute_window_epoch:.0f}. Submit net settlement on-chain.", manifest_digest=cls._hash(f"EXPIRED:{channel.channel_id}") ) # Gate 4: Off-chain active digest = cls._hash(json.dumps({ "channel": channel.channel_id, "balance_a": str(channel.off_chain_balance_a), "balance_b": str(channel.off_chain_balance_b), "dispute_until": f"{channel.dispute_window_epoch:.0f}", "gas": f"{ctx.gas_price_gwei:.2f}" }, sort_keys=True)) return SettlementManifest( verdict=SettlementVerdict.OFF_CHAIN_ACTIVE, directive=f"PROCEED: Use off-chain settlement. Dispute until {channel.dispute_window_epoch:.0f}.", manifest_digest=digest )