# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-silent-scope-escalation-during-forget-then-reinsert-cycles-in-audited-memory-stores.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-silent-scope-escalation-during-forget-then-reinsert-cycles-in-audited-memory-stores.warrant.json
# Markdown SHA-256:   f8a2f30e7e0e630a5b206ab4b82052258bde78f502a44506a258e4f6e2d7e40c
# Code AST SHA-256:   b5b8736003c5ca03fba8f31ae9b89f63eb26259e299dd248739272a321299fe2
# Verified Classes:   ScopeEnvelopeViolation, Principal, CanonicalKey, ForgetEnvelope, AuditedMemoryStore
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Iterable


class ScopeEnvelopeViolation(RuntimeError):
    """Raised when a write attempts to occupy a slot whose forget-envelope excludes it."""


@dataclass(frozen=True)
class Principal:
    name: str
    tenants: frozenset[str]
    agents: frozenset[str]


@dataclass(frozen=True)
class CanonicalKey:
    tenant: str
    agent: str
    session: str
    key: str


@dataclass
class ForgetEnvelope:
    key: CanonicalKey
    authorized: frozenset[Principal]
    superseded: bool = False


@dataclass
class AuditedMemoryStore:
    _rows: dict[CanonicalKey, str] = field(default_factory=dict)
    _envelopes: dict[CanonicalKey, ForgetEnvelope] = field(default_factory=dict)
    _log: list[tuple[str, CanonicalKey, str]] = field(default_factory=list)

    def _canonicalize(self, p: Principal, key: CanonicalKey) -> CanonicalKey:
        # Canonicalization must not drop the agent dimension; if it did, the
        # envelope would attach to a label that no longer identifies the slot.
        if key.tenant not in p.tenants or key.agent not in p.agents:
            raise ScopeEnvelopeViolation(
                f"principal {p.name} lacks scope for {key}"
            )
        return key

    def forget(self, p: Principal, key: CanonicalKey) -> None:
        ck = self._canonicalize(p, key)
        # Snapshot authority at forget time; do not recompute at write time.
        envelope = ForgetEnvelope(key=ck, authorized=frozenset({p}))
        self._envelopes[ck] = envelope
        self._rows.pop(ck, None)
        self._log.append(("FORGET", ck, p.name))

    def write(self, p: Principal, key: CanonicalKey, value: str) -> None:
        ck = self._canonicalize(p, key)
        envelope = self._envelopes.get(ck)
        if envelope is not None and not envelope.superseded:
            if p not in envelope.authorized:
                raise ScopeEnvelopeViolation(
                    f"principal {p.name} may not reinsert {ck}: "
                    f"forget-envelope held by "
                    f"{sorted(x.name for x in envelope.authorized)}"
                )
            envelope.superseded = True
        self._rows[ck] = value
        self._log.append(("WRITE", ck, p.name))

    def audit_replay(self) -> Iterable[tuple[str, CanonicalKey, str]]:
        yield from self._log


def reassign_session(
    store: AuditedMemoryStore,
    old_owner: Principal,
    new_owner: Principal,
    key: CanonicalKey,
    value: str,
) -> None:
    store.forget(old_owner, key)
    try:
        store.write(new_owner, key, value)
    except ScopeEnvelopeViolation as exc:
        # Surface the stale-field / cross-agent bug instead of silently
        # laundering authority through the recycled slot.
        raise ScopeEnvelopeViolation(
            f"reassignment of {key} rejected: {exc}"
        ) from exc