# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-silent-sse-resumption-divergence-in-mcp-http-native-streamable-transport.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-silent-sse-resumption-divergence-in-mcp-http-native-streamable-transport.warrant.json
# Markdown SHA-256:   83f8f5d539def046749eebea96b7b778d7cbe2182cc11303f6a4f23e3fccb797
# Code AST SHA-256:   24fd91321b36c3d09e3c3b3faf1b26e6972e7f042893a18bb6cbfebba37bf12f
# Verified Classes:   SSEStreamDivergenceError, StreamCheckpoint
# Defensive Guards:   5 fail-closed invariants (zero external deps)

import hashlib
from dataclasses import dataclass


class SSEStreamDivergenceError(RuntimeError):
    """Raised when an SSE reconnect cannot be proven to preserve session continuity."""
    pass


@dataclass
class StreamCheckpoint:
    session_id: str
    last_processed_id: int
    last_event_digest: str


def _digest(event_id: int, payload: bytes) -> str:
    return hashlib.sha256(f"{event_id}:".encode() + payload).hexdigest()


def validate_resumption(
    checkpoint: StreamCheckpoint,
    reconnect_session_id: str,
    first_event_id: int,
    first_event_payload: bytes,
    server_replay_epoch: str | None,
    client_replay_epoch: str | None,
) -> None:
    """
    Validate the first event received after an SSE reconnect.

    Raises SSEStreamDivergenceError if the stream cannot be proven continuous.
    Returns None only when the stream is provisionally continuous.
    """
    if reconnect_session_id != checkpoint.session_id:
        raise SSEStreamDivergenceError(
            f"session id changed on reconnect: "
            f"expected {checkpoint.session_id!r}, got {reconnect_session_id!r}"
        )

    if server_replay_epoch != client_replay_epoch:
        raise SSEStreamDivergenceError(
            f"replay epoch mismatch: client={client_replay_epoch!r}, "
            f"server={server_replay_epoch!r}; replay buffer was reset"
        )

    expected = checkpoint.last_processed_id + 1
    if first_event_id <= checkpoint.last_processed_id:
        raise SSEStreamDivergenceError(
            f"replay reset: first event id {first_event_id} <= "
            f"last processed {checkpoint.last_processed_id}"
        )
    if first_event_id > expected:
        raise SSEStreamDivergenceError(
            f"event gap on resume: expected id {expected}, got {first_event_id}; "
            f"events {expected}..{first_event_id - 1} were lost"
        )

    # id matches; verify the server replayed the *same* event we last saw
    # only if the server claims to replay the boundary event. If it does not
    # replay the boundary, we cannot cross-check content and must rely on
    # monotonicity alone -- which is why the epoch check above is mandatory.
    if _digest(first_event_id, first_event_payload) == checkpoint.last_event_digest:
        # server replayed the last event we already processed; caller must dedup.
        return

    # Provisionally continuous: id == expected and epoch matched.
    return


def resume_stream(
    checkpoint: StreamCheckpoint,
    reconnect_session_id: str,
    first_event_id: int,
    first_event_payload: bytes,
    server_replay_epoch: str | None,
    client_replay_epoch: str | None,
) -> None:
    try:
        validate_resumption(
            checkpoint,
            reconnect_session_id,
            first_event_id,
            first_event_payload,
            server_replay_epoch,
            client_replay_epoch,
        )
    except SSEStreamDivergenceError:
        # Caller must re-initialize the session rather than continue.
        raise