popoto.fields.tombstone_store

popoto.fields.tombstone_store

TombstoneStore — field-layer keeper of the tombstone keyspace (#649).

Relocates the raw-Redis tombstone bookkeeping that used to live inline in popoto.recipes.memory_lifecycle (issue #491) into the field layer, as part of the #630 "route recipes through the field layer" series. This is a relocation, not a redesign: every method here issues the exact same Redis commands, in the same order, as the recipe code it replaces, so wire behavior is byte-identical.

The tombstone keyspace is deliberately kept OUTSIDE the model's own keyspace ($TOMB:{Model}:data / $TOMB:{Model}:index rather than any {Model}:* key) so that no query, index scan, or key-set walk can ever surface a tombstoned record. That property must be preserved exactly — tombstones are NOT a popoto Model; that was explicitly considered and rejected.

Keyspace

$TOMB:{Model}:data — hash, redis_key -> msgpack-packed entry bytes $TOMB:{Model}:index — zset, redis_key -> death timestamp (score)

Tombstone dataclass

Durable record that a memory was forgotten, and what it looked like.

Forgetting tombstones rather than deletes so an aggressive low-confidence forget policy stays reversible (Risk 6) and so each death becomes negative evidence a future write path can consult (#494).

Attributes:

Name	Type	Description
redis_key	str	The forgotten record's Redis key. Restore handle.
fingerprint	str	ExistenceFilter fingerprint of the dead record — the identity token #494 matches new writes against.
tier	str	Tier the record held at death.
importance_at_death	float	Importance score at the moment of forgetting.
confidence_at_death	Optional[float]	ConfidenceField value at death, or None if the model carries no confidence signal.
evidence_count	int	Observations backing that confidence.
dismissal_count	int	Contradiction/dismissal count at death.
tombstoned_at	float	Unix timestamp of the forgetting.
reason	str	Free-form marker for what triggered it.

Source code in src/popoto/fields/tombstone_store.py

@dataclass
class Tombstone:
    """Durable record that a memory was forgotten, and what it looked like.

    Forgetting tombstones rather than deletes so an aggressive low-confidence
    forget policy stays reversible (Risk 6) and so each death becomes negative
    evidence a future write path can consult (#494).

    Attributes:
        redis_key: The forgotten record's Redis key. Restore handle.
        fingerprint: ExistenceFilter fingerprint of the dead record — the
            identity token #494 matches new writes against.
        tier: Tier the record held at death.
        importance_at_death: Importance score at the moment of forgetting.
        confidence_at_death: ConfidenceField value at death, or None if the
            model carries no confidence signal.
        evidence_count: Observations backing that confidence.
        dismissal_count: Contradiction/dismissal count at death.
        tombstoned_at: Unix timestamp of the forgetting.
        reason: Free-form marker for what triggered it.
    """

    redis_key: str
    fingerprint: str
    tier: str
    importance_at_death: float
    confidence_at_death: Optional[float]
    evidence_count: int
    dismissal_count: int
    tombstoned_at: float
    reason: str = "policy"

TombstoneStore

Owns the $TOMB:{Model}:* keyspace for a single model class.

Field-layer keeper of tombstone reads/writes. Every method issues the exact same Redis commands, in the same order, as the raw-Redis recipe code it replaces (see the table in issue #649). Callers (e.g. MemoryLifecycle) are responsible for msgpack-packing entries before calling archive() and for unpacking what get_entry/get_entries return — this store deals in raw bytes at that boundary, matching where the recipe drew the line.

Parameters:

Name	Type	Description	Default
model_class	Any	The Popoto Model class whose tombstones this store manages. Only model_class.__name__ is used, to derive the keyspace.	required

Source code in src/popoto/fields/tombstone_store.py

class TombstoneStore:
    """Owns the ``$TOMB:{Model}:*`` keyspace for a single model class.

    Field-layer keeper of tombstone reads/writes. Every method issues the
    exact same Redis commands, in the same order, as the raw-Redis recipe
    code it replaces (see the table in issue #649). Callers (e.g.
    ``MemoryLifecycle``) are responsible for msgpack-packing entries before
    calling ``archive()`` and for unpacking what ``get_entry``/``get_entries``
    return — this store deals in raw bytes at that boundary, matching where
    the recipe drew the line.

    Args:
        model_class: The Popoto Model class whose tombstones this store
            manages. Only ``model_class.__name__`` is used, to derive the
            keyspace.
    """

    def __init__(self, model_class: Any):
        self.model_class = model_class

    def _backend(self) -> Any:
        """The model's backend when it is not Redis (#759 M4): the archive
        is then that backend's ``popoto_tombstone`` table (``_tomb``
        ``field_call`` adapters), still outside the model's own table so no
        query can surface a tombstoned record."""
        from ..backends.routing import non_redis_backend

        return non_redis_backend(self.model_class)

    def _call(self, backend: Any, op: str, *args: Any) -> Any:
        return backend.field_call(self.model_class._meta.spec, "_tomb", op, *args)

    def keys(self) -> Tuple[str, str]:
        """Return the (data hash, recency index) Redis keys for tombstones."""
        name = self.model_class.__name__
        return (
            f"{TOMBSTONE_KEY_PREFIX}:{name}:data",
            f"{TOMBSTONE_KEY_PREFIX}:{name}:index",
        )

    def archive(self, redis_key: str, entry_bytes: bytes, ts: float) -> None:
        """Write a tombstone entry: HSET the data hash, ZADD the recency index.

        Both commands are queued in one transactional pipeline (``HSET`` then
        ``ZADD``, matching the original write order) and executed together.
        """
        backend = self._backend()
        if backend is not None:
            self._call(backend, "archive", redis_key, entry_bytes, ts)
            return
        data_key, index_key = self.keys()
        pipeline = _batch()
        pipeline.hset(data_key, redis_key, entry_bytes)
        pipeline.zadd(index_key, {redis_key: ts})
        pipeline.execute()

    def count(self) -> int:
        """Return the number of retained tombstones (``ZCARD`` on the index)."""
        backend = self._backend()
        if backend is not None:
            return int(self._call(backend, "count"))
        _, index_key = self.keys()
        return int(_sync(get_REDIS_DB().zcard(index_key)))

    def oldest_keys(self, n: int) -> List[str]:
        """Return up to ``n`` oldest tombstoned keys (``ZRANGE`` 0..n-1)."""
        backend = self._backend()
        if backend is not None:
            return list(self._call(backend, "oldest", n))
        _, index_key = self.keys()
        raw = get_REDIS_DB().zrange(index_key, 0, n - 1)
        return _decoded_members(raw)

    def newest_keys(self, stop: int) -> List[str]:
        """Return keys newest-death-first up to ``stop`` (``ZREVRANGE`` 0..stop)."""
        backend = self._backend()
        if backend is not None:
            return list(self._call(backend, "newest", stop))
        _, index_key = self.keys()
        raw = get_REDIS_DB().zrevrange(index_key, 0, stop)
        return _decoded_members(raw)

    def evict(self, keys: List[str]) -> None:
        """Remove tombstones for ``keys``: ``HDEL`` data + ``ZREM`` index.

        Both commands are queued in one transactional pipeline (``HDEL`` then
        ``ZREM``, matching the original eviction order) and executed together.
        """
        backend = self._backend()
        if backend is not None:
            self._call(backend, "evict", list(keys))
            return
        data_key, index_key = self.keys()
        pipeline = _batch()
        pipeline.hdel(data_key, *keys)
        pipeline.zrem(index_key, *keys)
        pipeline.execute()

    def get_entry(self, redis_key: str) -> Any:
        """Return the raw stored entry bytes for one key (``HGET``), or None."""
        backend = self._backend()
        if backend is not None:
            return self._call(backend, "entries", [redis_key])[0]
        data_key, _ = self.keys()
        return get_REDIS_DB().hget(data_key, redis_key)

    def get_entries(self, keys: List[str]) -> List[Any]:
        """Return raw stored entry bytes for ``keys`` (``HMGET``).

        Preserves argument order, with ``None`` holes for missing members —
        callers rely on positional correspondence with ``keys``.
        """
        backend = self._backend()
        if backend is not None:
            return list(self._call(backend, "entries", list(keys)))
        data_key, _ = self.keys()
        return _sync(get_REDIS_DB().hmget(data_key, keys))

    def purge(self, redis_key: str) -> bool:
        """Drop one tombstone permanently: ``HDEL`` data + ``ZREM`` index.

        Both commands are queued in one transactional pipeline and executed
        together. Returns True if a tombstone was actually removed.
        """
        backend = self._backend()
        if backend is not None:
            return bool(self._call(backend, "purge", redis_key))
        data_key, index_key = self.keys()
        pipeline = _batch()
        pipeline.hdel(data_key, redis_key)
        pipeline.zrem(index_key, redis_key)
        removed = pipeline.execute()
        return bool(removed and removed[0])

    def purge_all(self) -> int:
        """Drop every retained tombstone for this model: one ``DEL`` over both keys.

        The count read is best-effort and its failure is swallowed: the caller
        this was relocated from (``MemoryLifecycle.purge_all_tombstones``) read
        the count through an error-swallowing helper and issued the ``DEL``
        regardless. A failed count must not leave the tombstones in place — the
        return value is a report, the delete is the job.
        """
        backend = self._backend()
        if backend is not None:
            return int(self._call(backend, "purge_all"))
        try:
            count = self.count()
        except Exception:
            count = 0
        get_REDIS_DB().delete(*self.keys())
        return count

keys()

Return the (data hash, recency index) Redis keys for tombstones.

Source code in src/popoto/fields/tombstone_store.py

def keys(self) -> Tuple[str, str]:
    """Return the (data hash, recency index) Redis keys for tombstones."""
    name = self.model_class.__name__
    return (
        f"{TOMBSTONE_KEY_PREFIX}:{name}:data",
        f"{TOMBSTONE_KEY_PREFIX}:{name}:index",
    )

archive(redis_key, entry_bytes, ts)

Write a tombstone entry: HSET the data hash, ZADD the recency index.

Both commands are queued in one transactional pipeline (HSET then ZADD, matching the original write order) and executed together.

Source code in src/popoto/fields/tombstone_store.py

def archive(self, redis_key: str, entry_bytes: bytes, ts: float) -> None:
    """Write a tombstone entry: HSET the data hash, ZADD the recency index.

    Both commands are queued in one transactional pipeline (``HSET`` then
    ``ZADD``, matching the original write order) and executed together.
    """
    backend = self._backend()
    if backend is not None:
        self._call(backend, "archive", redis_key, entry_bytes, ts)
        return
    data_key, index_key = self.keys()
    pipeline = _batch()
    pipeline.hset(data_key, redis_key, entry_bytes)
    pipeline.zadd(index_key, {redis_key: ts})
    pipeline.execute()

count()

Return the number of retained tombstones (ZCARD on the index).

Source code in src/popoto/fields/tombstone_store.py

def count(self) -> int:
    """Return the number of retained tombstones (``ZCARD`` on the index)."""
    backend = self._backend()
    if backend is not None:
        return int(self._call(backend, "count"))
    _, index_key = self.keys()
    return int(_sync(get_REDIS_DB().zcard(index_key)))

oldest_keys(n)

Return up to n oldest tombstoned keys (ZRANGE 0..n-1).

Source code in src/popoto/fields/tombstone_store.py

def oldest_keys(self, n: int) -> List[str]:
    """Return up to ``n`` oldest tombstoned keys (``ZRANGE`` 0..n-1)."""
    backend = self._backend()
    if backend is not None:
        return list(self._call(backend, "oldest", n))
    _, index_key = self.keys()
    raw = get_REDIS_DB().zrange(index_key, 0, n - 1)
    return _decoded_members(raw)

newest_keys(stop)

Return keys newest-death-first up to stop (ZREVRANGE 0..stop).

Source code in src/popoto/fields/tombstone_store.py

def newest_keys(self, stop: int) -> List[str]:
    """Return keys newest-death-first up to ``stop`` (``ZREVRANGE`` 0..stop)."""
    backend = self._backend()
    if backend is not None:
        return list(self._call(backend, "newest", stop))
    _, index_key = self.keys()
    raw = get_REDIS_DB().zrevrange(index_key, 0, stop)
    return _decoded_members(raw)

evict(keys)

Remove tombstones for keys: HDEL data + ZREM index.

Both commands are queued in one transactional pipeline (HDEL then ZREM, matching the original eviction order) and executed together.

Source code in src/popoto/fields/tombstone_store.py

def evict(self, keys: List[str]) -> None:
    """Remove tombstones for ``keys``: ``HDEL`` data + ``ZREM`` index.

    Both commands are queued in one transactional pipeline (``HDEL`` then
    ``ZREM``, matching the original eviction order) and executed together.
    """
    backend = self._backend()
    if backend is not None:
        self._call(backend, "evict", list(keys))
        return
    data_key, index_key = self.keys()
    pipeline = _batch()
    pipeline.hdel(data_key, *keys)
    pipeline.zrem(index_key, *keys)
    pipeline.execute()

get_entry(redis_key)

Return the raw stored entry bytes for one key (HGET), or None.

Source code in src/popoto/fields/tombstone_store.py

def get_entry(self, redis_key: str) -> Any:
    """Return the raw stored entry bytes for one key (``HGET``), or None."""
    backend = self._backend()
    if backend is not None:
        return self._call(backend, "entries", [redis_key])[0]
    data_key, _ = self.keys()
    return get_REDIS_DB().hget(data_key, redis_key)

get_entries(keys)

Return raw stored entry bytes for keys (HMGET).

Preserves argument order, with None holes for missing members — callers rely on positional correspondence with keys.

Source code in src/popoto/fields/tombstone_store.py

def get_entries(self, keys: List[str]) -> List[Any]:
    """Return raw stored entry bytes for ``keys`` (``HMGET``).

    Preserves argument order, with ``None`` holes for missing members —
    callers rely on positional correspondence with ``keys``.
    """
    backend = self._backend()
    if backend is not None:
        return list(self._call(backend, "entries", list(keys)))
    data_key, _ = self.keys()
    return _sync(get_REDIS_DB().hmget(data_key, keys))

purge(redis_key)

Drop one tombstone permanently: HDEL data + ZREM index.

Both commands are queued in one transactional pipeline and executed together. Returns True if a tombstone was actually removed.

Source code in src/popoto/fields/tombstone_store.py

def purge(self, redis_key: str) -> bool:
    """Drop one tombstone permanently: ``HDEL`` data + ``ZREM`` index.

    Both commands are queued in one transactional pipeline and executed
    together. Returns True if a tombstone was actually removed.
    """
    backend = self._backend()
    if backend is not None:
        return bool(self._call(backend, "purge", redis_key))
    data_key, index_key = self.keys()
    pipeline = _batch()
    pipeline.hdel(data_key, redis_key)
    pipeline.zrem(index_key, redis_key)
    removed = pipeline.execute()
    return bool(removed and removed[0])

purge_all()

Drop every retained tombstone for this model: one DEL over both keys.

The count read is best-effort and its failure is swallowed: the caller this was relocated from (MemoryLifecycle.purge_all_tombstones) read the count through an error-swallowing helper and issued the DEL regardless. A failed count must not leave the tombstones in place — the return value is a report, the delete is the job.

Source code in src/popoto/fields/tombstone_store.py

def purge_all(self) -> int:
    """Drop every retained tombstone for this model: one ``DEL`` over both keys.

    The count read is best-effort and its failure is swallowed: the caller
    this was relocated from (``MemoryLifecycle.purge_all_tombstones``) read
    the count through an error-swallowing helper and issued the ``DEL``
    regardless. A failed count must not leave the tombstones in place — the
    return value is a report, the delete is the job.
    """
    backend = self._backend()
    if backend is not None:
        return int(self._call(backend, "purge_all"))
    try:
        count = self.count()
    except Exception:
        count = 0
    get_REDIS_DB().delete(*self.keys())
    return count