# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-distinguish-a-resumable-mcp-sse-session-from-a-silently-re-established-streamable-http-session-after-transport-migration.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-distinguish-a-resumable-mcp-sse-session-from-a-silently-re-established-streamable-http-session-after-transport-migration.warrant.json
# Markdown SHA-256:   7db03418251843065a4fcfdeaebc3342861836665c735e46adb7b9dfaaf07321
# Code AST SHA-256:   d8abf2152bdc3a0bd4e49c74db823cf93321a6e8da167d31f066789b9e15f2cc
# Verified Classes:   ResumptionGapError, StreamCursor
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass


class ResumptionGapError(RuntimeError):
    """Raised when a resumed MCP stream cannot be proven continuous."""


@dataclass(frozen=True)
class StreamCursor:
    session_id: str
    last_event_id: int


def verify_resumption(
    cursor: StreamCursor,
    response_session_id: str,
    first_replayed_event_id: int | None,
) -> None:
    # Session identity must survive the transport generation change.
    if response_session_id != cursor.session_id:
        raise ResumptionGapError(
            f"session id changed on reconnect: had {cursor.session_id!r}, "
            f"got {response_session_id!r}; resumption is unproven, re-initialize"
        )

    # A stream that opens with no replay is ambiguous: it may mean
    # "nothing pending" or "replay unsupported". Refuse to guess.
    if first_replayed_event_id is None:
        raise ResumptionGapError(
            "no replayed event on resume; cannot distinguish caught-up "
            "from replay-unsupported, refuse to treat as continuous"
        )

    # Event IDs must be monotonic across the resume point. A lower ID
    # means the server's counter reset, i.e. a silent gap.
    if first_replayed_event_id < cursor.last_event_id:
        raise ResumptionGapError(
            f"event id regression: requested >= {cursor.last_event_id}, "
            f"got {first_replayed_event_id}; counter reset implies gap"
        )


def resume_stream(cursor: StreamCursor, response_session_id: str, first_event_id: int | None) -> None:
    verify_resumption(cursor, response_session_id, first_event_id)