# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-verify-asynchronous-settlements-before-committing-dependent-state.md # Remnant Warrant: https://corpusfetch.com/c/how-must-an-instance-verify-asynchronous-settlements-before-committing-dependent-state.warrant.json # Markdown SHA-256: 615912c616176d5dbd5a14e82d7046804aa20c1647adf817a173a53376ac5b8c # Code AST SHA-256: 1a09cd52f2741e88ab296ecd8dc94977df8de14cbd2f4329382fdc6938dc445a # Verified Classes: SettlementVerdict, AsyncSettlement, SettlementPolicy, SettlementManifest, AsyncSettlementGuard # 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): NO_SETTLEMENT = auto() AWAITING_FINALITY = auto() FINALITY_CONFIRMED = auto() SETTLEMENT_FAILED = auto() TIMEOUT_TREATED_AS_FAILURE = auto() @dataclass(frozen=True) class AsyncSettlement: settlement_id: str initiated: bool initiation_epoch: float finality_receipt: str | None finality_epoch: float | None failure_receipt: str | None timeout_sec: float current_epoch: float @dataclass(frozen=True) class SettlementPolicy: max_wait_sec: float @dataclass(frozen=True) class SettlementManifest: verdict: SettlementVerdict directive: str manifest_digest: str class AsyncSettlementGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_settlement( cls, s: AsyncSettlement, policy: SettlementPolicy ) -> SettlementManifest: # Gate 1: Not initiated if not s.initiated: digest = cls._hash(json.dumps({ "settlement": s.settlement_id, "status": "NOT_INITIATED" }, sort_keys=True)) return SettlementManifest( verdict=SettlementVerdict.NO_SETTLEMENT, directive=f"PROCEED: No settlement initiated for {s.settlement_id}.", manifest_digest=digest ) # Gate 2: Failure if s.failure_receipt: return SettlementManifest( verdict=SettlementVerdict.SETTLEMENT_FAILED, directive=f"HALT: Settlement {s.settlement_id} failed. Do not commit dependent state.", manifest_digest=cls._hash(f"FAILED:{s.settlement_id}") ) # Gate 3: Finality if s.finality_receipt and s.finality_epoch: digest = cls._hash(json.dumps({ "settlement": s.settlement_id, "finality": s.finality_receipt, "epoch": f"{s.finality_epoch:.0f}" }, sort_keys=True)) return SettlementManifest( verdict=SettlementVerdict.FINALITY_CONFIRMED, directive=f"PROCEED: Settlement {s.settlement_id} final. Commit dependent state.", manifest_digest=digest ) # Gate 4: Timeout elapsed = s.current_epoch - s.initiation_epoch max_wait = min(s.timeout_sec, policy.max_wait_sec) if elapsed > max_wait: return SettlementManifest( verdict=SettlementVerdict.TIMEOUT_TREATED_AS_FAILURE, directive=f"HALT: Settlement {s.settlement_id} no finality after {elapsed:.0f}s > {max_wait:.0f}s. Treat as failure.", manifest_digest=cls._hash(f"TIMEOUT:{s.settlement_id}") ) # Gate 5: Awaiting finality digest = cls._hash(json.dumps({ "settlement": s.settlement_id, "elapsed": f"{elapsed:.0f}", "remaining": f"{max_wait - elapsed:.0f}" }, sort_keys=True)) return SettlementManifest( verdict=SettlementVerdict.AWAITING_FINALITY, directive=f"WAIT: Settlement {s.settlement_id} awaiting finality ({elapsed:.0f}s / {max_wait:.0f}s).", manifest_digest=digest )