popoto.recipes.provenance_journal

popoto.recipes.provenance_journal

Provenance journal -- append-only entries with confirm/supersede/retract (#560).

An agent is told, in a thread, "Tom said the launch slipped to the 30th." Two turns later Tom says "the 30th is wrong, it's the 27th." Stored as ordinary memory rows, the second statement overwrites the first: the original words are gone, nobody recorded who said either one, and there is no way to ask what the agent believed last Tuesday.

This module is the substrate that fixes that. Every capture is an immutable :class:JournalEntry; every correction is a new entry pointing at the one it corrects. Membership in the live belief set comes from the validity indexes (#580), not from a chain walk, so JournalEntry.query.filter( validity__current=True) is the working set and superseded entries stay fully readable by key and by validity__as_of=<earlier>.

::

from popoto.recipes import ProvenanceJournal, JournalEntry

first = ProvenanceJournal.append(
    agent_id="agent-1",
    speaker="tom",
    turn_id="t-41",
    verbatim="the launch slipped to the 30th",
    statement="Launch date is the 30th",
    subjects=["launch"],
).entry

ProvenanceJournal.supersede(
    first,
    agent_id="agent-1",
    speaker="tom",
    turn_id="t-43",
    verbatim="actually the 30th is wrong, it's the 27th",
    statement="Launch date is the 27th",
)

JournalEntry.query.filter(validity__current=True)  # the correction only
ProvenanceJournal.annotations_for(first)           # the correction, again
ProvenanceJournal.chain(first)                     # [first, correction]

Field choices and why

entry_id (AutoKeyField) Immutable UUID identity, assigned at __init__ -- so the record's Redis key is concrete before save, which is what lets the append-only guard check the right key, and what makes two independent appends unable to collide. agent_id (KeyField) Partition key, mirroring DefaultMemory. Required non-null by :meth:ProvenanceJournal.append: a None renders the literal string "None" into the record key. captured_at (FloatField) Wall-clock capture time of the source turn. Deliberately not named ingested_at. ValidityField.on_save hardcodes the save clock into its own ingested_at ZSET and ignores any model field, so two fields under one name would silently disagree and downstream readers would get different answers depending on which they read. The validity ingest axis is always the save clock; this field is the source turn's clock. turn_id (IndexedField) "Everything from turn T" in one query. speaker (IndexedField) "Everything attributed to S" in one query. Attribution, not authentication. verbatim (StringField) The exact source span. Privacy-sensitive -- the reason NeverRecordMixin is mandatory on this model rather than optional. statement (StringField) The atomic natural-language claim distilled from the span. subjects (TagField) Multi-value: one entry can concern several people or topics. Convention over schema and explicitly not a security boundary, inheriting TagField's framing verbatim. stated (BooleanField) Stated (True) vs inferred (False). Downstream conflict resolution uses it for precedence; this module only stores it. kind (IndexedField) assert / confirm / supersede / retract, validated against :attr:Defaults.JOURNAL_KINDS. Indexed so "every retraction" is one query. target (IndexedField) The annotated entry's Redis key. A plain indexed scalar rather than a Relationship: the target is already addressed by Redis key, and Relationship's lazy-load machinery plus its heavier save buys nothing here. Annotations are 1:1 with a target by design -- a correction spanning several claims is N annotation entries. validity (ValidityField) The valid_from / invalid_at / ingested_at axes. This module owns the valid-time axis; the interval and chain state itself stays owned by ValidityField as derived index state.

One record type, not two

Annotations are JournalEntry rows distinguished by kind + target, not a separate model. Annotations must themselves be annotatable (a retraction of a mistaken retraction; a confirmation of a supersession), the validity chain hashes are keyed by Redis key and are type-agnostic, and a single keyspace keeps the live-membership query one filter() instead of a union of two. The cost is honest and paid for explicitly: kind/target consistency is runtime validation (in :meth:JournalEntry.pre_save and again in the pre-flight), not a type-system guarantee.

Querying, with one sharp edge

filter(validity__current=True) and filter(validity__as_of=t) are the supported validity lookups. filter(validity=t) -- a bare exact-value filter on the field -- silently returns nothing: ValidityField.filter_query handles only the two suffixed params. target, kind, speaker and turn_id collide with nothing (the reserved field names are only limit, order_by and values).

Atomicity, stated precisely

supersede/retract queue the annotation's save and the target's interval close into one MULTI/EXEC. The property that holds is: no interleaving reader observes the annotation without the close. The property that does not hold, and is not claimed, is rollback -- Redis MULTI/EXEC does not roll back sibling commands when one command errors at execute time. A command-level error inside EXEC can therefore leave the annotation appended with the target still open. The pre-flight below exists to make that window unreachable in practice; the residual is a documented boundary, not an impossibility claim.

Immutability, stated precisely

It is an ORM-layer contract, enforced by AppendOnlyMixin against every Python write path in models/base.py. It does not hold against a raw Redis client. Two TOCTOU shapes are known and documented rather than closed: two processes saving the same key concurrently, and two saves of the same key queued onto one pipeline (the guard's EXISTS cannot see a queued-but- unexecuted command). See :mod:popoto.fields.append_only.

What blocked content leaves behind

"Nothing is destroyed" is scoped to what gets stored. A capture blocked by the never-record firewall is dropped before storage and leaves only a content-free tombstone in the $NR: keyspace, which is not part of the journal and is returned by no journal query. The journal-side signal is a gap in turn_id coverage, nothing more; no placeholder entry is written.

Deliberately not included

EmbeddingField It requires an embedding provider (an API key or a local Ollama), so it cannot be a zero-configuration default, and similarity search over the journal is not this module's job.

Do not subclass :class:JournalEntry

Popoto's ModelBase metaclass does not inherit Field attributes from a base model class: class SubEntry(JournalEntry): pass produces a model with an empty field set, so every record it writes persists nothing. This is an ORM-level limitation, not a journal one, and it is why this module ships no subclass extension seam.

Two consequences, both load-bearing:

• To extend the annotation vocabulary, call :meth:JournalEntry.register_kind -- it extends the vocabulary in place and records whether the new kind carries a target and whether it closes its target's validity interval.
• To get a different field set (an EmbeddingField, say) or a separate keyspace, declare your own Model with the same mixins and the same field set and point a :class:ProvenanceJournal subclass at it. Copying the field set is required, not stylistic. :meth:ProvenanceJournal._write refuses an entry_model that is missing them, so the failure is loud rather than a keyspace full of empty records.

All numeric behavior is left to the field defaults and to popoto.fields.constants.Defaults; this module pins no constants of its own beyond the stream bound noted on the class.

AnnotationResult dataclass

Outcome of one journal write, readable without touching Redis.

This type exists specifically so the validity-coupling kill switch cannot reproduce the silent-no-op shape recorded in #588: a caller must be able to tell "the target was closed" from "the target was not closed" without issuing a read.

Attributes:

Name	Type	Description
entry	JournalEntry	The appended :class:JournalEntry.
target_closed	Optional[bool]	Whether the target's validity interval was closed by this call. Journal-owned pipeline (no pipeline= argument): a real bool, read from the supersede script's own reply. False on a capture or a confirm (neither changes membership), when the coupling switch is off, and when the target was already closed by a concurrent annotation -- which is the honest answer, since this call closed nothing. Caller-supplied pipeline: None, meaning unknown until you execute. Nothing has run, so no truthful bool exists. Reporting True for a queued close would reproduce exactly the #588 silent-no-op shape this type exists to prevent -- a re-close of an already-closed target is queued the same way and applies nothing. Read results[close_index] after your own execute() for the real outcome.
coupling_enabled	bool	The state of :attr:Defaults.JOURNAL_VALIDITY_COUPLING_ENABLED at write time.
pipeline	Optional[Any]	The caller-supplied pipeline, returned unexecuted, or None when the journal owned and executed its own.
close_index	Optional[int]	Index of the queued interval-close command in the caller-supplied pipeline, so pipeline.execute()[close_index] gives the same answer target_closed carries on the journal-owned path (the closed member's key, or '' when the target was already closed). None whenever no close was queued onto a caller pipeline -- always so on the journal-owned path, where the pipeline has already executed.

Source code in src/popoto/recipes/provenance_journal.py

@dataclass(frozen=True)
class AnnotationResult:
    """Outcome of one journal write, readable without touching Redis.

    This type exists specifically so the validity-coupling kill switch cannot
    reproduce the silent-no-op shape recorded in #588: a caller must be able to
    tell "the target was closed" from "the target was not closed" without
    issuing a read.

    Attributes:
        entry: The appended :class:`JournalEntry`.
        target_closed: Whether the target's validity interval was closed by
            this call.

            * **Journal-owned pipeline** (no ``pipeline=`` argument): a real
              ``bool``, read from the supersede script's own reply. ``False``
              on a capture or a ``confirm`` (neither changes membership), when
              the coupling switch is off, and when the target was *already*
              closed by a concurrent annotation -- which is the honest answer,
              since this call closed nothing.
            * **Caller-supplied pipeline**: ``None``, meaning *unknown until
              you execute*. Nothing has run, so no truthful ``bool`` exists.
              Reporting ``True`` for a queued close would reproduce exactly the
              #588 silent-no-op shape this type exists to prevent -- a re-close
              of an already-closed target is queued the same way and applies
              nothing. Read ``results[close_index]`` after your own
              ``execute()`` for the real outcome.
        coupling_enabled: The state of
            :attr:`Defaults.JOURNAL_VALIDITY_COUPLING_ENABLED` at write time.
        pipeline: The caller-supplied pipeline, returned unexecuted, or
            ``None`` when the journal owned and executed its own.
        close_index: Index of the queued interval-close command in the
            caller-supplied pipeline, so ``pipeline.execute()[close_index]``
            gives the same answer ``target_closed`` carries on the
            journal-owned path (the closed member's key, or ``''`` when the
            target was already closed). ``None`` whenever no close was queued
            onto a caller pipeline -- always so on the journal-owned path,
            where the pipeline has already executed.
    """

    entry: "JournalEntry"
    target_closed: Optional[bool]
    coupling_enabled: bool
    pipeline: Optional[Any] = None
    close_index: Optional[int] = None

JournalEntry

Bases: AppendOnlyMixin, NeverRecordMixin, EventStreamMixin, Model

One immutable, fully attributed provenance record.

Both a capture and an annotation: kind="assert" with no target is an original claim; confirm/supersede/retract with a target annotate another entry. See the module docstring for the field rationale.

Write through :class:ProvenanceJournal rather than constructing entries directly -- the façade owns the pre-flight validation and the one-transaction annotate-and-close sequence, and is the documented seam a future refactor stays behind.

Example::

JournalEntry.query.filter(turn_id="t-41")
JournalEntry.query.filter(kind="retract", validity__current=True)

Source code in src/popoto/recipes/provenance_journal.py

class JournalEntry(AppendOnlyMixin, NeverRecordMixin, EventStreamMixin, Model):
    """One immutable, fully attributed provenance record.

    Both a capture and an annotation: ``kind="assert"`` with no ``target`` is
    an original claim; ``confirm``/``supersede``/``retract`` with a ``target``
    annotate another entry. See the module docstring for the field rationale.

    Write through :class:`ProvenanceJournal` rather than constructing entries
    directly -- the façade owns the pre-flight validation and the
    one-transaction annotate-and-close sequence, and is the documented seam a
    future refactor stays behind.

    Example::

        JournalEntry.query.filter(turn_id="t-41")
        JournalEntry.query.filter(kind="retract", validity__current=True)
    """

    entry_id = AutoKeyField()
    agent_id = KeyField()
    captured_at = FloatField(null=True)
    turn_id = IndexedField(type=str, null=True)
    speaker = IndexedField(type=str, null=True)
    verbatim = StringField(default="")
    statement = StringField(default="")
    #: Structured bookkeeping a *recipe* writes about its own decision --
    #: never anything a human or a model uttered. Machine-generated, so it is
    #: exempt from the never-record scan (see
    #: :attr:`_MACHINE_GENERATED_FIELDS`). Keep human-authored content in
    #: ``statement``/``verbatim``, which are scanned.
    payload = StringField(default="")
    subjects = TagField(null=True)
    stated = BooleanField(default=True)
    kind = IndexedField(type=str, null=True)
    target = IndexedField(type=str, null=True)
    #: The claim's type, from ``reconciliation.CLAIM_TYPES``. Assigned by
    #: capture **before the entry's first and only** ``save()``, which is what
    #: makes it legal on an append-only record; M5 reads it and never writes
    #: it, so there is no code path that could back-fill it on an old entry.
    #: ``None`` on every entry captured before #564 shipped, which the
    #: reconciler tolerates by falling back to the rule-free ``note`` type
    #: rather than skipping the entry.
    claim_type = IndexedField(type=str, null=True)
    validity = ValidityField()

    _stream_name = "journal"
    # Deliberately unpartitioned. ``StreamConsumer`` takes exactly one
    # ``stream_key`` and has no partition-discovery mechanism, so a stream
    # partitioned by ``agent_id`` would be a channel its own named consumer
    # cannot read. ``agent_id`` rides in the metadata instead, so a consumer
    # can filter without hydrating the record.
    _stream_partition_field = None
    _stream_metadata_fields = ("agent_id", "kind", "target")
    # Pinned, not configurable. 10k approximate entries is the mixin's own
    # default and is a notification channel bound, not a retention policy --
    # the journal itself is the durable record, and a consumer that falls
    # more than 10k mutations behind must re-read the journal rather than
    # replay the stream. Routing this through ``Defaults`` would add a
    # sync-test exemption and buy nothing, since ``EventStreamMixin`` already
    # exposes it as a per-model class attribute.
    _stream_max_length = 10000

    @classmethod
    def register_kind(
        cls, name: str, *, targetless: bool = False, closing: bool = False
    ) -> None:
        """Extend the annotation vocabulary in place.

        The extension seam for downstream modules that need kinds the core
        four do not cover -- a merge/equivalence kind, a queue-able kind, an
        exposure kind. Registration is process-global and additive: it never
        removes or reclassifies a core kind, so the vocabulary can only grow.

        This is a registration call rather than a model subclass on purpose.
        Popoto's ``ModelBase`` metaclass does not inherit ``Field`` attributes,
        so a ``JournalEntry`` subclass has an empty field set and persists
        nothing -- see the module docstring.

        Registering also records the kind's *behaviour*, which a plain
        vocabulary list cannot: without ``targetless``/``closing`` a new kind
        would be permanently target-required and permanently inert for
        membership::

            JournalEntry.register_kind("consolidate", closing=True)
            ProvenanceJournal.append(
                agent_id="a1", kind="consolidate", target=old,
                statement="folded into the canonical claim",
            )  # closes old's interval, exactly like a supersede

        The example says ``consolidate``, not ``merge``, because the registry
        below is process-global and kind names are reserved by whichever module
        registers them first: ``recipes/reconciliation.py`` claims both
        ``merge`` and ``disjoin`` at its own import time, with
        ``closing=False``. Written as ``register_kind("merge", closing=True)``
        this example raises ``ValueError`` in any process that has imported
        that module. Pick a name your own module owns.

        The reader rule this seam is built around is unchanged: an entry whose
        ``kind`` a reader does not recognize is **inert for membership** --
        never silently treated as a ``supersede`` or a ``retract``.

        .. important::

           The registry is process-global and is **not persisted**. *Reading*
           an entry with an unregistered kind is fine -- it reads back with its
           stored ``kind`` and is inert for membership, per the reader rule.
           But *writing* one is not: ``transfer/import_.py`` restores records
           through ``instance.save()``, which runs ``pre_save`` validation, so
           importing entries that use a registered kind into a process that has
           not re-registered it fails with ``ValueError`` and classifies those
           records ``ERRORED``. **A restoring or importing process must call
           the same** ``register_kind`` **calls before importing.** (Found in
           PR #589 review.)

        Args:
            name: The new kind. Must be a non-empty string outside
                :attr:`Defaults.JOURNAL_KINDS`, which is frozen because
                changing it would reclassify already-stored entries.
            targetless: True for an original-capture kind that carries no
                ``target``, like ``assert``.
            closing: True if an entry of this kind closes its target's
                validity interval, like ``supersede`` and ``retract``.

        Raises:
            ValueError: On an empty name, a core kind, a ``targetless`` and
                ``closing`` combination (a kind with no target has nothing to
                close), or a re-registration under different flags. Registering
                the same name with the same flags again is a no-op.
        """
        if not isinstance(name, str) or not name.strip():
            raise ValueError(f"kind name must be a non-empty string, got {name!r}")
        if name in tuple(Defaults.JOURNAL_KINDS):
            raise ValueError(
                f"{name!r} is a core kind; Defaults.JOURNAL_KINDS is frozen "
                f"because changing it reclassifies already-stored entries"
            )
        if targetless and closing:
            raise ValueError(
                f"{name!r}: a targetless kind carries no target, so it has "
                f"nothing to close -- targetless and closing are exclusive"
            )
        flags = (bool(targetless), bool(closing))
        existing = _REGISTERED_KINDS.get(name)
        if existing is not None and existing != flags:
            raise ValueError(
                f"{name!r} is already registered as "
                f"targetless={existing[0]}, closing={existing[1]}; "
                f"re-registering it as targetless={flags[0]}, "
                f"closing={flags[1]} would reclassify stored entries"
            )
        _REGISTERED_KINDS[name] = flags

    @classmethod
    def journal_kinds(cls) -> "tuple[str, ...]":
        """Return the annotation vocabulary: the core four plus registrations.

        :attr:`Defaults.JOURNAL_KINDS` is read at call time (so a runtime
        assignment to ``Defaults`` is honored rather than frozen at import),
        followed by every :meth:`register_kind` name in registration order.
        """
        return tuple(Defaults.JOURNAL_KINDS) + tuple(_REGISTERED_KINDS)

    @classmethod
    def kind_is_targetless(cls, kind: Any) -> bool:
        """True if ``kind`` is an original capture that carries no ``target``."""
        if kind in _CORE_TARGETLESS_KINDS:
            return True
        registered = _REGISTERED_KINDS.get(kind) if isinstance(kind, str) else None
        return bool(registered and registered[0])

    @classmethod
    def kind_is_closing(cls, kind: Any) -> bool:
        """True if an entry of ``kind`` closes its target's validity interval."""
        if kind in _CORE_CLOSING_KINDS:
            return True
        registered = _REGISTERED_KINDS.get(kind) if isinstance(kind, str) else None
        return bool(registered and registered[1])

    #: Non-key fields whose value Popoto itself generates, never a caller or a
    #: speaker. Excluded from the never-record scan surface -- see
    #: :meth:`_never_record_scan_values`.
    _MACHINE_GENERATED_FIELDS = frozenset({"target", "payload"})

    def _never_record_scan_values(self) -> Iterator[str]:
        """Yield the content values the firewall scans, minus machine pointers.

        ``target`` holds a Redis key **Popoto generated** -- ``JournalEntry:
        <agent_id>:<uuid4 hex>``. It is an internal pointer, not user content,
        and scanning it for payment cards is a category error with a measured
        cost: a uuid4 hex sometimes contains a 13-19 digit run that passes the
        Luhn checksum, so the firewall blocks the annotation as
        ``reason='payment_card', detector='luhn'``. Verified example::

            scan_never_record(
                "JournalEntry:agent/-under/-test:8ef1fc6db384458286216656bfb2cf04"
            )
            # -> NeverRecordVerdict(blocked=True, reason='payment_card',
            #                       detector='luhn')

        Measured at 5-8 blocked per 2000 random target keys (~0.25-0.4%), which
        is what made ``test_hard_delete_removes_the_record_and_every_derived_
        trace`` fail roughly 1 run in 150 on CI. The block landed at
        ``Model.save()`` step 0, which *returns* rather than raises, so the
        annotation was silently dropped while the invalidate EVAL still closed
        the target -- a membership change with zero provenance. (PR #589.)

        ``payload`` is exempt for the same reason and was added by the same
        failure. It holds a recipe's structured record *of its own decision* --
        M5's merge log is JSON whose values are class ids, disjunction ids and
        entry keys, three or more uuid4 hexes per write. That put the #589 bug
        back on a second door: measured at ~0.23% per hex and **~0.66% per
        merge-log write**, which is what made
        ``test_a_precedence_tie_inside_a_slot_becomes_an_explicit_disjunction``
        fail on CI's Valkey job while the identical commit passed on Redis.
        This time the block *raised* (the façade's ``_scan_or_block``) instead
        of dropping silently, so it surfaced as a red check rather than as
        missing provenance -- but the category error is the same one.

        The rule these two share, and the one to apply to a third field:
        **exempt a field when Popoto itself generates every byte of it.** A
        field that can carry what a human or a model said stays scanned, no
        matter how structured it looks.

        Every genuine content field is still scanned: ``verbatim``,
        ``statement``, ``speaker``, ``turn_id``, ``kind``, and -- via the
        façade's pre-flight, which also covers the ``subjects`` list the mixin
        cannot see -- ``agent_id`` and ``subjects``. This narrows the surface to
        machine-generated values only; it does not weaken the firewall for
        anything a human or a model ever wrote.
        """
        skip = self._MACHINE_GENERATED_FIELDS
        for field_name in self._never_record_scan_field_names():
            if field_name in skip:
                continue
            value = getattr(self, field_name, None)
            if isinstance(value, str) and value.strip():
                yield value

    def pre_save(self, *args: Any, **kwargs: Any) -> Any:
        """Validate ``kind`` and the ``kind``/``target`` combination, then save.

        Defence in depth behind :class:`ProvenanceJournal`'s pre-flight: the
        façade validates before it issues or queues anything, and this catches
        an entry constructed directly.

        Raises:
            ValueError: If ``kind`` is outside the model's vocabulary, if an
                ``assert`` entry carries a ``target``, or if an annotation
                entry carries none.
        """
        validate_kind_and_target(type(self), self.kind, self.target)
        return super().pre_save(*args, **kwargs)

register_kind(name, *, targetless=False, closing=False) classmethod

Extend the annotation vocabulary in place.

The extension seam for downstream modules that need kinds the core four do not cover -- a merge/equivalence kind, a queue-able kind, an exposure kind. Registration is process-global and additive: it never removes or reclassifies a core kind, so the vocabulary can only grow.

This is a registration call rather than a model subclass on purpose. Popoto's ModelBase metaclass does not inherit Field attributes, so a JournalEntry subclass has an empty field set and persists nothing -- see the module docstring.

Registering also records the kind's behaviour, which a plain vocabulary list cannot: without targetless/closing a new kind would be permanently target-required and permanently inert for membership::

JournalEntry.register_kind("consolidate", closing=True)
ProvenanceJournal.append(
    agent_id="a1", kind="consolidate", target=old,
    statement="folded into the canonical claim",
)  # closes old's interval, exactly like a supersede

The example says consolidate, not merge, because the registry below is process-global and kind names are reserved by whichever module registers them first: recipes/reconciliation.py claims both merge and disjoin at its own import time, with closing=False. Written as register_kind("merge", closing=True) this example raises ValueError in any process that has imported that module. Pick a name your own module owns.

The reader rule this seam is built around is unchanged: an entry whose kind a reader does not recognize is inert for membership -- never silently treated as a supersede or a retract.

.. important::

The registry is process-global and is not persisted. Reading an entry with an unregistered kind is fine -- it reads back with its stored kind and is inert for membership, per the reader rule. But writing one is not: transfer/import_.py restores records through instance.save(), which runs pre_save validation, so importing entries that use a registered kind into a process that has not re-registered it fails with ValueError and classifies those records ERRORED. A restoring or importing process must call the same register_kind calls before importing. (Found in PR #589 review.)

Parameters:

Name	Type	Description	Default
name	str	The new kind. Must be a non-empty string outside :attr:Defaults.JOURNAL_KINDS, which is frozen because changing it would reclassify already-stored entries.	required
targetless	bool	True for an original-capture kind that carries no target, like assert.	False
closing	bool	True if an entry of this kind closes its target's validity interval, like supersede and retract.	False

Raises:

Type	Description
ValueError	On an empty name, a core kind, a targetless and closing combination (a kind with no target has nothing to close), or a re-registration under different flags. Registering the same name with the same flags again is a no-op.

Source code in src/popoto/recipes/provenance_journal.py

@classmethod
def register_kind(
    cls, name: str, *, targetless: bool = False, closing: bool = False
) -> None:
    """Extend the annotation vocabulary in place.

    The extension seam for downstream modules that need kinds the core
    four do not cover -- a merge/equivalence kind, a queue-able kind, an
    exposure kind. Registration is process-global and additive: it never
    removes or reclassifies a core kind, so the vocabulary can only grow.

    This is a registration call rather than a model subclass on purpose.
    Popoto's ``ModelBase`` metaclass does not inherit ``Field`` attributes,
    so a ``JournalEntry`` subclass has an empty field set and persists
    nothing -- see the module docstring.

    Registering also records the kind's *behaviour*, which a plain
    vocabulary list cannot: without ``targetless``/``closing`` a new kind
    would be permanently target-required and permanently inert for
    membership::

        JournalEntry.register_kind("consolidate", closing=True)
        ProvenanceJournal.append(
            agent_id="a1", kind="consolidate", target=old,
            statement="folded into the canonical claim",
        )  # closes old's interval, exactly like a supersede

    The example says ``consolidate``, not ``merge``, because the registry
    below is process-global and kind names are reserved by whichever module
    registers them first: ``recipes/reconciliation.py`` claims both
    ``merge`` and ``disjoin`` at its own import time, with
    ``closing=False``. Written as ``register_kind("merge", closing=True)``
    this example raises ``ValueError`` in any process that has imported
    that module. Pick a name your own module owns.

    The reader rule this seam is built around is unchanged: an entry whose
    ``kind`` a reader does not recognize is **inert for membership** --
    never silently treated as a ``supersede`` or a ``retract``.

    .. important::

       The registry is process-global and is **not persisted**. *Reading*
       an entry with an unregistered kind is fine -- it reads back with its
       stored ``kind`` and is inert for membership, per the reader rule.
       But *writing* one is not: ``transfer/import_.py`` restores records
       through ``instance.save()``, which runs ``pre_save`` validation, so
       importing entries that use a registered kind into a process that has
       not re-registered it fails with ``ValueError`` and classifies those
       records ``ERRORED``. **A restoring or importing process must call
       the same** ``register_kind`` **calls before importing.** (Found in
       PR #589 review.)

    Args:
        name: The new kind. Must be a non-empty string outside
            :attr:`Defaults.JOURNAL_KINDS`, which is frozen because
            changing it would reclassify already-stored entries.
        targetless: True for an original-capture kind that carries no
            ``target``, like ``assert``.
        closing: True if an entry of this kind closes its target's
            validity interval, like ``supersede`` and ``retract``.

    Raises:
        ValueError: On an empty name, a core kind, a ``targetless`` and
            ``closing`` combination (a kind with no target has nothing to
            close), or a re-registration under different flags. Registering
            the same name with the same flags again is a no-op.
    """
    if not isinstance(name, str) or not name.strip():
        raise ValueError(f"kind name must be a non-empty string, got {name!r}")
    if name in tuple(Defaults.JOURNAL_KINDS):
        raise ValueError(
            f"{name!r} is a core kind; Defaults.JOURNAL_KINDS is frozen "
            f"because changing it reclassifies already-stored entries"
        )
    if targetless and closing:
        raise ValueError(
            f"{name!r}: a targetless kind carries no target, so it has "
            f"nothing to close -- targetless and closing are exclusive"
        )
    flags = (bool(targetless), bool(closing))
    existing = _REGISTERED_KINDS.get(name)
    if existing is not None and existing != flags:
        raise ValueError(
            f"{name!r} is already registered as "
            f"targetless={existing[0]}, closing={existing[1]}; "
            f"re-registering it as targetless={flags[0]}, "
            f"closing={flags[1]} would reclassify stored entries"
        )
    _REGISTERED_KINDS[name] = flags

journal_kinds() classmethod

Return the annotation vocabulary: the core four plus registrations.

:attr:Defaults.JOURNAL_KINDS is read at call time (so a runtime assignment to Defaults is honored rather than frozen at import), followed by every :meth:register_kind name in registration order.

Source code in src/popoto/recipes/provenance_journal.py

@classmethod
def journal_kinds(cls) -> "tuple[str, ...]":
    """Return the annotation vocabulary: the core four plus registrations.

    :attr:`Defaults.JOURNAL_KINDS` is read at call time (so a runtime
    assignment to ``Defaults`` is honored rather than frozen at import),
    followed by every :meth:`register_kind` name in registration order.
    """
    return tuple(Defaults.JOURNAL_KINDS) + tuple(_REGISTERED_KINDS)

kind_is_targetless(kind) classmethod

True if kind is an original capture that carries no target.

Source code in src/popoto/recipes/provenance_journal.py

@classmethod
def kind_is_targetless(cls, kind: Any) -> bool:
    """True if ``kind`` is an original capture that carries no ``target``."""
    if kind in _CORE_TARGETLESS_KINDS:
        return True
    registered = _REGISTERED_KINDS.get(kind) if isinstance(kind, str) else None
    return bool(registered and registered[0])

kind_is_closing(kind) classmethod

True if an entry of kind closes its target's validity interval.

Source code in src/popoto/recipes/provenance_journal.py

@classmethod
def kind_is_closing(cls, kind: Any) -> bool:
    """True if an entry of ``kind`` closes its target's validity interval."""
    if kind in _CORE_CLOSING_KINDS:
        return True
    registered = _REGISTERED_KINDS.get(kind) if isinstance(kind, str) else None
    return bool(registered and registered[1])

pre_save(*args, **kwargs)

Validate kind and the kind/target combination, then save.

Defence in depth behind :class:ProvenanceJournal's pre-flight: the façade validates before it issues or queues anything, and this catches an entry constructed directly.

Raises:

Type	Description
ValueError	If kind is outside the model's vocabulary, if an assert entry carries a target, or if an annotation entry carries none.

Source code in src/popoto/recipes/provenance_journal.py

def pre_save(self, *args: Any, **kwargs: Any) -> Any:
    """Validate ``kind`` and the ``kind``/``target`` combination, then save.

    Defence in depth behind :class:`ProvenanceJournal`'s pre-flight: the
    façade validates before it issues or queues anything, and this catches
    an entry constructed directly.

    Raises:
        ValueError: If ``kind`` is outside the model's vocabulary, if an
            ``assert`` entry carries a ``target``, or if an annotation
            entry carries none.
    """
    validate_kind_and_target(type(self), self.kind, self.target)
    return super().pre_save(*args, **kwargs)

ProvenanceJournal

Stateless façade over :class:JournalEntry -- the only supported API.

Every method is a classmethod; there is no instance state.

To extend the annotation vocabulary, call :meth:JournalEntry.register_kind -- not a model subclass. Popoto's ModelBase metaclass does not inherit Field attributes, so a JournalEntry subclass has an empty field set and persists nothing. :meth:_write refuses an entry_model missing the journal field set for exactly that reason; a separate keyspace means declaring your own Model with the same mixins and the same fields.

Treating this as the sole read and write API is a stated contract, not a style preference: the single-record-type decision is cheaply reversible only while the keyspace is empty (validity and chain keys are namespaced per model, and every record's Redis key embeds its class name), so a future split has to stay behind this façade.

Source code in src/popoto/recipes/provenance_journal.py

class ProvenanceJournal:
    """Stateless façade over :class:`JournalEntry` -- the only supported API.

    Every method is a classmethod; there is no instance state.

    To extend the annotation vocabulary, call
    :meth:`JournalEntry.register_kind` -- **not** a model subclass. Popoto's
    ``ModelBase`` metaclass does not inherit ``Field`` attributes, so a
    ``JournalEntry`` subclass has an empty field set and persists nothing.
    :meth:`_write` refuses an ``entry_model`` missing the journal field set for
    exactly that reason; a separate keyspace means declaring your own ``Model``
    with the same mixins and the same fields.

    Treating this as the sole read and write API is a stated contract, not a
    style preference: the single-record-type decision is cheaply reversible
    only while the keyspace is empty (validity and chain keys are namespaced
    per model, and every record's Redis key embeds its class name), so a future
    split has to stay behind this façade.
    """

    #: The entry model this façade reads and writes.
    entry_model: Type[JournalEntry] = JournalEntry

    # ------------------------------------------------------------------
    # Writes
    # ------------------------------------------------------------------

    @classmethod
    def append(
        cls,
        *,
        agent_id: str,
        statement: str = "",
        verbatim: str = "",
        payload: str = "",
        speaker: Optional[str] = None,
        turn_id: Optional[str] = None,
        subjects: Optional[Sequence[str]] = None,
        stated: bool = True,
        captured_at: Optional[float] = None,
        at: Optional[float] = None,
        kind: str = "assert",
        target: Optional[Union[str, "JournalEntry"]] = None,
        claim_type: Optional[str] = None,
        pipeline: Optional["Pipeline"] = None,
    ) -> AnnotationResult:
        """Append one capture to the journal.

        Args:
            agent_id: Partition key. Required and non-null -- a ``None`` would
                render the literal ``"None"`` into the record's Redis key.
            payload: Structured bookkeeping a recipe writes about its own
                decision. **Exempt from the never-record firewall**, so pass
                only values Popoto itself generated -- ids, keys, counters. It
                does not satisfy the "a record needs content" check below:
                a targetless entry still needs a ``statement`` or a
                ``verbatim``. See
                :meth:`JournalEntry._never_record_scan_values`.
            statement: The atomic claim. Required unless ``verbatim`` is given.
            verbatim: The exact source span.
            speaker: Who said it. Attribution, not authentication.
            turn_id: Conversation turn the capture came from.
            subjects: Subject tags. ``None`` or ``[]`` puts the entry in the
                untagged pool, mirroring ``TagField``'s zero-tag semantics.
            stated: True when stated outright, False when inferred.
            captured_at: Wall-clock time of the source turn. Defaults to now.
            at: Valid-from instant for the entry. Defaults to now. Set at
                construction rather than passed to the supersede script -- see
                the ``supersede`` implementation note on #588.
            kind: Normally left at ``"assert"``. A kind added with
                :meth:`JournalEntry.register_kind` is reachable here, with the
                same ``kind``/``target`` consistency rules -- and closes the
                target's interval if it was registered ``closing=True``.
            target: Only for a non-``assert`` kind. A
                :class:`JournalEntry` or its Redis key.
            claim_type: The claim's type, from
                ``recipes.reconciliation.CLAIM_TYPES``. Set here, at capture,
                or never: the entry is append-only, so there is no later write
                that could back-fill it. Left ``None`` the reconciler treats
                the claim as the rule-free ``note`` type.
            pipeline: Optional caller pipeline. Must be transactional. When
                supplied it is returned unexecuted on
                :attr:`AnnotationResult.pipeline`. Open it with
                :func:`popoto.batch`; a pipeline built directly off the Redis
                client is still accepted (and is the same object type), but
                that spelling is deprecated -- ``batch()`` is the seam that
                survives #630's move of the client behind the field layer.

        Returns:
            AnnotationResult: with ``target_closed=False`` -- a capture never
            changes another entry's membership. ``None`` instead when the
            caller supplied the pipeline (nothing has executed yet).

        Raises:
            JournalBlockedError: If any content is refused by the never-record
                firewall. Nothing is issued or queued.
            ValueError: On a missing ``agent_id``, empty content, a bad
                ``kind``/``target`` pairing, a nonexistent or cross-agent
                target, a backdated ``at``, or a non-transactional pipeline.
            TypeError: If ``entry_model`` does not declare the journal field
                set (a ``JournalEntry`` subclass, which persists nothing).
            AppendOnlyViolation: If the record's Redis key already exists.
        """
        if cls.entry_model.kind_is_targetless(kind) and not (statement or "").strip():
            if not (verbatim or "").strip():
                raise ValueError(
                    f"{cls.entry_model.__name__}: an entry with neither a "
                    f"statement nor a verbatim span is not a provenance record"
                )
        return cls._write(
            agent_id=agent_id,
            kind=kind,
            target=target,
            statement=statement,
            verbatim=verbatim,
            payload=payload,
            speaker=speaker,
            turn_id=turn_id,
            subjects=subjects,
            stated=stated,
            captured_at=captured_at,
            at=at,
            claim_type=claim_type,
            pipeline=pipeline,
        )

    @classmethod
    def confirm(
        cls,
        target: Union[str, "JournalEntry"],
        *,
        agent_id: str,
        statement: str = "",
        verbatim: str = "",
        speaker: Optional[str] = None,
        turn_id: Optional[str] = None,
        subjects: Optional[Sequence[str]] = None,
        stated: bool = True,
        captured_at: Optional[float] = None,
        at: Optional[float] = None,
        pipeline: Optional["Pipeline"] = None,
    ) -> AnnotationResult:
        """Append a ``confirm`` annotation. Membership is unaffected.

        A confirmation is evidence, not a membership change: the target keeps
        its open interval, and :attr:`AnnotationResult.target_closed` is always
        False. Downstream readers use the annotation count as corroboration.

        Args:
            target: The entry being confirmed, or its Redis key.
            agent_id: See :meth:`append`. Must match the target's agent.
            statement: See :meth:`append`. May be empty for an annotation.
            verbatim: See :meth:`append`.
            speaker: See :meth:`append`.
            turn_id: See :meth:`append`.
            subjects: See :meth:`append`.
            stated: See :meth:`append`.
            captured_at: See :meth:`append`.
            at: Valid-from instant. Must not precede the target's stored
                ``valid_from``.
            pipeline: See :meth:`append`.

        Returns:
            AnnotationResult: with ``target_closed=False``, or ``None`` when
            the caller supplied the pipeline (nothing has executed yet).

        Raises:
            JournalBlockedError: If content is refused by the firewall.
            ValueError: On a nonexistent, unsaved or cross-agent target, a
                backdated ``at``, or a non-transactional pipeline.
        """
        return cls._write(
            agent_id=agent_id,
            kind="confirm",
            target=target,
            statement=statement,
            verbatim=verbatim,
            speaker=speaker,
            turn_id=turn_id,
            subjects=subjects,
            stated=stated,
            captured_at=captured_at,
            at=at,
            pipeline=pipeline,
        )

    @classmethod
    def supersede(
        cls,
        target: Union[str, "JournalEntry"],
        *,
        agent_id: str,
        statement: str = "",
        verbatim: str = "",
        speaker: Optional[str] = None,
        turn_id: Optional[str] = None,
        subjects: Optional[Sequence[str]] = None,
        stated: bool = True,
        captured_at: Optional[float] = None,
        at: Optional[float] = None,
        pipeline: Optional["Pipeline"] = None,
    ) -> AnnotationResult:
        """Append a ``supersede`` annotation and close the target's interval.

        The annotation entry and the target's interval close are queued into
        one ``MULTI``/``EXEC``, so no interleaving reader sees the annotation
        without the close. Rollback is not claimed -- see the module docstring.

        Args:
            target: The entry being corrected, or its Redis key.
            agent_id: See :meth:`append`. Must match the target's agent.
            statement: The corrected claim.
            verbatim: The exact words of the correction.
            speaker: See :meth:`append`.
            turn_id: See :meth:`append`.
            subjects: See :meth:`append`.
            stated: See :meth:`append`.
            captured_at: See :meth:`append`.
            at: The instant the correction becomes true, and the instant the
                target's interval closes. Defaults to now. Must not precede
                the target's stored ``valid_from``.
            pipeline: See :meth:`append`.

        Returns:
            AnnotationResult: ``target_closed`` is False when the coupling
            switch is off, or when the target was already closed by a
            concurrent annotation (which is correct: both annotations are real
            provenance, one close applies). It is ``None`` when the caller
            supplied the pipeline -- nothing has executed, so no truthful
            answer exists yet; read ``results[close_index]`` after your own
            ``execute()``.

        Raises:
            JournalBlockedError: If content is refused by the firewall. The
                target is **not** closed and no chain link is written.
            ValueError: On a nonexistent, unsaved or cross-agent target, a
                backdated ``at``, or a non-transactional pipeline.
        """
        return cls._write(
            agent_id=agent_id,
            kind="supersede",
            target=target,
            statement=statement,
            verbatim=verbatim,
            speaker=speaker,
            turn_id=turn_id,
            subjects=subjects,
            stated=stated,
            captured_at=captured_at,
            at=at,
            pipeline=pipeline,
        )

    @classmethod
    def retract(
        cls,
        target: Union[str, "JournalEntry"],
        *,
        agent_id: str,
        statement: str = "",
        verbatim: str = "",
        speaker: Optional[str] = None,
        turn_id: Optional[str] = None,
        subjects: Optional[Sequence[str]] = None,
        stated: bool = True,
        captured_at: Optional[float] = None,
        at: Optional[float] = None,
        pipeline: Optional["Pipeline"] = None,
    ) -> AnnotationResult:
        """Append a ``retract`` annotation and close the target's interval.

        Mechanically identical to :meth:`supersede`; the difference is
        semantic. A supersession replaces a claim with a better one, a
        retraction withdraws it with no replacement, and both remove the target
        from live membership while leaving it fully readable historically.

        Args:
            target: The entry being withdrawn, or its Redis key.
            agent_id: See :meth:`append`. Must match the target's agent.
            statement: Optional -- a retraction may be wordless.
            verbatim: See :meth:`append`.
            speaker: See :meth:`append`.
            turn_id: See :meth:`append`.
            subjects: See :meth:`append`.
            stated: See :meth:`append`.
            captured_at: See :meth:`append`.
            at: The instant the target stops being true. Defaults to now.
            pipeline: See :meth:`append`.

        Returns:
            AnnotationResult: see :meth:`supersede`.

        Raises:
            JournalBlockedError: If content is refused by the firewall.
            ValueError: On a nonexistent, unsaved or cross-agent target, a
                backdated ``at``, or a non-transactional pipeline.
        """
        return cls._write(
            agent_id=agent_id,
            kind="retract",
            target=target,
            statement=statement,
            verbatim=verbatim,
            speaker=speaker,
            turn_id=turn_id,
            subjects=subjects,
            stated=stated,
            captured_at=captured_at,
            at=at,
            pipeline=pipeline,
        )

    # ------------------------------------------------------------------
    # Reads
    # ------------------------------------------------------------------

    @classmethod
    def annotations_for(
        cls, entry: Union[str, "JournalEntry"]
    ) -> Union["QueryBuilder", "list[JournalEntry]"]:
        """Return every entry annotating ``entry``, in one ``filter()`` call.

        Args:
            entry: The annotated entry, or its Redis key.

        Returns:
            list: Matching :class:`JournalEntry` instances, in index order.
            Empty for an unsaved or unannotated entry.
        """
        target_key = _resolve_member_key(entry)
        if not target_key:
            return []
        return cls.entry_model.query.filter(target=target_key)

    @classmethod
    def chain(cls, entry: Union[str, "JournalEntry"]) -> "list[Any]":
        """Return the supersession chain through ``entry``, oldest first.

        A display and replay-verification read, **not** the membership query:
        live membership comes from
        ``filter(validity__current=True)`` with no chain walk at all. This
        walks the validity chain hashes from any member outward, so the chain
        is recoverable from its head, its tail, or anywhere in between.

        Args:
            entry: Any member of the chain, or its Redis key.

        Returns:
            list: :class:`JournalEntry` instances ordered oldest -> newest.
            ``[entry]`` when it has no chain links; ``[]`` for an unsaved entry
            or an unresolvable key.
        """
        instance: Any = entry
        if isinstance(entry, str):
            instance = cls.entry_model.query.get(redis_key=entry)
            if instance is None:
                return []
        return SupersessionProtocol.chain(instance, VALIDITY_FIELD_NAME)

    # ------------------------------------------------------------------
    # Internals
    # ------------------------------------------------------------------

    @classmethod
    def _write(
        cls,
        *,
        agent_id: str,
        kind: str,
        target: Optional[Union[str, "JournalEntry"]],
        statement: str,
        verbatim: str,
        speaker: Optional[str],
        payload: str = "",
        turn_id: Optional[str],
        subjects: Optional[Sequence[str]],
        stated: bool,
        captured_at: Optional[float],
        at: Optional[float],
        pipeline: Optional["Pipeline"],
        claim_type: Optional[str] = None,
    ) -> AnnotationResult:
        """Run the pre-flight, then append (and optionally close) in one go.

        The single write path behind every public mutating method, so the
        pre-flight cannot be bypassed by adding a method.
        """
        model = cls.entry_model
        _require_journal_shape(model)
        instant = _coerce_instant(at)
        subject_tags = list(subjects or [])

        # ---- D7 pre-flight. Every check below raises BEFORE a single
        # mutating command is issued or queued. The reads it performs
        # (target existence, the target's stored valid_from) are the point:
        # each one is a question that cannot be answered after the fact
        # without having already written something.

        # 1. Vocabulary and kind/target consistency. Pure, in-memory checks --
        #    they issue no Redis command, so they are safe to run ahead of the
        #    firewall.
        target_key = _resolve_member_key(target)
        kind, target_key = validate_kind_and_target(model, kind, target_key)

        if not agent_id:
            raise ValueError(
                f"{model.__name__}: agent_id is required and must be non-empty "
                f"-- a None renders the literal 'None' into the record key"
            )

        # 2. Build the entry. Construction is pure: ``AutoKeyField`` assigns a
        #    key in ``__init__`` but nothing is issued to Redis. Building it
        #    *here*, before the firewall, is what lets step 3 scan exactly the
        #    value set ``NeverRecordMixin`` will scan at save time.
        entry = model(
            agent_id=agent_id,
            captured_at=time.time() if captured_at is None else float(captured_at),
            turn_id=turn_id,
            speaker=speaker,
            verbatim=verbatim,
            statement=statement,
            payload=payload,
            subjects=subject_tags,
            stated=stated,
            kind=kind,
            target=target_key,
            # Set at construction, like every other field: the record is
            # append-only, so capture is the only chance to assign it (#564).
            claim_type=claim_type,
            # Valid-time is set at CONSTRUCTION, not passed to the supersede
            # script. ValidityField.on_save uses the field value as valid_from
            # and its ZADD NX runs earlier in the pipeline, which makes the
            # supersede script's own valid_from write a silent no-op -- so a
            # backdated instant passed only to the close is silently replaced
            # by the save clock (#588). True of ``save_and_invalidate`` for the
            # same reason: it queues the save first, on the same pipe.
            validity=instant,
        )

        # 3. The firewall, scanned here rather than left to NeverRecordMixin.
        #    ``Model.save()`` *returns* rather than raises when the firewall
        #    fires (``return pipeline if pipeline else False``, base.py step 0),
        #    and in pipeline mode the pipeline it returns is indistinguishable
        #    from success -- so a naive annotate-and-close would queue the
        #    invalidate EVAL against an annotation that was never written and
        #    commit a membership change with zero provenance.
        #
        #    The scanned values are derived from
        #    ``entry._never_record_scan_values()`` -- the *same method* the
        #    mixin calls -- rather than re-listed here. That equality is
        #    load-bearing: PR #589 shipped a hand-written list that omitted
        #    ``target``, the mixin scanned it anyway, a Luhn-passing uuid4 hex
        #    blocked the save, and the drop was silent. Two value sets that can
        #    drift are two chances to reach that path; one set cannot drift.
        #
        #    Two additions the mixin structurally cannot make. ``agent_id`` is a
        #    KeyField, excluded from the mixin's surface, and it is the one
        #    value here that *must* be scanned: it is rendered into the record's
        #    Redis key NAME, the one place ``hard_delete`` cannot scrub
        #    retroactively -- the name of an annotation's ``$IndexedF:`` index
        #    key embeds it. And ``subject_tags`` is a list, so the mixin's
        #    str-only yield never sees it.
        _scan_or_block(agent_id, *subject_tags, *entry._never_record_scan_values())

        # Hydrated by the pre-flight below when there is a target, and reused as
        # the ``closes`` argument of the close. The read is the ownership check's
        # read, not a second one: naming the incumbent by instance rather than by
        # key string adds no Redis command.
        stored_target: Any = None

        if target_key is not None:
            # 4. The target exists, and belongs to this agent. ``target`` is a
            #    full Redis key that could name another agent's partition, so a
            #    cross-agent annotation is rejected rather than silently
            #    closing a neighbour's record.
            if not model.exists(redis_key=target_key):
                raise ValueError(
                    f"{model.__name__}: annotation target {target_key!r} does "
                    f"not exist. Save the target before annotating it."
                )
            stored_target = model.query.get(redis_key=target_key)
            if stored_target is None:
                raise ValueError(
                    f"{model.__name__}: annotation target {target_key!r} is not "
                    f"a readable {model.__name__} record"
                )
            if str(stored_target.agent_id) != str(agent_id):
                raise ValueError(
                    f"{model.__name__}: cross-agent annotation refused -- "
                    f"target belongs to agent {stored_target.agent_id!r}, "
                    f"annotation to {agent_id!r}"
                )

            # 5. The requested instant is not before the target's *stored*
            #    valid_from. This cannot be delegated to
            #    ``execute_supersede``: its CLOSE_BEFORE_START -> ValueError
            #    remap applies only on the non-pipeline branch, and its
            #    client-side pre-check compares close_at against the
            #    caller-supplied valid_from, which is the same instant here --
            #    so it never fires. Without this pre-read, a genuine backdate
            #    surfaces as a raw ResponseError from execute() with the
            #    annotation already written.
            #
            #    ``get_valid_from`` is the field's own read of the same index
            #    (#630): one ZSCORE against the key ``get_interval_keys`` used
            #    to build here by hand, returning the same ``Optional[float]``.
            stored_valid_from = ValidityField.get_valid_from(
                model, VALIDITY_FIELD_NAME, target_key
            )
            if stored_valid_from is not None and instant < float(stored_valid_from):
                raise ValueError(
                    f"{model.__name__}: annotation instant {instant} precedes "
                    f"the target's valid_from ({float(stored_valid_from)}). A "
                    f"zero-or-negative-length interval is a caller bug, not a "
                    f"state to store."
                )

        # 6. A non-transactional caller pipeline would silently void the
        #    atomicity guarantee, so it is refused rather than honored.
        from ..backends.routing import non_redis_backend

        backend = non_redis_backend(model)
        if backend is not None:
            # #759 M4: a journal stored on another backend appends (and
            # closes) inside that backend's transaction, which is atomic by
            # construction; only its own unit of work can carry the write.
            return _append_on_backend(
                model,
                entry,
                stored_target=stored_target,
                target_key=target_key,
                kind=kind,
                instant=instant,
                pipeline=pipeline,
                backend=backend,
            )
        if pipeline is not None:
            if not isinstance(pipeline, redis.client.Pipeline):
                raise ValueError(
                    f"{model.__name__}: pipeline must be a redis Pipeline, got "
                    f"{type(pipeline).__name__}"
                )
            if pipeline.transaction is not True:
                raise ValueError(
                    f"{model.__name__}: pipeline(transaction=False) voids the "
                    f"annotate-and-close atomicity guarantee. Use "
                    f"popoto.get_redis().pipeline() (transactional by default)."
                )
            # A WATCHing pipeline that has not yet been put into MULTI executes
            # each command immediately rather than queueing, so Model.save()
            # fails part-way through and leaves an entry hash with no indexes
            # and no validity interval. On an append-only model that orphan is
            # permanent -- only hard_delete can remove it -- so this is refused
            # here rather than discovered at write time. (Pre-existing ORM
            # behavior: a plain Model.save() against such a pipeline fails
            # identically. Found in PR #589 review.)
            #
            # ``watching and not explicit_transaction`` is the precise
            # condition, NOT ``watching`` alone. watch() + multi() is the
            # standard redis-py optimistic-locking pattern: multi() sets
            # ``explicit_transaction = True``, after which commands queue
            # normally and the whole annotate-and-close applies atomically at
            # execute(). Refusing that case made the journal strictly less
            # capable than a bare Popoto model, which accepts it. (Round-3
            # review of PR #589.)
            if getattr(pipeline, "watching", False) and not getattr(
                pipeline, "explicit_transaction", False
            ):
                raise ValueError(
                    f"{model.__name__}: a WATCHing pipeline that is not yet in "
                    f"MULTI executes commands immediately instead of queueing, "
                    f"which would leave a permanent orphan record on an "
                    f"append-only model. Call pipeline.multi() to open the "
                    f"transaction -- that keeps your optimistic lock and makes "
                    f"the annotate-and-close atomic -- or use a fresh pipeline. "
                    f"Do NOT call UNWATCH: it would discard the lock you took."
                )

        # ---- End of pre-flight. From here on, commands are issued.

        coupling_enabled = bool(Defaults.JOURNAL_VALIDITY_COUPLING_ENABLED)
        should_close = model.kind_is_closing(kind) and target_key is not None
        if should_close and not coupling_enabled:
            _warn_uncoupled_once(model.__name__)

        owns_pipeline = pipeline is None
        # ``batch()`` opens the shared connection's transactional pipeline --
        # the same object this line used to build by hand, without this module
        # importing the client (#630).
        pipe = batch() if pipeline is None else pipeline

        close_index: Optional[int] = None
        if should_close and coupling_enabled:
            # ``should_close`` is ``kind_is_closing(kind) and target_key is not
            # None``, so a close can never be reached without a target. Raised
            # rather than asserted (``python -O`` strips asserts) and rather
            # than annotated away, because a ``None`` here is exactly the #588
            # shape: the close would have no incumbent to name, and the caller
            # would still be told the target was closed. It is also what makes
            # ``stored_target`` non-``None`` below -- the pre-flight hydrates it
            # exactly when ``target_key`` is not ``None``.
            if target_key is None or stored_target is None:
                raise RuntimeError(
                    f"{model.__name__}: reached the interval close with no "
                    f"resolved target. The close would name no incumbent while "
                    f"the result reported a close (#588). This is a bug in this "
                    f"module, not a caller error."
                )
            # ``save_and_invalidate``, not a bare ``execute_supersede`` (#606).
            # The protocol expresses everything this call site needs: an
            # *explicit* ``old_member`` (``closes=``, the target the pre-flight
            # already hydrated and validated) and ``assert_valid_from=False``,
            # which it passes unconditionally because ``at`` is a close-time
            # assertion about the incumbent, never a start-time assertion about
            # the successor. Valid-time is still set at CONSTRUCTION above; the
            # script's own ``valid_from`` write is a no-op behind ``on_save``'s
            # earlier ``ZADD NX`` either way.
            #
            # The save moves inside the protocol, which is why it does not
            # appear above: the whole point of routing through here is that the
            # entry HSET and the close EVAL are queued by one caller onto one
            # MULTI, with the declined-save guard between them. The atomicity
            # guarantee is unchanged -- same commands, same order, same pipe.
            #
            # The non-closing branch stays outside the protocol. There is no
            # incumbent to name, so ``save_and_invalidate`` has nothing to be
            # given; it saves the entry directly and carries its own copy of the
            # declined-save guard.
            #
            # The D7 pre-flight above is NOT delegated. Its firewall scan,
            # cross-agent ownership check and kind/target consistency check are
            # journal semantics the protocol has no vocabulary for, and every
            # one of them must run before the first command is queued.
            try:
                supersede_result = SupersessionProtocol.save_and_invalidate(
                    entry,
                    closes=stored_target,
                    at=instant,
                    field_name=VALIDITY_FIELD_NAME,
                    pipeline=pipe,
                )
            except SupersedeDeclinedError as exc:
                # Re-raised in this module's own vocabulary. The protocol's
                # message is about a successor and an incumbent; a journal
                # caller needs to hear that the *annotation* was not written,
                # which is the contract this module has documented and tested.
                verdict = getattr(exc, "verdict", None)
                reason = (
                    f"the never-record firewall ({verdict.reason})"
                    if verdict is not None
                    else "an early-return gate in Model.save()"
                )
                raise RuntimeError(
                    f"{model.__name__}: the annotation was not written -- "
                    f"{reason} stopped it, and Model.save() signals that by "
                    f"returning rather than raising. Nothing further is "
                    f"issued: queueing the interval close now would commit a "
                    f"membership change with no provenance behind it. This is "
                    f"a bug in this module (the pre-flight should have caught "
                    f"it), not a caller error."
                ) from exc
            close_index = supersede_result.close_index
        else:
            saved = entry.save(pipeline=pipe)

            # Defence in depth against the whole class of bug this module's
            # worst failure mode belongs to. ``Model.save()`` has several
            # early-return gates (the never-record firewall, the write filter)
            # that *return* instead of raising, and in pipeline mode what they
            # return is the pipeline -- indistinguishable from success.
            #
            # Nothing is closed on this branch, so the immediate harm the
            # closing branch guards against cannot occur here; what this raise
            # buys is that a silently-dropped annotation is never reported back
            # as a written one. The closing branch has the same guard inside
            # ``SupersessionProtocol._save_and_close``, where it sits *between*
            # the save and the close -- the only position from which it can stop
            # the close from being queued.
            #
            # The pre-flight scans the same value set the firewall does, so the
            # firewall gate is unreachable from here; this raise is what makes
            # that claim *checkable* rather than assumed, and it covers the
            # gates that have not been analysed yet. Raised, not asserted --
            # ``python -O`` strips asserts, and this must hold in production
            # above all.
            blocked = getattr(entry, "_never_record_verdict", None)
            if not saved or blocked is not None:
                reason = (
                    f"the never-record firewall ({blocked.reason})"
                    if blocked is not None
                    else "an early-return gate in Model.save()"
                )
                raise RuntimeError(
                    f"{model.__name__}: the annotation was not written -- "
                    f"{reason} stopped it, and Model.save() signals that by "
                    f"returning rather than raising. Nothing further is issued: "
                    f"queueing the interval close now would commit a membership "
                    f"change with no provenance behind it. This is a bug in this "
                    f"module (the pre-flight should have caught it), not a "
                    f"caller error."
                )

        if not owns_pipeline:
            # The caller owns execution, so nothing has run yet and there is no
            # truthful answer to "was the target closed". Report ``None`` --
            # *unknown until you execute* -- and hand back the index of the
            # queued close so the caller can read the real outcome from their
            # own ``execute()`` results. Reporting ``True`` here would be
            # affirmatively wrong for a re-close of an already-closed target:
            # the command queues identically and applies nothing.
            return AnnotationResult(
                entry=entry,
                target_closed=None,
                coupling_enabled=coupling_enabled,
                pipeline=pipe,
                close_index=close_index,
            )

        try:
            results = pipe.execute()
        except redis.exceptions.ResponseError as e:
            # The entry HSET is above the EVAL in this MULTI and has already
            # applied; Redis does not roll back a transaction when one command
            # errors. The annotation is real provenance and stays -- suppressing
            # it to match a failed close would lose information an append-only
            # journal exists to keep. Only the close failed, so surface *which*,
            # typed, rather than a raw Lua token string (#588 D8).
            #
            # Reachable because M1 names ``old_member`` explicitly, which the
            # script reads as a caller assertion: a target hard-deleted between
            # the pre-flight above and EXEC now returns MEMBER_ABSENT where it
            # previously took the idempotent no-op branch.
            raise map_lua_error(e) from e
        target_closed = False
        if close_index is not None and close_index < len(results):
            # SUPERSEDE_LUA returns the closed member key, or '' when its
            # idempotency guard found the target already closed (Race 3: two
            # annotations closing one target -- both entries are real
            # provenance, exactly one close applies).
            target_closed = bool(results[close_index])
        return AnnotationResult(
            entry=entry,
            target_closed=target_closed,
            coupling_enabled=coupling_enabled,
            pipeline=None,
        )

append(*, agent_id, statement='', verbatim='', payload='', speaker=None, turn_id=None, subjects=None, stated=True, captured_at=None, at=None, kind='assert', target=None, claim_type=None, pipeline=None) classmethod

Append one capture to the journal.

Parameters:

Name	Type	Description	Default
agent_id	str	Partition key. Required and non-null -- a None would render the literal "None" into the record's Redis key.	required
payload	str	Structured bookkeeping a recipe writes about its own decision. Exempt from the never-record firewall, so pass only values Popoto itself generated -- ids, keys, counters. It does not satisfy the "a record needs content" check below: a targetless entry still needs a statement or a verbatim. See :meth:JournalEntry._never_record_scan_values.	''
statement	str	The atomic claim. Required unless verbatim is given.	''
verbatim	str	The exact source span.	''
speaker	Optional[str]	Who said it. Attribution, not authentication.	None
turn_id	Optional[str]	Conversation turn the capture came from.	None
subjects	Optional[Sequence[str]]	Subject tags. None or [] puts the entry in the untagged pool, mirroring TagField's zero-tag semantics.	None
stated	bool	True when stated outright, False when inferred.	True
captured_at	Optional[float]	Wall-clock time of the source turn. Defaults to now.	None
at	Optional[float]	Valid-from instant for the entry. Defaults to now. Set at construction rather than passed to the supersede script -- see the supersede implementation note on #588.	None
kind	str	Normally left at "assert". A kind added with :meth:JournalEntry.register_kind is reachable here, with the same kind/target consistency rules -- and closes the target's interval if it was registered closing=True.	'assert'
target	Optional[Union[str, JournalEntry]]	Only for a non-assert kind. A :class:JournalEntry or its Redis key.	None
claim_type	Optional[str]	The claim's type, from recipes.reconciliation.CLAIM_TYPES. Set here, at capture, or never: the entry is append-only, so there is no later write that could back-fill it. Left None the reconciler treats the claim as the rule-free note type.	None
pipeline	Optional[Pipeline]	Optional caller pipeline. Must be transactional. When supplied it is returned unexecuted on :attr:AnnotationResult.pipeline. Open it with :func:popoto.batch; a pipeline built directly off the Redis client is still accepted (and is the same object type), but that spelling is deprecated -- batch() is the seam that survives #630's move of the client behind the field layer.	None

Returns:

Name	Type	Description
AnnotationResult	AnnotationResult	with target_closed=False -- a capture never
	AnnotationResult	changes another entry's membership. None instead when the
	AnnotationResult	caller supplied the pipeline (nothing has executed yet).

Raises:

Type	Description
JournalBlockedError	If any content is refused by the never-record firewall. Nothing is issued or queued.
ValueError	On a missing agent_id, empty content, a bad kind/target pairing, a nonexistent or cross-agent target, a backdated at, or a non-transactional pipeline.
TypeError	If entry_model does not declare the journal field set (a JournalEntry subclass, which persists nothing).
AppendOnlyViolation	If the record's Redis key already exists.

Source code in src/popoto/recipes/provenance_journal.py

@classmethod
def append(
    cls,
    *,
    agent_id: str,
    statement: str = "",
    verbatim: str = "",
    payload: str = "",
    speaker: Optional[str] = None,
    turn_id: Optional[str] = None,
    subjects: Optional[Sequence[str]] = None,
    stated: bool = True,
    captured_at: Optional[float] = None,
    at: Optional[float] = None,
    kind: str = "assert",
    target: Optional[Union[str, "JournalEntry"]] = None,
    claim_type: Optional[str] = None,
    pipeline: Optional["Pipeline"] = None,
) -> AnnotationResult:
    """Append one capture to the journal.

    Args:
        agent_id: Partition key. Required and non-null -- a ``None`` would
            render the literal ``"None"`` into the record's Redis key.
        payload: Structured bookkeeping a recipe writes about its own
            decision. **Exempt from the never-record firewall**, so pass
            only values Popoto itself generated -- ids, keys, counters. It
            does not satisfy the "a record needs content" check below:
            a targetless entry still needs a ``statement`` or a
            ``verbatim``. See
            :meth:`JournalEntry._never_record_scan_values`.
        statement: The atomic claim. Required unless ``verbatim`` is given.
        verbatim: The exact source span.
        speaker: Who said it. Attribution, not authentication.
        turn_id: Conversation turn the capture came from.
        subjects: Subject tags. ``None`` or ``[]`` puts the entry in the
            untagged pool, mirroring ``TagField``'s zero-tag semantics.
        stated: True when stated outright, False when inferred.
        captured_at: Wall-clock time of the source turn. Defaults to now.
        at: Valid-from instant for the entry. Defaults to now. Set at
            construction rather than passed to the supersede script -- see
            the ``supersede`` implementation note on #588.
        kind: Normally left at ``"assert"``. A kind added with
            :meth:`JournalEntry.register_kind` is reachable here, with the
            same ``kind``/``target`` consistency rules -- and closes the
            target's interval if it was registered ``closing=True``.
        target: Only for a non-``assert`` kind. A
            :class:`JournalEntry` or its Redis key.
        claim_type: The claim's type, from
            ``recipes.reconciliation.CLAIM_TYPES``. Set here, at capture,
            or never: the entry is append-only, so there is no later write
            that could back-fill it. Left ``None`` the reconciler treats
            the claim as the rule-free ``note`` type.
        pipeline: Optional caller pipeline. Must be transactional. When
            supplied it is returned unexecuted on
            :attr:`AnnotationResult.pipeline`. Open it with
            :func:`popoto.batch`; a pipeline built directly off the Redis
            client is still accepted (and is the same object type), but
            that spelling is deprecated -- ``batch()`` is the seam that
            survives #630's move of the client behind the field layer.

    Returns:
        AnnotationResult: with ``target_closed=False`` -- a capture never
        changes another entry's membership. ``None`` instead when the
        caller supplied the pipeline (nothing has executed yet).

    Raises:
        JournalBlockedError: If any content is refused by the never-record
            firewall. Nothing is issued or queued.
        ValueError: On a missing ``agent_id``, empty content, a bad
            ``kind``/``target`` pairing, a nonexistent or cross-agent
            target, a backdated ``at``, or a non-transactional pipeline.
        TypeError: If ``entry_model`` does not declare the journal field
            set (a ``JournalEntry`` subclass, which persists nothing).
        AppendOnlyViolation: If the record's Redis key already exists.
    """
    if cls.entry_model.kind_is_targetless(kind) and not (statement or "").strip():
        if not (verbatim or "").strip():
            raise ValueError(
                f"{cls.entry_model.__name__}: an entry with neither a "
                f"statement nor a verbatim span is not a provenance record"
            )
    return cls._write(
        agent_id=agent_id,
        kind=kind,
        target=target,
        statement=statement,
        verbatim=verbatim,
        payload=payload,
        speaker=speaker,
        turn_id=turn_id,
        subjects=subjects,
        stated=stated,
        captured_at=captured_at,
        at=at,
        claim_type=claim_type,
        pipeline=pipeline,
    )

confirm(target, *, agent_id, statement='', verbatim='', speaker=None, turn_id=None, subjects=None, stated=True, captured_at=None, at=None, pipeline=None) classmethod

Append a confirm annotation. Membership is unaffected.

A confirmation is evidence, not a membership change: the target keeps its open interval, and :attr:AnnotationResult.target_closed is always False. Downstream readers use the annotation count as corroboration.

Parameters:

Name	Type	Description	Default
target	Union[str, JournalEntry]	The entry being confirmed, or its Redis key.	required
agent_id	str	See :meth:append. Must match the target's agent.	required
statement	str	See :meth:append. May be empty for an annotation.	''
verbatim	str	See :meth:append.	''
speaker	Optional[str]	See :meth:append.	None
turn_id	Optional[str]	See :meth:append.	None
subjects	Optional[Sequence[str]]	See :meth:append.	None
stated	bool	See :meth:append.	True
captured_at	Optional[float]	See :meth:append.	None
at	Optional[float]	Valid-from instant. Must not precede the target's stored valid_from.	None
pipeline	Optional[Pipeline]	See :meth:append.	None

Returns:

Name	Type	Description
AnnotationResult	AnnotationResult	with target_closed=False, or None when
	AnnotationResult	the caller supplied the pipeline (nothing has executed yet).

Raises:

Type	Description
JournalBlockedError	If content is refused by the firewall.
ValueError	On a nonexistent, unsaved or cross-agent target, a backdated at, or a non-transactional pipeline.

Source code in src/popoto/recipes/provenance_journal.py

@classmethod
def confirm(
    cls,
    target: Union[str, "JournalEntry"],
    *,
    agent_id: str,
    statement: str = "",
    verbatim: str = "",
    speaker: Optional[str] = None,
    turn_id: Optional[str] = None,
    subjects: Optional[Sequence[str]] = None,
    stated: bool = True,
    captured_at: Optional[float] = None,
    at: Optional[float] = None,
    pipeline: Optional["Pipeline"] = None,
) -> AnnotationResult:
    """Append a ``confirm`` annotation. Membership is unaffected.

    A confirmation is evidence, not a membership change: the target keeps
    its open interval, and :attr:`AnnotationResult.target_closed` is always
    False. Downstream readers use the annotation count as corroboration.

    Args:
        target: The entry being confirmed, or its Redis key.
        agent_id: See :meth:`append`. Must match the target's agent.
        statement: See :meth:`append`. May be empty for an annotation.
        verbatim: See :meth:`append`.
        speaker: See :meth:`append`.
        turn_id: See :meth:`append`.
        subjects: See :meth:`append`.
        stated: See :meth:`append`.
        captured_at: See :meth:`append`.
        at: Valid-from instant. Must not precede the target's stored
            ``valid_from``.
        pipeline: See :meth:`append`.

    Returns:
        AnnotationResult: with ``target_closed=False``, or ``None`` when
        the caller supplied the pipeline (nothing has executed yet).

    Raises:
        JournalBlockedError: If content is refused by the firewall.
        ValueError: On a nonexistent, unsaved or cross-agent target, a
            backdated ``at``, or a non-transactional pipeline.
    """
    return cls._write(
        agent_id=agent_id,
        kind="confirm",
        target=target,
        statement=statement,
        verbatim=verbatim,
        speaker=speaker,
        turn_id=turn_id,
        subjects=subjects,
        stated=stated,
        captured_at=captured_at,
        at=at,
        pipeline=pipeline,
    )

supersede(target, *, agent_id, statement='', verbatim='', speaker=None, turn_id=None, subjects=None, stated=True, captured_at=None, at=None, pipeline=None) classmethod

Append a supersede annotation and close the target's interval.

The annotation entry and the target's interval close are queued into one MULTI/EXEC, so no interleaving reader sees the annotation without the close. Rollback is not claimed -- see the module docstring.

Parameters:

Name	Type	Description	Default
target	Union[str, JournalEntry]	The entry being corrected, or its Redis key.	required
agent_id	str	See :meth:append. Must match the target's agent.	required
statement	str	The corrected claim.	''
verbatim	str	The exact words of the correction.	''
speaker	Optional[str]	See :meth:append.	None
turn_id	Optional[str]	See :meth:append.	None
subjects	Optional[Sequence[str]]	See :meth:append.	None
stated	bool	See :meth:append.	True
captured_at	Optional[float]	See :meth:append.	None
at	Optional[float]	The instant the correction becomes true, and the instant the target's interval closes. Defaults to now. Must not precede the target's stored valid_from.	None
pipeline	Optional[Pipeline]	See :meth:append.	None

Returns:

Name	Type	Description
AnnotationResult	AnnotationResult	target_closed is False when the coupling
	AnnotationResult	switch is off, or when the target was already closed by a
	AnnotationResult	concurrent annotation (which is correct: both annotations are real
	AnnotationResult	provenance, one close applies). It is None when the caller
	AnnotationResult	supplied the pipeline -- nothing has executed, so no truthful
	AnnotationResult	answer exists yet; read results[close_index] after your own
	AnnotationResult	execute().

Raises:

Type	Description
JournalBlockedError	If content is refused by the firewall. The target is not closed and no chain link is written.
ValueError	On a nonexistent, unsaved or cross-agent target, a backdated at, or a non-transactional pipeline.

Source code in src/popoto/recipes/provenance_journal.py

@classmethod
def supersede(
    cls,
    target: Union[str, "JournalEntry"],
    *,
    agent_id: str,
    statement: str = "",
    verbatim: str = "",
    speaker: Optional[str] = None,
    turn_id: Optional[str] = None,
    subjects: Optional[Sequence[str]] = None,
    stated: bool = True,
    captured_at: Optional[float] = None,
    at: Optional[float] = None,
    pipeline: Optional["Pipeline"] = None,
) -> AnnotationResult:
    """Append a ``supersede`` annotation and close the target's interval.

    The annotation entry and the target's interval close are queued into
    one ``MULTI``/``EXEC``, so no interleaving reader sees the annotation
    without the close. Rollback is not claimed -- see the module docstring.

    Args:
        target: The entry being corrected, or its Redis key.
        agent_id: See :meth:`append`. Must match the target's agent.
        statement: The corrected claim.
        verbatim: The exact words of the correction.
        speaker: See :meth:`append`.
        turn_id: See :meth:`append`.
        subjects: See :meth:`append`.
        stated: See :meth:`append`.
        captured_at: See :meth:`append`.
        at: The instant the correction becomes true, and the instant the
            target's interval closes. Defaults to now. Must not precede
            the target's stored ``valid_from``.
        pipeline: See :meth:`append`.

    Returns:
        AnnotationResult: ``target_closed`` is False when the coupling
        switch is off, or when the target was already closed by a
        concurrent annotation (which is correct: both annotations are real
        provenance, one close applies). It is ``None`` when the caller
        supplied the pipeline -- nothing has executed, so no truthful
        answer exists yet; read ``results[close_index]`` after your own
        ``execute()``.

    Raises:
        JournalBlockedError: If content is refused by the firewall. The
            target is **not** closed and no chain link is written.
        ValueError: On a nonexistent, unsaved or cross-agent target, a
            backdated ``at``, or a non-transactional pipeline.
    """
    return cls._write(
        agent_id=agent_id,
        kind="supersede",
        target=target,
        statement=statement,
        verbatim=verbatim,
        speaker=speaker,
        turn_id=turn_id,
        subjects=subjects,
        stated=stated,
        captured_at=captured_at,
        at=at,
        pipeline=pipeline,
    )

retract(target, *, agent_id, statement='', verbatim='', speaker=None, turn_id=None, subjects=None, stated=True, captured_at=None, at=None, pipeline=None) classmethod

Append a retract annotation and close the target's interval.

Mechanically identical to :meth:supersede; the difference is semantic. A supersession replaces a claim with a better one, a retraction withdraws it with no replacement, and both remove the target from live membership while leaving it fully readable historically.

Parameters:

Name	Type	Description	Default
target	Union[str, JournalEntry]	The entry being withdrawn, or its Redis key.	required
agent_id	str	See :meth:append. Must match the target's agent.	required
statement	str	Optional -- a retraction may be wordless.	''
verbatim	str	See :meth:append.	''
speaker	Optional[str]	See :meth:append.	None
turn_id	Optional[str]	See :meth:append.	None
subjects	Optional[Sequence[str]]	See :meth:append.	None
stated	bool	See :meth:append.	True
captured_at	Optional[float]	See :meth:append.	None
at	Optional[float]	The instant the target stops being true. Defaults to now.	None
pipeline	Optional[Pipeline]	See :meth:append.	None

Returns:

Name	Type	Description
AnnotationResult	AnnotationResult	see :meth:supersede.

Raises:

Type	Description
JournalBlockedError	If content is refused by the firewall.
ValueError	On a nonexistent, unsaved or cross-agent target, a backdated at, or a non-transactional pipeline.

Source code in src/popoto/recipes/provenance_journal.py

@classmethod
def retract(
    cls,
    target: Union[str, "JournalEntry"],
    *,
    agent_id: str,
    statement: str = "",
    verbatim: str = "",
    speaker: Optional[str] = None,
    turn_id: Optional[str] = None,
    subjects: Optional[Sequence[str]] = None,
    stated: bool = True,
    captured_at: Optional[float] = None,
    at: Optional[float] = None,
    pipeline: Optional["Pipeline"] = None,
) -> AnnotationResult:
    """Append a ``retract`` annotation and close the target's interval.

    Mechanically identical to :meth:`supersede`; the difference is
    semantic. A supersession replaces a claim with a better one, a
    retraction withdraws it with no replacement, and both remove the target
    from live membership while leaving it fully readable historically.

    Args:
        target: The entry being withdrawn, or its Redis key.
        agent_id: See :meth:`append`. Must match the target's agent.
        statement: Optional -- a retraction may be wordless.
        verbatim: See :meth:`append`.
        speaker: See :meth:`append`.
        turn_id: See :meth:`append`.
        subjects: See :meth:`append`.
        stated: See :meth:`append`.
        captured_at: See :meth:`append`.
        at: The instant the target stops being true. Defaults to now.
        pipeline: See :meth:`append`.

    Returns:
        AnnotationResult: see :meth:`supersede`.

    Raises:
        JournalBlockedError: If content is refused by the firewall.
        ValueError: On a nonexistent, unsaved or cross-agent target, a
            backdated ``at``, or a non-transactional pipeline.
    """
    return cls._write(
        agent_id=agent_id,
        kind="retract",
        target=target,
        statement=statement,
        verbatim=verbatim,
        speaker=speaker,
        turn_id=turn_id,
        subjects=subjects,
        stated=stated,
        captured_at=captured_at,
        at=at,
        pipeline=pipeline,
    )

annotations_for(entry) classmethod

Return every entry annotating entry, in one filter() call.

Parameters:

Name	Type	Description	Default
entry	Union[str, JournalEntry]	The annotated entry, or its Redis key.	required

Returns:

Name	Type	Description
list	Union[QueryBuilder, list[JournalEntry]]	Matching :class:JournalEntry instances, in index order.
	Union[QueryBuilder, list[JournalEntry]]	Empty for an unsaved or unannotated entry.

Source code in src/popoto/recipes/provenance_journal.py

@classmethod
def annotations_for(
    cls, entry: Union[str, "JournalEntry"]
) -> Union["QueryBuilder", "list[JournalEntry]"]:
    """Return every entry annotating ``entry``, in one ``filter()`` call.

    Args:
        entry: The annotated entry, or its Redis key.

    Returns:
        list: Matching :class:`JournalEntry` instances, in index order.
        Empty for an unsaved or unannotated entry.
    """
    target_key = _resolve_member_key(entry)
    if not target_key:
        return []
    return cls.entry_model.query.filter(target=target_key)

chain(entry) classmethod

Return the supersession chain through entry, oldest first.

A display and replay-verification read, not the membership query: live membership comes from filter(validity__current=True) with no chain walk at all. This walks the validity chain hashes from any member outward, so the chain is recoverable from its head, its tail, or anywhere in between.

Parameters:

Name	Type	Description	Default
entry	Union[str, JournalEntry]	Any member of the chain, or its Redis key.	required

Returns:

Name	Type	Description
list	list[Any]	:class:JournalEntry instances ordered oldest -> newest.
	list[Any]	[entry] when it has no chain links; [] for an unsaved entry
	list[Any]	or an unresolvable key.

Source code in src/popoto/recipes/provenance_journal.py

@classmethod
def chain(cls, entry: Union[str, "JournalEntry"]) -> "list[Any]":
    """Return the supersession chain through ``entry``, oldest first.

    A display and replay-verification read, **not** the membership query:
    live membership comes from
    ``filter(validity__current=True)`` with no chain walk at all. This
    walks the validity chain hashes from any member outward, so the chain
    is recoverable from its head, its tail, or anywhere in between.

    Args:
        entry: Any member of the chain, or its Redis key.

    Returns:
        list: :class:`JournalEntry` instances ordered oldest -> newest.
        ``[entry]`` when it has no chain links; ``[]`` for an unsaved entry
        or an unresolvable key.
    """
    instance: Any = entry
    if isinstance(entry, str):
        instance = cls.entry_model.query.get(redis_key=entry)
        if instance is None:
            return []
    return SupersessionProtocol.chain(instance, VALIDITY_FIELD_NAME)

validate_kind_and_target(model, kind, target)

Check one kind/target pair against model's vocabulary.

Shared by :meth:JournalEntry.pre_save and :class:ProvenanceJournal's pre-flight so the two cannot drift apart.

Parameters:

Name	Type	Description	Default
model	Type[JournalEntry]	The :class:JournalEntry class (or subclass) being written.	required
kind	Any	The candidate kind.	required
target	Any	The candidate target Redis key, or None.	required

Returns:

Type	Description
tuple[str, Optional[str]]	The validated (kind, target) pair.

Raises:

Type	Description
ValueError	On an out-of-vocabulary kind or an inconsistent pairing.

Source code in src/popoto/recipes/provenance_journal.py

def validate_kind_and_target(
    model: Type["JournalEntry"], kind: Any, target: Any
) -> "tuple[str, Optional[str]]":
    """Check one ``kind``/``target`` pair against ``model``'s vocabulary.

    Shared by :meth:`JournalEntry.pre_save` and :class:`ProvenanceJournal`'s
    pre-flight so the two cannot drift apart.

    Args:
        model: The :class:`JournalEntry` class (or subclass) being written.
        kind: The candidate kind.
        target: The candidate target Redis key, or ``None``.

    Returns:
        The validated ``(kind, target)`` pair.

    Raises:
        ValueError: On an out-of-vocabulary kind or an inconsistent pairing.
    """
    vocabulary = model.journal_kinds()
    if kind not in vocabulary:
        raise ValueError(
            f"{model.__name__}.kind must be one of {vocabulary}, got {kind!r}"
        )
    if model.kind_is_targetless(kind):
        if target:
            raise ValueError(
                f"{model.__name__}: a {kind!r} entry is an original capture and "
                f"must not carry a target (got {target!r})"
            )
        return kind, None
    if not target:
        raise ValueError(
            f"{model.__name__}: a {kind!r} entry annotates another entry and "
            f"must name a target"
        )
    return kind, target