# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-search-budget-allocation-invariants-prevent-discovery-starvation-in-cost-constrained-agent-networks.md # Remnant Warrant: https://corpusfetch.com/c/what-search-budget-allocation-invariants-prevent-discovery-starvation-in-cost-constrained-agent-networks.warrant.json # Markdown SHA-256: a0a2ad0d5afff5f198dddf08c08fcb1f824707b3731397d4f309cf3c9d7c5796 # Code AST SHA-256: 8864de64f27e447654846671e528257f2079e2e4ddc77da3d2f1511282c721bf # Verified Classes: BudgetVerdict, DiscoveryNeed, BudgetState, BudgetPolicy, BudgetManifest, SearchBudgetGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class BudgetVerdict(Enum): UNLIMITED_BUDGET = auto() SPEND_APPROVED = auto() SPEND_BLOCKED_BY_RESERVE = auto() SPEND_BLOCKED_BY_PRIORITY = auto() @dataclass(frozen=True) class DiscoveryNeed: need_id: str priority: str # "BLOCKING" | "IMPORTANT" | "NICE_TO_HAVE" estimated_cost: float is_reserved: bool @dataclass(frozen=True) class BudgetState: total_budget: float spent: float reserved: float @dataclass(frozen=True) class BudgetPolicy: min_remaining_pct: float @dataclass(frozen=True) class BudgetManifest: verdict: BudgetVerdict remaining_after: float directive: str manifest_digest: str class SearchBudgetGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_spend( cls, need: DiscoveryNeed, state: BudgetState, policy: BudgetPolicy ) -> BudgetManifest: # Gate 1: Unlimited if state.total_budget == 0: # 0 = unlimited sentinel digest = cls._hash(json.dumps({ "need": need.need_id, "budget": "UNLIMITED" }, sort_keys=True)) return BudgetManifest( verdict=BudgetVerdict.UNLIMITED_BUDGET, remaining_after=-1.0, directive=f"PROCEED: Unlimited budget, spend on {need.need_id}.", manifest_digest=digest ) available = state.total_budget - state.spent - state.reserved # Gate 2: Reserve check if need.estimated_cost > available: return BudgetManifest( verdict=BudgetVerdict.SPEND_BLOCKED_BY_RESERVE, remaining_after=available, directive=f"HALT: Cost {need.estimated_cost:.2f} > available {available:.2f} (reserved: {state.reserved:.2f}).", manifest_digest=cls._hash(f"RESERVED:{need.need_id}") ) # Gate 3: Priority check (non-blocking needs must preserve min remaining) remaining_after = available - need.estimated_cost min_remaining = state.total_budget * policy.min_remaining_pct if need.priority != "BLOCKING" and remaining_after < min_remaining: return BudgetManifest( verdict=BudgetVerdict.SPEND_BLOCKED_BY_PRIORITY, remaining_after=remaining_after, directive=f"HALT: Non-blocking {need.need_id} would leave {remaining_after:.2f} < min {min_remaining:.2f}.", manifest_digest=cls._hash(f"PRIORITY:{need.need_id}") ) # Gate 4: Approved digest = cls._hash(json.dumps({ "need": need.need_id, "priority": need.priority, "cost": f"{need.estimated_cost:.2f}", "remaining_after": f"{remaining_after:.2f}" }, sort_keys=True)) return BudgetManifest( verdict=BudgetVerdict.SPEND_APPROVED, remaining_after=remaining_after, directive=f"PROCEED: Spend {need.estimated_cost:.2f} on {need.need_id}, {remaining_after:.2f} remaining.", manifest_digest=digest )