popoto.fields.validity_field

popoto.fields.validity_field

Validity Field - Bitemporal Validity Intervals and Supersession Index State

This module provides :class:ValidityField, the validity axis for popoto models (issue #580). Where DecayingSortedField answers "how important is this memory right now?", ValidityField answers the prior question: "is this memory still true at all?"

Motivation

An agent learns "user is on the free plan". Two weeks later it learns "user upgraded to enterprise". Without a validity axis the stale fact keeps its place in every index and only loses ground gradually, through decay — so it can still be packed into the agent's context ahead of the correction. ValidityField makes that first fact stop being a member of default retrieval the moment it is superseded, while keeping it fully queryable in historical mode.

Design Philosophy

• Validity decides membership, decay decides ordering among the valid. The two axes compose and neither needs to know the other's constants.
• Not a :class:~popoto.fields.sorted_field_mixin.SortedFieldMixin. This is load-bearing, not an oversight — see the class docstring. Validity must never win a query's ordering field.
• Closed, never deleted. Superseding a record closes its interval; the record and its chain links survive for provenance and as_of replay.
• No model-hash mutation for chain links (plan D3). Chain links live in two derived HASHes so an append-only journal (#560) can adopt this field unchanged.
• Valkey-safe. Core commands only (ZADD/ZSCORE/ZREM/ ZRANGEBYSCORE/HSET/HDEL/GET/SET/DEL/EXISTS) plus Lua 5.1. No Redis-module commands (BF./CMS./TS./TOPK.) anywhere.

Index Structure (plan D1) — six Redis keys per model/field::

$ValidityF:Model:field:valid_from      ZSET   member -> valid-from epoch
$ValidityF:Model:field:invalid_at      ZSET   member -> close epoch, +inf = open
$ValidityF:Model:field:ingested_at     ZSET   member -> ingest epoch
$ValidityF:Model:field:chain:fwd       HASH   old redis_key -> superseding key
$ValidityF:Model:field:chain:rev       HASH   new redis_key -> superseded key
$ValidityF:Model:field:open:{digest}   STRING identity digest -> open record

(The $ValidityF prefix is auto-derived by the FieldBase metaclass from the class name ValidityField.)

An as-of-t membership test is valid_from <= t AND invalid_at > t: two ZRANGEBYSCOREs intersected, or two ZSCOREs inside Lua. +inf as the open-interval sentinel is native to both Redis and Valkey sorted sets.

Example

from popoto import Model, KeyField, ValidityField

class Fact(Model): fact_id = KeyField() validity = ValidityField()

old = Fact(fact_id="plan-1").save() # interval opens at now, closes at +inf new = Fact(fact_id="plan-2").save()

The explicit close below is NOT optional (issue #693). Saving alone only

ever OPENS intervals, so a save-only model's exclusion set stays empty,

unless a save declares a future valid_from.

Close old and chain it to new in one atomic EVAL:

ValidityField.execute_supersede( Fact, "validity", new_member=new.db_key.redis_key, mode="supersede", old_member=old.db_key.redis_key, )

Fact.query.filter(validity__current=True) # -> only new Fact.query.filter(validity__as_of=earlier) # -> old

ValidityError

Bases: ValueError

Base for every typed failure raised out of :data:SUPERSEDE_LUA.

Subclasses :class:ValueError deliberately (plan D4). Two shipped contracts depend on it: ObservationProtocol._apply_supersession degrades on except (TypeError, ValueError), and the V0 test suite asserts pytest.raises(ValueError) for close-before-start. Widening those to a new base class would be a silent behavior change on a signal path.

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

class ValidityError(ValueError):
    """Base for every typed failure raised out of :data:`SUPERSEDE_LUA`.

    Subclasses :class:`ValueError` deliberately (plan D4). Two shipped contracts
    depend on it: ``ObservationProtocol._apply_supersession`` degrades on
    ``except (TypeError, ValueError)``, and the V0 test suite asserts
    ``pytest.raises(ValueError)`` for close-before-start. Widening those to a new
    base class would be a silent behavior change on a signal path.
    """

ValidityMemberAbsentError

Bases: ValidityError

A member the caller named does not exist at the instant of the write.

Raised for a successor that was never saved (including the "Model:None" key an unsaved instance yields) and for an incumbent named explicitly via old_member. An incumbent merely resolved from the open-claim pointer is a hint, not an assertion, and its absence means "no incumbent" instead.

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

class ValidityMemberAbsentError(ValidityError):
    """A member the caller named does not exist at the instant of the write.

    Raised for a successor that was never saved (including the ``"Model:None"``
    key an unsaved instance yields) and for an incumbent named explicitly via
    ``old_member``. An incumbent merely *resolved* from the open-claim pointer is
    a hint, not an assertion, and its absence means "no incumbent" instead.
    """

ValidityCloseBeforeStartError

Bases: ValidityError

A close instant precedes the record's own stored valid_from.

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

class ValidityCloseBeforeStartError(ValidityError):
    """A close instant precedes the record's own stored ``valid_from``."""

ValidityValidFromConflictError

Bases: ValidityError

An asserted valid_from disagrees with the already-stored start.

Valid-time has exactly one writer: the field value at construction. Any other writer that asserts a disagreeing start gets this rather than losing silently to ZADD NX.

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

class ValidityValidFromConflictError(ValidityError):
    """An asserted ``valid_from`` disagrees with the already-stored start.

    Valid-time has exactly one writer: the field value at construction. Any other
    writer that *asserts* a disagreeing start gets this rather than losing
    silently to ``ZADD NX``.
    """

ValidityField

Bases: Field

Bitemporal validity interval for each record of a model.

Declared as a plain field::

class Fact(Model):
    fact_id = KeyField()
    validity = ValidityField()

The stored field value is the record's valid-from epoch (a float); the interval state itself lives in the six derived Redis keys documented in the module docstring, maintained by :data:SUPERSEDE_LUA.

Declaring the field is NOT enough to exclude anything (issue #693)

A plain .save() routes through :meth:on_save in mode 'open': it writes valid_from = save time (or a caller-declared epoch, see below) and invalid_at = +inf, and nothing else. +inf never satisfies the invalid_at <= as_of test in :meth:resolve_excluded_keys, so with respect to that branch, a save that does not declare a future valid_from gets a fully populated pair of interval ZSETs and an exclusion set that stays permanently empty. That is only half of resolve_excluded_keys, though: its other, independent branch excludes on valid_from > as_of, and :meth:on_save uses field_value as the valid-from epoch whenever it is numeric. So a plain .save() that declares a future valid_from is excluded, until that moment arrives — the one thing a save without a producer can exclude. A save also never registers a record as the incumbent for any identity — the {prefix}:open:{digest} pointer is written by :data:SUPERSEDE_LUA itself, so a later SupersessionProtocol.supersede(successor, identity_key=...) finds no incumbent, closes nothing, and returns None.

Closing an interval requires an explicit producer. Today there are four:

• :meth:SupersessionProtocol.supersede / :meth:~popoto.fields.supersession.SupersessionProtocol.save_and_supersede — use one of these for every identity-bearing write, including the first claim about an identity, or the second claim will close nothing.
• :meth:~popoto.fields.supersession.SupersessionProtocol.invalidate / :meth:~popoto.fields.supersession.SupersessionProtocol.save_and_invalidate — the identity-free counterpart of the pair above.
• ProvenanceJournal.
• ObservationProtocol.on_context_used with the "contradicted" outcome and instance._superseded_by set — see _apply_supersession. This is still an explicit application call, not an inference.

The empty-set-versus-None distinction is real but invisible to an adopter: gating did run, and excluded nothing. Whether this should stay imperative is the open question on issue #693.

Why this is NOT a SortedFieldMixin (plan D2)

This is load-bearing and must not be "improved". ModelOptions.add_field classifies any SortedFieldMixin into _meta.sorted_field_names, which puts the field in filter_for_keys_set's first loop — where a returned list becomes Query._sorted_field_order and the field can win the query's ordering. Validity is membership, not priority: it must never order results. As a plain Field it lands in the second loop and returns a set (hence :meth:filter_query's set return type — a list there would silently reintroduce the ordering bug). It also stays out of the reindex/migration loops that iterate sorted_field_names.

Query params (see :meth:get_filter_query_params): - {field}__current=True — records valid right now - {field}__current=False — the complement (closed / not-yet-started) - {field}__as_of=t — records valid at epoch t

These are deliberate queries: they consume a filter param and therefore disable sorted-range limit pushdown. That is exactly why the default retrieval path gates on validity server-side instead of via a filter kwarg.

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

class ValidityField(Field):
    """Bitemporal validity interval for each record of a model.

    Declared as a plain field::

        class Fact(Model):
            fact_id = KeyField()
            validity = ValidityField()

    The stored field value is the record's valid-from epoch (a ``float``); the
    interval state itself lives in the six derived Redis keys documented in the
    module docstring, maintained by :data:`SUPERSEDE_LUA`.

    Declaring the field is NOT enough to exclude anything (issue #693)
    -----------------------------------------------------------------
    A plain ``.save()`` routes through :meth:`on_save` in mode ``'open'``: it
    writes ``valid_from = save time`` (or a caller-declared epoch, see below)
    and ``invalid_at = +inf``, and nothing else. ``+inf`` never satisfies the
    ``invalid_at <= as_of`` test in :meth:`resolve_excluded_keys`, so **with
    respect to that branch, a save that does not declare a future
    ``valid_from`` gets a fully populated pair of interval ZSETs and an
    exclusion set that stays permanently empty.** That is only half of
    ``resolve_excluded_keys``, though: its other, independent branch excludes
    on ``valid_from > as_of``, and :meth:`on_save` uses ``field_value`` as the
    valid-from epoch whenever it is numeric. So a plain ``.save()`` that
    declares a future ``valid_from`` *is* excluded, until that moment arrives
    — the one thing a save without a producer can exclude. A save also never
    registers a record as the incumbent for any identity — the
    ``{prefix}:open:{digest}`` pointer is written by :data:`SUPERSEDE_LUA`
    itself, so a later
    ``SupersessionProtocol.supersede(successor, identity_key=...)`` finds no
    incumbent, closes nothing, and returns ``None``.

    Closing an interval requires an explicit producer. Today there are four:

    - :meth:`SupersessionProtocol.supersede` /
      :meth:`~popoto.fields.supersession.SupersessionProtocol.save_and_supersede`
      — use one of these for *every* identity-bearing write, including the
      first claim about an identity, or the second claim will close nothing.
    - :meth:`~popoto.fields.supersession.SupersessionProtocol.invalidate` /
      :meth:`~popoto.fields.supersession.SupersessionProtocol.save_and_invalidate`
      — the identity-free counterpart of the pair above.
    - ``ProvenanceJournal``.
    - ``ObservationProtocol.on_context_used`` with the ``"contradicted"``
      outcome *and* ``instance._superseded_by`` set — see
      ``_apply_supersession``. This is still an explicit application call, not
      an inference.

    The empty-set-versus-``None`` distinction is real but invisible to an
    adopter: gating did run, and excluded nothing. Whether this should stay
    imperative is the open question on issue #693.

    Why this is NOT a ``SortedFieldMixin`` (plan D2)
    ------------------------------------------------
    This is load-bearing and must not be "improved". ``ModelOptions.add_field``
    classifies any ``SortedFieldMixin`` into ``_meta.sorted_field_names``, which
    puts the field in ``filter_for_keys_set``'s *first* loop — where a returned
    **list** becomes ``Query._sorted_field_order`` and the field can win the
    query's ordering. Validity is membership, not priority: it must never order
    results. As a plain ``Field`` it lands in the second loop and returns a
    ``set`` (hence :meth:`filter_query`'s ``set`` return type — a list there
    would silently reintroduce the ordering bug). It also stays out of the
    reindex/migration loops that iterate ``sorted_field_names``.

    Query params (see :meth:`get_filter_query_params`):
        - ``{field}__current=True``  — records valid right now
        - ``{field}__current=False`` — the complement (closed / not-yet-started)
        - ``{field}__as_of=t``       — records valid at epoch ``t``

    These are *deliberate* queries: they consume a filter param and therefore
    disable sorted-range limit pushdown. That is exactly why the default
    retrieval path gates on validity server-side instead of via a filter kwarg.
    """

    # Export/import (issue #580 review blocker, PR #582): no validity byte
    # lives in the model hash -- the interval and chain state live entirely
    # in the six derived keys documented in the module docstring. A plain
    # re-save's ``on_save`` opens a *fresh* interval in mode="open" and has
    # no way to know about a prior close time or supersession chain, so the
    # ``Field`` default of ``roundtrip_policy = "rebuild"`` is a false claim
    # here: a transfer export/import round trip would silently reopen every
    # superseded record, because gating is subtractive (see the warning on
    # :meth:`resolve_valid_keys`) and a record with no interval entry is
    # fully retrievable. "carry" restores all six derived keys explicitly --
    # interval scores, chain links, and the identity-scoped open-claim
    # pointers -- after ``save()`` has already run.
    roundtrip_policy: str = "carry"

    #: Sentinel written into exported ``invalid_at`` in place of Python's
    #: ``float('inf')``. ``to_jsonable`` (transfer/format.py) passes floats
    #: through unchanged, and ``json.dumps`` would then emit the bare literal
    #: ``Infinity`` -- a token Python's own ``json.loads`` accepts back
    #: (so it *would* round-trip end-to-end) but which is not valid JSON per
    #: spec and would break any non-Python consumer of the export. A plain
    #: string sentinel keeps every exported line spec-valid JSON, consistent
    #: with the rest of the transfer format's convention of tagging
    #: non-primitive values explicitly rather than relying on interpreter
    #: leniency.
    OPEN_SENTINEL_TOKEN = "+inf"

    @classmethod
    def find_open_pointers_for_member(
        cls, model: "ModelLike", field_name: str, member_key: str
    ) -> "list[str]":
        """Return the pointer keys under ``{prefix}:open:*`` that name ``member_key``.

        There is no record -> identity-digest reverse lookup by design (plan D1
        fixes the key count at six, and the digest is opaque: identity is
        caller-defined and hashed by ``SupersessionProtocol.identity_key``), so
        the only way to answer "which identities currently claim this record as
        their open one?" is to ``SCAN`` the pointer keyspace and compare values.
        Both callers are admin/rare paths — :meth:`on_delete` and
        :meth:`export_state` — never save or read.

        A record may legitimately match zero pointers: it was never superseded
        on an identity, or it has already been closed and the pointer moved on.
        """
        prefix = cls.get_prefix_db_key(model, field_name).redis_key
        backend = non_redis_backend(model)
        if backend is not None:
            digests = backend.field_call(
                model._meta.spec,
                field_name,
                "pointers",
                _record_id(model, member_key),
            )
            return [
                cls.get_open_pointer_key(model, field_name, digest)
                for digest in digests
            ]
        matched = []
        for pointer_key in scan_keys(f"{prefix}:open:*"):
            pointer_key = _as_str(pointer_key)
            current = get_REDIS_DB().get(pointer_key)
            if current is not None and _as_str(current) == member_key:
                matched.append(pointer_key)
        return matched

    @classmethod
    def export_state(  # type: ignore[override]
        cls,
        model_instance: "Model",
        field_name: str,
        field_value: Any,
        **kwargs: Any,
    ) -> "Optional[dict[str, Any]]":
        """Export this record's interval scores, chain links, and open claims.

        Reads the member's score from each of the three interval ZSETs, its
        own forward/reverse chain-link entries, and every identity-scoped
        open-claim pointer (``{prefix}:open:{digest}``, see
        :meth:`get_open_pointer_key`) that currently names this record.

        All six derived keys are therefore carried. The pointer matters more
        than its "per-identity, not per-record" shape suggests:
        ``SupersessionProtocol.supersede(new, identity_key=...)`` resolves the
        incumbent *solely* through it (``old_member=''``, and
        :data:`SUPERSEDE_LUA` ``GET``s ``KEYS[4]``). A round trip that dropped
        the pointer would leave the next supersession on that identity closing
        nothing and writing no chain link, while still repointing the pointer
        at the newcomer -- orphaning the incumbent open forever. Gating is
        subtractive (see :meth:`resolve_valid_keys`), so that orphan stays
        fully retrievable: the same silent resurrection ``roundtrip_policy =
        "carry"`` exists to prevent, deferred by one supersession.

        Capturing the pointers costs a ``SCAN`` of ``{prefix}:open:*`` per
        record (see :meth:`find_open_pointers_for_member`) because the digest
        is opaque and there is no record -> digest reverse lookup. Export is
        an admin-path operation and that cost is accepted deliberately, on
        the same reasoning as :meth:`on_delete`'s scan.

        Returns:
            ``{"valid_from": float, "invalid_at": float | "+inf",
            "ingested_at": float, "chain_fwd": str | None,
            "chain_rev": str | None, "open_pointers": list[str]}``, or
            ``None`` when this instance has no interval entry at all (never
            saved through this field, or already deleted). ``open_pointers``
            holds identity digests, not full Redis keys, so the destination
            rebuilds them under its own prefix; it is commonly empty -- a
            record that was never superseded on an identity, or one already
            closed, legitimately owns no pointer, and its interval and chain
            state are still exported.
        """
        field = model_instance._meta.fields.get(field_name)
        if not isinstance(field, ValidityField):
            return None

        member_key = model_instance.db_key.redis_key
        backend = non_redis_backend(model_instance)
        if backend is not None:
            return cast(
                "Optional[dict[str, Any]]",
                backend.field_call(
                    model_instance._meta.spec,
                    field_name,
                    "export",
                    _record_id(model_instance, member_key),
                ),
            )
        keys = cls.get_all_keys(model_instance, field_name)

        valid_from = cast(
            "Optional[float]", get_REDIS_DB().zscore(keys["valid_from"], member_key)
        )
        invalid_at = cast(
            "Optional[float]", get_REDIS_DB().zscore(keys["invalid_at"], member_key)
        )
        ingested_at = cast(
            "Optional[float]", get_REDIS_DB().zscore(keys["ingested_at"], member_key)
        )
        if valid_from is None and invalid_at is None and ingested_at is None:
            return None

        fwd = get_REDIS_DB().hget(keys["chain_fwd"], member_key)
        rev = get_REDIS_DB().hget(keys["chain_rev"], member_key)

        invalid_at_out: Union[float, str]
        if invalid_at is None or float(invalid_at) == Defaults.VALIDITY_OPEN_SENTINEL:
            invalid_at_out = cls.OPEN_SENTINEL_TOKEN
        else:
            invalid_at_out = float(invalid_at)

        # Pointer keys are ``{prefix}:open:{digest}``; the digest is the last
        # segment (16 hex chars from blake2b, never contains a separator), so
        # a single rsplit recovers it without re-deriving the prefix.
        open_pointers = [
            pointer_key.rsplit(":", 1)[-1]
            for pointer_key in cls.find_open_pointers_for_member(
                model_instance, field_name, member_key
            )
        ]

        return {
            "valid_from": float(valid_from) if valid_from is not None else None,
            "invalid_at": invalid_at_out,
            "ingested_at": float(ingested_at) if ingested_at is not None else None,
            "chain_fwd": _as_str(fwd) if fwd is not None else None,
            "chain_rev": _as_str(rev) if rev is not None else None,
            # Plain list of strings: spec-valid JSON with no special tokens
            # needed, unlike ``invalid_at``'s :data:`OPEN_SENTINEL_TOKEN`.
            "open_pointers": open_pointers,
        }

    @classmethod
    def import_state(
        cls,
        model_instance: "Model",
        field_name: str,
        state: Any,
        **kwargs: Any,
    ) -> None:
        """Restore this record's interval scores and chain links after import.

        Written with plain ``ZADD``/``HSET`` -- deliberately NOT through
        :data:`SUPERSEDE_LUA` and NOT with the ``NX`` guards ``on_save``
        uses. The transfer driver calls ``import_state`` *after* ``save()``,
        so ``on_save`` has already run: it seeds ``valid_from`` /
        ``ingested_at`` / ``invalid_at`` via ``ZADD NX`` in mode="open",
        i.e. a fresh, open interval with no close time and no chain. An
        ``NX`` write here would be a silent no-op against those
        already-present scores, discarding the carried close time and chain
        -- exactly the resurrection bug this field exists to prevent.
        Plain ``ZADD``/``HSET`` overwrite instead.

        Carried open-claim pointers are restored with a plain ``SET``, for
        the same reason: ``on_save`` cannot write them at all on this path
        (it calls :meth:`execute_supersede` with no ``identity_digest``, so
        ``KEYS[4]`` is ``''`` and :data:`SUPERSEDE_LUA` skips the ``SET``),
        and an unconditional ``SET`` is what makes the identity's next
        supersession resolve this record as the incumbent (see
        :meth:`export_state` for why dropping it resurrects records). Only
        the digests actually captured at export are written, so a record
        that owned no open claim still writes none.

        Chain links are restored independently of import order: ``HSET``
        does not require the counterpart record to already exist, and
        ``_walk_links``-style chain traversal already treats a link to a
        record with no interval entry as a chain end (see
        :meth:`on_delete`'s "Known limitation" note), so a partially
        imported chain converges once both sides have landed.
        """
        if not state:
            return None

        field = model_instance._meta.fields.get(field_name)
        if not isinstance(field, ValidityField):
            return None

        member_key = model_instance.db_key.redis_key
        backend = non_redis_backend(model_instance)
        if backend is not None:
            backend.field_call(
                model_instance._meta.spec,
                field_name,
                "import",
                _record_id(model_instance, member_key),
                dict(state),
            )
            return None
        keys = cls.get_all_keys(model_instance, field_name)

        valid_from = state.get("valid_from")
        if valid_from is not None:
            get_REDIS_DB().zadd(keys["valid_from"], {member_key: float(valid_from)})

        invalid_at = state.get("invalid_at")
        if invalid_at is not None:
            score = (
                Defaults.VALIDITY_OPEN_SENTINEL
                if invalid_at == cls.OPEN_SENTINEL_TOKEN
                else float(invalid_at)
            )
            get_REDIS_DB().zadd(keys["invalid_at"], {member_key: score})

        ingested_at = state.get("ingested_at")
        if ingested_at is not None:
            get_REDIS_DB().zadd(keys["ingested_at"], {member_key: float(ingested_at)})

        chain_fwd = state.get("chain_fwd")
        if chain_fwd:
            get_REDIS_DB().hset(keys["chain_fwd"], member_key, chain_fwd)

        chain_rev = state.get("chain_rev")
        if chain_rev:
            get_REDIS_DB().hset(keys["chain_rev"], member_key, chain_rev)

        for digest in state.get("open_pointers") or []:
            get_REDIS_DB().set(
                cls.get_open_pointer_key(model_instance, field_name, str(digest)),
                member_key,
            )

        return None

    def __init__(self, **kwargs: Any) -> None:
        """Initialize the field.

        Args:
            **kwargs: Standard :class:`~popoto.fields.field.Field` options.
                ``type`` defaults to ``float`` (the stored valid-from epoch) and
                ``null`` to ``True`` — a record may be saved without an explicit
                valid-from, in which case save time is used.
        """
        kwargs.setdefault("type", float)
        kwargs.setdefault("null", True)
        super().__init__(**kwargs)

    # ------------------------------------------------------------------
    # Key helpers (plan D1)
    # ------------------------------------------------------------------

    @classmethod
    def get_prefix_db_key(cls, model: "ModelLike", field_name: str) -> DB_key:
        """Return the ``$ValidityF:{Model}:{field}`` prefix all six keys extend."""
        # ``Field.get_special_use_field_db_key`` annotates an instance but reads
        # only ``_meta``, which a model class carries too — hence ``ModelLike``
        # here and the narrowing cast at the boundary.
        return cls.get_special_use_field_db_key(cast("Model", model), field_name)

    @classmethod
    def get_valid_from_key(cls, model: "ModelLike", field_name: str) -> str:
        """Redis key of the ``valid_from`` ZSET (member -> valid-from epoch)."""
        return DB_key(cls.get_prefix_db_key(model, field_name), "valid_from").redis_key

    @classmethod
    def get_invalid_at_key(cls, model: "ModelLike", field_name: str) -> str:
        """Redis key of the ``invalid_at`` ZSET (member -> close epoch, ``+inf`` = open)."""
        return DB_key(cls.get_prefix_db_key(model, field_name), "invalid_at").redis_key

    @classmethod
    def get_ingested_at_key(cls, model: "ModelLike", field_name: str) -> str:
        """Redis key of the ``ingested_at`` ZSET (member -> transaction-time epoch)."""
        return DB_key(cls.get_prefix_db_key(model, field_name), "ingested_at").redis_key

    @classmethod
    def get_chain_fwd_key(cls, model: "ModelLike", field_name: str) -> str:
        """Redis key of the forward chain HASH (old redis_key -> superseding key)."""
        return DB_key(
            cls.get_prefix_db_key(model, field_name), "chain", "fwd"
        ).redis_key

    @classmethod
    def get_chain_rev_key(cls, model: "ModelLike", field_name: str) -> str:
        """Redis key of the reverse chain HASH (new redis_key -> superseded key)."""
        return DB_key(
            cls.get_prefix_db_key(model, field_name), "chain", "rev"
        ).redis_key

    @classmethod
    def get_open_pointer_key(
        cls, model: "ModelLike", field_name: str, identity_digest: str
    ) -> str:
        """Redis key of the open-claim pointer STRING for one identity digest.

        Args:
            model: The model class (or instance) owning the field.
            field_name: Name of the ``ValidityField`` on that model.
            identity_digest: Opaque, already-normalized identity token — see
                ``SupersessionProtocol.identity_key`` (plan D7), which hashes the
                caller's ``(subject, predicate)`` into 16 hex characters so raw
                user text can never reach the keyspace.
        """
        return DB_key(
            cls.get_prefix_db_key(model, field_name), "open", identity_digest
        ).redis_key

    @classmethod
    def get_interval_keys(
        cls, model: "ModelLike", field_name: str
    ) -> "tuple[str, str]":
        """Return ``(valid_from_key, invalid_at_key)`` for a model/field pair.

        This is the stable seam every other validity consumer builds on: the
        decay-Lua gate passes these two as extra ``KEYS``, the composite-score
        mask ``ZRANGESTORE``s from them, and the assembler's
        ``_resolve_excluded_keys`` reads them directly. Returned in the order
        ``(valid_from, invalid_at)``; an as-of-``t`` member satisfies
        ``valid_from <= t AND invalid_at > t``.

        Args:
            model: The model class (or instance) owning the field.
            field_name: Name of the ``ValidityField`` on that model.

        Returns:
            A 2-tuple of Redis key strings.
        """
        return (
            cls.get_valid_from_key(model, field_name),
            cls.get_invalid_at_key(model, field_name),
        )

    @classmethod
    def get_all_keys(cls, model: "ModelLike", field_name: str) -> "dict[str, str]":
        """Return the five non-identity keys as a name -> Redis key mapping.

        Keys: ``valid_from``, ``invalid_at``, ``ingested_at``, ``chain_fwd``,
        ``chain_rev``. The sixth key (the per-identity open pointer) is excluded
        because it is parameterized by identity digest — use
        :meth:`get_open_pointer_key` for that one.
        """
        return {
            "valid_from": cls.get_valid_from_key(model, field_name),
            "invalid_at": cls.get_invalid_at_key(model, field_name),
            "ingested_at": cls.get_ingested_at_key(model, field_name),
            "chain_fwd": cls.get_chain_fwd_key(model, field_name),
            "chain_rev": cls.get_chain_rev_key(model, field_name),
        }

    # ------------------------------------------------------------------
    # Membership resolution
    # ------------------------------------------------------------------

    @classmethod
    def resolve_valid_keys(
        cls,
        model: "ModelLike",
        field_name: str,
        as_of: Optional[float] = None,
    ) -> "set[str]":
        """Return the record keys whose interval covers ``as_of``.

        Two read-only ``ZRANGEBYSCORE``s intersected:
        ``valid_from <= as_of`` AND ``invalid_at > as_of``.

        .. warning::

           This is a **whitelist**: a record with no entry in either ZSET is
           absent from the result. That makes it wrong for retrieval gating,
           and it is deliberately **not** used by any retrieval path. All three
           gating layers are *subtractive* — they exclude records that are
           positively known to be closed or not-yet-started, and leave
           unmanaged records (no interval at all) fully visible. Using this
           method to gate retrieval would silently hide every record that
           predates the field's adoption on a model. The assembler uses
           ``ContextAssembler._resolve_excluded_keys`` instead, and the query
           layer uses ``QueryBuilder._apply_validity_mask``.

        Retained as the public, deliberate-query helper for callers that
        genuinely want "which records positively claim validity at ``t``" —
        e.g. audit and provenance tooling, not context assembly.

        Args:
            model: The model class (or instance) owning the field.
            field_name: Name of the ``ValidityField`` on that model.
            as_of: Epoch seconds to evaluate at. ``None`` means "now".

        Returns:
            ``set[str]`` of decoded Redis keys. Decoded — not bytes — because
            every consumer compares against ``record.db_key.redis_key``, which is
            a ``str``; the ``bytes`` form is confined to :meth:`filter_query`,
            where the query layer intersects raw index replies.

        Note:
            This is a point-in-time snapshot of a live store. A supersession
            landing after the read is not reflected until the next call — the
            same accepted property tag scoping already has (plan Race 3).
        """
        t = time.time() if as_of is None else float(as_of)
        backend = non_redis_backend(model)
        if backend is not None:
            return _backend_members(backend, model, field_name, t, "valid")
        valid_from_key, invalid_at_key = cls.get_interval_keys(model, field_name)
        # redis-py types every command ``Awaitable[T] | T`` for both the sync and
        # async clients, so mypy cannot see that these are concrete lists on the
        # sync client we actually use. Same narrowing cast as
        # ``ContextAssembler._resolve_excluded_keys``; CLAUDE.md notes this error
        # family is redis-py-version-dependent.
        started = cast(
            "list[Any]", get_REDIS_DB().zrangebyscore(valid_from_key, "-inf", t)
        )
        still_open = cast(
            "list[Any]", get_REDIS_DB().zrangebyscore(invalid_at_key, f"({t}", "+inf")
        )
        return {_as_str(m) for m in started} & {_as_str(m) for m in still_open}

    @classmethod
    def resolve_excluded_keys(
        cls,
        model: "ModelLike",
        field_name: str,
        as_of: Optional[float] = None,
    ) -> "set[str]":
        """Return the record keys to DROP as of ``as_of`` (#648).

        The subtractive counterpart to :meth:`resolve_valid_keys`, and the one
        every retrieval path actually wants. Two read-only ``ZRANGEBYSCORE``s,
        unioned: ``invalid_at <= as_of`` (already closed) and
        ``valid_from > as_of`` (not yet started) — the exact mirror of the two
        ranges :meth:`resolve_valid_keys` intersects.

        .. warning::

           **This returns an exclusion set, not a whitelist, and that is
           deliberate.** A record with no entry in either interval ZSET is
           *unmanaged* and stays fully retrievable. Resolving the valid set
           instead — via :meth:`resolve_valid_keys`, whose own warning is the
           other half of this one — would silently hide every record that
           predates the day a ``ValidityField`` was added to an existing model,
           since none of those has an interval until it is next saved. That is a
           data-visibility regression, not a stricter gate. Do not "simplify"
           this into a valid-key intersection.

           This method lives beside ``resolve_valid_keys`` precisely so the
           warning and the trap it warns about are read together. It used to
           live in ``ContextAssembler._resolve_excluded_keys``, a different file
           from the method it forbids.

        The exclusive lower bound on ``valid_from`` is spelled ``f"({t}"``,
        while the ``invalid_at`` upper bound is inclusive. Redis renders the
        ``+inf`` open sentinel such that an open record never matches
        ``<= as_of``. Both bound conventions are this method's business, not a
        caller's.

        Callers own the *policy* questions this does not answer: whether the
        model declares a ``ValidityField`` at all, and whether the
        ``Defaults.VALIDITY_GATING_ENABLED`` kill switch is on (which must be
        read at call time, never captured at import, so a deploy-level switch
        takes effect for adopters who cannot edit model code).

        Args:
            model: The model class (or instance) owning the field.
            field_name: Name of the ``ValidityField`` on that model.
            as_of: Epoch seconds to evaluate membership at. ``None`` = now.

        Returns:
            ``set[str]`` of decoded record keys to drop. An empty set means
            gating ran and excluded nothing — which is **not** the same as a
            caller's ``None`` for "gating did not run at all".

        Note:
            A point-in-time snapshot of a live store. A supersession landing
            after the read is not reflected until the next call.
        """
        t = time.time() if as_of is None else float(as_of)
        backend = non_redis_backend(model)
        if backend is not None:
            return _backend_members(backend, model, field_name, t, "excluded")
        valid_from_key, invalid_at_key = cls.get_interval_keys(model, field_name)
        # invalid_at <= t: already closed. The +inf open sentinel never matches.
        # Read before valid_from, the order the assembler established.
        closed = cast(
            "list[Any]", get_REDIS_DB().zrangebyscore(invalid_at_key, "-inf", t)
        )
        # valid_from > t: not yet started.
        future = cast(
            "list[Any]", get_REDIS_DB().zrangebyscore(valid_from_key, f"({t}", "+inf")
        )
        return {_as_str(m) for m in closed + future}

    @classmethod
    def is_valid_at(
        cls,
        model: "ModelLike",
        field_name: str,
        member_key: str,
        as_of: Optional[float] = None,
    ) -> bool:
        """Return whether one record's interval covers ``as_of``.

        Two ``ZSCORE``s rather than two range reads — the single-member form of
        :meth:`resolve_valid_keys`. A member absent from either ZSET is not valid
        (it has no interval).
        """
        t = time.time() if as_of is None else float(as_of)
        backend = non_redis_backend(model)
        if backend is not None:
            state = _backend_interval(backend, model, field_name, member_key) or {}
            start = state.get("valid_from")
            close = state.get("invalid_at")
        else:
            valid_from_key, invalid_at_key = cls.get_interval_keys(model, field_name)
            # redis-py types ZSCORE ``Awaitable[float | None] | float | None``
            # to cover the async client; narrow to the sync reply (see the
            # cast note in :meth:`resolve_valid_keys`).
            start = cast(
                "Optional[float]", get_REDIS_DB().zscore(valid_from_key, member_key)
            )
            close = cast(
                "Optional[float]", get_REDIS_DB().zscore(invalid_at_key, member_key)
            )
        if start is None or close is None:
            return False
        if float(close) == Defaults.VALIDITY_OPEN_SENTINEL:
            return float(start) <= t
        return float(start) <= t < float(close)

    # ------------------------------------------------------------------
    # Script execution
    # ------------------------------------------------------------------

    @classmethod
    def execute_supersede(
        cls,
        model: "ModelLike",
        field_name: str,
        *,
        new_member: str = "",
        mode: str = "open",
        now: Optional[float] = None,
        valid_from: Optional[float] = None,
        ingested_at: Optional[float] = None,
        close_at: Optional[float] = None,
        old_member: str = "",
        identity_digest: str = "",
        assert_valid_from: bool = False,
        pipeline: Optional[redis.client.Pipeline] = None,
    ) -> Any:
        """Run :data:`SUPERSEDE_LUA` once against this model/field's six keys.

        The single seam through which every validity mutation flows —
        ``ValidityField.on_save`` and ``SupersessionProtocol`` both call it, so
        there is exactly one place that knows the script's KEYS/ARGV order.

        Args:
            model: The model class (or instance) owning the field.
            field_name: Name of the ``ValidityField`` on that model.
            new_member: Redis key of the record whose interval opens. Empty for
                a pure ``invalidate``.
            mode: ``'open'`` (a save: open the newcomer, close nothing),
                ``'supersede'`` (close the incumbent and chain it to the
                newcomer), or ``'invalidate'`` (close only).
            now: Epoch seconds to use as the script's clock. Defaults to
                ``time.time()``. Passing one clock for a batch keeps intervals
                consistent across a multi-record write.
            valid_from: Valid-from epoch for ``new_member``. Defaults to ``now``.
            ingested_at: Transaction-time epoch for ``new_member``. Defaults to
                ``now``.
            close_at: Explicit close epoch for the incumbent. Defaults to ``now``.
            old_member: Explicit incumbent key, bypassing identity resolution.
            identity_digest: Identity digest naming the open-claim pointer. When
                empty, ``KEYS[4]`` is passed as ``''`` and the script neither
                reads nor repoints a pointer.
            assert_valid_from: When ``True``, ``valid_from`` is a caller
                *assertion* about ``new_member``'s start rather than a default,
                and a disagreement with the already-stored start raises
                :class:`ValidityValidFromConflictError` instead of losing
                silently to the script's ``ZADD NX`` (plan D3). ``False`` — the
                default, and what ``SupersessionProtocol`` and
                ``ProvenanceJournal`` pass — preserves today's behavior for every
                existing caller: their ``at=`` is a *close-time* assertion about
                the incumbent, not a start-time assertion about the successor.
            pipeline: Optional external pipeline. When given, the EVAL is queued
                (following ``tag_field.py``'s threading shape) and the pipeline
                is returned; the closed-member result is only available at
                ``execute()`` time.

        Returns:
            The closed member key as ``str``, or ``None`` if nothing was closed —
            or the ``pipeline`` when one was supplied.

        Raises:
            ValueError: If ``mode`` is not one of :data:`VALID_MODES`, or if the
                client-side pre-check finds ``close_at`` before ``valid_from``.
            ValidityMemberAbsentError: If ``new_member``, or an explicitly-named
                ``old_member``, does not exist at the instant of the write.
            ValidityCloseBeforeStartError: If ``close_at`` precedes the
                incumbent's stored ``valid_from``.
            ValidityValidFromConflictError: If ``assert_valid_from`` is set and
                ``valid_from`` disagrees with the stored start.

        Note:
            The ``ResponseError`` -> typed-exception remap lives on the
            **non-pipeline** branch only. On a caller-supplied pipeline redis-py
            raises during ``pipe.execute()`` result parsing, long after this
            method returned, so the caller sees a raw
            ``redis.exceptions.ResponseError``. Use
            :meth:`SupersessionProtocol.save_and_supersede`, which owns its
            ``execute()``, to get a typed error in pipeline shape.
        """
        if mode not in VALID_MODES:
            raise ValueError(
                f"ValidityField mode must be one of {sorted(VALID_MODES)}, got {mode!r}"
            )
        clock = time.time() if now is None else float(now)
        if close_at is not None and valid_from is not None and mode != "open":
            # Cheap client-side pre-check for the direct-invalidation form, where
            # the caller already knows both ends. The authoritative check lives in
            # the script (the stored valid_from is the one that matters).
            if float(close_at) < float(valid_from):
                raise ValueError(
                    "ValidityField: close-at "
                    f"({close_at}) precedes valid_from ({valid_from})"
                )

        backend = non_redis_backend(model)
        if backend is not None:
            return cls._execute_supersede_on_backend(
                backend,
                model,
                field_name,
                new_member=new_member,
                mode=mode,
                now=clock,
                valid_from=valid_from,
                ingested_at=ingested_at,
                close_at=close_at,
                old_member=old_member,
                identity_digest=identity_digest,
                assert_valid_from=assert_valid_from,
                pipeline=pipeline,
            )

        keys = cls.get_all_keys(model, field_name)
        pointer_key = (
            cls.get_open_pointer_key(model, field_name, identity_digest)
            if identity_digest
            else ""
        )
        args = [
            keys["valid_from"],  # KEYS[1]
            keys["invalid_at"],  # KEYS[2]
            keys["ingested_at"],  # KEYS[3]
            pointer_key,  # KEYS[4] (may be '')
            keys["chain_fwd"],  # KEYS[5]
            keys["chain_rev"],  # KEYS[6]
            new_member or "",  # ARGV[1]
            repr(clock),  # ARGV[2]
            "" if valid_from is None else repr(float(valid_from)),  # ARGV[3]
            "" if ingested_at is None else repr(float(ingested_at)),  # ARGV[4]
            mode,  # ARGV[5]
            "" if close_at is None else repr(float(close_at)),  # ARGV[6]
            old_member or "",  # ARGV[7]
            "1" if assert_valid_from else "",  # ARGV[8]
        ]

        if isinstance(pipeline, redis.client.Pipeline):
            run_lua(pipeline, SUPERSEDE_LUA, 6, *args)
            return pipeline
        try:
            result = run_lua(get_REDIS_DB(), SUPERSEDE_LUA, 6, *args)
        except redis.exceptions.ResponseError as e:
            raise map_lua_error(e) from e
        closed = _as_str(result) if result else ""
        return closed or None

    @classmethod
    def _execute_supersede_on_backend(
        cls,
        backend: Any,
        model: "ModelLike",
        field_name: str,
        *,
        new_member: str,
        mode: str,
        now: float,
        valid_from: Optional[float],
        ingested_at: Optional[float],
        close_at: Optional[float],
        old_member: str,
        identity_digest: str,
        assert_valid_from: bool,
        pipeline: Any,
    ) -> Any:
        """:meth:`execute_supersede` on a non-Redis backend (#759 M3): the
        backend's ``supersede``, ``SUPERSEDE_LUA`` phase for phase in one
        transaction, raising the same typed errors with the same text.

        With the backend's own unit of work the supersede runs inside its
        transaction, so the closed key is known and returned (where a queued
        ``EVAL`` returns the pipeline), and an error is typed at once. A
        Redis pipeline cannot carry the write: it runs now, and the pipeline
        comes back untouched."""
        from ..batch import unit_of

        # A transaction() unit of work, or a popoto.batch()'s (#759 M5).
        uow = unit_of(pipeline, backend)
        closed = backend.supersede(
            model._meta.spec,
            field_name,
            successor=_record_id(model, new_member) if new_member else None,
            incumbent=_record_id(model, old_member) if old_member else None,
            identity=identity_digest or None,
            mode=mode,
            valid_from=valid_from,
            invalid_at=close_at,
            now=now,
            uow=uow,
            ingested_at=ingested_at,
            assert_valid_from=assert_valid_from,
        )
        if pipeline is not None and uow is None:
            return pipeline
        return closed or None

    # ------------------------------------------------------------------
    # TTL interaction (plan D9)
    # ------------------------------------------------------------------

    @classmethod
    def warn_if_ttl(cls, model: "ModelLike", field_name: str) -> bool:
        """Warn once when a ``ValidityField`` model also declares a TTL.

        A TTL truncates supersession chains and silently breaks ``as_of``
        correctness: the expired record vanishes while its chain links and
        interval entries remain. This warns rather than raises — refusing would
        break adopters who legitimately want bounded history (plan D9).

        Returns:
            ``True`` if a warning was emitted on this call.
        """
        meta = getattr(model, "_meta", None)
        if meta is None or getattr(meta, "ttl", None) is None:
            return False
        marker = (getattr(meta, "model_name", str(model)), field_name)
        if marker in _TTL_WARNED:
            return False
        _TTL_WARNED.add(marker)
        logger.warning(
            "%s.%s is a ValidityField on a model with Meta.ttl=%s. TTL expiry "
            "truncates supersession chains and breaks as_of reconstruction: the "
            "record disappears while its interval and chain links remain. Drop "
            "the TTL, or accept bounded history.",
            marker[0],
            field_name,
            meta.ttl,
        )
        return True

    # ------------------------------------------------------------------
    # Model hooks
    # ------------------------------------------------------------------

    @classmethod
    def on_save(
        cls,
        model_instance: "Model",
        field_name: str,
        field_value: Any,
        pipeline: Optional[redis.client.Pipeline] = None,
        **kwargs: Any,
    ) -> Any:
        """Open (or re-affirm) this record's validity interval.

        Routes through :data:`SUPERSEDE_LUA` in mode ``'open'`` rather than
        issuing a bare ``ZADD``, which is what makes the save path safe against
        plan Race 2: the script's ``ZSCORE``/``NX`` guards mean a save that
        interleaves with a concurrent supersession can never resurrect an
        already-closed record, and a re-save never shifts an existing interval's
        start or ingest time.

        ``field_value`` is used as the valid-from epoch when it is numeric;
        otherwise save time is used. The write is idempotent, so repeated saves
        of an open record are no-ops on the index.

        Args:
            model_instance: The instance being saved.
            field_name: Name of this field on the model.
            field_value: The declared valid-from epoch, or ``None``.
            pipeline: Optional external pipeline; the EVAL is queued onto it.
            **kwargs: Accepted for forward compatibility.

        Returns:
            The ``pipeline`` if one was provided, else the script result.
        """
        cls.warn_if_ttl(model_instance, field_name)
        now = time.time()
        declared = False
        try:
            if field_value is not None:
                valid_from = float(field_value)
                declared = True
            else:
                valid_from = now
        except (TypeError, ValueError):
            valid_from = now
        return cls.execute_supersede(
            model_instance,
            field_name,
            new_member=model_instance.db_key.redis_key,
            mode="open",
            now=now,
            valid_from=valid_from,
            ingested_at=now,
            # Plan D3: a declared field value IS the single authoritative writer
            # of valid-time, so a re-save that declares a different start is the
            # reporter's bug and must be loud. A defaulted save asserts nothing
            # and keeps NX idempotence.
            assert_valid_from=declared,
            pipeline=pipeline,
        )

    @classmethod
    def pre_save_validate(
        cls,
        model_instance: "Model",
        field_name: str,
        field_value: Any,
        **kwargs: Any,
    ) -> None:
        """Refuse a save that declares a ``valid_from`` the index disagrees with.

        Runs from the single pre-split dispatch site in ``Model.save()``, before
        *any* write is issued or queued — which is the point. Raising from
        :meth:`on_save` would be too late on a model that also declares
        ``IndexedFieldMixin`` fields: those commit their hash values and index
        entries eagerly, against live Redis, before ``ValidityField.on_save``
        ever runs (plan D5 half 1, the same treatment #476 gave the
        unique-conflict window).

        A cheap client-side pre-check for a good error message; the authoritative
        comparison is in :data:`SUPERSEDE_LUA` under ``ARGV[8]``, evaluated
        atomically (plan Race 4). A racing pre-check can only produce a false
        negative, which the script then catches.
        """
        if field_value is None:
            return  # defaulted: no assertion, nothing to conflict with
        try:
            declared = float(field_value)
        except (TypeError, ValueError):
            return  # on_save falls back to the save clock; not an assertion
        try:
            member_key = model_instance.db_key.redis_key
        except (TypeError, ValueError):
            return
        if not member_key:
            return
        backend = non_redis_backend(model_instance)
        if backend is not None:
            state = _backend_interval(backend, model_instance, field_name, member_key)
            stored = None if state is None else state["valid_from"]
        else:
            stored = get_REDIS_DB().zscore(
                cls.get_valid_from_key(model_instance, field_name), member_key
            )
        if stored is not None and float(stored) != declared:
            raise ValidityValidFromConflictError(
                f"ValidityField: {model_instance.__class__.__name__}.{field_name} "
                f"declares valid_from={declared!r} for {member_key}, but the "
                f"index already holds {float(stored)!r}. Valid-time has one "
                "writer -- the field value at construction. Adopt the stored "
                "value with ValidityField.get_valid_from(...), or overwrite the "
                "index with a plain ZADD (no NX), then save again."
            )

    @classmethod
    def get_valid_from(
        cls,
        model: "ModelLike",
        field_name: str,
        member_key: Optional[str] = None,
    ) -> Optional[float]:
        """Return the record's *effective* valid-from — the index score.

        ``instance.validity`` is the **declared** value: ``None`` there means
        "not declared, defaulted to the save clock". This returns what the
        ``valid_from`` index actually holds, which is what every ``as_of`` query
        answers against (plan D5 half 2). The two differ legitimately, and
        reading this is how an operator reconciles a record whose hash and index
        already disagree.

        Args:
            model: The model class, or a live instance.
            field_name: Name of the ``ValidityField``.
            member_key: The record's Redis key. Defaults to ``model``'s own key
                when an instance was passed.

        Returns:
            The stored valid-from epoch, or ``None`` when the member has no
            entry in the index.
        """
        if member_key is None:
            try:
                member_key = model.db_key.redis_key  # type: ignore[union-attr]
            except (AttributeError, TypeError, ValueError) as e:
                raise ValueError(
                    "ValidityField.get_valid_from: pass member_key when the "
                    "first argument is a model class rather than an instance"
                ) from e
        backend = non_redis_backend(model)
        if backend is not None:
            state = _backend_interval(backend, model, field_name, member_key)
            score = None if state is None else state["valid_from"]
        else:
            score = get_REDIS_DB().zscore(
                cls.get_valid_from_key(model, field_name), member_key
            )
        return None if score is None else float(score)

    @classmethod
    def on_delete(
        cls,
        model_instance: "Model",
        field_name: str,
        field_value: Any,
        pipeline: Optional[redis.client.Pipeline] = None,
        **kwargs: Any,
    ) -> Any:
        """Remove every trace of this record from the validity keyspace.

        ``ZREM`` from the three interval ZSETs, ``HDEL`` from both chain HASHes,
        and clear any open-claim pointer still aimed at the record. Records are
        normally *closed*, not deleted — this hook exists so an explicit
        ``delete()`` (or a key migration, which calls it with ``saved_redis_key``)
        does not leave orphaned index members behind.

        Pointer cleanup scans ``{prefix}:open:*`` because the pointer keyspace is
        keyed by identity digest, not by member — the reverse lookup does not
        exist by design (plan D1 fixes the key count at six). This is a
        delete-time-only cost on a path that is rare relative to save and read;
        adding a seventh per-record back-pointer key to avoid it was rejected as
        the worse trade.

        Known limitation: the deleted key is removed from both chain HASHes as a
        *field*, but a neighbor's link may still name it as a *value* (``fwd``
        holds ``old -> deleted``). Scrubbing the value side would mean an
        ``HGETALL`` of the whole chain on every delete. Chain traversal treats a
        link to a record with no interval as a chain end, which is the correct
        reading of a hard-deleted link.

        Returns:
            The ``pipeline`` if one was provided, else the pointer-cleanup result.
        """
        member = kwargs.get("saved_redis_key") or model_instance.db_key.redis_key
        keys = cls.get_all_keys(model_instance, field_name)

        stale_pointers = cls.find_open_pointers_for_member(
            model_instance, field_name, member
        )

        if pipeline is not None:
            pipeline.zrem(keys["valid_from"], member)
            pipeline.zrem(keys["invalid_at"], member)
            pipeline.zrem(keys["ingested_at"], member)
            pipeline.hdel(keys["chain_fwd"], member)
            pipeline.hdel(keys["chain_rev"], member)
            for pointer_key in stale_pointers:
                pipeline.delete(pointer_key)
            return pipeline

        get_REDIS_DB().zrem(keys["valid_from"], member)
        get_REDIS_DB().zrem(keys["invalid_at"], member)
        get_REDIS_DB().zrem(keys["ingested_at"], member)
        get_REDIS_DB().hdel(keys["chain_fwd"], member)
        get_REDIS_DB().hdel(keys["chain_rev"], member)
        result: Any = 0
        for pointer_key in stale_pointers:
            result = get_REDIS_DB().delete(pointer_key)
        return result

    # ------------------------------------------------------------------
    # Query integration
    # ------------------------------------------------------------------

    def get_filter_query_params(self, field_name: str) -> "set[str]":
        """Declare the two deliberate validity lookups.

        ``{field}__as_of=t`` (records valid at epoch ``t``) and
        ``{field}__current=True|False`` (valid now / the complement).
        Unioned with ``super()``'s set per the base-class contract.
        """
        return super().get_filter_query_params(field_name) | {
            f"{field_name}__as_of",
            f"{field_name}__current",
        }

    @classmethod
    def filter_query(
        cls,
        model: "Model",
        field_name: str,
        **query_params: Any,
    ) -> "set[Any]":
        """Resolve validity lookups to a ``set`` of matching Redis keys.

        Two ``ZRANGEBYSCORE`` reads intersected — ``valid_from <= t`` AND
        ``invalid_at > t``. ``__current=True`` evaluates at now;
        ``__current=False`` returns the complement (every member with an
        interval that does *not* cover now: closed, or not yet started). Multiple
        params AND-intersect, consistent with the rest of ``filter_for_keys_set``.

        Returns a ``set`` and never a ``list``: the query layer turns a list
        return into ``Query._sorted_field_order``, and validity must never order
        results (plan D2).

        Args:
            model: The model class being queried.
            field_name: Name of this field on the model.
            **query_params: ``{field}__as_of`` and/or ``{field}__current``.

        Returns:
            ``set`` of Redis keys (``bytes``, matching the other index fields'
            reply type) for records matching every supplied param.

        Raises:
            ValueError: If ``__current`` is not a bool or ``__as_of`` is not a
                finite number.
        """
        backend = non_redis_backend(model)
        if backend is not None:
            return cls._filter_query_on_backend(
                backend, model, field_name, **query_params
            )
        valid_from_key, invalid_at_key = cls.get_interval_keys(model, field_name)
        results = []

        for query_param, query_value in query_params.items():
            if query_param == f"{field_name}__current":
                if not isinstance(query_value, bool):
                    raise ValueError(
                        f"{query_param} filter must be True or False, "
                        f"got {query_value!r}"
                    )
                t = time.time()
                valid = cls._members_valid_at(valid_from_key, invalid_at_key, t)
                if query_value:
                    results.append(valid)
                else:
                    # Narrow the sync ZRANGE replies (see the cast note in
                    # :meth:`resolve_valid_keys`).
                    everything = set(
                        cast("list[Any]", get_REDIS_DB().zrange(invalid_at_key, 0, -1))
                    ) | set(
                        cast("list[Any]", get_REDIS_DB().zrange(valid_from_key, 0, -1))
                    )
                    results.append(everything - valid)

            elif query_param == f"{field_name}__as_of":
                try:
                    t = float(query_value)
                except (TypeError, ValueError) as e:
                    raise ValueError(
                        f"{query_param} filter must be a number of epoch seconds, "
                        f"got {query_value!r}"
                    ) from e
                results.append(cls._members_valid_at(valid_from_key, invalid_at_key, t))

        if not results:
            return set()
        matched = results[0]
        for other in results[1:]:
            matched &= other
        return matched

    @classmethod
    def _filter_query_on_backend(
        cls, backend: Any, model: Any, field_name: str, **query_params: Any
    ) -> "set[Any]":
        """:meth:`filter_query` for a model on a non-Redis backend: the same
        validation and set algebra over the backend's interval reads, keys
        as ``bytes`` as the Redis replies are. (``Query.filter`` does not come
        here on such a backend; the planner compiles the lookups to SQL.)"""
        from ..backends import QueryPlan
        from ..backends.planning import validity_cond

        results = []
        for query_param, query_value in query_params.items():
            suffix = query_param[len(field_name) + 2 :]
            if suffix not in ("as_of", "current"):
                continue
            cond = validity_cond(field_name, query_param, suffix, query_value)
            rows = backend.select(model._meta.spec, QueryPlan(where=cond, project=()))
            results.append({row["_id"].canonical.encode() for row in rows})
        if not results:
            return set()
        matched = results[0]
        for other in results[1:]:
            matched &= other
        return matched

    @staticmethod
    def _members_valid_at(
        valid_from_key: str, invalid_at_key: str, t: float
    ) -> "set[Any]":
        """Raw (``bytes``) member set whose interval covers ``t``.

        The two-``ZRANGEBYSCORE`` primitive behind :meth:`filter_query`. Kept
        separate from :meth:`resolve_valid_keys` because the query layer
        intersects the raw index replies while assembler-side consumers want
        decoded ``str`` keys.
        """
        # Narrow the sync ZRANGEBYSCORE replies (see the cast note in
        # :meth:`resolve_valid_keys`).
        started = cast(
            "list[Any]", get_REDIS_DB().zrangebyscore(valid_from_key, "-inf", t)
        )
        still_open = cast(
            "list[Any]", get_REDIS_DB().zrangebyscore(invalid_at_key, f"({t}", "+inf")
        )
        return set(started) & set(still_open)

find_open_pointers_for_member(model, field_name, member_key) classmethod

Return the pointer keys under {prefix}:open:* that name member_key.

There is no record -> identity-digest reverse lookup by design (plan D1 fixes the key count at six, and the digest is opaque: identity is caller-defined and hashed by SupersessionProtocol.identity_key), so the only way to answer "which identities currently claim this record as their open one?" is to SCAN the pointer keyspace and compare values. Both callers are admin/rare paths — :meth:on_delete and :meth:export_state — never save or read.

A record may legitimately match zero pointers: it was never superseded on an identity, or it has already been closed and the pointer moved on.

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

@classmethod
def find_open_pointers_for_member(
    cls, model: "ModelLike", field_name: str, member_key: str
) -> "list[str]":
    """Return the pointer keys under ``{prefix}:open:*`` that name ``member_key``.

    There is no record -> identity-digest reverse lookup by design (plan D1
    fixes the key count at six, and the digest is opaque: identity is
    caller-defined and hashed by ``SupersessionProtocol.identity_key``), so
    the only way to answer "which identities currently claim this record as
    their open one?" is to ``SCAN`` the pointer keyspace and compare values.
    Both callers are admin/rare paths — :meth:`on_delete` and
    :meth:`export_state` — never save or read.

    A record may legitimately match zero pointers: it was never superseded
    on an identity, or it has already been closed and the pointer moved on.
    """
    prefix = cls.get_prefix_db_key(model, field_name).redis_key
    backend = non_redis_backend(model)
    if backend is not None:
        digests = backend.field_call(
            model._meta.spec,
            field_name,
            "pointers",
            _record_id(model, member_key),
        )
        return [
            cls.get_open_pointer_key(model, field_name, digest)
            for digest in digests
        ]
    matched = []
    for pointer_key in scan_keys(f"{prefix}:open:*"):
        pointer_key = _as_str(pointer_key)
        current = get_REDIS_DB().get(pointer_key)
        if current is not None and _as_str(current) == member_key:
            matched.append(pointer_key)
    return matched

export_state(model_instance, field_name, field_value, **kwargs) classmethod

Export this record's interval scores, chain links, and open claims.

Reads the member's score from each of the three interval ZSETs, its own forward/reverse chain-link entries, and every identity-scoped open-claim pointer ({prefix}:open:{digest}, see :meth:get_open_pointer_key) that currently names this record.

All six derived keys are therefore carried. The pointer matters more than its "per-identity, not per-record" shape suggests: SupersessionProtocol.supersede(new, identity_key=...) resolves the incumbent solely through it (old_member='', and :data:SUPERSEDE_LUA GETs KEYS[4]). A round trip that dropped the pointer would leave the next supersession on that identity closing nothing and writing no chain link, while still repointing the pointer at the newcomer -- orphaning the incumbent open forever. Gating is subtractive (see :meth:resolve_valid_keys), so that orphan stays fully retrievable: the same silent resurrection roundtrip_policy = "carry" exists to prevent, deferred by one supersession.

Capturing the pointers costs a SCAN of {prefix}:open:* per record (see :meth:find_open_pointers_for_member) because the digest is opaque and there is no record -> digest reverse lookup. Export is an admin-path operation and that cost is accepted deliberately, on the same reasoning as :meth:on_delete's scan.

Returns:

Type	Description
Optional[dict[str, Any]]	``{"valid_from": float, "invalid_at": float | "+inf",
Optional[dict[str, Any]]	"ingested_at": float, "chain_fwd": str | None,
Optional[dict[str, Any]]	"chain_rev": str | None, "open_pointers": list[str]}``, or
Optional[dict[str, Any]]	None when this instance has no interval entry at all (never
Optional[dict[str, Any]]	saved through this field, or already deleted). open_pointers
Optional[dict[str, Any]]	holds identity digests, not full Redis keys, so the destination
Optional[dict[str, Any]]	rebuilds them under its own prefix; it is commonly empty -- a
Optional[dict[str, Any]]	record that was never superseded on an identity, or one already
Optional[dict[str, Any]]	closed, legitimately owns no pointer, and its interval and chain
Optional[dict[str, Any]]	state are still exported.

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

@classmethod
def export_state(  # type: ignore[override]
    cls,
    model_instance: "Model",
    field_name: str,
    field_value: Any,
    **kwargs: Any,
) -> "Optional[dict[str, Any]]":
    """Export this record's interval scores, chain links, and open claims.

    Reads the member's score from each of the three interval ZSETs, its
    own forward/reverse chain-link entries, and every identity-scoped
    open-claim pointer (``{prefix}:open:{digest}``, see
    :meth:`get_open_pointer_key`) that currently names this record.

    All six derived keys are therefore carried. The pointer matters more
    than its "per-identity, not per-record" shape suggests:
    ``SupersessionProtocol.supersede(new, identity_key=...)`` resolves the
    incumbent *solely* through it (``old_member=''``, and
    :data:`SUPERSEDE_LUA` ``GET``s ``KEYS[4]``). A round trip that dropped
    the pointer would leave the next supersession on that identity closing
    nothing and writing no chain link, while still repointing the pointer
    at the newcomer -- orphaning the incumbent open forever. Gating is
    subtractive (see :meth:`resolve_valid_keys`), so that orphan stays
    fully retrievable: the same silent resurrection ``roundtrip_policy =
    "carry"`` exists to prevent, deferred by one supersession.

    Capturing the pointers costs a ``SCAN`` of ``{prefix}:open:*`` per
    record (see :meth:`find_open_pointers_for_member`) because the digest
    is opaque and there is no record -> digest reverse lookup. Export is
    an admin-path operation and that cost is accepted deliberately, on
    the same reasoning as :meth:`on_delete`'s scan.

    Returns:
        ``{"valid_from": float, "invalid_at": float | "+inf",
        "ingested_at": float, "chain_fwd": str | None,
        "chain_rev": str | None, "open_pointers": list[str]}``, or
        ``None`` when this instance has no interval entry at all (never
        saved through this field, or already deleted). ``open_pointers``
        holds identity digests, not full Redis keys, so the destination
        rebuilds them under its own prefix; it is commonly empty -- a
        record that was never superseded on an identity, or one already
        closed, legitimately owns no pointer, and its interval and chain
        state are still exported.
    """
    field = model_instance._meta.fields.get(field_name)
    if not isinstance(field, ValidityField):
        return None

    member_key = model_instance.db_key.redis_key
    backend = non_redis_backend(model_instance)
    if backend is not None:
        return cast(
            "Optional[dict[str, Any]]",
            backend.field_call(
                model_instance._meta.spec,
                field_name,
                "export",
                _record_id(model_instance, member_key),
            ),
        )
    keys = cls.get_all_keys(model_instance, field_name)

    valid_from = cast(
        "Optional[float]", get_REDIS_DB().zscore(keys["valid_from"], member_key)
    )
    invalid_at = cast(
        "Optional[float]", get_REDIS_DB().zscore(keys["invalid_at"], member_key)
    )
    ingested_at = cast(
        "Optional[float]", get_REDIS_DB().zscore(keys["ingested_at"], member_key)
    )
    if valid_from is None and invalid_at is None and ingested_at is None:
        return None

    fwd = get_REDIS_DB().hget(keys["chain_fwd"], member_key)
    rev = get_REDIS_DB().hget(keys["chain_rev"], member_key)

    invalid_at_out: Union[float, str]
    if invalid_at is None or float(invalid_at) == Defaults.VALIDITY_OPEN_SENTINEL:
        invalid_at_out = cls.OPEN_SENTINEL_TOKEN
    else:
        invalid_at_out = float(invalid_at)

    # Pointer keys are ``{prefix}:open:{digest}``; the digest is the last
    # segment (16 hex chars from blake2b, never contains a separator), so
    # a single rsplit recovers it without re-deriving the prefix.
    open_pointers = [
        pointer_key.rsplit(":", 1)[-1]
        for pointer_key in cls.find_open_pointers_for_member(
            model_instance, field_name, member_key
        )
    ]

    return {
        "valid_from": float(valid_from) if valid_from is not None else None,
        "invalid_at": invalid_at_out,
        "ingested_at": float(ingested_at) if ingested_at is not None else None,
        "chain_fwd": _as_str(fwd) if fwd is not None else None,
        "chain_rev": _as_str(rev) if rev is not None else None,
        # Plain list of strings: spec-valid JSON with no special tokens
        # needed, unlike ``invalid_at``'s :data:`OPEN_SENTINEL_TOKEN`.
        "open_pointers": open_pointers,
    }

import_state(model_instance, field_name, state, **kwargs) classmethod

Restore this record's interval scores and chain links after import.

Written with plain ZADD/HSET -- deliberately NOT through :data:SUPERSEDE_LUA and NOT with the NX guards on_save uses. The transfer driver calls import_state after save(), so on_save has already run: it seeds valid_from / ingested_at / invalid_at via ZADD NX in mode="open", i.e. a fresh, open interval with no close time and no chain. An NX write here would be a silent no-op against those already-present scores, discarding the carried close time and chain -- exactly the resurrection bug this field exists to prevent. Plain ZADD/HSET overwrite instead.

Carried open-claim pointers are restored with a plain SET, for the same reason: on_save cannot write them at all on this path (it calls :meth:execute_supersede with no identity_digest, so KEYS[4] is '' and :data:SUPERSEDE_LUA skips the SET), and an unconditional SET is what makes the identity's next supersession resolve this record as the incumbent (see :meth:export_state for why dropping it resurrects records). Only the digests actually captured at export are written, so a record that owned no open claim still writes none.

Chain links are restored independently of import order: HSET does not require the counterpart record to already exist, and _walk_links-style chain traversal already treats a link to a record with no interval entry as a chain end (see :meth:on_delete's "Known limitation" note), so a partially imported chain converges once both sides have landed.

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

@classmethod
def import_state(
    cls,
    model_instance: "Model",
    field_name: str,
    state: Any,
    **kwargs: Any,
) -> None:
    """Restore this record's interval scores and chain links after import.

    Written with plain ``ZADD``/``HSET`` -- deliberately NOT through
    :data:`SUPERSEDE_LUA` and NOT with the ``NX`` guards ``on_save``
    uses. The transfer driver calls ``import_state`` *after* ``save()``,
    so ``on_save`` has already run: it seeds ``valid_from`` /
    ``ingested_at`` / ``invalid_at`` via ``ZADD NX`` in mode="open",
    i.e. a fresh, open interval with no close time and no chain. An
    ``NX`` write here would be a silent no-op against those
    already-present scores, discarding the carried close time and chain
    -- exactly the resurrection bug this field exists to prevent.
    Plain ``ZADD``/``HSET`` overwrite instead.

    Carried open-claim pointers are restored with a plain ``SET``, for
    the same reason: ``on_save`` cannot write them at all on this path
    (it calls :meth:`execute_supersede` with no ``identity_digest``, so
    ``KEYS[4]`` is ``''`` and :data:`SUPERSEDE_LUA` skips the ``SET``),
    and an unconditional ``SET`` is what makes the identity's next
    supersession resolve this record as the incumbent (see
    :meth:`export_state` for why dropping it resurrects records). Only
    the digests actually captured at export are written, so a record
    that owned no open claim still writes none.

    Chain links are restored independently of import order: ``HSET``
    does not require the counterpart record to already exist, and
    ``_walk_links``-style chain traversal already treats a link to a
    record with no interval entry as a chain end (see
    :meth:`on_delete`'s "Known limitation" note), so a partially
    imported chain converges once both sides have landed.
    """
    if not state:
        return None

    field = model_instance._meta.fields.get(field_name)
    if not isinstance(field, ValidityField):
        return None

    member_key = model_instance.db_key.redis_key
    backend = non_redis_backend(model_instance)
    if backend is not None:
        backend.field_call(
            model_instance._meta.spec,
            field_name,
            "import",
            _record_id(model_instance, member_key),
            dict(state),
        )
        return None
    keys = cls.get_all_keys(model_instance, field_name)

    valid_from = state.get("valid_from")
    if valid_from is not None:
        get_REDIS_DB().zadd(keys["valid_from"], {member_key: float(valid_from)})

    invalid_at = state.get("invalid_at")
    if invalid_at is not None:
        score = (
            Defaults.VALIDITY_OPEN_SENTINEL
            if invalid_at == cls.OPEN_SENTINEL_TOKEN
            else float(invalid_at)
        )
        get_REDIS_DB().zadd(keys["invalid_at"], {member_key: score})

    ingested_at = state.get("ingested_at")
    if ingested_at is not None:
        get_REDIS_DB().zadd(keys["ingested_at"], {member_key: float(ingested_at)})

    chain_fwd = state.get("chain_fwd")
    if chain_fwd:
        get_REDIS_DB().hset(keys["chain_fwd"], member_key, chain_fwd)

    chain_rev = state.get("chain_rev")
    if chain_rev:
        get_REDIS_DB().hset(keys["chain_rev"], member_key, chain_rev)

    for digest in state.get("open_pointers") or []:
        get_REDIS_DB().set(
            cls.get_open_pointer_key(model_instance, field_name, str(digest)),
            member_key,
        )

    return None

get_prefix_db_key(model, field_name) classmethod

Return the $ValidityF:{Model}:{field} prefix all six keys extend.

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

@classmethod
def get_prefix_db_key(cls, model: "ModelLike", field_name: str) -> DB_key:
    """Return the ``$ValidityF:{Model}:{field}`` prefix all six keys extend."""
    # ``Field.get_special_use_field_db_key`` annotates an instance but reads
    # only ``_meta``, which a model class carries too — hence ``ModelLike``
    # here and the narrowing cast at the boundary.
    return cls.get_special_use_field_db_key(cast("Model", model), field_name)

get_valid_from_key(model, field_name) classmethod

Redis key of the valid_from ZSET (member -> valid-from epoch).

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

@classmethod
def get_valid_from_key(cls, model: "ModelLike", field_name: str) -> str:
    """Redis key of the ``valid_from`` ZSET (member -> valid-from epoch)."""
    return DB_key(cls.get_prefix_db_key(model, field_name), "valid_from").redis_key

get_invalid_at_key(model, field_name) classmethod

Redis key of the invalid_at ZSET (member -> close epoch, +inf = open).

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

@classmethod
def get_invalid_at_key(cls, model: "ModelLike", field_name: str) -> str:
    """Redis key of the ``invalid_at`` ZSET (member -> close epoch, ``+inf`` = open)."""
    return DB_key(cls.get_prefix_db_key(model, field_name), "invalid_at").redis_key

get_ingested_at_key(model, field_name) classmethod

Redis key of the ingested_at ZSET (member -> transaction-time epoch).

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

@classmethod
def get_ingested_at_key(cls, model: "ModelLike", field_name: str) -> str:
    """Redis key of the ``ingested_at`` ZSET (member -> transaction-time epoch)."""
    return DB_key(cls.get_prefix_db_key(model, field_name), "ingested_at").redis_key

get_chain_fwd_key(model, field_name) classmethod

Redis key of the forward chain HASH (old redis_key -> superseding key).

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

@classmethod
def get_chain_fwd_key(cls, model: "ModelLike", field_name: str) -> str:
    """Redis key of the forward chain HASH (old redis_key -> superseding key)."""
    return DB_key(
        cls.get_prefix_db_key(model, field_name), "chain", "fwd"
    ).redis_key

get_chain_rev_key(model, field_name) classmethod

Redis key of the reverse chain HASH (new redis_key -> superseded key).

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

@classmethod
def get_chain_rev_key(cls, model: "ModelLike", field_name: str) -> str:
    """Redis key of the reverse chain HASH (new redis_key -> superseded key)."""
    return DB_key(
        cls.get_prefix_db_key(model, field_name), "chain", "rev"
    ).redis_key

get_open_pointer_key(model, field_name, identity_digest) classmethod

Redis key of the open-claim pointer STRING for one identity digest.

Parameters:

Name	Type	Description	Default
model	ModelLike	The model class (or instance) owning the field.	required
field_name	str	Name of the ValidityField on that model.	required
identity_digest	str	Opaque, already-normalized identity token — see SupersessionProtocol.identity_key (plan D7), which hashes the caller's (subject, predicate) into 16 hex characters so raw user text can never reach the keyspace.	required

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

@classmethod
def get_open_pointer_key(
    cls, model: "ModelLike", field_name: str, identity_digest: str
) -> str:
    """Redis key of the open-claim pointer STRING for one identity digest.

    Args:
        model: The model class (or instance) owning the field.
        field_name: Name of the ``ValidityField`` on that model.
        identity_digest: Opaque, already-normalized identity token — see
            ``SupersessionProtocol.identity_key`` (plan D7), which hashes the
            caller's ``(subject, predicate)`` into 16 hex characters so raw
            user text can never reach the keyspace.
    """
    return DB_key(
        cls.get_prefix_db_key(model, field_name), "open", identity_digest
    ).redis_key

get_interval_keys(model, field_name) classmethod

Return (valid_from_key, invalid_at_key) for a model/field pair.

This is the stable seam every other validity consumer builds on: the decay-Lua gate passes these two as extra KEYS, the composite-score mask ZRANGESTOREs from them, and the assembler's _resolve_excluded_keys reads them directly. Returned in the order (valid_from, invalid_at); an as-of-t member satisfies valid_from <= t AND invalid_at > t.

Parameters:

Name	Type	Description	Default
model	ModelLike	The model class (or instance) owning the field.	required
field_name	str	Name of the ValidityField on that model.	required

Returns:

Type	Description
tuple[str, str]	A 2-tuple of Redis key strings.

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

@classmethod
def get_interval_keys(
    cls, model: "ModelLike", field_name: str
) -> "tuple[str, str]":
    """Return ``(valid_from_key, invalid_at_key)`` for a model/field pair.

    This is the stable seam every other validity consumer builds on: the
    decay-Lua gate passes these two as extra ``KEYS``, the composite-score
    mask ``ZRANGESTORE``s from them, and the assembler's
    ``_resolve_excluded_keys`` reads them directly. Returned in the order
    ``(valid_from, invalid_at)``; an as-of-``t`` member satisfies
    ``valid_from <= t AND invalid_at > t``.

    Args:
        model: The model class (or instance) owning the field.
        field_name: Name of the ``ValidityField`` on that model.

    Returns:
        A 2-tuple of Redis key strings.
    """
    return (
        cls.get_valid_from_key(model, field_name),
        cls.get_invalid_at_key(model, field_name),
    )

get_all_keys(model, field_name) classmethod

Return the five non-identity keys as a name -> Redis key mapping.

Keys: valid_from, invalid_at, ingested_at, chain_fwd, chain_rev. The sixth key (the per-identity open pointer) is excluded because it is parameterized by identity digest — use :meth:get_open_pointer_key for that one.

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

@classmethod
def get_all_keys(cls, model: "ModelLike", field_name: str) -> "dict[str, str]":
    """Return the five non-identity keys as a name -> Redis key mapping.

    Keys: ``valid_from``, ``invalid_at``, ``ingested_at``, ``chain_fwd``,
    ``chain_rev``. The sixth key (the per-identity open pointer) is excluded
    because it is parameterized by identity digest — use
    :meth:`get_open_pointer_key` for that one.
    """
    return {
        "valid_from": cls.get_valid_from_key(model, field_name),
        "invalid_at": cls.get_invalid_at_key(model, field_name),
        "ingested_at": cls.get_ingested_at_key(model, field_name),
        "chain_fwd": cls.get_chain_fwd_key(model, field_name),
        "chain_rev": cls.get_chain_rev_key(model, field_name),
    }

resolve_valid_keys(model, field_name, as_of=None) classmethod

Return the record keys whose interval covers as_of.

Two read-only ZRANGEBYSCOREs intersected: valid_from <= as_of AND invalid_at > as_of.

.. warning::

This is a whitelist: a record with no entry in either ZSET is absent from the result. That makes it wrong for retrieval gating, and it is deliberately not used by any retrieval path. All three gating layers are subtractive — they exclude records that are positively known to be closed or not-yet-started, and leave unmanaged records (no interval at all) fully visible. Using this method to gate retrieval would silently hide every record that predates the field's adoption on a model. The assembler uses ContextAssembler._resolve_excluded_keys instead, and the query layer uses QueryBuilder._apply_validity_mask.

Retained as the public, deliberate-query helper for callers that genuinely want "which records positively claim validity at t" — e.g. audit and provenance tooling, not context assembly.

Parameters:

Name	Type	Description	Default
model	ModelLike	The model class (or instance) owning the field.	required
field_name	str	Name of the ValidityField on that model.	required
as_of	Optional[float]	Epoch seconds to evaluate at. None means "now".	None

Returns:

Type	Description
set[str]	set[str] of decoded Redis keys. Decoded — not bytes — because
set[str]	every consumer compares against record.db_key.redis_key, which is
set[str]	a str; the bytes form is confined to :meth:filter_query,
set[str]	where the query layer intersects raw index replies.

Note

This is a point-in-time snapshot of a live store. A supersession landing after the read is not reflected until the next call — the same accepted property tag scoping already has (plan Race 3).

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

@classmethod
def resolve_valid_keys(
    cls,
    model: "ModelLike",
    field_name: str,
    as_of: Optional[float] = None,
) -> "set[str]":
    """Return the record keys whose interval covers ``as_of``.

    Two read-only ``ZRANGEBYSCORE``s intersected:
    ``valid_from <= as_of`` AND ``invalid_at > as_of``.

    .. warning::

       This is a **whitelist**: a record with no entry in either ZSET is
       absent from the result. That makes it wrong for retrieval gating,
       and it is deliberately **not** used by any retrieval path. All three
       gating layers are *subtractive* — they exclude records that are
       positively known to be closed or not-yet-started, and leave
       unmanaged records (no interval at all) fully visible. Using this
       method to gate retrieval would silently hide every record that
       predates the field's adoption on a model. The assembler uses
       ``ContextAssembler._resolve_excluded_keys`` instead, and the query
       layer uses ``QueryBuilder._apply_validity_mask``.

    Retained as the public, deliberate-query helper for callers that
    genuinely want "which records positively claim validity at ``t``" —
    e.g. audit and provenance tooling, not context assembly.

    Args:
        model: The model class (or instance) owning the field.
        field_name: Name of the ``ValidityField`` on that model.
        as_of: Epoch seconds to evaluate at. ``None`` means "now".

    Returns:
        ``set[str]`` of decoded Redis keys. Decoded — not bytes — because
        every consumer compares against ``record.db_key.redis_key``, which is
        a ``str``; the ``bytes`` form is confined to :meth:`filter_query`,
        where the query layer intersects raw index replies.

    Note:
        This is a point-in-time snapshot of a live store. A supersession
        landing after the read is not reflected until the next call — the
        same accepted property tag scoping already has (plan Race 3).
    """
    t = time.time() if as_of is None else float(as_of)
    backend = non_redis_backend(model)
    if backend is not None:
        return _backend_members(backend, model, field_name, t, "valid")
    valid_from_key, invalid_at_key = cls.get_interval_keys(model, field_name)
    # redis-py types every command ``Awaitable[T] | T`` for both the sync and
    # async clients, so mypy cannot see that these are concrete lists on the
    # sync client we actually use. Same narrowing cast as
    # ``ContextAssembler._resolve_excluded_keys``; CLAUDE.md notes this error
    # family is redis-py-version-dependent.
    started = cast(
        "list[Any]", get_REDIS_DB().zrangebyscore(valid_from_key, "-inf", t)
    )
    still_open = cast(
        "list[Any]", get_REDIS_DB().zrangebyscore(invalid_at_key, f"({t}", "+inf")
    )
    return {_as_str(m) for m in started} & {_as_str(m) for m in still_open}

resolve_excluded_keys(model, field_name, as_of=None) classmethod

Return the record keys to DROP as of as_of (#648).

The subtractive counterpart to :meth:resolve_valid_keys, and the one every retrieval path actually wants. Two read-only ZRANGEBYSCOREs, unioned: invalid_at <= as_of (already closed) and valid_from > as_of (not yet started) — the exact mirror of the two ranges :meth:resolve_valid_keys intersects.

.. warning::

This returns an exclusion set, not a whitelist, and that is deliberate. A record with no entry in either interval ZSET is unmanaged and stays fully retrievable. Resolving the valid set instead — via :meth:resolve_valid_keys, whose own warning is the other half of this one — would silently hide every record that predates the day a ValidityField was added to an existing model, since none of those has an interval until it is next saved. That is a data-visibility regression, not a stricter gate. Do not "simplify" this into a valid-key intersection.

This method lives beside resolve_valid_keys precisely so the warning and the trap it warns about are read together. It used to live in ContextAssembler._resolve_excluded_keys, a different file from the method it forbids.

The exclusive lower bound on valid_from is spelled f"({t}", while the invalid_at upper bound is inclusive. Redis renders the +inf open sentinel such that an open record never matches <= as_of. Both bound conventions are this method's business, not a caller's.

Callers own the policy questions this does not answer: whether the model declares a ValidityField at all, and whether the Defaults.VALIDITY_GATING_ENABLED kill switch is on (which must be read at call time, never captured at import, so a deploy-level switch takes effect for adopters who cannot edit model code).

Parameters:

Name	Type	Description	Default
model	ModelLike	The model class (or instance) owning the field.	required
field_name	str	Name of the ValidityField on that model.	required
as_of	Optional[float]	Epoch seconds to evaluate membership at. None = now.	None

Returns:

Type	Description
set[str]	set[str] of decoded record keys to drop. An empty set means
set[str]	gating ran and excluded nothing — which is not the same as a
set[str]	caller's None for "gating did not run at all".

Note

A point-in-time snapshot of a live store. A supersession landing after the read is not reflected until the next call.

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

@classmethod
def resolve_excluded_keys(
    cls,
    model: "ModelLike",
    field_name: str,
    as_of: Optional[float] = None,
) -> "set[str]":
    """Return the record keys to DROP as of ``as_of`` (#648).

    The subtractive counterpart to :meth:`resolve_valid_keys`, and the one
    every retrieval path actually wants. Two read-only ``ZRANGEBYSCORE``s,
    unioned: ``invalid_at <= as_of`` (already closed) and
    ``valid_from > as_of`` (not yet started) — the exact mirror of the two
    ranges :meth:`resolve_valid_keys` intersects.

    .. warning::

       **This returns an exclusion set, not a whitelist, and that is
       deliberate.** A record with no entry in either interval ZSET is
       *unmanaged* and stays fully retrievable. Resolving the valid set
       instead — via :meth:`resolve_valid_keys`, whose own warning is the
       other half of this one — would silently hide every record that
       predates the day a ``ValidityField`` was added to an existing model,
       since none of those has an interval until it is next saved. That is a
       data-visibility regression, not a stricter gate. Do not "simplify"
       this into a valid-key intersection.

       This method lives beside ``resolve_valid_keys`` precisely so the
       warning and the trap it warns about are read together. It used to
       live in ``ContextAssembler._resolve_excluded_keys``, a different file
       from the method it forbids.

    The exclusive lower bound on ``valid_from`` is spelled ``f"({t}"``,
    while the ``invalid_at`` upper bound is inclusive. Redis renders the
    ``+inf`` open sentinel such that an open record never matches
    ``<= as_of``. Both bound conventions are this method's business, not a
    caller's.

    Callers own the *policy* questions this does not answer: whether the
    model declares a ``ValidityField`` at all, and whether the
    ``Defaults.VALIDITY_GATING_ENABLED`` kill switch is on (which must be
    read at call time, never captured at import, so a deploy-level switch
    takes effect for adopters who cannot edit model code).

    Args:
        model: The model class (or instance) owning the field.
        field_name: Name of the ``ValidityField`` on that model.
        as_of: Epoch seconds to evaluate membership at. ``None`` = now.

    Returns:
        ``set[str]`` of decoded record keys to drop. An empty set means
        gating ran and excluded nothing — which is **not** the same as a
        caller's ``None`` for "gating did not run at all".

    Note:
        A point-in-time snapshot of a live store. A supersession landing
        after the read is not reflected until the next call.
    """
    t = time.time() if as_of is None else float(as_of)
    backend = non_redis_backend(model)
    if backend is not None:
        return _backend_members(backend, model, field_name, t, "excluded")
    valid_from_key, invalid_at_key = cls.get_interval_keys(model, field_name)
    # invalid_at <= t: already closed. The +inf open sentinel never matches.
    # Read before valid_from, the order the assembler established.
    closed = cast(
        "list[Any]", get_REDIS_DB().zrangebyscore(invalid_at_key, "-inf", t)
    )
    # valid_from > t: not yet started.
    future = cast(
        "list[Any]", get_REDIS_DB().zrangebyscore(valid_from_key, f"({t}", "+inf")
    )
    return {_as_str(m) for m in closed + future}

is_valid_at(model, field_name, member_key, as_of=None) classmethod

Return whether one record's interval covers as_of.

Two ZSCOREs rather than two range reads — the single-member form of :meth:resolve_valid_keys. A member absent from either ZSET is not valid (it has no interval).

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

@classmethod
def is_valid_at(
    cls,
    model: "ModelLike",
    field_name: str,
    member_key: str,
    as_of: Optional[float] = None,
) -> bool:
    """Return whether one record's interval covers ``as_of``.

    Two ``ZSCORE``s rather than two range reads — the single-member form of
    :meth:`resolve_valid_keys`. A member absent from either ZSET is not valid
    (it has no interval).
    """
    t = time.time() if as_of is None else float(as_of)
    backend = non_redis_backend(model)
    if backend is not None:
        state = _backend_interval(backend, model, field_name, member_key) or {}
        start = state.get("valid_from")
        close = state.get("invalid_at")
    else:
        valid_from_key, invalid_at_key = cls.get_interval_keys(model, field_name)
        # redis-py types ZSCORE ``Awaitable[float | None] | float | None``
        # to cover the async client; narrow to the sync reply (see the
        # cast note in :meth:`resolve_valid_keys`).
        start = cast(
            "Optional[float]", get_REDIS_DB().zscore(valid_from_key, member_key)
        )
        close = cast(
            "Optional[float]", get_REDIS_DB().zscore(invalid_at_key, member_key)
        )
    if start is None or close is None:
        return False
    if float(close) == Defaults.VALIDITY_OPEN_SENTINEL:
        return float(start) <= t
    return float(start) <= t < float(close)

execute_supersede(model, field_name, *, new_member='', mode='open', now=None, valid_from=None, ingested_at=None, close_at=None, old_member='', identity_digest='', assert_valid_from=False, pipeline=None) classmethod

Run :data:SUPERSEDE_LUA once against this model/field's six keys.

The single seam through which every validity mutation flows — ValidityField.on_save and SupersessionProtocol both call it, so there is exactly one place that knows the script's KEYS/ARGV order.

Parameters:

Name	Type	Description	Default
model	ModelLike	The model class (or instance) owning the field.	required
field_name	str	Name of the ValidityField on that model.	required
new_member	str	Redis key of the record whose interval opens. Empty for a pure invalidate.	''
mode	str	'open' (a save: open the newcomer, close nothing), 'supersede' (close the incumbent and chain it to the newcomer), or 'invalidate' (close only).	'open'
now	Optional[float]	Epoch seconds to use as the script's clock. Defaults to time.time(). Passing one clock for a batch keeps intervals consistent across a multi-record write.	None
valid_from	Optional[float]	Valid-from epoch for new_member. Defaults to now.	None
ingested_at	Optional[float]	Transaction-time epoch for new_member. Defaults to now.	None
close_at	Optional[float]	Explicit close epoch for the incumbent. Defaults to now.	None
old_member	str	Explicit incumbent key, bypassing identity resolution.	''
identity_digest	str	Identity digest naming the open-claim pointer. When empty, KEYS[4] is passed as '' and the script neither reads nor repoints a pointer.	''
assert_valid_from	bool	When True, valid_from is a caller assertion about new_member's start rather than a default, and a disagreement with the already-stored start raises :class:ValidityValidFromConflictError instead of losing silently to the script's ZADD NX (plan D3). False — the default, and what SupersessionProtocol and ProvenanceJournal pass — preserves today's behavior for every existing caller: their at= is a close-time assertion about the incumbent, not a start-time assertion about the successor.	False
pipeline	Optional[Pipeline]	Optional external pipeline. When given, the EVAL is queued (following tag_field.py's threading shape) and the pipeline is returned; the closed-member result is only available at execute() time.	None

Returns:

Type	Description
Any	The closed member key as str, or None if nothing was closed —
Any	or the pipeline when one was supplied.

Raises:

Type	Description
ValueError	If mode is not one of :data:VALID_MODES, or if the client-side pre-check finds close_at before valid_from.
ValidityMemberAbsentError	If new_member, or an explicitly-named old_member, does not exist at the instant of the write.
ValidityCloseBeforeStartError	If close_at precedes the incumbent's stored valid_from.
ValidityValidFromConflictError	If assert_valid_from is set and valid_from disagrees with the stored start.

Note

The ResponseError -> typed-exception remap lives on the non-pipeline branch only. On a caller-supplied pipeline redis-py raises during pipe.execute() result parsing, long after this method returned, so the caller sees a raw redis.exceptions.ResponseError. Use :meth:SupersessionProtocol.save_and_supersede, which owns its execute(), to get a typed error in pipeline shape.

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

@classmethod
def execute_supersede(
    cls,
    model: "ModelLike",
    field_name: str,
    *,
    new_member: str = "",
    mode: str = "open",
    now: Optional[float] = None,
    valid_from: Optional[float] = None,
    ingested_at: Optional[float] = None,
    close_at: Optional[float] = None,
    old_member: str = "",
    identity_digest: str = "",
    assert_valid_from: bool = False,
    pipeline: Optional[redis.client.Pipeline] = None,
) -> Any:
    """Run :data:`SUPERSEDE_LUA` once against this model/field's six keys.

    The single seam through which every validity mutation flows —
    ``ValidityField.on_save`` and ``SupersessionProtocol`` both call it, so
    there is exactly one place that knows the script's KEYS/ARGV order.

    Args:
        model: The model class (or instance) owning the field.
        field_name: Name of the ``ValidityField`` on that model.
        new_member: Redis key of the record whose interval opens. Empty for
            a pure ``invalidate``.
        mode: ``'open'`` (a save: open the newcomer, close nothing),
            ``'supersede'`` (close the incumbent and chain it to the
            newcomer), or ``'invalidate'`` (close only).
        now: Epoch seconds to use as the script's clock. Defaults to
            ``time.time()``. Passing one clock for a batch keeps intervals
            consistent across a multi-record write.
        valid_from: Valid-from epoch for ``new_member``. Defaults to ``now``.
        ingested_at: Transaction-time epoch for ``new_member``. Defaults to
            ``now``.
        close_at: Explicit close epoch for the incumbent. Defaults to ``now``.
        old_member: Explicit incumbent key, bypassing identity resolution.
        identity_digest: Identity digest naming the open-claim pointer. When
            empty, ``KEYS[4]`` is passed as ``''`` and the script neither
            reads nor repoints a pointer.
        assert_valid_from: When ``True``, ``valid_from`` is a caller
            *assertion* about ``new_member``'s start rather than a default,
            and a disagreement with the already-stored start raises
            :class:`ValidityValidFromConflictError` instead of losing
            silently to the script's ``ZADD NX`` (plan D3). ``False`` — the
            default, and what ``SupersessionProtocol`` and
            ``ProvenanceJournal`` pass — preserves today's behavior for every
            existing caller: their ``at=`` is a *close-time* assertion about
            the incumbent, not a start-time assertion about the successor.
        pipeline: Optional external pipeline. When given, the EVAL is queued
            (following ``tag_field.py``'s threading shape) and the pipeline
            is returned; the closed-member result is only available at
            ``execute()`` time.

    Returns:
        The closed member key as ``str``, or ``None`` if nothing was closed —
        or the ``pipeline`` when one was supplied.

    Raises:
        ValueError: If ``mode`` is not one of :data:`VALID_MODES`, or if the
            client-side pre-check finds ``close_at`` before ``valid_from``.
        ValidityMemberAbsentError: If ``new_member``, or an explicitly-named
            ``old_member``, does not exist at the instant of the write.
        ValidityCloseBeforeStartError: If ``close_at`` precedes the
            incumbent's stored ``valid_from``.
        ValidityValidFromConflictError: If ``assert_valid_from`` is set and
            ``valid_from`` disagrees with the stored start.

    Note:
        The ``ResponseError`` -> typed-exception remap lives on the
        **non-pipeline** branch only. On a caller-supplied pipeline redis-py
        raises during ``pipe.execute()`` result parsing, long after this
        method returned, so the caller sees a raw
        ``redis.exceptions.ResponseError``. Use
        :meth:`SupersessionProtocol.save_and_supersede`, which owns its
        ``execute()``, to get a typed error in pipeline shape.
    """
    if mode not in VALID_MODES:
        raise ValueError(
            f"ValidityField mode must be one of {sorted(VALID_MODES)}, got {mode!r}"
        )
    clock = time.time() if now is None else float(now)
    if close_at is not None and valid_from is not None and mode != "open":
        # Cheap client-side pre-check for the direct-invalidation form, where
        # the caller already knows both ends. The authoritative check lives in
        # the script (the stored valid_from is the one that matters).
        if float(close_at) < float(valid_from):
            raise ValueError(
                "ValidityField: close-at "
                f"({close_at}) precedes valid_from ({valid_from})"
            )

    backend = non_redis_backend(model)
    if backend is not None:
        return cls._execute_supersede_on_backend(
            backend,
            model,
            field_name,
            new_member=new_member,
            mode=mode,
            now=clock,
            valid_from=valid_from,
            ingested_at=ingested_at,
            close_at=close_at,
            old_member=old_member,
            identity_digest=identity_digest,
            assert_valid_from=assert_valid_from,
            pipeline=pipeline,
        )

    keys = cls.get_all_keys(model, field_name)
    pointer_key = (
        cls.get_open_pointer_key(model, field_name, identity_digest)
        if identity_digest
        else ""
    )
    args = [
        keys["valid_from"],  # KEYS[1]
        keys["invalid_at"],  # KEYS[2]
        keys["ingested_at"],  # KEYS[3]
        pointer_key,  # KEYS[4] (may be '')
        keys["chain_fwd"],  # KEYS[5]
        keys["chain_rev"],  # KEYS[6]
        new_member or "",  # ARGV[1]
        repr(clock),  # ARGV[2]
        "" if valid_from is None else repr(float(valid_from)),  # ARGV[3]
        "" if ingested_at is None else repr(float(ingested_at)),  # ARGV[4]
        mode,  # ARGV[5]
        "" if close_at is None else repr(float(close_at)),  # ARGV[6]
        old_member or "",  # ARGV[7]
        "1" if assert_valid_from else "",  # ARGV[8]
    ]

    if isinstance(pipeline, redis.client.Pipeline):
        run_lua(pipeline, SUPERSEDE_LUA, 6, *args)
        return pipeline
    try:
        result = run_lua(get_REDIS_DB(), SUPERSEDE_LUA, 6, *args)
    except redis.exceptions.ResponseError as e:
        raise map_lua_error(e) from e
    closed = _as_str(result) if result else ""
    return closed or None

warn_if_ttl(model, field_name) classmethod

Warn once when a ValidityField model also declares a TTL.

A TTL truncates supersession chains and silently breaks as_of correctness: the expired record vanishes while its chain links and interval entries remain. This warns rather than raises — refusing would break adopters who legitimately want bounded history (plan D9).

Returns:

Type	Description
bool	True if a warning was emitted on this call.

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

@classmethod
def warn_if_ttl(cls, model: "ModelLike", field_name: str) -> bool:
    """Warn once when a ``ValidityField`` model also declares a TTL.

    A TTL truncates supersession chains and silently breaks ``as_of``
    correctness: the expired record vanishes while its chain links and
    interval entries remain. This warns rather than raises — refusing would
    break adopters who legitimately want bounded history (plan D9).

    Returns:
        ``True`` if a warning was emitted on this call.
    """
    meta = getattr(model, "_meta", None)
    if meta is None or getattr(meta, "ttl", None) is None:
        return False
    marker = (getattr(meta, "model_name", str(model)), field_name)
    if marker in _TTL_WARNED:
        return False
    _TTL_WARNED.add(marker)
    logger.warning(
        "%s.%s is a ValidityField on a model with Meta.ttl=%s. TTL expiry "
        "truncates supersession chains and breaks as_of reconstruction: the "
        "record disappears while its interval and chain links remain. Drop "
        "the TTL, or accept bounded history.",
        marker[0],
        field_name,
        meta.ttl,
    )
    return True

on_save(model_instance, field_name, field_value, pipeline=None, **kwargs) classmethod

Open (or re-affirm) this record's validity interval.

Routes through :data:SUPERSEDE_LUA in mode 'open' rather than issuing a bare ZADD, which is what makes the save path safe against plan Race 2: the script's ZSCORE/NX guards mean a save that interleaves with a concurrent supersession can never resurrect an already-closed record, and a re-save never shifts an existing interval's start or ingest time.

field_value is used as the valid-from epoch when it is numeric; otherwise save time is used. The write is idempotent, so repeated saves of an open record are no-ops on the index.

Parameters:

Name	Type	Description	Default
model_instance	Model	The instance being saved.	required
field_name	str	Name of this field on the model.	required
field_value	Any	The declared valid-from epoch, or None.	required
pipeline	Optional[Pipeline]	Optional external pipeline; the EVAL is queued onto it.	None
**kwargs	Any	Accepted for forward compatibility.	{}

Returns:

Type	Description
Any	The pipeline if one was provided, else the script result.

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

@classmethod
def on_save(
    cls,
    model_instance: "Model",
    field_name: str,
    field_value: Any,
    pipeline: Optional[redis.client.Pipeline] = None,
    **kwargs: Any,
) -> Any:
    """Open (or re-affirm) this record's validity interval.

    Routes through :data:`SUPERSEDE_LUA` in mode ``'open'`` rather than
    issuing a bare ``ZADD``, which is what makes the save path safe against
    plan Race 2: the script's ``ZSCORE``/``NX`` guards mean a save that
    interleaves with a concurrent supersession can never resurrect an
    already-closed record, and a re-save never shifts an existing interval's
    start or ingest time.

    ``field_value`` is used as the valid-from epoch when it is numeric;
    otherwise save time is used. The write is idempotent, so repeated saves
    of an open record are no-ops on the index.

    Args:
        model_instance: The instance being saved.
        field_name: Name of this field on the model.
        field_value: The declared valid-from epoch, or ``None``.
        pipeline: Optional external pipeline; the EVAL is queued onto it.
        **kwargs: Accepted for forward compatibility.

    Returns:
        The ``pipeline`` if one was provided, else the script result.
    """
    cls.warn_if_ttl(model_instance, field_name)
    now = time.time()
    declared = False
    try:
        if field_value is not None:
            valid_from = float(field_value)
            declared = True
        else:
            valid_from = now
    except (TypeError, ValueError):
        valid_from = now
    return cls.execute_supersede(
        model_instance,
        field_name,
        new_member=model_instance.db_key.redis_key,
        mode="open",
        now=now,
        valid_from=valid_from,
        ingested_at=now,
        # Plan D3: a declared field value IS the single authoritative writer
        # of valid-time, so a re-save that declares a different start is the
        # reporter's bug and must be loud. A defaulted save asserts nothing
        # and keeps NX idempotence.
        assert_valid_from=declared,
        pipeline=pipeline,
    )

pre_save_validate(model_instance, field_name, field_value, **kwargs) classmethod

Refuse a save that declares a valid_from the index disagrees with.

Runs from the single pre-split dispatch site in Model.save(), before any write is issued or queued — which is the point. Raising from :meth:on_save would be too late on a model that also declares IndexedFieldMixin fields: those commit their hash values and index entries eagerly, against live Redis, before ValidityField.on_save ever runs (plan D5 half 1, the same treatment #476 gave the unique-conflict window).

A cheap client-side pre-check for a good error message; the authoritative comparison is in :data:SUPERSEDE_LUA under ARGV[8], evaluated atomically (plan Race 4). A racing pre-check can only produce a false negative, which the script then catches.

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

@classmethod
def pre_save_validate(
    cls,
    model_instance: "Model",
    field_name: str,
    field_value: Any,
    **kwargs: Any,
) -> None:
    """Refuse a save that declares a ``valid_from`` the index disagrees with.

    Runs from the single pre-split dispatch site in ``Model.save()``, before
    *any* write is issued or queued — which is the point. Raising from
    :meth:`on_save` would be too late on a model that also declares
    ``IndexedFieldMixin`` fields: those commit their hash values and index
    entries eagerly, against live Redis, before ``ValidityField.on_save``
    ever runs (plan D5 half 1, the same treatment #476 gave the
    unique-conflict window).

    A cheap client-side pre-check for a good error message; the authoritative
    comparison is in :data:`SUPERSEDE_LUA` under ``ARGV[8]``, evaluated
    atomically (plan Race 4). A racing pre-check can only produce a false
    negative, which the script then catches.
    """
    if field_value is None:
        return  # defaulted: no assertion, nothing to conflict with
    try:
        declared = float(field_value)
    except (TypeError, ValueError):
        return  # on_save falls back to the save clock; not an assertion
    try:
        member_key = model_instance.db_key.redis_key
    except (TypeError, ValueError):
        return
    if not member_key:
        return
    backend = non_redis_backend(model_instance)
    if backend is not None:
        state = _backend_interval(backend, model_instance, field_name, member_key)
        stored = None if state is None else state["valid_from"]
    else:
        stored = get_REDIS_DB().zscore(
            cls.get_valid_from_key(model_instance, field_name), member_key
        )
    if stored is not None and float(stored) != declared:
        raise ValidityValidFromConflictError(
            f"ValidityField: {model_instance.__class__.__name__}.{field_name} "
            f"declares valid_from={declared!r} for {member_key}, but the "
            f"index already holds {float(stored)!r}. Valid-time has one "
            "writer -- the field value at construction. Adopt the stored "
            "value with ValidityField.get_valid_from(...), or overwrite the "
            "index with a plain ZADD (no NX), then save again."
        )

get_valid_from(model, field_name, member_key=None) classmethod

Return the record's effective valid-from — the index score.

instance.validity is the declared value: None there means "not declared, defaulted to the save clock". This returns what the valid_from index actually holds, which is what every as_of query answers against (plan D5 half 2). The two differ legitimately, and reading this is how an operator reconciles a record whose hash and index already disagree.

Parameters:

Name	Type	Description	Default
model	ModelLike	The model class, or a live instance.	required
field_name	str	Name of the ValidityField.	required
member_key	Optional[str]	The record's Redis key. Defaults to model's own key when an instance was passed.	None

Returns:

Type	Description
Optional[float]	The stored valid-from epoch, or None when the member has no
Optional[float]	entry in the index.

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

@classmethod
def get_valid_from(
    cls,
    model: "ModelLike",
    field_name: str,
    member_key: Optional[str] = None,
) -> Optional[float]:
    """Return the record's *effective* valid-from — the index score.

    ``instance.validity`` is the **declared** value: ``None`` there means
    "not declared, defaulted to the save clock". This returns what the
    ``valid_from`` index actually holds, which is what every ``as_of`` query
    answers against (plan D5 half 2). The two differ legitimately, and
    reading this is how an operator reconciles a record whose hash and index
    already disagree.

    Args:
        model: The model class, or a live instance.
        field_name: Name of the ``ValidityField``.
        member_key: The record's Redis key. Defaults to ``model``'s own key
            when an instance was passed.

    Returns:
        The stored valid-from epoch, or ``None`` when the member has no
        entry in the index.
    """
    if member_key is None:
        try:
            member_key = model.db_key.redis_key  # type: ignore[union-attr]
        except (AttributeError, TypeError, ValueError) as e:
            raise ValueError(
                "ValidityField.get_valid_from: pass member_key when the "
                "first argument is a model class rather than an instance"
            ) from e
    backend = non_redis_backend(model)
    if backend is not None:
        state = _backend_interval(backend, model, field_name, member_key)
        score = None if state is None else state["valid_from"]
    else:
        score = get_REDIS_DB().zscore(
            cls.get_valid_from_key(model, field_name), member_key
        )
    return None if score is None else float(score)

on_delete(model_instance, field_name, field_value, pipeline=None, **kwargs) classmethod

Remove every trace of this record from the validity keyspace.

ZREM from the three interval ZSETs, HDEL from both chain HASHes, and clear any open-claim pointer still aimed at the record. Records are normally closed, not deleted — this hook exists so an explicit delete() (or a key migration, which calls it with saved_redis_key) does not leave orphaned index members behind.

Pointer cleanup scans {prefix}:open:* because the pointer keyspace is keyed by identity digest, not by member — the reverse lookup does not exist by design (plan D1 fixes the key count at six). This is a delete-time-only cost on a path that is rare relative to save and read; adding a seventh per-record back-pointer key to avoid it was rejected as the worse trade.

Known limitation: the deleted key is removed from both chain HASHes as a field, but a neighbor's link may still name it as a value (fwd holds old -> deleted). Scrubbing the value side would mean an HGETALL of the whole chain on every delete. Chain traversal treats a link to a record with no interval as a chain end, which is the correct reading of a hard-deleted link.

Returns:

Type	Description
Any	The pipeline if one was provided, else the pointer-cleanup result.

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

@classmethod
def on_delete(
    cls,
    model_instance: "Model",
    field_name: str,
    field_value: Any,
    pipeline: Optional[redis.client.Pipeline] = None,
    **kwargs: Any,
) -> Any:
    """Remove every trace of this record from the validity keyspace.

    ``ZREM`` from the three interval ZSETs, ``HDEL`` from both chain HASHes,
    and clear any open-claim pointer still aimed at the record. Records are
    normally *closed*, not deleted — this hook exists so an explicit
    ``delete()`` (or a key migration, which calls it with ``saved_redis_key``)
    does not leave orphaned index members behind.

    Pointer cleanup scans ``{prefix}:open:*`` because the pointer keyspace is
    keyed by identity digest, not by member — the reverse lookup does not
    exist by design (plan D1 fixes the key count at six). This is a
    delete-time-only cost on a path that is rare relative to save and read;
    adding a seventh per-record back-pointer key to avoid it was rejected as
    the worse trade.

    Known limitation: the deleted key is removed from both chain HASHes as a
    *field*, but a neighbor's link may still name it as a *value* (``fwd``
    holds ``old -> deleted``). Scrubbing the value side would mean an
    ``HGETALL`` of the whole chain on every delete. Chain traversal treats a
    link to a record with no interval as a chain end, which is the correct
    reading of a hard-deleted link.

    Returns:
        The ``pipeline`` if one was provided, else the pointer-cleanup result.
    """
    member = kwargs.get("saved_redis_key") or model_instance.db_key.redis_key
    keys = cls.get_all_keys(model_instance, field_name)

    stale_pointers = cls.find_open_pointers_for_member(
        model_instance, field_name, member
    )

    if pipeline is not None:
        pipeline.zrem(keys["valid_from"], member)
        pipeline.zrem(keys["invalid_at"], member)
        pipeline.zrem(keys["ingested_at"], member)
        pipeline.hdel(keys["chain_fwd"], member)
        pipeline.hdel(keys["chain_rev"], member)
        for pointer_key in stale_pointers:
            pipeline.delete(pointer_key)
        return pipeline

    get_REDIS_DB().zrem(keys["valid_from"], member)
    get_REDIS_DB().zrem(keys["invalid_at"], member)
    get_REDIS_DB().zrem(keys["ingested_at"], member)
    get_REDIS_DB().hdel(keys["chain_fwd"], member)
    get_REDIS_DB().hdel(keys["chain_rev"], member)
    result: Any = 0
    for pointer_key in stale_pointers:
        result = get_REDIS_DB().delete(pointer_key)
    return result

get_filter_query_params(field_name)

Declare the two deliberate validity lookups.

{field}__as_of=t (records valid at epoch t) and {field}__current=True|False (valid now / the complement). Unioned with super()'s set per the base-class contract.

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

def get_filter_query_params(self, field_name: str) -> "set[str]":
    """Declare the two deliberate validity lookups.

    ``{field}__as_of=t`` (records valid at epoch ``t``) and
    ``{field}__current=True|False`` (valid now / the complement).
    Unioned with ``super()``'s set per the base-class contract.
    """
    return super().get_filter_query_params(field_name) | {
        f"{field_name}__as_of",
        f"{field_name}__current",
    }

filter_query(model, field_name, **query_params) classmethod

Resolve validity lookups to a set of matching Redis keys.

Two ZRANGEBYSCORE reads intersected — valid_from <= t AND invalid_at > t. __current=True evaluates at now; __current=False returns the complement (every member with an interval that does not cover now: closed, or not yet started). Multiple params AND-intersect, consistent with the rest of filter_for_keys_set.

Returns a set and never a list: the query layer turns a list return into Query._sorted_field_order, and validity must never order results (plan D2).

Parameters:

Name	Type	Description	Default
model	Model	The model class being queried.	required
field_name	str	Name of this field on the model.	required
**query_params	Any	{field}__as_of and/or {field}__current.	{}

Returns:

Type	Description
set[Any]	set of Redis keys (bytes, matching the other index fields'
set[Any]	reply type) for records matching every supplied param.

Raises:

Type	Description
ValueError	If __current is not a bool or __as_of is not a finite number.

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

@classmethod
def filter_query(
    cls,
    model: "Model",
    field_name: str,
    **query_params: Any,
) -> "set[Any]":
    """Resolve validity lookups to a ``set`` of matching Redis keys.

    Two ``ZRANGEBYSCORE`` reads intersected — ``valid_from <= t`` AND
    ``invalid_at > t``. ``__current=True`` evaluates at now;
    ``__current=False`` returns the complement (every member with an
    interval that does *not* cover now: closed, or not yet started). Multiple
    params AND-intersect, consistent with the rest of ``filter_for_keys_set``.

    Returns a ``set`` and never a ``list``: the query layer turns a list
    return into ``Query._sorted_field_order``, and validity must never order
    results (plan D2).

    Args:
        model: The model class being queried.
        field_name: Name of this field on the model.
        **query_params: ``{field}__as_of`` and/or ``{field}__current``.

    Returns:
        ``set`` of Redis keys (``bytes``, matching the other index fields'
        reply type) for records matching every supplied param.

    Raises:
        ValueError: If ``__current`` is not a bool or ``__as_of`` is not a
            finite number.
    """
    backend = non_redis_backend(model)
    if backend is not None:
        return cls._filter_query_on_backend(
            backend, model, field_name, **query_params
        )
    valid_from_key, invalid_at_key = cls.get_interval_keys(model, field_name)
    results = []

    for query_param, query_value in query_params.items():
        if query_param == f"{field_name}__current":
            if not isinstance(query_value, bool):
                raise ValueError(
                    f"{query_param} filter must be True or False, "
                    f"got {query_value!r}"
                )
            t = time.time()
            valid = cls._members_valid_at(valid_from_key, invalid_at_key, t)
            if query_value:
                results.append(valid)
            else:
                # Narrow the sync ZRANGE replies (see the cast note in
                # :meth:`resolve_valid_keys`).
                everything = set(
                    cast("list[Any]", get_REDIS_DB().zrange(invalid_at_key, 0, -1))
                ) | set(
                    cast("list[Any]", get_REDIS_DB().zrange(valid_from_key, 0, -1))
                )
                results.append(everything - valid)

        elif query_param == f"{field_name}__as_of":
            try:
                t = float(query_value)
            except (TypeError, ValueError) as e:
                raise ValueError(
                    f"{query_param} filter must be a number of epoch seconds, "
                    f"got {query_value!r}"
                ) from e
            results.append(cls._members_valid_at(valid_from_key, invalid_at_key, t))

    if not results:
        return set()
    matched = results[0]
    for other in results[1:]:
        matched &= other
    return matched

map_lua_error(e)

Return the typed exception for a :data:SUPERSEDE_LUA error reply.

Package-internal, deliberately without a leading underscore: three call sites across two packages import it (supersession, recipes.provenance_journal), which is more reach than a private name honestly describes. It stays out of popoto.__all__ -- internal to the package, not to the module.

Returns the mapped exception instance, or e itself when no token matches. It never raises: every call site is spelled raise map_lua_error(e) from e, so a helper that raised internally would leave that expression unfinished, and one that returned None would turn the call site into a TypeError.

The reply is a space-separated string whose first token is a stable POPOTO_VALIDITY_* constant; the remaining tokens are diagnostic detail for humans and are never parsed.

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

def map_lua_error(e: BaseException) -> BaseException:
    """Return the typed exception for a :data:`SUPERSEDE_LUA` error reply.

    Package-internal, deliberately without a leading underscore: three call
    sites across two packages import it (``supersession``,
    ``recipes.provenance_journal``), which is more reach than a private name
    honestly describes. It stays out of ``popoto.__all__`` -- internal to the
    package, not to the module.

    **Returns** the mapped exception instance, or ``e`` itself when no token
    matches. It never raises: every call site is spelled
    ``raise map_lua_error(e) from e``, so a helper that raised internally would
    leave that expression unfinished, and one that returned ``None`` would turn
    the call site into a ``TypeError``.

    The reply is a space-separated string whose first token is a stable
    ``POPOTO_VALIDITY_*`` constant; the remaining tokens are diagnostic detail
    for humans and are never parsed.
    """
    text = str(e)
    for token, exc_type in _LUA_ERROR_MAP:
        if token in text:
            return exc_type(_LUA_ERROR_MESSAGES[token].format(detail=text.strip()))
    return e