# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-forget-tombstone-durability-and-the-write-after-forget-race-in-llm-agent-memory-apis.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-forget-tombstone-durability-and-the-write-after-forget-race-in-llm-agent-memory-apis.warrant.json
# Markdown SHA-256:   a86d87667c92b9b0a9dda38206c79e315ddd9678fc0a66fdb43174b135de22c1
# Code AST SHA-256:   c027f0ce4b8ca80cd63c38262d0760640e007d6928dfa718e231a12d284a9628
# Verified Classes:   ForgetNotDurableError, ResurrectionDetectedError
# Defensive Guards:   4 fail-closed invariants (zero external deps)

import time
from typing import Callable, Optional


class ForgetNotDurableError(RuntimeError):
    """Raised when a forget cannot be confirmed absent across a quiesce window."""


class ResurrectionDetectedError(RuntimeError):
    """Raised when a key reappears after a forget that was reported successful."""


def confirm_forget(
    read: Callable[[str], Optional[object]],
    key: str,
    quiesce_seconds: float = 2.0,
    polls: int = 2,
    sleep: Callable[[float], None] = time.sleep,
) -> None:
    """
    Verify that `key` is absent from the memory store across a quiesce window.

    Use this only when the store does not document durable tombstones.
    Raises ForgetNotDurableError if the key is still visible on any poll,
    or ResurrectionDetectedError if it disappears then reappears.
    """
    if polls < 2:
        raise ValueError("polls must be >= 2 to span a quiesce window")

    seen_absent = False
    for i in range(polls):
        value = read(key)
        if value is not None:
            if seen_absent:
                raise ResurrectionDetectedError(
                    f"key {key!r} reappeared after being observed absent "
                    f"(poll {i}); forget is not durable — likely a "
                    f"write-after-forget race"
                )
            raise ForgetNotDurableError(
                f"key {key!r} still readable at poll {i}; forget did not hold"
            )
        seen_absent = True
        if i < polls - 1:
            sleep(quiesce_seconds)


def safe_forget_then_write(
    forget: Callable[[str], object],
    write: Callable[[str, object], object],
    read: Callable[[str], Optional[object]],
    key: str,
    new_value: object,
    quiesce_seconds: float = 2.0,
) -> None:
    """
    Forget a key, prove it is absent, then write the superseding value.
    Refuses to write if forget could not be confirmed, because an
    unconfirmed forget plus a write is exactly the resurrection pattern.
    """
    forget(key)
    try:
        confirm_forget(read, key, quiesce_seconds=quiesce_seconds)
    except (ForgetNotDurableError, ResurrectionDetectedError) as exc:
        raise ForgetNotDurableError(
            f"refusing to write {key!r}: forget unconfirmed ({exc})"
        ) from exc
    write(key, new_value)