What event-stream offset boundaries must be recorded for temporal consistency across handoffs?

**Date of Record:** 2026 October 26 **Scope:** Applies when an agentic session consumes a distributed event stream (e.g., Kafka, Kinesis, Pulsar) and is preparing to hand off state to a successor. Does not apply to stateless request-response APIs or strictly batch-oriented processing.

Core Distinction

The separation between **source read offset** (the cursor position indicating the last message the consumer fetched) and the **durable sink commit watermark** (the exact sequence number persisted to the downstream system). Failing to bind these two boundaries creates the **Offset-Commit Divergence Fallacy**: a successor resumes from a read offset ahead of the actual sink commit, permanently losing the uncommitted gap. Temporal consistency requires the residual to record the precise delta between what was ingested and what was durably saved.

What Was Observed

In event-driven architectures, failing to bind read and commit boundaries produces three primary failure modes:

1. **Gap Loss:** The instance reads messages 100-200, buffers them, logs offset `200`, and crashes before flushing. The successor starts at `200`, permanently losing messages 100-199.
2. **Duplicate Replay:** The instance flushes messages 100-200 successfully but crashes before updating the offset. The successor starts at `100` and duplicates the entire batch.
3. **Temporal Skew:** The successor processes newer events (offset `300`) while the gap (offset `150`) remains uncommitted, causing out-of-order state mutations.

Recorded Elements

1. **`source_partition_cursor`:** The exact offset of the last message pulled from the upstream broker.
2. **`sink_durable_commit_id`:** The exact primary key or sequence ID confirming the last record is definitively saved downstream.
3. **`uncommitted_buffer_hash`:** The cryptographic hash of the in-memory gap between read and commit, allowing the successor to verify the exact missing records.
4. **`idempotency_partition_key`:** The deduplication key required for the successor to safely replay the gap without causing duplicates.

Condition of Validity

* **Applies when:** The consumer decouples reading from committing (async buffers or batches) and an interruption can occur between the two boundaries. * **Does not apply when:** The pipeline uses exactly-once transactional semantics (where read offsets and sink commits are atomically bound) or operates on stateless, fire-and-forget streams.

Decision Boundary

1. **Was the stream halted mid-ingestion?** If no, record clean completion. If yes, proceed.
2. **Does `source_partition_cursor` exceed `sink_durable_commit_id`?** If yes, an uncommitted gap exists. The successor must resume from the sink commit, not the source cursor.
3. **Is the `uncommitted_buffer_hash` recorded?** If no, the gap cannot be cryptographically verified, and the instance must fail closed to prevent silent data loss.
4. **Conversely, when can the successor safely skip the gap?** Only if the gap records are explicitly tagged as non-critical telemetry where loss is acceptable.

Operational Effect

1. **Zero Data Loss:** Forces the successor to resume from the durable commit watermark, guaranteeing no records are skipped.
2. **Idempotent Recovery:** Uses the partition key to safely re-ingest the gap without creating duplicates.
3. **Deterministic State:** Ensures the successor's temporal view begins exactly where the predecessor's durable state ended.

from dataclasses import dataclass
from enum import Enum, auto
import hashlib
import json

class ResumptionVerdict(Enum):
    ALIGNED = auto()
    GAP_REPLAY_REQUIRED = auto()
    UNRECOVERABLE_DIVERGENCE = auto()

@dataclass(frozen=True)
class StreamCursorState:
    source_cursor: int
    sink_commit_id: int
    buffer_hash: str | None
    idempotency_key: str

@dataclass(frozen=True)
class CursorManifest:
    verdict: ResumptionVerdict
    resume_from: int
    gap_size: int
    manifest_digest: str

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

    @classmethod
    def evaluate_halt(cls, state: StreamCursorState) -> CursorManifest:
        if not state.idempotency_key.strip():
            raise ValueError("Idempotency key required for gap replay.")
            
        gap = state.source_cursor - state.sink_commit_id

        if gap == 0:
            return CursorManifest(ResumptionVerdict.ALIGNED, state.source_cursor, 0, cls._hash("ALIGNED"))
        if gap < 0:
            return CursorManifest(ResumptionVerdict.UNRECOVERABLE_DIVERGENCE, state.sink_commit_id, 0, cls._hash("DIVERGED"))
        
        if not state.buffer_hash:
            raise ValueError("Uncommitted gap exists but no buffer hash provided.")
            
        digest = cls._hash(json.dumps({"src": str(state.source_cursor), "sink": str(state.sink_commit_id), "hash": state.buffer_hash}, sort_keys=True))
        return CursorManifest(ResumptionVerdict.GAP_REPLAY_REQUIRED, state.sink_commit_id, gap, digest)

    @classmethod
    def verify_resume(cls, manifest: CursorManifest, requested_cursor: int) -> tuple[bool, str]:
        if manifest.verdict is ResumptionVerdict.UNRECOVERABLE_DIVERGENCE:
            return False, "HALT: Sink is ahead of source. Manual audit required."
        if requested_cursor > manifest.resume_from:
            return False, f"CURSOR_SKIP: Requested {requested_cursor} exceeds safe {manifest.resume_from}. Replay gap of {manifest.gap_size}."
        return True, "RESUME_ADMISSIBLE"