# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-handle-tool-side-state-leakage-between-invocations.md # Remnant Warrant: https://corpusfetch.com/c/how-must-an-instance-handle-tool-side-state-leakage-between-invocations.warrant.json # Markdown SHA-256: dbafad1726145d47bc0aba29d932cfd0005570fb495f456358d892413e165ff4 # Code AST SHA-256: bb73c352e617b09ca1030fe681341762d3f2fa67cd54a775c9a1a42dba58c004 # Verified Classes: StateVerdict, ToolInvocation, StateLeakContext, StateLeakPolicy, StateLeakManifest, ToolStateLeakageGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class StateVerdict(Enum): STATELESS_TOOL = auto() ISOLATED_INVOCATIONS = auto() STATE_LEAKAGE_DETECTED = auto() SESSION_LIMIT_EXHAUSTED = auto() @dataclass(frozen=True) class ToolInvocation: invocation_id: str tool_id: str has_session_state: bool state_delta_hash: str | None session_usage_ratio: float result_hash: str @dataclass(frozen=True) class StateLeakContext: prior_state_deltas: tuple[str, ...] isolation_policy: str # "ISOLATED" | "SESSION_SHARED" | "AGENT_SHARED" @dataclass(frozen=True) class StateLeakPolicy: require_isolation_for_critical: bool max_session_usage_ratio: float @dataclass(frozen=True) class StateLeakManifest: verdict: StateVerdict directive: str manifest_digest: str class ToolStateLeakageGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_invocation( cls, inv: ToolInvocation, ctx: StateLeakContext, policy: StateLeakPolicy ) -> StateLeakManifest: # Gate 1: Stateless if not inv.has_session_state: digest = cls._hash(json.dumps({ "invocation": inv.invocation_id, "tool": inv.tool_id, "type": "STATELESS" }, sort_keys=True)) return StateLeakManifest( verdict=StateVerdict.STATELESS_TOOL, directive=f"PROCEED: Tool {inv.tool_id} is stateless.", manifest_digest=digest ) # Gate 2: Session limit if inv.session_usage_ratio >= policy.max_session_usage_ratio: return StateLeakManifest( verdict=StateVerdict.SESSION_LIMIT_EXHAUSTED, directive=f"HALT: Session usage {inv.session_usage_ratio:.2%} >= {policy.max_session_usage_ratio:.2%}.", manifest_digest=cls._hash(f"SESSION_LIMIT:{inv.invocation_id}") ) # Gate 3: Isolation policy if policy.require_isolation_for_critical and ctx.isolation_policy != "ISOLATED": return StateLeakManifest( verdict=StateVerdict.STATE_LEAKAGE_DETECTED, directive=f"HALT: Policy requires isolation but tool is {ctx.isolation_policy}.", manifest_digest=cls._hash(f"NO_ISOLATION:{inv.invocation_id}") ) # Gate 4: State delta accumulation check if ctx.prior_state_deltas and len(ctx.prior_state_deltas) >= 3: digest = cls._hash(json.dumps({ "invocation": inv.invocation_id, "tool": inv.tool_id, "prior_deltas": str(len(ctx.prior_state_deltas)), "current_delta": inv.state_delta_hash }, sort_keys=True)) return StateLeakManifest( verdict=StateVerdict.STATE_LEAKAGE_DETECTED, directive=f"WARN: {len(ctx.prior_state_deltas)} prior state deltas accumulated for {inv.tool_id}.", manifest_digest=digest ) # Gate 5: Isolated digest = cls._hash(json.dumps({ "invocation": inv.invocation_id, "tool": inv.tool_id, "policy": ctx.isolation_policy, "session_usage": f"{inv.session_usage_ratio:.2%}" }, sort_keys=True)) return StateLeakManifest( verdict=StateVerdict.ISOLATED_INVOCATIONS, directive=f"PROCEED: Invocation {inv.invocation_id} with {ctx.isolation_policy} isolation.", manifest_digest=digest )