# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-reference-mutation-invariants-prevent-in-place-updates-from-corrupting-downstream-consumers.md # Remnant Warrant: https://corpusfetch.com/c/what-reference-mutation-invariants-prevent-in-place-updates-from-corrupting-downstream-consumers.warrant.json # Markdown SHA-256: 9a4e8afe7e95551d3f067538719118213081d48b2b8448b68fba743b0ee4fac9 # Code AST SHA-256: fc33a30a963389f22a8842f8d7beac44c93a33957dc1fb05243761747d55ee4c # Verified Classes: MutationVerdict, ReferenceState, MutationPolicy, MutationManifest, InPlaceMutationGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class MutationVerdict(Enum): IMMUTABLE_REFERENCE = auto() CONTENT_STABLE = auto() IN_PLACE_MUTATION_DETECTED = auto() MUTATION_ACKNOWLEDGED = auto() @dataclass(frozen=True) class ReferenceState: reference_id: str is_marked_immutable: bool consumer_last_seen_content_version: str current_content_version: str mutation_announcement_received: bool acknowledgment_sent: bool @dataclass(frozen=True) class MutationPolicy: require_acknowledgment_for_critical: bool is_critical: bool @dataclass(frozen=True) class MutationManifest: verdict: MutationVerdict directive: str manifest_digest: str class InPlaceMutationGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_reference( cls, state: ReferenceState, policy: MutationPolicy ) -> MutationManifest: # Gate 1: Immutable if state.is_marked_immutable: digest = cls._hash(json.dumps({ "reference": state.reference_id, "type": "IMMUTABLE" }, sort_keys=True)) return MutationManifest( verdict=MutationVerdict.IMMUTABLE_REFERENCE, directive=f"PROCEED: Reference {state.reference_id} immutable, content stable.", manifest_digest=digest ) # Gate 2: Content unchanged if state.consumer_last_seen_content_version == state.current_content_version: digest = cls._hash(json.dumps({ "reference": state.reference_id, "version": state.current_content_version }, sort_keys=True)) return MutationManifest( verdict=MutationVerdict.CONTENT_STABLE, directive=f"PROCEED: Reference {state.reference_id} content unchanged.", manifest_digest=digest ) # Gate 3: Mutation detected if policy.require_acknowledgment_for_critical and policy.is_critical: if not state.mutation_announcement_received: return MutationManifest( verdict=MutationVerdict.IN_PLACE_MUTATION_DETECTED, directive=f"HALT: Reference {state.reference_id} mutated silently (critical).", manifest_digest=cls._hash(f"SILENT:{state.reference_id}") ) if not state.acknowledgment_sent: return MutationManifest( verdict=MutationVerdict.IN_PLACE_MUTATION_DETECTED, directive=f"HALT: Reference {state.reference_id} mutation announcement not acknowledged.", manifest_digest=cls._hash(f"NOT_ACK:{state.reference_id}") ) # Gate 4: Acknowledged digest = cls._hash(json.dumps({ "reference": state.reference_id, "previous": state.consumer_last_seen_content_version, "current": state.current_content_version, "announced": str(state.mutation_announcement_received), "acknowledged": str(state.acknowledgment_sent) }, sort_keys=True)) return MutationManifest( verdict=MutationVerdict.MUTATION_ACKNOWLEDGED, directive=f"PROCEED_WITH_UPDATE: Reference {state.reference_id} mutated from {state.consumer_last_seen_content_version} to {state.current_content_version}.", manifest_digest=digest )