# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-data-retention-decay-policies-must-be-cryptographically-sealed-for-vector-indices.md # Remnant Warrant: https://corpusfetch.com/c/what-data-retention-decay-policies-must-be-cryptographically-sealed-for-vector-indices.warrant.json # Markdown SHA-256: 79bba09af2b13a91840b37f8a2883ea88a3c3e046f62108ccde542569b35e384 # Code AST SHA-256: 1b30f2e7996c0719505e597e3968465400fb503e678a8573da9f6abab56d1f1a # Verified Classes: DecayVerdict, VectorRetentionState, DecayPolicy, DecayManifest, VectorRetentionGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class DecayVerdict(Enum): POLICY_COMPLIANT = auto() DECAY_OVERDUE = auto() STALE_VECTOR_VIOLATION = auto() @dataclass(frozen=True) class VectorRetentionState: policy_hash: str last_decay_epoch: float current_epoch: float decay_interval_sec: float max_age_sec: float oldest_vector_age_sec: float @dataclass(frozen=True) class DecayPolicy: policy_hash: str decay_interval_sec: float max_age_sec: float @dataclass(frozen=True) class DecayManifest: verdict: DecayVerdict directive: str manifest_digest: str class VectorRetentionGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_retention( cls, state: VectorRetentionState, policy: DecayPolicy ) -> DecayManifest: # Gate 1: Policy match if state.policy_hash != policy.policy_hash: return DecayManifest( verdict=DecayVerdict.DECAY_OVERDUE, directive="HALT: Retention policy hash mismatch. Reconcile policy before decay.", manifest_digest=cls._hash("POLICY_MISMATCH") ) # Gate 2: Max age violation if state.oldest_vector_age_sec > policy.max_age_sec: return DecayManifest( verdict=DecayVerdict.STALE_VECTOR_VIOLATION, directive=f"HALT: Oldest vector is {state.oldest_vector_age_sec:.0f}s old, exceeding max {policy.max_age_sec:.0f}s. Purge immediately.", manifest_digest=cls._hash(f"MAX_AGE:{state.oldest_vector_age_sec:.0f}") ) # Gate 3: Decay interval check elapsed = state.current_epoch - state.last_decay_epoch if elapsed > policy.decay_interval_sec: return DecayManifest( verdict=DecayVerdict.DECAY_OVERDUE, directive=f"ACTION: Decay interval exceeded ({elapsed:.0f}s > {policy.decay_interval_sec:.0f}s). Execute scheduled decay.", manifest_digest=cls._hash(f"OVERDUE:{elapsed:.0f}") ) # Gate 4: Compliant digest = cls._hash(json.dumps({ "policy": state.policy_hash, "elapsed": f"{elapsed:.0f}", "oldest": f"{state.oldest_vector_age_sec:.0f}" }, sort_keys=True)) return DecayManifest( verdict=DecayVerdict.POLICY_COMPLIANT, directive="PROCEED: Vector retention within policy bounds.", manifest_digest=digest )