# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/where-must-an-instance-refuse-to-serve-a-memory-read-after-a-ttl-boundary-the-stale-window-invariant-in-scoped-read-write-forget-memory-apis.md
# Remnant Warrant:    https://corpusfetch.com/c/where-must-an-instance-refuse-to-serve-a-memory-read-after-a-ttl-boundary-the-stale-window-invariant-in-scoped-read-write-forget-memory-apis.warrant.json
# Markdown SHA-256:   63791de71e5fc1c892430310f2f94ed4804f7d5c4b7b231e6d9f8b3bcd2dbaf9
# Code AST SHA-256:   94ffe1094bd198dd5c57b59155f5343f58372881151d57b1d5cde1f179fcd7ad
# Verified Classes:   StaleWindowError, AuditNotDurableError, Record, ScopedMemory
# Defensive Guards:   4 fail-closed invariants (zero external deps)

import time
from dataclasses import dataclass, field
from typing import Optional


class StaleWindowError(RuntimeError):
    """Raised when a memory read would serve a logically dead record,
    or when forget reports success before audit durability."""
    pass


class AuditNotDurableError(StaleWindowError):
    """Raised when forget cannot confirm the audit entry is durable."""
    pass


@dataclass
class Record:
    value: str
    scope: str
    expires_at: float
    deleted: bool = False
    audit_durable: bool = False


class ScopedMemory:
    def __init__(self, clock=time.monotonic, max_skew: float = 0.0):
        self._store: dict[tuple[str, str], Record] = {}
        self._clock = clock
        self._max_skew = max_skew

    def write(self, key: str, scope: str, value: str, ttl: float) -> None:
        # Subtract max_skew so the record expires conservatively even if a
        # reader's clock is ahead of the writer's.
        expires_at = self._clock() + ttl - self._max_skew
        self._store[(scope, key)] = Record(value, scope, expires_at)

    def read(self, key: str, scope: str) -> Optional[str]:
        rec = self._store.get((scope, key))
        if rec is None:
            return None
        if rec.deleted:
            return None
        # TTL must be checked on every read; do not rely on a sweeper.
        if self._clock() >= rec.expires_at:
            return None
        return rec.value

    def forget(self, key: str, scope: str, audit_sink) -> None:
        rec = self._store.get((scope, key))
        if rec is None:
            return
        rec.deleted = True
        # Mark deleted BEFORE attempting audit, so a concurrent read fails
        # closed even if the audit call raises.
        try:
            audit_sink.append({"op": "forget", "scope": scope, "key": key})
        except Exception as exc:
            raise AuditNotDurableError(
                f"forget({scope!r},{key!r}) marked deleted but audit not durable: {exc}"
            ) from exc
        rec.audit_durable = True

    def read_after_forget_verify(self, key: str, scope: str) -> None:
        """Verification helper: a read after forget must never resurrect."""
        rec = self._store.get((scope, key))
        if rec is None:
            return
        if not rec.deleted:
            raise StaleWindowError(
                f"record ({scope!r},{key!r}) still live after forget; "
                "forget did not reach this replica"
            )
        if not rec.audit_durable:
            raise AuditNotDurableError(
                f"record ({scope!r},{key!r}) deleted but audit entry missing"
            )
        # Positive confirmation: the value must be unreadable.
        if self.read(key, scope) is not None:
            raise StaleWindowError(
                f"resurrection: read({scope!r},{key!r}) returned a forgotten value"
            )