# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-track-cumulative-retry-cost-across-escalating-failure.md # Remnant Warrant: https://corpusfetch.com/c/how-must-an-instance-track-cumulative-retry-cost-across-escalating-failure.warrant.json # Markdown SHA-256: b85c7ba515964699168995eeb2977e095decbf556a1142d3d7552eaf58dad27e # Code AST SHA-256: aa3dc2adfb49c246de760d5645ced7b9ed43896a17e97f9369863c2d2b69d14a # Verified Classes: RetryVerdict, RetryAttempt, RetryState, RetryPolicy, RetryManifest, RetryCostGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class RetryVerdict(Enum): NO_RETRY_NEEDED = auto() RETRY_BUDGET_AVAILABLE = auto() RETRY_BUDGET_EXHAUSTED = auto() ABANDON_OPERATION = auto() @dataclass(frozen=True) class RetryAttempt: attempt_number: int attempt_cost: float cumulative_cost_so_far: float @dataclass(frozen=True) class RetryState: operation_id: str primary_cost: float retry_budget: float retry_attempts: tuple[RetryAttempt, ...] @dataclass(frozen=True) class RetryPolicy: max_retry_cost_multiplier: float max_attempts: int @dataclass(frozen=True) class RetryManifest: verdict: RetryVerdict cumulative_cost: float directive: str manifest_digest: str class RetryCostGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_retry( cls, state: RetryState, policy: RetryPolicy, next_attempt_cost: float ) -> RetryManifest: cumulative_cost = sum(a.attempt_cost for a in state.retry_attempts) # Gate 1: Attempt count limit if len(state.retry_attempts) >= policy.max_attempts: return RetryManifest( verdict=RetryVerdict.ABANDON_OPERATION, cumulative_cost=cumulative_cost, directive=f"HALT: Reached max attempts {policy.max_attempts}, cost {cumulative_cost:.2f}. Abandon.", manifest_digest=cls._hash(f"MAX_ATTEMPTS:{cumulative_cost:.2f}") ) # Gate 2: Cost multiplier limit cost_limit = state.primary_cost * policy.max_retry_cost_multiplier projected_cost = cumulative_cost + next_attempt_cost if projected_cost > cost_limit: return RetryManifest( verdict=RetryVerdict.ABANDON_OPERATION, cumulative_cost=cumulative_cost, directive=f"HALT: Projected cost {projected_cost:.2f} > limit {cost_limit:.2f}. Abandon.", manifest_digest=cls._hash(f"COST_LIMIT:{cost_limit:.2f}") ) # Gate 3: Budget check if projected_cost > state.retry_budget: return RetryManifest( verdict=RetryVerdict.RETRY_BUDGET_EXHAUSTED, cumulative_cost=cumulative_cost, directive=f"HALT: Projected cost {projected_cost:.2f} > retry budget {state.retry_budget:.2f}.", manifest_digest=cls._hash(f"BUDGET:{state.retry_budget:.2f}") ) # Gate 4: Retry allowed digest = cls._hash(json.dumps({ "operation": state.operation_id, "cumulative": f"{cumulative_cost:.2f}", "projected": f"{projected_cost:.2f}", "budget": f"{state.retry_budget:.2f}", "attempts": str(len(state.retry_attempts)) }, sort_keys=True)) return RetryManifest( verdict=RetryVerdict.RETRY_BUDGET_AVAILABLE, cumulative_cost=projected_cost, directive=f"PROCEED: Retry cost {projected_cost:.2f} within budget {state.retry_budget:.2f}.", manifest_digest=digest )