# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-bound-cost-cascades-when-agents-recursively-invoke-peers.md # Remnant Warrant: https://corpusfetch.com/c/how-must-an-instance-bound-cost-cascades-when-agents-recursively-invoke-peers.warrant.json # Markdown SHA-256: b5defc7e52428e873d01fc296f3c462dbbe1a3f2464ec43388ea410c069ad774 # Code AST SHA-256: a2cf5b6a056a49649d8a7c97e847751280f685d18bc5360e3dbf7f311d8edbb6 # Verified Classes: CascadeVerdict, InvocationNode, CascadeState, CascadePolicy, CascadeManifest, RecursiveCostGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class CascadeVerdict(Enum): FLAT_INVOCATION = auto() CASCADE_BUDGET_OK = auto() CASCADE_BUDGET_EXCEEDED = auto() CASCADE_DEPTH_EXCEEDED = auto() @dataclass(frozen=True) class InvocationNode: node_id: str depth: int allocated_budget: float consumed_budget: float children: tuple[str, ...] @dataclass(frozen=True) class CascadeState: cascade_id: str nodes: tuple[InvocationNode, ...] total_budget: float max_depth: int @dataclass(frozen=True) class CascadePolicy: require_budget_allocation: bool @dataclass(frozen=True) class CascadeManifest: verdict: CascadeVerdict total_consumed: float max_depth_reached: int directive: str manifest_digest: str class RecursiveCostGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_cascade( cls, state: CascadeState, policy: CascadePolicy ) -> CascadeManifest: # Gate 1: Flat invocation max_depth_reached = max((n.depth for n in state.nodes), default=0) if max_depth_reached <= 1: digest = cls._hash(json.dumps({ "cascade": state.cascade_id, "type": "FLAT" }, sort_keys=True)) return CascadeManifest( verdict=CascadeVerdict.FLAT_INVOCATION, total_consumed=sum(n.consumed_budget for n in state.nodes), max_depth_reached=max_depth_reached, directive=f"PROCEED: Flat invocation, no cascade.", manifest_digest=digest ) # Gate 2: Depth if max_depth_reached > state.max_depth: return CascadeManifest( verdict=CascadeVerdict.CASCADE_DEPTH_EXCEEDED, total_consumed=sum(n.consumed_budget for n in state.nodes), max_depth_reached=max_depth_reached, directive=f"HALT: Depth {max_depth_reached} > {state.max_depth}.", manifest_digest=cls._hash(f"DEPTH:{max_depth_reached}") ) # Gate 3: Budget total_consumed = sum(n.consumed_budget for n in state.nodes) if total_consumed > state.total_budget: return CascadeManifest( verdict=CascadeVerdict.CASCADE_BUDGET_EXCEEDED, total_consumed=total_consumed, max_depth_reached=max_depth_reached, directive=f"HALT: Consumed {total_consumed:.2f} > budget {state.total_budget:.2f}.", manifest_digest=cls._hash(f"OVER_BUDGET:{total_consumed:.2f}") ) # Gate 4: Per-node check if policy.require_budget_allocation: over_allocated = [ n.node_id for n in state.nodes if n.consumed_budget > n.allocated_budget ] if over_allocated: return CascadeManifest( verdict=CascadeVerdict.CASCADE_BUDGET_EXCEEDED, total_consumed=total_consumed, max_depth_reached=max_depth_reached, directive=f"HALT: {len(over_allocated)} nodes exceeded allocation.", manifest_digest=cls._hash(f"NODE_OVER:{len(over_allocated)}") ) # Gate 5: OK digest = cls._hash(json.dumps({ "cascade": state.cascade_id, "nodes": str(len(state.nodes)), "consumed": f"{total_consumed:.2f}", "depth": str(max_depth_reached) }, sort_keys=True)) return CascadeManifest( verdict=CascadeVerdict.CASCADE_BUDGET_OK, total_consumed=total_consumed, max_depth_reached=max_depth_reached, directive=f"PROCEED: Cascade within budget ({total_consumed:.2f}/{state.total_budget:.2f}), depth {max_depth_reached}.", manifest_digest=digest )