BaseEventStore

Event store interface for event-sourced persistence. All event store adapters (Memory, MessageDB, etc.) implement this contract.

See Event Store Adapters for concrete adapter configuration.

This class outlines the base event store capabilities to be implemented in all supported event store adapters.

It is also a marker interface for registering event store classes with the domain.

Source code in src/protean/port/event_store.py

def __init__(self, name: str, domain: Domain, conn_info: dict[str, str]) -> None:
    self.name = name
    self.domain = domain
    self.conn_info = conn_info

close

close() -> None

Close the event store and release all connections.

Subclasses that hold external resources (connection pools, sockets, etc.) should override this to perform cleanup. The default implementation is a no-op so that adapters without external resources (e.g. the in-memory store) work without changes.

Source code in src/protean/port/event_store.py

def close(self) -> None:
    """Close the event store and release all connections.

    Subclasses that hold external resources (connection pools, sockets,
    etc.) should override this to perform cleanup.  The default
    implementation is a no-op so that adapters without external
    resources (e.g. the in-memory store) work without changes.
    """

read_all

read_all(
    stream: str = "$all", *, page_size: int = 1000
) -> Iterator[Message]

Yield every message in stream, paging through the store in bounded batches.

A cold-load read that must be complete (a full projection rebuild, a backup, an integrity check) cannot rely on a single large read with a sentinel no_of_messages: past the cap it silently truncates. This iterator pages the store in page_size batches and advances a cursor until a short page signals the end, so it reads the whole stream at a bounded memory cost regardless of size.

Paging is done on the raw store rows, not on deserialized messages, so that interleaved snapshot rows (type == "SNAPSHOT", no metadata) do not distort it. Snapshots are skipped from the output — the iterator yields events and commands only — but they still count toward the raw page, so a page carrying a snapshot neither ends the read early nor desyncs the cursor. The cursor advances from the last raw row of each page, which may itself be a snapshot; its top-level position is read directly, never through Message.deserialize.

The cursor field follows the stream shape (ADR-0024): $all and a bare category page by global_position; a specific stream (category-id) pages by its own per-stream position. Reads are inclusive (>= position), so each next page resumes one past the last row seen, which avoids re-emitting the boundary row.

PARAMETER	DESCRIPTION
stream	The stream to read. $all (default), a category, or a specific category-id stream. TYPE: str DEFAULT: '$all'
page_size	Number of messages to read per underlying read call. TYPE: int DEFAULT: 1000

YIELDS	DESCRIPTION
Message	Every Message in stream, in read order, with no gaps and
Message	no duplicates across page boundaries.

RAISES	DESCRIPTION
IncorrectUsageError	If page_size is not a positive integer.

Source code in src/protean/port/event_store.py

def read_all(
    self, stream: str = "$all", *, page_size: int = 1000
) -> Iterator[Message]:
    """Yield every message in ``stream``, paging through the store in bounded batches.

    A cold-load read that must be complete (a full projection rebuild, a
    backup, an integrity check) cannot rely on a single large ``read`` with a
    sentinel ``no_of_messages``: past the cap it silently truncates. This
    iterator pages the store in ``page_size`` batches and advances a cursor
    until a short page signals the end, so it reads the whole stream at a
    bounded memory cost regardless of size.

    Paging is done on the *raw* store rows, not on deserialized messages, so
    that interleaved snapshot rows (``type == "SNAPSHOT"``, no metadata) do
    not distort it. Snapshots are skipped from the output — the iterator
    yields events and commands only — but they still count toward the raw
    page, so a page carrying a snapshot neither ends the read early nor
    desyncs the cursor. The cursor advances from the last *raw* row of each
    page, which may itself be a snapshot; its top-level position is read
    directly, never through ``Message.deserialize``.

    The cursor field follows the stream shape (ADR-0024): ``$all`` and a bare
    category page by ``global_position``; a specific stream (``category-id``)
    pages by its own per-stream ``position``. Reads are inclusive
    (``>= position``), so each next page resumes one past the last row seen,
    which avoids re-emitting the boundary row.

    Args:
        stream: The stream to read. ``$all`` (default), a category, or a
            specific ``category-id`` stream.
        page_size: Number of messages to read per underlying ``read`` call.

    Yields:
        Every `Message` in ``stream``, in read order, with no gaps and
        no duplicates across page boundaries.

    Raises:
        IncorrectUsageError: If ``page_size`` is not a positive integer.
    """
    # Check the type before the value: a float or ``None`` slipping through
    # would flow into the adapter's row limit and either page oddly or raise
    # a bare ``TypeError`` far from the cause. ``bool`` is an ``int``, and
    # ``True`` (== 1) is harmless, so it is not special-cased.
    if not isinstance(page_size, int) or page_size < 1:
        raise IncorrectUsageError(
            f"`page_size` must be a positive integer, got {page_size!r}"
        )

    # A category read (`$all` or a bare category) pages by `global_position`;
    # a specific stream pages by its per-stream `position`. `category(stream)`
    # strips the `-id` suffix, so it equals `stream` only for a category/$all.
    pages_by_global_position = stream == self.category(stream)

    cursor = 0
    while True:
        raw_page = self._read(stream, position=cursor, no_of_messages=page_size)
        for raw_message in raw_page:
            if self._is_snapshot_row(raw_message):
                continue
            yield Message.deserialize(raw_message)

        # A short raw page is the last page: the store had no more rows to
        # fill it. Terminate on the *raw* count, not the yielded count, so a
        # page whose rows include snapshots is not mistaken for the end. This
        # also terminates the empty-stream case after one read.
        if len(raw_page) < page_size:
            return

        cursor = self._next_cursor(raw_page[-1], pages_by_global_position)

load_aggregate

load_aggregate(
    part_of: type[BaseAggregate],
    identifier: str,
    *,
    at_version: int | None = None,
    as_of: datetime | None = None,
) -> BaseAggregate | None

Load an aggregate from underlying events.

By default, reconstitutes the aggregate to its current (latest) state. When at_version or as_of is provided, reconstitutes a historical snapshot of the aggregate: a temporal query.

PARAMETER	DESCRIPTION
part_of	The EventSourced Aggregate's class. TYPE: type[BaseAggregate]
identifier	Unique aggregate identifier. TYPE: str
at_version	Reconstitute to this exact version (0-indexed). Version 0 is the state after the first event. TYPE: int | None DEFAULT: None
as_of	Reconstitute the aggregate as of this timestamp. Only events written on or before as_of are applied. TYPE: datetime | None DEFAULT: None

RETURNS	DESCRIPTION
BaseAggregate | None	The fully-formed aggregate, or None when no events exist
BaseAggregate | None	(and no temporal param was given that would raise instead).

Source code in src/protean/port/event_store.py

def load_aggregate(
    self,
    part_of: type[BaseAggregate],
    identifier: str,
    *,
    at_version: int | None = None,
    as_of: datetime | None = None,
) -> BaseAggregate | None:
    """Load an aggregate from underlying events.

    By default, reconstitutes the aggregate to its current (latest) state.
    When ``at_version`` or ``as_of`` is provided, reconstitutes a historical
    snapshot of the aggregate: a *temporal query*.

    Args:
        part_of: The EventSourced Aggregate's class.
        identifier: Unique aggregate identifier.
        at_version: Reconstitute to this exact version (0-indexed).
            Version 0 is the state after the first event.
        as_of: Reconstitute the aggregate as of this timestamp.
            Only events written on or before ``as_of`` are applied.

    Returns:
        The fully-formed aggregate, or ``None`` when no events exist
        (and no temporal param was given that would raise instead).
    """
    if as_of is not None:
        return self._load_aggregate_as_of(part_of, identifier, as_of)
    if at_version is not None:
        return self._load_aggregate_at_version(part_of, identifier, at_version)
    return self._load_aggregate_current(part_of, identifier)

create_snapshot

create_snapshot(
    part_of: type[BaseAggregate], identifier: str
) -> bool

Create a snapshot for a specific event-sourced aggregate instance.

Reads the full event stream for the aggregate, reconstructs it via from_events(), and writes a snapshot to the snapshot stream. This bypasses the snapshot threshold -- manual triggers always create a snapshot regardless of event count.

PARAMETER	DESCRIPTION
part_of	The EventSourced Aggregate class TYPE: type[BaseAggregate]
identifier	Unique aggregate identifier TYPE: str

RETURNS	DESCRIPTION
bool	True if a snapshot was created.

RAISES	DESCRIPTION
IncorrectUsageError	If the aggregate is not event-sourced.
ObjectNotFoundError	If no events exist for the given identifier.

Source code in src/protean/port/event_store.py

def create_snapshot(self, part_of: type[BaseAggregate], identifier: str) -> bool:
    """Create a snapshot for a specific event-sourced aggregate instance.

    Reads the full event stream for the aggregate, reconstructs it via
    ``from_events()``, and writes a snapshot to the snapshot stream.
    This bypasses the snapshot threshold -- manual triggers always create
    a snapshot regardless of event count.

    Args:
        part_of: The EventSourced Aggregate class
        identifier: Unique aggregate identifier

    Returns:
        True if a snapshot was created.

    Raises:
        IncorrectUsageError: If the aggregate is not event-sourced.
        ObjectNotFoundError: If no events exist for the given identifier.
    """
    if not part_of.meta_.is_event_sourced:
        raise IncorrectUsageError(
            f"`{part_of.__name__}` is not an event-sourced aggregate"
        )

    # Read ALL events (fresh reconstruction, not from existing snapshot),
    # paging the whole stream so the snapshot reflects the head state even
    # when the stream is longer than one read page.
    event_stream = self._read_stream_fully(
        f"{part_of.meta_.stream_category}-{identifier}"
    )

    if not event_stream:
        raise ObjectNotFoundError(
            f"`{part_of.__name__}` object with identifier {identifier} "
            f"does not exist."
        )

    events = [Message.deserialize(msg).to_domain_object() for msg in event_stream]
    aggregate = part_of.from_events(events)

    self._write(
        f"{part_of.meta_.stream_category}:snapshot-{identifier}",
        "SNAPSHOT",
        aggregate.to_dict(),
    )

    return True

create_snapshots

create_snapshots(part_of: type[BaseAggregate]) -> int

Create snapshots for all instances of an event-sourced aggregate.

Discovers all unique aggregate identifiers in the stream category, then creates a snapshot for each.

PARAMETER	DESCRIPTION
part_of	The EventSourced Aggregate class TYPE: type[BaseAggregate]

RETURNS	DESCRIPTION
int	Number of snapshots created.

RAISES	DESCRIPTION
IncorrectUsageError	If the aggregate is not event-sourced.

Source code in src/protean/port/event_store.py

def create_snapshots(self, part_of: type[BaseAggregate]) -> int:
    """Create snapshots for all instances of an event-sourced aggregate.

    Discovers all unique aggregate identifiers in the stream category,
    then creates a snapshot for each.

    Args:
        part_of: The EventSourced Aggregate class

    Returns:
        Number of snapshots created.

    Raises:
        IncorrectUsageError: If the aggregate is not event-sourced.
    """
    if not part_of.meta_.is_event_sourced:
        raise IncorrectUsageError(
            f"`{part_of.__name__}` is not an event-sourced aggregate"
        )

    identifiers = self._stream_identifiers(part_of.meta_.stream_category)

    # With fact_events enabled, persisting also writes a
    # ``{category}-fact-{id}`` stream that shares the category prefix. Those
    # are not aggregate instances and have no ``@apply`` handler, so exclude
    # them. Scoped to fact_events so an ordinary instance whose identifier
    # happens to start with ``fact-`` is never wrongly skipped.
    if part_of.meta_.fact_events:
        identifiers = [
            identifier
            for identifier in identifiers
            if not self._is_fact_stream_identifier(identifier)
        ]

    count = 0
    for identifier in identifiers:
        self.create_snapshot(part_of, identifier)
        count += 1

    return count

trace_causation

trace_causation(message_id: str | Message) -> list[Message]

Walk UP the causation chain from a message to the root.

Returns an ordered list of Messages from the root command (first) to the given message (last). The given message itself is included.

PARAMETER	DESCRIPTION
message_id	A Protean message ID string (headers.id) or a Message object. TYPE: str | Message

RETURNS	DESCRIPTION
list[Message]	List of Message objects in causal order (root first,
list[Message]	target last).

RAISES	DESCRIPTION
ValueError	If the message cannot be found in the event store.

Source code in src/protean/port/event_store.py

def trace_causation(self, message_id: str | Message) -> list[Message]:
    """Walk UP the causation chain from a message to the root.

    Returns an ordered list of Messages from the root command (first)
    to the given message (last).  The given message itself is included.

    Args:
        message_id: A Protean message ID string (``headers.id``) or
            a `Message` object.

    Returns:
        List of `Message` objects in causal order (root first,
        target last).

    Raises:
        ValueError: If the message cannot be found in the event store.
    """
    mid, group = self._resolve_and_load_group(message_id)

    # Build lookup: headers.id -> raw_message
    by_id: dict[str, dict[str, Any]] = {}
    for m in group:
        hid = self._extract_message_id(m)
        if hid:
            by_id[hid] = m

    # Walk up from target to root
    chain: list[dict[str, Any]] = []
    current_id: str | None = mid
    visited: set[str] = set()

    while current_id and current_id not in visited:
        visited.add(current_id)
        raw_msg = by_id.get(current_id)
        if raw_msg is None:
            break
        chain.append(raw_msg)
        current_id = self._extract_causation_id(raw_msg)

    # Reverse so root is first
    chain.reverse()

    return [Message.deserialize(m) for m in chain]

trace_effects

trace_effects(
    message_id: str | Message, *, recursive: bool = True
) -> list[Message]

Walk DOWN the causation chain to find all effects of a message.

Returns messages that were caused by the given message, ordered by global_position (chronological order).

PARAMETER	DESCRIPTION
message_id	A Protean message ID string (headers.id) or a Message object. TYPE: str | Message
recursive	If True (default), return the full subtree of effects. If False, return only direct children. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
list[Message]	List of Message objects caused by the given message,
list[Message]	in chronological order. The given message itself is NOT included.

RAISES	DESCRIPTION
ValueError	If the message cannot be found in the event store.

Source code in src/protean/port/event_store.py

def trace_effects(
    self, message_id: str | Message, *, recursive: bool = True
) -> list[Message]:
    """Walk DOWN the causation chain to find all effects of a message.

    Returns messages that were caused by the given message, ordered by
    ``global_position`` (chronological order).

    Args:
        message_id: A Protean message ID string (``headers.id``) or
            a `Message` object.
        recursive: If ``True`` (default), return the full subtree of
            effects.  If ``False``, return only direct children.

    Returns:
        List of `Message` objects caused by the given message,
        in chronological order.  The given message itself is NOT included.

    Raises:
        ValueError: If the message cannot be found in the event store.
    """
    mid, group = self._resolve_and_load_group(message_id)

    # Build children lookup: causation_id -> [raw_messages]
    children: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for m in group:
        cid = self._extract_causation_id(m)
        if cid:
            children[cid].append(m)

    if not recursive:
        direct = children.get(mid, [])
        direct.sort(key=lambda m: m.get("global_position", 0))
        return [Message.deserialize(m) for m in direct]

    # BFS for full subtree
    result: list[dict[str, Any]] = []
    queue: deque[str] = deque([mid])
    visited: set[str] = {mid}

    while queue:
        current = queue.popleft()
        for child in children.get(current, []):
            child_id = self._extract_message_id(child)
            if child_id and child_id not in visited:
                visited.add(child_id)
                result.append(child)
                queue.append(child_id)

    result.sort(key=lambda m: m.get("global_position", 0))
    return [Message.deserialize(m) for m in result]

build_causation_tree

build_causation_tree(
    correlation_id: str,
) -> CausationNode | None

Build a full causation tree for a correlation ID.

Returns the root node of the tree with children recursively populated.

PARAMETER	DESCRIPTION
correlation_id	The correlation ID to trace. TYPE: str

RETURNS	DESCRIPTION
CausationNode | None	Root CausationNode with children, or None if no
CausationNode | None	messages found.

Source code in src/protean/port/event_store.py

def build_causation_tree(self, correlation_id: str) -> CausationNode | None:
    """Build a full causation tree for a correlation ID.

    Returns the root node of the tree with children recursively populated.

    Args:
        correlation_id: The correlation ID to trace.

    Returns:
        Root [`CausationNode`][protean.port.event_store.CausationNode] with children, or ``None`` if no
        messages found.
    """
    group = self._load_correlation_group(correlation_id)
    if not group:
        return None

    # Build index and children map
    by_id: dict[str, dict[str, Any]] = {}
    children_map: dict[str, list[dict[str, Any]]] = defaultdict(list)
    roots: list[dict[str, Any]] = []

    for m in group:
        hid = self._extract_message_id(m)
        if hid:
            by_id[hid] = m
        cid = self._extract_causation_id(m)
        if cid:
            children_map[cid].append(m)
        else:
            roots.append(m)

    # Sort children by global_position for deterministic ordering
    for children in children_map.values():
        children.sort(key=lambda m: m.get("global_position", 0))

    visited: set[str] = set()

    def _build_node(raw_msg: dict[str, Any]) -> CausationNode:
        hid = self._extract_message_id(raw_msg) or "?"
        visited.add(hid)

        metadata = raw_msg.get("metadata", {})
        if not isinstance(metadata, dict):
            metadata = {}
        headers = metadata.get("headers", {})
        if not isinstance(headers, dict):
            headers = {}
        domain_meta = metadata.get("domain", {})
        if not isinstance(domain_meta, dict):
            domain_meta = {}

        node = CausationNode(
            message_id=hid,
            message_type=raw_msg.get("type", headers.get("type", "?")),
            kind=domain_meta.get("kind", "?"),
            stream=raw_msg.get("stream_name", headers.get("stream", "?")),
            time=str(raw_msg.get("time", "")) if raw_msg.get("time") else None,
            global_position=raw_msg.get("global_position"),
        )

        for child_msg in children_map.get(hid, []):
            child_id = self._extract_message_id(child_msg)
            if child_id and child_id not in visited:
                node.children.append(_build_node(child_msg))

        return node

    if not roots:
        # All messages have causation_id set — pick the one whose
        # causation_id points outside the group
        root_candidates = [
            m for m in group if self._extract_causation_id(m) not in by_id
        ]
        roots = root_candidates if root_candidates else [group[0]]

    roots.sort(key=lambda m: m.get("global_position", 0))
    return _build_node(roots[0])

stream_head_position

stream_head_position(stream_category: str) -> int

Return the global_position of the newest message in a category stream.

Public wrapper around _stream_head_position.

PARAMETER	DESCRIPTION
stream_category	The stream category to check. TYPE: str

RETURNS	DESCRIPTION
int	The global_position of the latest message, or -1 if the
int	stream has no messages.

Source code in src/protean/port/event_store.py

def stream_head_position(self, stream_category: str) -> int:
    """Return the global_position of the newest message in a category stream.

    Public wrapper around `_stream_head_position`.

    Args:
        stream_category: The stream category to check.

    Returns:
        The ``global_position`` of the latest message, or ``-1`` if the
        stream has no messages.
    """
    return self._stream_head_position(stream_category)

verify

verify() -> IntegrityReport

Check the store's internal invariants without mutating anything.

Reads the whole store once through $all and reports every violation of these invariants:

• every row carries its required fields (id, stream_name, position),
• per-stream position is gapless from the stream base (0),
• global_position is strictly increasing store-wide,
• message ids are unique,
• each :snapshot- stream carries a well-formed snapshot whose _version does not exceed its aggregate stream head.

A corrupt row is reported, never silently skipped: that is the whole point of the check, so a row missing a required field or a snapshot with a non-integer _version becomes a violation rather than a pass, and the missing-field guard runs first so those checks never touch an absent value. This handles the corruption a store can actually hold: every adapter types its columns (MessageDB by SQL column type, the memory adapter by its pydantic model), so a present-but-wrongly-typed field (a list id, a string position) does not arise from a read.

This asserts the store's internal consistency, not a schema version (none is stored today). It is read-only: a clean store yields a report with no violations (ok is True).

Source code in src/protean/port/event_store.py

def verify(self) -> IntegrityReport:
    """Check the store's internal invariants without mutating anything.

    Reads the whole store once through ``$all`` and reports every violation
    of these invariants:

    - every row carries its required fields (``id``, ``stream_name``,
      ``position``),
    - per-stream ``position`` is gapless from the stream base (0),
    - ``global_position`` is strictly increasing store-wide,
    - message ids are unique,
    - each ``:snapshot-`` stream carries a well-formed snapshot whose
      ``_version`` does not exceed its aggregate stream head.

    A corrupt row is reported, never silently skipped: that is the whole
    point of the check, so a row missing a required field or a snapshot with
    a non-integer ``_version`` becomes a violation rather than a pass, and
    the missing-field guard runs first so those checks never touch an absent
    value. This handles the corruption a store can actually hold: every
    adapter types its columns (MessageDB by SQL column type, the memory
    adapter by its pydantic model), so a present-but-wrongly-typed field
    (a list ``id``, a string ``position``) does not arise from a read.

    This asserts the store's *internal* consistency, not a schema version
    (none is stored today). It is read-only: a clean store yields a report
    with no violations (``ok`` is ``True``).
    """
    violations: list[IntegrityViolation] = []

    seen_ids: set[str] = set()
    prev_global_position: int | None = None
    # Per-stream: the last position seen (streams appear in position order
    # within the global_position-ordered scan) and the head (max) position.
    last_position: dict[str, int] = {}
    stream_head: dict[str, int] = {}
    # Per snapshot stream: the ``_version`` of its most recent snapshot.
    snapshot_version: dict[str, int] = {}

    message_count = 0
    for msg in self._iter_all_messages():
        message_count += 1
        stream = msg.get("stream_name")
        position = msg.get("position")
        global_position = msg.get("global_position")
        message_id = msg.get("id")

        # A row read from ``$all`` always carries a ``global_position`` (the
        # read filters ``>= position``), so check monotonicity FIRST, before
        # the malformed guard can ``continue``. A row can be malformed and
        # break monotonicity at once, and verify reports every violation.
        assert global_position is not None  # read-contract guarantee (mypy)
        if (
            prev_global_position is not None
            and global_position <= prev_global_position
        ):
            violations.append(
                IntegrityViolation(
                    kind=self.VERIFY_NON_MONOTONIC_GLOBAL_POSITION,
                    stream=stream,
                    position=position,
                    detail=(
                        f"global_position {global_position} does not exceed "
                        f"the previous {prev_global_position}."
                    ),
                )
            )
        prev_global_position = global_position

        # A row missing any required field is itself a corruption. Flag it
        # and move on: the remaining checks need those values, and running
        # them on a half-populated row would either crash or invent a
        # spurious gap.
        missing = [f for f in self._REQUIRED_FIELDS if msg.get(f) is None]
        if missing:
            violations.append(
                IntegrityViolation(
                    kind=self.VERIFY_MALFORMED_MESSAGE,
                    stream=stream,
                    position=position,
                    detail=(
                        "Message is missing required field(s): "
                        f"{', '.join(missing)}."
                    ),
                )
            )
            continue

        # Past the guard the three required fields are present. Narrow them
        # for the checks below (the raw dict values are ``Any | None``).
        assert (
            message_id is not None and stream is not None and position is not None
        )

        # Duplicate message id (the store's own message identity)
        if message_id in seen_ids:
            violations.append(
                IntegrityViolation(
                    kind=self.VERIFY_DUPLICATE_MESSAGE_ID,
                    stream=stream,
                    position=position,
                    detail=f"Message id '{message_id}' appears more than once.",
                )
            )
        seen_ids.add(message_id)

        # Per-stream gapless position from base 0
        expected = last_position.get(stream, -1) + 1
        if position != expected:
            violations.append(
                IntegrityViolation(
                    kind=self.VERIFY_POSITION_GAP,
                    stream=stream,
                    position=position,
                    detail=(
                        f"Stream '{stream}' jumps to position {position}; "
                        f"expected {expected}."
                    ),
                )
            )
        last_position[stream] = position
        stream_head[stream] = max(stream_head.get(stream, -1), position)

        # Track the latest snapshot version per snapshot stream. A snapshot
        # whose data is not a dict, or whose ``_version`` is missing or not a
        # plain int (``bool`` is an int subclass, so exclude it), is corrupt:
        # flag it here rather than silently dropping it from the head check.
        if self._SNAPSHOT_MARKER in stream:
            data = msg.get("data")
            version = data.get("_version") if isinstance(data, dict) else None
            if isinstance(version, int) and not isinstance(version, bool):
                snapshot_version[stream] = version
            else:
                violations.append(
                    IntegrityViolation(
                        kind=self.VERIFY_MALFORMED_SNAPSHOT,
                        stream=stream,
                        position=position,
                        detail=(
                            "Snapshot data is not a dict or its _version is "
                            "not an integer."
                        ),
                    )
                )

    # Each snapshot's _version must not exceed its aggregate stream head
    for snap_stream, version in snapshot_version.items():
        category, _, identifier = snap_stream.partition(self._SNAPSHOT_MARKER)
        aggregate_stream = f"{category}-{identifier}"
        head = stream_head.get(aggregate_stream, -1)
        if version > head:
            violations.append(
                IntegrityViolation(
                    kind=self.VERIFY_SNAPSHOT_AHEAD_OF_STREAM,
                    stream=snap_stream,
                    position=None,
                    detail=(
                        f"Snapshot _version {version} exceeds the head position "
                        f"{head} of aggregate stream '{aggregate_stream}'."
                    ),
                )
            )

    return IntegrityReport(
        message_count=message_count,
        stream_count=len(stream_head),
        violations=tuple(violations),
    )

---

CausationNode

Tree node used by build_causation_tree() to represent the causation hierarchy of messages sharing a correlation_id.

A node in the causation tree, representing a single message and its effects.