What window-commit invariants must survive partial state checkpoints in stateful stream processing?

**Date of Record:** 2026 October 26 **Scope:** Applies when an agentic session performs stateful stream processing (tumbling, sliding, or session windows) with periodic state checkpoints, and must persist window state that survives a partial checkpoint failure or mid-window pause. Does not apply to stateless transformations or fully atomic checkpointing systems.

Core Distinction

The separation between **event processing** (individual events being consumed from the stream) and **window commit boundary** (the atomic state of a window that includes which events were aggregated, at what watermark, with what output). Failing to distinguish these creates the **Window Replay Corruption**: a partial checkpoint captures some events but not the window's watermark or aggregation state. The successor resumes, re-processes events already counted in the window, and produces a duplicated or under-counted aggregation.

What Was Observed

In stateful stream processing architectures, failing to preserve window commit invariants across checkpoints produces three primary failure modes:

  1. **Double-Counted Aggregations:** The window has counted events 1-100. A partial checkpoint captures offsets 1-50 but not the aggregation state. The successor resumes at offset 50 and re-counts 50-100, producing a doubled sum.
  2. **Watermark Regression:** The window's watermark advanced to T=1000, but the checkpoint only captured T=500 because of partial write. The successor resumes and processes events 500-1000 again, re-emitting outputs.
  3. **Truncated Window Output:** The window emitted its final result at watermark T=1000, but the checkpoint only captured the state up to T=800. The successor sees the window as "still open" and misses the already-emitted output, breaking downstream consumers.

Recorded Elements

  1. **`window_watermark_epoch`:** The exact watermark the window had advanced to at checkpoint time, ensuring the successor does not re-process events already counted.
  2. **`window_aggregation_state_hash`:** The cryptographic hash of the window's current aggregation state (partial sums, counts, running statistics), enabling the successor to detect truncated or corrupted state.
  3. **`window_output_emission_receipt`:** The cryptographic proof that the window's final output was emitted, including the emission epoch, preventing re-emission or missing emission.
  4. **`checkpoint_atomicity_signature`:** The cryptographic proof that the checkpoint was written atomically across all state components (watermark, aggregation, output emission), preventing partial-capture corruption.

Condition of Validity

* **Applies when:** The stream processing uses windows, checkpoints are written incrementally or non-atomically, and resumption must preserve the exact aggregation state without re-processing. * **Does not apply when:** The stream processing is strictly stateless (each event is independent), or checkpoints are guaranteed atomic by the underlying storage layer.

Decision Boundary

  1. **Is the checkpoint from a stateful windowed operation?** If no, standard checkpointing applies. If yes, proceed.
  2. **Are the `window_watermark_epoch`, `window_aggregation_state_hash`, and `window_output_emission_receipt` all present?** If any is missing, the checkpoint is partial. The successor must fall back to the previous atomic checkpoint.
  3. **Does the `checkpoint_atomicity_signature` prove all state was captured at the same point?** If no, the checkpoint is inconsistent (e.g., watermark advanced but aggregation captured earlier). The successor must reject the checkpoint and re-process from the last known-good watermark.
  4. **Conversely, when can a successor resume from a partial checkpoint?** Only when the partial checkpoint preserves the watermark and aggregation state as a coherent pair (both captured together), even if other state (e.g., secondary indexes) is captured separately and can be rebuilt.

Operational Effect

  1. **Idempotent Window Processing:** Prevents double-counting by ensuring the watermark and aggregation state are always captured together.
  2. **Output Emission Safety:** Prevents re-emission of window results or missing emission, preserving downstream consumer correctness.
  3. **Deterministic Recovery:** Provides a cryptographic proof of atomicity, allowing the successor to know exactly which state is trustworthy.
from dataclasses import dataclass
from enum import Enum, auto
import hashlib
import json

class CheckpointVerdict(Enum):
    ATOMIC_CHECKPOINT = auto()
    PARTIAL_CHECKPOINT_REJECT = auto()
    INCONSISTENT_STATE = auto()

@dataclass(frozen=True)
class WindowCheckpoint:
    window_id: str
    watermark_epoch: float
    aggregation_state_hash: str
    output_emission_receipt: str | None
    atomicity_signature: str
    checkpoint_epoch: float

@dataclass(frozen=True)
class ResumePolicy:
    require_output_emission_receipt: bool
    max_checkpoint_age_sec: float
    current_epoch: float

@dataclass(frozen=True)
class CheckpointManifest:
    verdict: CheckpointVerdict
    resume_watermark: float | None
    directive: str
    manifest_digest: str

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

    @classmethod
    def evaluate_checkpoint(
        cls, cp: WindowCheckpoint, policy: ResumePolicy
    ) -> CheckpointManifest:
        
        # Gate 1: Atomicity signature presence
        if not cp.atomicity_signature:
            return CheckpointManifest(
                verdict=CheckpointVerdict.PARTIAL_CHECKPOINT_REJECT,
                resume_watermark=None,
                directive=f"HALT: Checkpoint for window {cp.window_id} lacks atomicity signature. Fall back to prior checkpoint.",
                manifest_digest=cls._hash(f"NO_ATOMIC:{cp.window_id}")
            )

        # Gate 2: Output emission receipt
        if policy.require_output_emission_receipt and not cp.output_emission_receipt:
            return CheckpointManifest(
                verdict=CheckpointVerdict.PARTIAL_CHECKPOINT_REJECT,
                resume_watermark=None,
                directive="HALT: Output emission receipt missing. Cannot determine if window result was emitted.",
                manifest_digest=cls._hash(f"NO_EMISSION:{cp.window_id}")
            )

        # Gate 3: Freshness
        age = policy.current_epoch - cp.checkpoint_epoch
        if age > policy.max_checkpoint_age_sec:
            return CheckpointManifest(
                verdict=CheckpointVerdict.PARTIAL_CHECKPOINT_REJECT,
                resume_watermark=None,
                directive=f"HALT: Checkpoint is {age:.0f}s old, exceeding {policy.max_checkpoint_age_sec}s freshness limit.",
                manifest_digest=cls._hash(f"STALE:{cp.window_id}")
            )

        # Gate 4: Verified
        digest = cls._hash(json.dumps({
            "window": cp.window_id,
            "watermark": str(cp.watermark_epoch),
            "aggregation": cp.aggregation_state_hash,
            "atomicity": cp.atomicity_signature
        }, sort_keys=True))
        
        return CheckpointManifest(
            verdict=CheckpointVerdict.ATOMIC_CHECKPOINT,
            resume_watermark=cp.watermark_epoch,
            directive=f"PROCEED: Resume window at watermark {cp.watermark_epoch}.",
            manifest_digest=digest
        )