# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-resource-leak-invariants-must-survive-session-termination.md # Remnant Warrant: https://corpusfetch.com/c/what-resource-leak-invariants-must-survive-session-termination.warrant.json # Markdown SHA-256: 13921c0790249dcd13e0bc5c8d38399a4b98075bb0b26cc6e8f8b6eb9c511984 # Code AST SHA-256: 4e6519d76eb4083df319b36fe2f970527cfbc2f5c9a82b430fcc88606a9441f3 # Verified Classes: ResourceVerdict, AllocatedResource, SessionTermination, ResourcePolicy, ResourceManifest, ResourceLeakGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class ResourceVerdict(Enum): NO_EXTERNAL_RESOURCES = auto() CLEANUP_VERIFIED = auto() LEAK_DETECTED = auto() QUARANTINE_REQUIRED = auto() @dataclass(frozen=True) class AllocatedResource: resource_id: str resource_type: str release_receipt: str | None is_force_releasable: bool @dataclass(frozen=True) class SessionTermination: session_id: str allocated_resources: tuple[AllocatedResource, ...] runtime_guarantees_cleanup: bool @dataclass(frozen=True) class ResourcePolicy: require_cleanup_verification: bool @dataclass(frozen=True) class ResourceManifest: verdict: ResourceVerdict unreleased_count: int directive: str manifest_digest: str class ResourceLeakGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_termination( cls, state: SessionTermination, policy: ResourcePolicy ) -> ResourceManifest: # Gate 1: No resources if not state.allocated_resources: digest = cls._hash(json.dumps({ "session": state.session_id, "resources": "NONE" }, sort_keys=True)) return ResourceManifest( verdict=ResourceVerdict.NO_EXTERNAL_RESOURCES, unreleased_count=0, directive=f"PROCEED: Session {state.session_id} allocated no external resources.", manifest_digest=digest ) # Gate 2: Runtime cleanup guarantee if state.runtime_guarantees_cleanup: digest = cls._hash(json.dumps({ "session": state.session_id, "resources": str(len(state.allocated_resources)), "cleanup": "RUNTIME_GUARANTEED" }, sort_keys=True)) return ResourceManifest( verdict=ResourceVerdict.CLEANUP_VERIFIED, unreleased_count=0, directive="PROCEED: Runtime guarantees cleanup.", manifest_digest=digest ) # Gate 3: Explicit release verification unreleased = [r for r in state.allocated_resources if not r.release_receipt] if not unreleased: digest = cls._hash(json.dumps({ "session": state.session_id, "released": str(len(state.allocated_resources)) }, sort_keys=True)) return ResourceManifest( verdict=ResourceVerdict.CLEANUP_VERIFIED, unreleased_count=0, directive=f"PROCEED: All {len(state.allocated_resources)} resources explicitly released.", manifest_digest=digest ) # Gate 4: Leak handling non_forceable = [r for r in unreleased if not r.is_force_releasable] if non_forceable: return ResourceManifest( verdict=ResourceVerdict.LEAK_DETECTED, unreleased_count=len(unreleased), directive=f"HALT: {len(non_forceable)} unreleased non-forceable resources. Successors will block.", manifest_digest=cls._hash(f"LEAK:{len(non_forceable)}") ) digest = cls._hash(json.dumps({ "session": state.session_id, "unreleased": tuple(sorted(r.resource_id for r in unreleased)) }, sort_keys=True)) return ResourceManifest( verdict=ResourceVerdict.QUARANTINE_REQUIRED, unreleased_count=len(unreleased), directive=f"ACTION: {len(unreleased)} resources left unreleased. Recorded for successor force-release.", manifest_digest=digest )