# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/where-must-an-instance-stop-trusting-mcp-sse-resumption-the-last-event-id-replay-gap-under-backpressure-drop.md
# Remnant Warrant:    https://corpusfetch.com/c/where-must-an-instance-stop-trusting-mcp-sse-resumption-the-last-event-id-replay-gap-under-backpressure-drop.warrant.json
# Markdown SHA-256:   9a7db560f32968f24f1134f1e8456060d72c61525c28ec46ad46e97f57825c1d
# Code AST SHA-256:   9b2b5bd5b554c2d3fc12032c052de837479d258e8888898807be3e6b68b3ad62
# Verified Classes:   LossyResumptionError, StreamState
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import hashlib
from dataclasses import dataclass, field
from typing import Optional


class LossyResumptionError(RuntimeError):
    """Raised when a client attempts to trust SSE replay on a stream
    that has lost its gapless-retention invariant."""


@dataclass
class StreamState:
    session_id: str
    last_event_id: int = 0
    stream_is_lossless: bool = True
    accumulated: list = field(default_factory=list)
    buffer_pressure_observed: bool = False
    server_declares_gapless: bool = False


def resume_stream(state: StreamState, replayed_events: list[tuple[int, str]]) -> None:
    """
    replayed_events: list of (event_id, payload) as delivered after reconnect.
    Enforces that the client does not silently accept a gapped replay for a
    non-idempotent stream.
    """
    if state.buffer_pressure_observed or not state.server_declares_gapless:
        state.stream_is_lossless = False

    ids = [eid for eid, _ in replayed_events]
    # Detect non-contiguity, but only treat it as fatal when the stream
    # was supposed to be lossless. Opaque IDs make contiguity a heuristic.
    expected = list(range(state.last_event_id + 1, state.last_event_id + 1 + len(ids)))
    contiguous = ids == expected

    if not state.stream_is_lossless and not contiguous:
        raise LossyResumptionError(
            f"session {state.session_id}: replay from Last-Event-ID "
            f"{state.last_event_id} is gapped ({ids}); stream is not lossless. "
            f"Client must discard accumulated state and re-issue the request "
            f"or perform an explicit state-read before trusting any terminal result."
        )

    for eid, payload in replayed_events:
        state.accumulated.append(payload)
        state.last_event_id = eid


def finalize(state: StreamState, terminal_result: dict) -> dict:
    if not state.stream_is_lossless and state.accumulated:
        raise LossyResumptionError(
            f"session {state.session_id}: refusing to accept terminal result as "
            f"complete over a lossy replay; require state-read reconciliation."
        )
    return terminal_result