# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-preserve-decision-lineage-when-audit-trails-are-compressed.md # Remnant Warrant: https://corpusfetch.com/c/how-must-an-instance-preserve-decision-lineage-when-audit-trails-are-compressed.warrant.json # Markdown SHA-256: e5742883cf7ad617cf1e1d3a46e52f64f40a3aea14ac172d44d6a42c310b88dd # Code AST SHA-256: 6163f669aac98f7d698c521be37d76cd947d94ff09f74c491239565034881012 # Verified Classes: LineageVerdict, Decision, CompressionState, LineagePolicy, LineageManifest, AuditTrailCompressionGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class LineageVerdict(Enum): NO_COMPRESSION = auto() LINEAGE_PRESERVED = auto() LINEAGE_LOST = auto() RECONSTRUCTION_AVAILABLE = auto() @dataclass(frozen=True) class Decision: decision_id: str evidence_hashes: tuple[str, ...] binding_hash: str | None @dataclass(frozen=True) class CompressionState: decisions: tuple[Decision, ...] aggregation_tree_present: bool post_compression_verification_passed: bool @dataclass(frozen=True) class LineagePolicy: require_lineage_preservation: bool @dataclass(frozen=True) class LineageManifest: verdict: LineageVerdict directive: str manifest_digest: str class AuditTrailCompressionGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_compression( cls, state: CompressionState, policy: LineagePolicy ) -> LineageManifest: # Gate 1: No compression, no concern if not state.aggregation_tree_present and not policy.require_lineage_preservation: digest = cls._hash(json.dumps({ "decisions": str(len(state.decisions)), "compression": "NONE" }, sort_keys=True)) return LineageManifest( verdict=LineageVerdict.NO_COMPRESSION, directive="PROCEED: No compression, full audit trail preserved.", manifest_digest=digest ) # Gate 2: Binding hash presence unbound = [d for d in state.decisions if not d.binding_hash] if unbound: return LineageManifest( verdict=LineageVerdict.LINEAGE_LOST, directive=f"HALT: {len(unbound)} decisions lack binding hash. Lineage will be lost.", manifest_digest=cls._hash(f"UNBOUND:{len(unbound)}") ) # Gate 3: Aggregation tree if policy.require_lineage_preservation and not state.aggregation_tree_present: return LineageManifest( verdict=LineageVerdict.LINEAGE_LOST, directive="HALT: Compression lacks aggregation tree. Cannot reconstruct per-decision evidence.", manifest_digest=cls._hash("NO_TREE") ) # Gate 4: Post-compression verification if not state.post_compression_verification_passed: return LineageManifest( verdict=LineageVerdict.RECONSTRUCTION_AVAILABLE, directive="WAIT: Post-compression verification pending.", manifest_digest=cls._hash("VERIFICATION_PENDING") ) # Gate 5: Verified digest = cls._hash(json.dumps({ "decisions": str(len(state.decisions)), "tree": "PRESENT", "verification": "PASSED" }, sort_keys=True)) return LineageManifest( verdict=LineageVerdict.LINEAGE_PRESERVED, directive=f"PROCEED: Lineage preserved for {len(state.decisions)} decisions.", manifest_digest=digest )