# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-treat-a-forget-acknowledgement-as-a-completion-signal-versus-a-request-acknowledgement-in-scoped-retention-memory-apis.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-treat-a-forget-acknowledgement-as-a-completion-signal-versus-a-request-acknowledgement-in-scoped-retention-memory-apis.warrant.json
# Markdown SHA-256:   9988f1016a24054815c249193ba60cd53541c3dd1b763a1977dcb9915c95ab9e
# Code AST SHA-256:   75c3fb440bd19a8c3a4b75813b3eac9b91f7ac3a31be629ccd47f6def3c8d99e
# Verified Classes:   ForgetEffectuationTimeout, ScopeLaunderingError, ForgetAck, ScopedMemory
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import time
from dataclasses import dataclass


class ForgetEffectuationTimeout(RuntimeError):
    """Raised when a scoped forget is accepted but the scope is not verified
    empty within the store's advertised effectuation bound."""
    pass


class ScopeLaunderingError(RuntimeError):
    """Raised when an agent attempts to write derived content into a scope
    (or its descendant) that is under post-deletion quarantine."""
    pass


@dataclass(frozen=True)
class ForgetAck:
    token: str
    scope: str
    effectuation_bound_s: float


def is_descendant(candidate: str, ancestor: str) -> bool:
    return candidate == ancestor or candidate.startswith(ancestor.rstrip("/") + "/")


class ScopedMemory:
    """Minimal model of a scoped-retention memory API with async effectuation."""

    def __init__(self, store, advertised_bound_s: float = 30.0):
        self._store = store
        self._bound = advertised_bound_s
        self._quarantine: dict[str, ForgetAck] = {}

    def forget(self, scope: str) -> ForgetAck:
        token = self._store.register_deletion(scope)
        ack = ForgetAck(token=token, scope=scope, effectuation_bound_s=self._bound)
        self._quarantine[scope] = ack
        return ack

    def _probe_empty(self, scope: str) -> bool:
        return self._store.read_scope(scope) == []

    def await_effectuation(self, ack: ForgetAck, poll_s: float = 0.25) -> None:
        deadline = time.monotonic() + ack.effectuation_bound_s
        while time.monotonic() < deadline:
            if self._probe_empty(ack.scope):
                self._quarantine.pop(ack.scope, None)
                return
            time.sleep(poll_s)
        raise ForgetEffectuationTimeout(
            f"scope {ack.scope!r} not empty after {ack.effectuation_bound_s}s "
            f"(token {ack.token}); treat as consistency violation, do not derive"
        )

    def write(self, scope: str, record: dict) -> None:
        for quarantined in self._quarantine:
            if is_descendant(scope, quarantined):
                raise ScopeLaunderingError(
                    f"write to {scope!r} is a descendant of quarantined scope "
                    f"{quarantined!r}; token does not cover this record"
                )
        self._store.write(scope, record)