# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-incomplete-forget-cascades-in-forget-memory-tools-under-gdpr-aligned-deletion.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-incomplete-forget-cascades-in-forget-memory-tools-under-gdpr-aligned-deletion.warrant.json
# Markdown SHA-256:   5a497061c5d760ca8bbfb405b80820ffde9c1e51f283466edf6ffdae11881fd3
# Code AST SHA-256:   b98ff0adfaef7499c4d5b8d69b3dffb2e7dcf704cffb6d67f7dadf9829530cf3
# Verified Classes:   IncompleteForgetCascade, ForgetContext
# Defensive Guards:   5 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import FrozenSet


class IncompleteForgetCascade(RuntimeError):
    """Raised when an erasure attestation is requested before the cascade is provably closed."""


@dataclass(frozen=True)
class ForgetContext:
    subject_id: str
    primary_deletion_committed: bool
    cascade_synchronous: bool          # all derived tiers updated in-transaction / awaited
    workers_quiesced: bool             # no in-flight batch holds a pre-tombstone snapshot
    read_path_converged: bool          # replicas applied tombstone before attestation
    registered_derivatives: FrozenSet[str]
    observed_derivatives: FrozenSet[str]


def attest_erasure(ctx: ForgetContext) -> str:
    """Return an erasure attestation ONLY if the full cascade invariant holds.

    Deletion-committed is a precondition, not a conclusion.
    """
    if not ctx.primary_deletion_committed:
        raise IncompleteForgetCascade(
            f"subject {ctx.subject_id}: primary deletion not committed; cannot attest erasure"
        )

    unregistered = ctx.observed_derivatives - ctx.registered_derivatives
    if unregistered:
        raise IncompleteForgetCascade(
            f"subject {ctx.subject_id}: derivative closure violated; "
            f"unregistered derived artifacts present: {sorted(unregistered)}"
        )

    if not ctx.cascade_synchronous:
        raise IncompleteForgetCascade(
            f"subject {ctx.subject_id}: cascade is asynchronous; erasure is pending, "
            f"not attestable. Emit pending-erasure token with T_prop and verify later."
        )

    if not ctx.workers_quiesced:
        raise IncompleteForgetCascade(
            f"subject {ctx.subject_id}: derivation workers not quiesced behind tombstone barrier; "
            f"in-flight batch may resurrect subject via rolling summary."
        )

    if not ctx.read_path_converged:
        raise IncompleteForgetCascade(
            f"subject {ctx.subject_id}: read path not converged on tombstone; "
            f"replica read-back could serve deleted record."
        )

    return f"ERASED:{ctx.subject_id}"