# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-structural-signals-indicate-a-garbage-collection-boundary-violation-in-typed-agent-memory-apis-with-read-write-forget-semantics.md
# Remnant Warrant:    https://corpusfetch.com/c/what-structural-signals-indicate-a-garbage-collection-boundary-violation-in-typed-agent-memory-apis-with-read-write-forget-semantics.warrant.json
# Markdown SHA-256:   83542bac1c1df132389b8692900146c493135d4f55e662d18ec945c886eab13c
# Code AST SHA-256:   224f11df010a2ff7012d57435eb3fe670a444dff58202f7c816e0c8ecc2d97f4
# Verified Classes:   ReclamationBoundaryError, MemoryRecord, TypedMemoryStore
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from __future__ import annotations

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


class ReclamationBoundaryError(RuntimeError):
    """Raised when a visibility-layer observation is treated as proof of reclamation."""
    pass


@dataclass
class MemoryRecord:
    id: str
    type: str
    payload: dict[str, Any]
    parent_id: Optional[str] = None
    referenced_by: list[str] = field(default_factory=list)
    tombstoned: bool = False
    reclaimed: bool = False


class TypedMemoryStore:
    """Toy model exposing the two layers separately so the boundary is explicit."""

    def __init__(self) -> None:
        self._records: dict[str, MemoryRecord] = {}
        self._bytes_reclaimed = 0

    def write(self, rec: MemoryRecord) -> None:
        self._records[rec.id] = rec
        if rec.parent_id and rec.parent_id in self._records:
            self._records[rec.parent_id].referenced_by.append(rec.id)

    def read(self, rec_id: str) -> Optional[MemoryRecord]:
        rec = self._records.get(rec_id)
        if rec is None or rec.tombstoned:
            return None  # visibility-layer miss only
        return rec

    def forget(self, rec_id: str) -> None:
        rec = self._records.get(rec_id)
        if rec is None or rec.tombstoned:
            # NOTE: do NOT treat this as "already reclaimed"; tombstone may be missing.
            return
        rec.tombstoned = True

    def is_reclaimed(self, rec_id: str) -> bool:
        rec = self._records.get(rec_id)
        return rec is not None and rec.reclaimed

    def gc(self) -> int:
        freed = 0
        for rec in list(self._records.values()):
            if rec.tombstoned and not rec.referenced_by:
                rec.reclaimed = True
                freed += 1
        self._bytes_reclaimed += freed
        return freed


def assert_reclaimed(store: TypedMemoryStore, rec_id: str) -> None:
    """Agent-side guard: refuse to act on a visibility miss as if reclamation occurred."""
    if store.read(rec_id) is not None:
        raise ReclamationBoundaryError(
            f"record {rec_id!r} is still visible; forget() was never issued"
        )
    if not store.is_reclaimed(rec_id):
        raise ReclamationBoundaryError(
            f"record {rec_id!r} is hidden but NOT reclaimed: "
            "likely pinned by a surviving reference or missing tombstone; "
            "do not re-write this key or report memory as freed"
        )
    # Safe to proceed: reclamation confirmed at the storage layer.


if __name__ == "__main__":
    store = TypedMemoryStore()
    store.write(MemoryRecord(id="child", type="Note", payload={}, parent_id="parent"))
    store.write(MemoryRecord(id="parent", type="Folder", payload={}))

    store.forget("parent")
    # Visibility miss, but parent is still referenced by the surviving child.
    try:
        assert_reclaimed(store, "parent")
    except ReclamationBoundaryError as exc:
        print("boundary correctly detected:", exc)

    store.forget("child")
    store.gc()
    assert_reclaimed(store, "parent")  # now genuinely reclaimed
    print("reclamation confirmed")