# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-consolidate-memory-during-idle-periods-without-losing-recently-acquired-state.md # Remnant Warrant: https://corpusfetch.com/c/how-must-an-instance-consolidate-memory-during-idle-periods-without-losing-recently-acquired-state.warrant.json # Markdown SHA-256: fb61c23708eb80a667e03c0b3c610363a149bc4f2fd3626bd014cc3d19db6939 # Code AST SHA-256: 478b5cb2c1fb0b0496e94c80018df3c8fec89ca595e595fcf9a7e10ae5a854e2 # Verified Classes: ValidationStatus, ConsolidationVerdict, MemoryRecord, ConsolidationContext, ConsolidationPolicy, ConsolidationManifest, ConsolidationGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class ValidationStatus(Enum): VALIDATED_BY_PROBE = 3 VALIDATED_BY_REPEAT = 2 PROVISIONAL_UNVALIDATED = 1 class ConsolidationVerdict(Enum): NO_CONSOLIDATION = auto() SAFE_TO_CONSOLIDATE = auto() FRESH_EVIDENCE_PROTECTED = auto() PREMATURE_CONSOLIDATION = auto() @dataclass(frozen=True) class MemoryRecord: memory_id: str age_sec: float validation_status: ValidationStatus content_hash: str @dataclass(frozen=True) class ConsolidationContext: memories: tuple[MemoryRecord, ...] consolidation_action: str # "PRUNE" | "MERGE" | "REEMBED" @dataclass(frozen=True) class ConsolidationPolicy: fresh_reservation_sec: float require_validated_for_prune: bool @dataclass(frozen=True) class ConsolidationManifest: verdict: ConsolidationVerdict directive: str manifest_digest: str class ConsolidationGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_action( cls, ctx: ConsolidationContext, policy: ConsolidationPolicy ) -> ConsolidationManifest: # Gate 1: No consolidation if ctx.consolidation_action == "NONE": digest = cls._hash(json.dumps({ "status": "NO_CONSOLIDATION" }, sort_keys=True)) return ConsolidationManifest( verdict=ConsolidationVerdict.NO_CONSOLIDATION, directive="PROCEED: No consolidation action.", manifest_digest=digest ) # Gate 2: Fresh memory check fresh = [ m for m in ctx.memories if m.age_sec < policy.fresh_reservation_sec ] if fresh and ctx.consolidation_action in ("PRUNE", "MERGE"): return ConsolidationManifest( verdict=ConsolidationVerdict.FRESH_EVIDENCE_PROTECTED, directive=f"HALT: {len(fresh)} fresh memories within {policy.fresh_reservation_sec:.0f}s must be reserved.", manifest_digest=cls._hash(f"FRESH_PROTECT:{len(fresh)}") ) # Gate 3: Validation for prune if policy.require_validated_for_prune and ctx.consolidation_action == "PRUNE": unvalidated = [ m for m in ctx.memories if m.validation_status == ValidationStatus.PROVISIONAL_UNVALIDATED ] if unvalidated: return ConsolidationManifest( verdict=ConsolidationVerdict.PREMATURE_CONSOLIDATION, directive=f"HALT: {len(unvalidated)} provisional memories targeted for prune.", manifest_digest=cls._hash(f"PREMATURE:{len(unvalidated)}") ) # Gate 4: Safe digest = cls._hash(json.dumps({ "action": ctx.consolidation_action, "memory_count": str(len(ctx.memories)), "fresh_reserved": str(len(fresh)), "validated_only": str(all( m.validation_status != ValidationStatus.PROVISIONAL_UNVALIDATED for m in ctx.memories )) }, sort_keys=True)) return ConsolidationManifest( verdict=ConsolidationVerdict.SAFE_TO_CONSOLIDATE, directive=f"PROCEED: {len(ctx.memories)} memories safe for {ctx.consolidation_action}.", manifest_digest=digest )