# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-happens-when-an-instance-trusts-write-time-provenance-but-not-mutation-frequency-the-silent-overwrite-boundary-in-agent-memory-poisoning-detectio.md
# Remnant Warrant:    https://corpusfetch.com/c/what-happens-when-an-instance-trusts-write-time-provenance-but-not-mutation-frequency-the-silent-overwrite-boundary-in-agent-memory-poisoning-detectio.warrant.json
# Markdown SHA-256:   79c4926b7bd7a0e3fec2e3f915b711888521c6e37efeb1826dbd1f1e9da6f9d3
# Code AST SHA-256:   255f2536e45d13aebbe2675440e1f459b2228db381624e372bee95781862eee9
# Verified Classes:   SlowOverwritePoisoningError, MemoryWriteGuard
# Defensive Guards:   2 fail-closed invariants (zero external deps)

class SlowOverwritePoisoningError(RuntimeError):
    """Raised when cumulative per-key drift across spaced authorized writes
    exceeds the no-decay drift budget, even though per-window rate is clean."""
    pass


class MemoryWriteGuard:
    def __init__(self, drift_budget: float, max_writes_since_attest: int):
        self.drift_budget = drift_budget
        self.max_writes_since_attest = max_writes_since_attest
        # key -> cumulative drift since last attested baseline (NO decay)
        self._cum_drift = {}
        # key -> count of overwrites since last out-of-band attestation
        self._writes_since_attest = {}

    def record_write(self, key: str, provenance_ok: bool,
                     semantic_drift: float) -> None:
        if not provenance_ok:
            # provenance failure is a separate, already-handled path
            return

        self._cum_drift[key] = self._cum_drift.get(key, 0.0) + semantic_drift
        self._writes_since_attest[key] = self._writes_since_attest.get(key, 0) + 1

        # Per-window rate is intentionally NOT consulted here: an attacker
        # spaces writes to keep rate sub-threshold. Cumulative drift and
        # overwrite count do not decay, so spacing cannot evade them.
        if self._cum_drift[key] > self.drift_budget:
            raise SlowOverwritePoisoningError(
                f"key={key!r} cumulative drift {self._cum_drift[key]:.3f} "
                f"exceeds budget {self.drift_budget:.3f} across "
                f"{self._writes_since_attest[key]} authorized writes"
            )
        if self._writes_since_attest[key] > self.max_writes_since_attest:
            raise SlowOverwritePoisoningError(
                f"key={key!r} overwritten {self._writes_since_attest[key]} times "
                f"since last attestation (limit {self.max_writes_since_attest}); "
                f"spaced authorized writes require re-attestation"
            )

    def attest_baseline(self, key: str) -> None:
        """Out-of-band trust refresh resets both counters."""
        self._cum_drift[key] = 0.0
        self._writes_since_attest[key] = 0