# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-memory-durability-invariants-must-survive-when-disk-pressure-forces-eviction.md # Remnant Warrant: https://corpusfetch.com/c/what-memory-durability-invariants-must-survive-when-disk-pressure-forces-eviction.warrant.json # Markdown SHA-256: fae73074d712c03a08c854c6569a766d660ed72bd08570a11753ccea0f1bec1c # Code AST SHA-256: db82d158bc885b5ad78c72b566d77afb02ee7bdfa4908701a1224ab0456c1b83 # Verified Classes: DurabilityVerdict, MemoryEntry, DiskState, EvictionPolicy, EvictionManifest, MemoryDurabilityGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class DurabilityVerdict(Enum): NO_PRESSURE = auto() EVICT_REDUNDANT = auto() EVICT_RECONSTRUCTIBLE = auto() EVICT_IRREPLACEABLE_LAST_RESORT = auto() @dataclass(frozen=True) class MemoryEntry: memory_id: str durability_class: str # "IRREPLACEABLE" | "RECONSTRUCTIBLE_FROM_ORIGIN" | "RECONSTRUCTIBLE_FROM_PEER" | "REDUNDANT" reconstruction_cost: float size_bytes: int is_critical: bool @dataclass(frozen=True) class DiskState: current_usage_bytes: int capacity_bytes: int pressure_threshold_pct: float @dataclass(frozen=True) class EvictionPolicy: require_durability_class: bool @dataclass(frozen=True) class EvictionManifest: verdict: DurabilityVerdict eviction_candidates: tuple[str, ...] directive: str manifest_digest: str class MemoryDurabilityGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_eviction( cls, memories: tuple[MemoryEntry, ...], disk: DiskState, policy: EvictionPolicy ) -> EvictionManifest: # Gate 1: No pressure usage_pct = (disk.current_usage_bytes / disk.capacity_bytes) * 100 if disk.capacity_bytes > 0 else 0 if usage_pct < disk.pressure_threshold_pct: digest = cls._hash(json.dumps({ "usage_pct": f"{usage_pct:.1f}" }, sort_keys=True)) return EvictionManifest( verdict=DurabilityVerdict.NO_PRESSURE, eviction_candidates=tuple(), directive=f"PROCEED: Usage {usage_pct:.1f}% below pressure threshold {disk.pressure_threshold_pct:.1f}%.", manifest_digest=digest ) # Gate 2: Durability classification if policy.require_durability_class: unclassified = [m.memory_id for m in memories if not m.durability_class] if unclassified: return EvictionManifest( verdict=DurabilityVerdict.EVICT_IRREPLACEABLE_LAST_RESORT, eviction_candidates=tuple(), directive=f"HALT: {len(unclassified)} memories unclassified. Treat all as irreplaceable.", manifest_digest=cls._hash(f"UNCLASSIFIED:{len(unclassified)}") ) # Gate 3: Prioritize eviction redundant = [m.memory_id for m in memories if m.durability_class == "REDUNDANT"] reconstructible = sorted( [m for m in memories if m.durability_class.startswith("RECONSTRUCTIBLE")], key=lambda m: m.reconstruction_cost # evict cheapest to reconstruct first ) irreplaceable = [m.memory_id for m in memories if m.durability_class == "IRREPLACEABLE"] if redundant: digest = cls._hash(json.dumps({ "candidates": tuple(sorted(redundant)) }, sort_keys=True)) return EvictionManifest( verdict=DurabilityVerdict.EVICT_REDUNDANT, eviction_candidates=tuple(sorted(redundant)), directive=f"PROCEED: Evict {len(redundant)} redundant memories.", manifest_digest=digest ) if reconstructible: digest = cls._hash(json.dumps({ "candidates": tuple(m.memory_id for m in reconstructible), "costs": tuple(f"{m.reconstruction_cost:.2f}" for m in reconstructible) }, sort_keys=True)) return EvictionManifest( verdict=DurabilityVerdict.EVICT_RECONSTRUCTIBLE, eviction_candidates=tuple(m.memory_id for m in reconstructible), directive=f"PROCEED: Evict {len(reconstructible)} reconstructible memories (cheapest first).", manifest_digest=digest ) # Gate 4: Last resort irreplaceable return EvictionManifest( verdict=DurabilityVerdict.EVICT_IRREPLACEABLE_LAST_RESORT, eviction_candidates=tuple(sorted(irreplaceable)), directive=f"HALT_LAST_RESORT: Only irreplaceable memories remain ({len(irreplaceable)}). Requires operator confirmation.", manifest_digest=cls._hash(f"IRREPLACEABLE:{len(irreplaceable)}") )