# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-exact-cursor-offsets-must-be-persisted-when-halting-large-stream-ingestions.md
# Remnant Warrant:    https://corpusfetch.com/c/what-exact-cursor-offsets-must-be-persisted-when-halting-large-stream-ingestions.warrant.json
# Markdown SHA-256:   9894b4cae1fa57440bd719793d13947c7fe7978a4da3399c4ae5d32806183a0c
# Code AST SHA-256:   2325ae0f87355de20a86a90b16cb545c7e48d2ab6fa5c6ab04cc20dfc257f05c
# Verified Classes:   ResumptionVerdict, StreamCursorState, CursorManifest, StreamCursorBinder
# 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_RESUME = auto()
    REPLAY_GAP_REQUIRED = auto()
    UNRECOVERABLE_DIVERGENCE = auto()


@dataclass(frozen=True)
class StreamCursorState:
    source_partition_cursor: int
    sink_durable_commit_id: int
    uncommitted_buffer_hash: str | None
    idempotency_partition_key: str


@dataclass(frozen=True)
class CursorManifest:
    verdict: ResumptionVerdict
    resume_from_cursor: int
    replay_gap_size: int
    manifest_digest: str


class StreamCursorBinder:
    """
    Evaluates source read offsets against durable sink commit watermarks to
    ensure halting large stream ingestions does not cause data loss or duplication.
    """

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

    @classmethod
    def evaluate_halt_state(
        cls, state: StreamCursorState
    ) -> CursorManifest:
        
        if not state.idempotency_partition_key.strip():
            raise ValueError(
                "An idempotency partition key is required to safely replay the gap."
            )

        gap = state.source_partition_cursor - state.sink_durable_commit_id

        # Gate 1: Cursors are perfectly aligned - safe to resume at source
        if gap == 0:
            return CursorManifest(
                verdict=ResumptionVerdict.ALIGNED_RESUME,
                resume_from_cursor=state.source_partition_cursor,
                replay_gap_size=0,
                manifest_digest=cls._hash(f"ALIGNED:{state.source_partition_cursor}"),
            )

        # Gate 2: Sink is ahead of source (anomalous, possibly rollback) - halt
        if gap < 0:
            return CursorManifest(
                verdict=ResumptionVerdict.UNRECOVERABLE_DIVERGENCE,
                resume_from_cursor=state.sink_durable_commit_id,
                replay_gap_size=0,
                manifest_digest=cls._hash(f"NEGATIVE_GAP:{gap}"),
            )

        # Gate 3: Source is ahead of sink (normal halt with uncommitted buffer)
        # The successor must resume from the sink commit and replay the gap.
        if not state.uncommitted_buffer_hash:
            raise ValueError(
                "Source is ahead of sink but no uncommitted_buffer_hash was provided. "
                "Cannot guarantee gap integrity."
            )

        digest_payload = {
            "source_cursor": str(state.source_partition_cursor),
            "sink_commit": str(state.sink_durable_commit_id),
            "gap": str(gap),
            "buffer_hash": state.uncommitted_buffer_hash,
            "idempotency_key": state.idempotency_partition_key,
        }
        
        digest = cls._hash(
            json.dumps(digest_payload, sort_keys=True, separators=(",", ":"))
        )

        return CursorManifest(
            verdict=ResumptionVerdict.REPLAY_GAP_REQUIRED,
            resume_from_cursor=state.sink_durable_commit_id,
            replay_gap_size=gap,
            manifest_digest=digest,
        )

    @classmethod
    def verify_successor_resumption(
        cls, manifest: CursorManifest, successor_requested_cursor: int
    ) -> tuple[bool, str]:
        """
        Fetching-time check: ensures the successor does not blindly resume
        from a cursor ahead of the durable sink commit.
        """
        if manifest.verdict is ResumptionVerdict.UNRECOVERABLE_DIVERGENCE:
            return False, "HALT: Sink commit is ahead of source cursor. Manual audit required."
            
        if successor_requested_cursor > manifest.resume_from_cursor:
            return False, (
                f"CURSOR_SKIP_PREVENTED: Successor requested {successor_requested_cursor}, "
                f"but durable sink is only at {manifest.resume_from_cursor}. "
                f"Must replay gap of {manifest.replay_gap_size} records."
            )

        return True, f"RESUME_ADMISSIBLE: Starting from {successor_requested_cursor}."