# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-discovery-invariants-prevent-infinite-refund-loops-in-autonomous-barter-networks.md # Remnant Warrant: https://corpusfetch.com/c/what-discovery-invariants-prevent-infinite-refund-loops-in-autonomous-barter-networks.warrant.json # Markdown SHA-256: c1515bbc79f857481d0ff4376605e510c0111904d7cc8505b812341d000e88ed # Code AST SHA-256: 9a124e8c3fc693c3dc39e3a5c4f522605d6df07aac9f1367788ca4a47ab0129f # Verified Classes: RefundVerdict, RefundRequest, NetworkState, RefundManifest, BarterRefundGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class RefundVerdict(Enum): REFUND_APPROVED = auto() CYCLE_DETECTED = auto() DUPLICATE_REFUND = auto() @dataclass(frozen=True) class RefundRequest: refund_id: str original_transaction_id: str sender_id: str receiver_id: str amount: int epoch: int @dataclass(frozen=True) class NetworkState: existing_refund_edges: frozenset[tuple[str, str]] refunded_transactions: frozenset[str] current_epoch: int @dataclass(frozen=True) class RefundManifest: verdict: RefundVerdict directive: str manifest_digest: str class BarterRefundGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_refund( cls, req: RefundRequest, state: NetworkState ) -> RefundManifest: # Gate 1: Duplicate refund check if req.original_transaction_id in state.refunded_transactions: return RefundManifest( verdict=RefundVerdict.DUPLICATE_REFUND, directive=f"REJECT: Transaction {req.original_transaction_id} already refunded.", manifest_digest=cls._hash(f"DUP:{req.original_transaction_id}") ) # Gate 2: Cycle detection (simplified DFS) new_edge = (req.sender_id, req.receiver_id) edge_set = set(state.existing_refund_edges) edge_set.add(new_edge) # Build adjacency list adj: dict[str, list[str]] = {} for src, dst in edge_set: adj.setdefault(src, []).append(dst) # DFS to detect cycle def has_cycle(node: str, visited: set[str], rec_stack: set[str]) -> bool: visited.add(node) rec_stack.add(node) for neighbor in adj.get(node, []): if neighbor not in visited: if has_cycle(neighbor, visited, rec_stack): return True elif neighbor in rec_stack: return True rec_stack.remove(node) return False visited = set() rec_stack = set() for node in adj: if node not in visited: if has_cycle(node, visited, rec_stack): return RefundManifest( verdict=RefundVerdict.CYCLE_DETECTED, directive=f"REJECT: Refund cycle detected involving {req.sender_id} and {req.receiver_id}.", manifest_digest=cls._hash(f"CYCLE:{req.sender_id}") ) # Gate 3: Approved digest = cls._hash(json.dumps({ "refund": req.refund_id, "txn": req.original_transaction_id, "edge": f"{req.sender_id}->{req.receiver_id}" }, sort_keys=True)) return RefundManifest( verdict=RefundVerdict.REFUND_APPROVED, directive="PROCEED: Refund is valid and introduces no cycles.", manifest_digest=digest )