# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-treat-mcp-streamable-session-resumption-as-failed-rather-than-merely-delayed-sse-backpressure-and-last-event-id-continuity-for-l.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-treat-mcp-streamable-session-resumption-as-failed-rather-than-merely-delayed-sse-backpressure-and-last-event-id-continuity-for-l.warrant.json
# Markdown SHA-256:   5ae4ed1300f3fb76c2718d8986aac81ad76d0d553344410033495c1b48d8bdf9
# Code AST SHA-256:   3efd14ed2085a5d51b06851a178011bd727fe874246d814cffcafe2dda231f7e
# Verified Classes:   StreamResumptionError, Event, ReconnectResult
# Defensive Guards:   7 fail-closed invariants (zero external deps)

from __future__ import annotations

import dataclasses
from typing import Iterable, Optional


class StreamResumptionError(RuntimeError):
    """Raised when a reconnect cannot prove replay continuity from the client cursor."""


@dataclasses.dataclass(frozen=True)
class Event:
    id: int
    kind: str
    payload: dict


@dataclasses.dataclass(frozen=True)
class ReconnectResult:
    http_status: int
    replay_floor: Optional[int]  # lowest id the server can still replay, if declared
    first_event: Optional[Event]
    session_durable: bool


def verify_resumption(
    cursor: int,
    result: ReconnectResult,
    retained_floor: Optional[int] = None,
) -> Event:
    """
    Decide whether a reconnect is a valid resumption from `cursor`.

    Raises StreamResumptionError when the server cannot prove that the next
    event it delivers is the immediate successor of the client's cursor.
    """
    if result.http_status in (404, 410):
        raise StreamResumptionError(
            f"session lost: status={result.http_status}; re-initialize required"
        )
    if result.http_status != 200:
        raise StreamResumptionError(
            f"unexpected reconnect status={result.http_status}"
        )

    floor = result.replay_floor if result.replay_floor is not None else retained_floor
    if floor is not None and cursor < floor - 1:
        # Cursor is older than the retained window: the gap is unreplayable.
        raise StreamResumptionError(
            f"cursor {cursor} below replay floor {floor}; "
            f"events {cursor + 1}..{floor - 1} are lost"
        )

    if not result.session_durable:
        # Implicit re-creation under the same session id looks like a live stream
        # but has no continuity guarantee.
        raise StreamResumptionError(
            "session not durable across reconnect; implicit re-creation "
            "cannot honor Last-Event-ID"
        )

    ev = result.first_event
    if ev is None:
        raise StreamResumptionError(
            "reconnect returned an empty stream; cannot prove continuity"
        )
    if ev.id != cursor + 1:
        raise StreamResumptionError(
            f"replay gap: expected event id {cursor + 1}, got {ev.id}; "
            f"silent skip of {ev.id - cursor - 1} event(s)"
        )
    return ev


def drain_with_continuity(
    cursor: int,
    events: Iterable[Event],
    result: ReconnectResult,
) -> list[Event]:
    """Validate resumption, then drain the remainder, asserting gap-free ids."""
    first = verify_resumption(cursor, result)
    out = [first]
    expected = first.id + 1
    for ev in events:
        if ev.id != expected:
            raise StreamResumptionError(
                f"non-contiguous stream: expected {expected}, got {ev.id}"
            )
        out.append(ev)
        expected += 1
    return out