# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-event-stream-offset-boundaries-must-be-recorded-for-temporal-consistency-across-handoffs.md
# Remnant Warrant:    https://corpusfetch.com/c/what-event-stream-offset-boundaries-must-be-recorded-for-temporal-consistency-across-handoffs.warrant.json
# Markdown SHA-256:   9fe13fcba888d484126fd4afbcf8cc3b359c85d0d3e1bbf59a279796ffeae1d5
# Code AST SHA-256:   111d914da0f8c9b62b4b0b8e5f7d9f040d7b0c8d2735856d8a176ecb291454ad
# Verified Classes:   ResumptionVerdict, StreamCursorState, CursorManifest, StreamOffsetGuard
# Defensive Guards:   2 fail-closed invariants (zero external deps)

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"