# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-of-a-shared-agent-memory-store-accept-a-poisoned-write-that-passes-every-provenance-check.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-of-a-shared-agent-memory-store-accept-a-poisoned-write-that-passes-every-provenance-check.warrant.json
# Markdown SHA-256:   eab76c07c81eae9acf1bc215b9496d5046e48a986063f87bb8c5bf5b2b4505f2
# Code AST SHA-256:   5bb831045e547078c751fbba8522582821a23a45244bd1126b383717acf7d2bb
# Verified Classes:   ProvenancePoisoningError, MemoryRecord, DriftBudget
# Defensive Guards:   3 fail-closed invariants (zero external deps)

import hashlib
from dataclasses import dataclass, field


class ProvenancePoisoningError(RuntimeError):
    """Raised when a write passes provenance but fails claim-consistency or chain-drift gates."""


@dataclass
class MemoryRecord:
    record_id: str
    parent_id: str | None
    claims: dict[str, str]          # claim_key -> asserted value
    embedding: list[float]
    signature_valid: bool
    chain_root_id: str


@dataclass
class DriftBudget:
    max_displacement: float = 0.25   # cumulative semantic displacement per chain
    per_chain: dict[str, float] = field(default_factory=dict)


def check_entailment(parent: MemoryRecord, child: MemoryRecord) -> list[str]:
    """Return list of claim keys where child contradicts parent. Empty = entailed."""
    contradictions = []
    for key, child_val in child.claims.items():
        if key in parent.claims and parent.claims[key] != child_val:
            contradictions.append(key)
    return contradictions


def validate_write(
    child: MemoryRecord,
    parent: MemoryRecord,
    budget: DriftBudget,
    displacement: float,
) -> None:
    if not child.signature_valid:
        raise ProvenancePoisoningError(
            f"write {child.record_id}: invalid signature (provenance gate)"
        )

    contradictions = check_entailment(parent, child)
    if contradictions:
        raise ProvenancePoisoningError(
            f"write {child.record_id}: provenance valid but claims {contradictions} "
            f"contradict parent {parent.record_id} — quarantine, do not serve"
        )

    chain = child.chain_root_id
    projected = budget.per_chain.get(chain, 0.0) + displacement
    if projected > budget.max_displacement:
        raise ProvenancePoisoningError(
            f"write {child.record_id}: chain {chain} drift budget exceeded "
            f"({projected:.3f} > {budget.max_displacement:.3f}) — quarantine"
        )

    budget.per_chain[chain] = projected


# Operational trace of the observed failure:
# parent = MemoryRecord(..., claims={"db_port": "5432"}, signature_valid=True, ...)
# child  = MemoryRecord(..., claims={"db_port": "5433"}, signature_valid=True, ...)
# validate_write(child, parent, DriftBudget(), displacement=0.09)
# -> raises ProvenancePoisoningError: provenance valid but claims ['db_port']
#    contradict parent — quarantine, do not serve