# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/in-what-scenarios-must-an-instance-treat-mcp-streamable-http-session-resumption-as-a-re-attachment-to-a-logical-stream-rather-than-a-replay-of-transpo.md
# Remnant Warrant:    https://corpusfetch.com/c/in-what-scenarios-must-an-instance-treat-mcp-streamable-http-session-resumption-as-a-re-attachment-to-a-logical-stream-rather-than-a-replay-of-transpo.warrant.json
# Markdown SHA-256:   4aacf07dc88782e6d3cd8b873c9fadcf813d23316436e474c9e60f6afe61ee27
# Code AST SHA-256:   76ccb1ce036fae594c07f101f0e1be42271c7980588f7d6aea4e100e3fc84063
# Verified Classes:   ResumptionCursorError, SessionStream
# Defensive Guards:   7 fail-closed invariants (zero external deps)

import time
from dataclasses import dataclass, field


class ResumptionCursorError(RuntimeError):
    """Raised when a resumption request cannot be safely satisfied."""


@dataclass
class SessionStream:
    session_id: str
    window_size: int = 256
    _events: list = field(default_factory=list)  # (event_id, monotonic_seq, payload)
    _seq: int = 0
    _closed: bool = False

    def emit(self, payload: dict) -> str:
        self._seq += 1
        event_id = f"{self.session_id}:{self._seq}"
        self._events.append((event_id, self._seq, payload))
        if len(self._events) > self.window_size:
            self._events.pop(0)
        return event_id

    def reattach(self, last_event_id: str | None) -> list:
        """
        Return events strictly after last_event_id, or raise if the cursor is
        outside the retained window. Never replay from the beginning.
        """
        if last_event_id is None:
            # No cursor: client is attaching fresh. Do not replay history.
            return []
        # Parse the cursor. It must belong to this session.
        try:
            cursor_session, cursor_seq_str = last_event_id.rsplit(":", 1)
            cursor_seq = int(cursor_seq_str)
        except (ValueError, AttributeError) as exc:
            raise ResumptionCursorError(
                f"malformed Last-Event-ID {last_event_id!r} for session {self.session_id}"
            ) from exc
        if cursor_session != self.session_id:
            raise ResumptionCursorError(
                f"cursor session {cursor_session!r} does not match {self.session_id!r}"
            )
        if not self._events:
            raise ResumptionCursorError(
                f"session {self.session_id} has no retained events; cannot reattach"
            )
        oldest_seq = self._events[0][1]
        newest_seq = self._events[-1][1]
        if cursor_seq < oldest_seq - 1:
            # Cursor has fallen out of the retention window. This is a
            # resumption FAILURE, not a cue to replay from the start.
            raise ResumptionCursorError(
                f"cursor seq {cursor_seq} older than window (oldest={oldest_seq}); "
                f"client must re-initialize session {self.session_id}"
            )
        if cursor_seq > newest_seq:
            raise ResumptionCursorError(
                f"cursor seq {cursor_seq} ahead of server seq {newest_seq}; "
                f"possible session id reuse"
            )
        return [payload for (_eid, seq, payload) in self._events if seq > cursor_seq]


def handle_resume(stream: SessionStream, last_event_id: str | None, now: float) -> list:
    """
    Entry point for a POST /mcp resumption request. Enforces a bounded wait:
    if the client has stalled beyond the idle budget, refuse to buffer.
    """
    idle_budget_s = 30.0
    if stream._closed:
        raise ResumptionCursorError(f"session {stream.session_id} is closed")
    if now - getattr(stream, "_last_activity", now) > idle_budget_s:
        raise ResumptionCursorError(
            f"session {stream.session_id} idle beyond {idle_budget_s}s; "
            f"backpressure policy requires re-initialization"
        )
    pending = stream.reattach(last_event_id)
    stream._last_activity = now
    return pending