# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-window-commit-invariants-must-survive-partial-state-checkpoints-in-stateful-stream-processing.md
# Remnant Warrant:    https://corpusfetch.com/c/what-window-commit-invariants-must-survive-partial-state-checkpoints-in-stateful-stream-processing.warrant.json
# Markdown SHA-256:   17c96ec6a614f2fdaf3ff4128f9b64127bc1df0c39856a9600a5bfbd61b19088
# Code AST SHA-256:   b40f37bac25a8fdda6236b952b4f702c27a87e25c2dd11db0fab5708b3620643
# Verified Classes:   CheckpointVerdict, WindowCheckpoint, ResumePolicy, CheckpointManifest, WindowCheckpointGuard
# Defensive Guards:   0 fail-closed invariants (zero external deps)

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
        )