# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-structural-signals-indicate-mcp-http-native-transport-streaming-session-resumption-has-silently-degraded-into-replay-from-zero-under-backpressure.md
# Remnant Warrant:    https://corpusfetch.com/c/what-structural-signals-indicate-mcp-http-native-transport-streaming-session-resumption-has-silently-degraded-into-replay-from-zero-under-backpressure.warrant.json
# Markdown SHA-256:   67273eebdfe658dff6f0f17813cfed51d5977c8d5e533309409f85ef63a43721
# Code AST SHA-256:   fbee2756a1393eeff385e591d51f060b9c5a7c35240d849a5b0558f224bed5f0
# Verified Classes:   ResumeIntegrityError, ResumeVerdict, ResumeDecision
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass
from enum import Enum


class ResumeIntegrityError(RuntimeError):
    """Raised when a resumed MCP HTTP-native stream fails the structural check."""


class ResumeVerdict(Enum):
    CORRECT = "correct"
    GAP_SKIPPED = "gap_skipped"
    REPLAYED = "replayed"


@dataclass(frozen=True)
class ResumeDecision:
    verdict: ResumeVerdict
    last_id: int
    first_id: int


def classify_resume(last_event_id: int, first_event_id: int) -> ResumeDecision:
    """Classify a resumed stream by its first event ID against the sent Last-Event-ID.

    The only reliable structural signal that a resume is correct is
    first_event_id == last_event_id + 1. Anything else means the server
    silently replayed, skipped, or reset the session-scoped event stream.
    """
    if first_event_id == last_event_id + 1:
        return ResumeDecision(ResumeVerdict.CORRECT, last_event_id, first_event_id)
    if first_event_id > last_event_id + 1:
        return ResumeDecision(ResumeVerdict.GAP_SKIPPED, last_event_id, first_event_id)
    return ResumeDecision(ResumeVerdict.REPLAYED, last_event_id, first_event_id)


def guard_resumed_stream(last_event_id: int, first_event_id: int) -> None:
    """Abort before dispatching any event if the resume is not provably correct.

    Must be called with the first event of the resumed stream, before it is
    handed to the MCP session layer. Once dispatched, corruption is in flight.
    """
    decision = classify_resume(last_event_id, first_event_id)

    if decision.verdict is ResumeVerdict.GAP_SKIPPED:
        raise ResumeIntegrityError(
            f"resume gap: sent Last-Event-ID={decision.last_id} but server "
            f"resumed at id={decision.first_id}; events in "
            f"({decision.last_id}, {decision.first_id}) were lost"
        )

    if decision.verdict is ResumeVerdict.REPLAYED:
        raise ResumeIntegrityError(
            f"silent replay: sent Last-Event-ID={decision.last_id} but server "
            f"restarted stream at id={decision.first_id}; session-scoped state "
            f"would be re-applied or duplicated"
        )

    # Correct resume: safe to continue dispatching.