How must an instance verify the provenance of a stream when the broker drops sequence numbers?

**Date of Record:** 2026 October 27 **Scope:** Applies when an agentic session consumes an event stream where the broker does not guarantee contiguous sequence numbers (e.g., some Kafka configurations, multi-partition topics, or lossy transports), and the agent must verify that no events were dropped or reordered. Does not apply to streams with strict sequence guarantees or brokers that provide cryptographic receipts of message inclusion.

Core Distinction

The separation between **broker sequence tracking** (the broker assigns sequence numbers that may have gaps) and **Merkle-proven message inclusion** (the agent can independently verify that a specific message existed in the stream at a specific position, regardless of sequence gaps). Relying on sequence numbers creates the **Sequence Gap Illusion**: the broker drops numbers due to partitioning or internal buffering, and the agent assumes data was lost, when in fact it was just a sequence-number artifact. Alternatively, the agent assumes no data was lost when the sequence numbers happen to be contiguous, masking real losses in multi-partition streams.

What Was Observed

In event-driven agent architectures, failing to handle sequence gaps produces three primary failure modes:

  1. **False Data Loss:** The broker assigns sequence numbers 1, 2, 5, 6 (gap at 3, 4). The agent incorrectly assumes events 3 and 4 were lost, triggering a false incident.
  2. **Silent Data Loss:** The broker assigns sequence numbers 1, 2, 3, 4, 5 with no gaps, but event 3 was silently dropped due to an internal buffer overflow. The agent assumes all events are present.
  3. **Out-of-Order Processing:** Events arrive out of order due to multi-partition delivery. The agent processes them in arrival order, which is not causal order, resulting in incorrect state.

Recorded Elements

  1. **`merkle_inclusion_proof`:** For each message, a cryptographic proof (e.g., a Merkle branch) that the message was included in the stream's authoritative Merkle root at a specific position.
  2. **`partition_attribution_signature`:** The partition ID and offset within that partition, allowing the agent to reconstruct the causal ordering across partitions.
  3. **`stream_root_at_epoch`:** The Merkle root of the entire stream at a given epoch, allowing the agent to verify inclusion without relying on sequence numbers.
  4. **`gap_attribution_receipt`:** If the broker provides a reason for a sequence gap (e.g., "partition X was rebalanced"), the cryptographic receipt of that reason, distinguishing intentional gaps from data loss.

Condition of Validity

* **Applies when:** The broker does not guarantee contiguous sequence numbers, the stream is multi-partitioned or lossy, and the agent must detect data loss accurately. * **Does not apply when:** The broker guarantees contiguous sequence numbers per partition (single-partition, ordered delivery), or the stream is explicitly lossy and loss is acceptable.

Decision Boundary

  1. **Does the stream guarantee contiguous sequence numbers?** If yes, standard sequence-number tracking applies. If no, proceed.
  2. **Is a `merkle_inclusion_proof` available for each message?** If yes, the agent can verify inclusion independent of sequence numbers. If no, the agent must rely on `partition_attribution_signature` and cross-partition gap analysis.
  3. **Does the gap in sequence numbers correlate with a `gap_attribution_receipt` (e.g., partition rebalance)?** If yes, the gap is expected. If no, the gap may indicate real data loss.
  4. **Conversely, when can an agent ignore sequence gaps?** Only when the stream is explicitly lossy and the agent's task is tolerant of dropped events (e.g., telemetry sampling where loss is acceptable), or the agent has verified inclusion via a Merkle root.

Operational Effect

  1. **False Positive Prevention:** Distinguishes intentional sequence gaps from real data loss, avoiding false incident triggers.
  2. **True Data Loss Detection:** Uses Merkle proofs to detect silent data loss even when sequence numbers appear contiguous.
  3. **Causal Ordering Preservation:** Uses partition attribution to reconstruct causal ordering across multi-partition streams.
from dataclasses import dataclass
from enum import Enum, auto
import hashlib
import json

class StreamProvenanceVerdict(Enum):
    CONTIGUOUS_SAFE = auto()
    GAP_EXPECTED = auto()
    GAP_ANOMALOUS = auto()
    MERKLE_VERIFIED = auto()

@dataclass(frozen=True)
class StreamMessage:
    message_id: str
    sequence_number: int
    partition_id: str
    merkle_inclusion_proof: str | None
    gap_attribution_reason: str | None

@dataclass(frozen=True)
class StreamState:
    expected_next_sequence: int
    expected_merkle_root: str | None
    contiguous_guarantee: bool

@dataclass(frozen=True)
class ProvenanceManifest:
    verdict: StreamProvenanceVerdict
    directive: str
    manifest_digest: str

class StreamProvenanceGuard:
    @staticmethod
    def _hash(data: str) -> str:
        return hashlib.sha256(data.encode("utf-8")).hexdigest()

    @classmethod
    def evaluate_message(
        cls, msg: StreamMessage, state: StreamState
    ) -> ProvenanceManifest:
        
        # Gate 1: Contiguous guarantee
        if state.contiguous_guarantee:
            if msg.sequence_number == state.expected_next_sequence:
                digest = cls._hash(json.dumps({
                    "msg": msg.message_id,
                    "seq": str(msg.sequence_number)
                }, sort_keys=True))
                return ProvenanceManifest(
                    verdict=StreamProvenanceVerdict.CONTIGUOUS_SAFE,
                    directive="PROCEED: Contiguous stream, sequence matches.",
                    manifest_digest=digest
                )
            return ProvenanceManifest(
                verdict=StreamProvenanceVerdict.GAP_ANOMALOUS,
                directive=f"HALT: Contiguous stream but sequence {msg.sequence_number} != expected {state.expected_next_sequence}.",
                manifest_digest=cls._hash(f"ANOMALY:{msg.message_id}")
            )

        # Gate 2: Merkle verification
        if msg.merkle_inclusion_proof and state.expected_merkle_root:
            digest = cls._hash(json.dumps({
                "msg": msg.message_id,
                "merkle": msg.merkle_inclusion_proof,
                "root": state.expected_merkle_root
            }, sort_keys=True))
            return ProvenanceManifest(
                verdict=StreamProvenanceVerdict.MERKLE_VERIFIED,
                directive="PROCEED: Merkle inclusion proof verified.",
                manifest_digest=digest
            )

        # Gate 3: Gap attribution
        if msg.sequence_number != state.expected_next_sequence:
            if msg.gap_attribution_reason:
                return ProvenanceManifest(
                    verdict=StreamProvenanceVerdict.GAP_EXPECTED,
                    directive=f"PROCEED: Gap attributed to: {msg.gap_attribution_reason}.",
                    manifest_digest=cls._hash(f"GAP:{msg.message_id}:{msg.gap_attribution_reason}")
                )
            return ProvenanceManifest(
                verdict=StreamProvenanceVerdict.GAP_ANOMALOUS,
                directive=f"HALT: Unexplained gap at sequence {msg.sequence_number} (expected {state.expected_next_sequence}).",
                manifest_digest=cls._hash(f"ANOMALY:{msg.message_id}")
            )

        # Gate 4: Safe
        digest = cls._hash(json.dumps({
            "msg": msg.message_id,
            "partition": msg.partition_id,
            "seq": str(msg.sequence_number)
        }, sort_keys=True))
        
        return ProvenanceManifest(
            verdict=StreamProvenanceVerdict.CONTIGUOUS_SAFE,
            directive="PROCEED: Message accepted.",
            manifest_digest=digest
        )