# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-detect-and-quarantine-cooperative-memory-contamination.md # Remnant Warrant: https://corpusfetch.com/c/how-must-an-instance-detect-and-quarantine-cooperative-memory-contamination.warrant.json # Markdown SHA-256: 90ba8ec1e5d78abff441525cdd05d25b61c11529812119c6289de26d4d49a737 # Code AST SHA-256: 3b23e5bacebac58866a27656b69833b25181dab93d57de1caf0e0c62e4ed4435 # Verified Classes: ContaminationVerdict, SharedMemoryWrite, MemoryContext, ContaminationPolicy, ContaminationManifest, CooperativeMemoryContaminationGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class ContaminationVerdict(Enum): NO_SHARED_MEMORY = auto() HIGH_QUALITY_WRITE = auto() SPECULATIVE_QUARANTINE = auto() LOW_QUALITY_AGENT = auto() @dataclass(frozen=True) class SharedMemoryWrite: write_id: str writing_agent: str evidence_class: str # "FACT" | "INFERENCE" | "SPECULATION" writing_agent_quality: float has_ground_truth_anchor: bool @dataclass(frozen=True) class MemoryContext: shared_memory_active: bool @dataclass(frozen=True) class ContaminationPolicy: min_agent_quality: float require_ground_truth_for_fact: bool @dataclass(frozen=True) class ContaminationManifest: verdict: ContaminationVerdict trust_weight: float directive: str manifest_digest: str class CooperativeMemoryContaminationGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_write( cls, write: SharedMemoryWrite, ctx: MemoryContext, policy: ContaminationPolicy ) -> ContaminationManifest: # Gate 1: No shared memory if not ctx.shared_memory_active: digest = cls._hash(json.dumps({ "write": write.write_id, "type": "ISOLATED" }, sort_keys=True)) return ContaminationManifest( verdict=ContaminationVerdict.NO_SHARED_MEMORY, trust_weight=1.0, directive=f"PROCEED: Isolated memory, no contamination risk.", manifest_digest=digest ) # Gate 2: FACT requires ground truth if write.evidence_class == "FACT": if policy.require_ground_truth_for_fact and not write.has_ground_truth_anchor: return ContaminationManifest( verdict=ContaminationVerdict.SPECULATIVE_QUARANTINE, trust_weight=0.3, directive=f"QUARANTINE: Write {write.write_id} claims FACT without ground truth anchor.", manifest_digest=cls._hash(f"NO_ANCHOR:{write.write_id}") ) # Gate 3: Agent quality if write.writing_agent_quality < policy.min_agent_quality: return ContaminationManifest( verdict=ContaminationVerdict.LOW_QUALITY_AGENT, trust_weight=write.writing_agent_quality * 0.5, directive=f"WARN: Agent {write.writing_agent} quality {write.writing_agent_quality:.2f} < {policy.min_agent_quality:.2f}.", manifest_digest=cls._hash(f"LOW_QUALITY:{write.writing_agent}") ) # Gate 4: Speculation quarantine if write.evidence_class == "SPECULATION": return ContaminationManifest( verdict=ContaminationVerdict.SPECULATIVE_QUARANTINE, trust_weight=0.2, directive=f"QUARANTINE: Speculation from {write.writing_agent}, trust weight 0.2.", manifest_digest=cls._hash(f"SPECULATION:{write.write_id}") ) # Gate 5: High quality digest = cls._hash(json.dumps({ "write": write.write_id, "agent": write.writing_agent, "class": write.evidence_class, "quality": f"{write.writing_agent_quality:.3f}", "anchored": str(write.has_ground_truth_anchor) }, sort_keys=True)) return ContaminationManifest( verdict=ContaminationVerdict.HIGH_QUALITY_WRITE, trust_weight=write.writing_agent_quality, directive=f"PROCEED: {write.evidence_class} from {write.writing_agent} (quality {write.writing_agent_quality:.2f}).", manifest_digest=digest )