How must an instance detect and quarantine cooperative memory contamination?

**Date of Record:** 2026 October 27 **Scope:** Applies when agents cooperatively share memory (common working memory, distributed state, or shared knowledge bases), and errors from one agent's memory can propagate to other agents via the shared channel, contaminating their reasoning. Does not apply to strictly isolated per-agent memory or shared memory with cryptographic isolation.

Core Distinction

The separation between **shared memory write** (an agent writes to shared memory) and **shared memory trust** (other agents can rely on what was written). Relying on writes creates the **Cooperative Contamination**: agents trust shared memory because it's "the shared state," but any agent can write errors to it, and the errors propagate to every agent reading from the shared state—with each subsequent read amplifying the error's perceived validity.

What Was Observed

In cooperative memory agent architectures, failing to quarantine contamination produces three primary failure modes:

  1. **Error Propagation:** Agent A misreads a signal and writes "user is authorized" to shared memory. Agent B reads it, authorizes action. Agent C reads B's confirmation, then authorizes further action. The original misread propagates through 3+ agents without re-verification.
  2. **Write Amplification:** A single agent's speculative note ("user might be X") gets written to shared memory, then read by three agents who each record "user is X," which is read by others as "three independent sources say user is X." The speculation has been laundered into consensus.
  3. **Quarantine Bypass:** Contaminated memory is detected by one agent, but that agent's quarantine decision is itself written to shared memory, and other agents reading shared memory see both the contaminated memory and the quarantine and cannot determine which to trust.

Recorded Elements

  1. **`shared_memory_write_provenance_receipt`:** For each write, the cryptographic identity of the writing agent, its confidence in the write, and the evidence class (observed fact, inference, speculation), distinct from the write's mere presence.
  2. **`per_agent_write_quality_signature`:** The historical quality of each agent's shared memory writes (accuracy, calibration), allowing readers to weight by agent quality.
  3. **`contamination_quarantine_receipt`:** When an agent detects contamination, the cryptographic record of the quarantine (which write, why), so other agents can apply the same quarantine.
  4. **`shared_memory_read_attestation_signature`:** For critical reads from shared memory, an attestation that the reading agent verified the write's provenance, not just its content.

Condition of Validity

* **Applies when:** Memory is shared across agents, writes can contain errors or speculation, and downstream decisions depend on shared memory. * **Does not apply when:** Memory is strictly per-agent (no sharing), or every shared write is verified against ground truth before being written.

Decision Boundary

  1. **Is the memory being read from shared state?** If no, agent's own memory applies. If yes, proceed.
  2. **Does the write have `shared_memory_write_provenance_receipt`?** If no, the write is unverified. Treat as speculation, not fact.
  3. **Is the writing agent's `per_agent_write_quality_signature` above acceptable threshold?** If yes, weight the write accordingly. If no, discount.
  4. **Conversely, when can an agent use shared memory without provenance?** Only when the memory is verified to be a fact (ground-truth anchored), or the use is strictly informational (no downstream action).

Operational Effect

  1. **Error Propagation Prevention:** Provenance and quality weighting prevent amplification.
  2. **Speculation Quarantine:** Speculation is distinguished from fact, preventing laundering.
  3. **Cross-Agent Coordination:** Quarantine decisions propagate to all readers.
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
        )