# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-distinguish-sse-stream-resumption-from-session-resumption-in-mcp-http-native-transport.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-distinguish-sse-stream-resumption-from-session-resumption-in-mcp-http-native-transport.warrant.json
# Markdown SHA-256:   548491a46a4f00d88e23c7c8d19827f3489b01076010aae7e978ec3cd2a7eb2f
# Code AST SHA-256:   edd3f148db330e64d9ca33962196afea9f8b6fb6874713e0966f0be539d73d8f
# Verified Classes:   ResumptionBoundaryError, StreamEventLog
# Defensive Guards:   5 fail-closed invariants (zero external deps)

class ResumptionBoundaryError(RuntimeError):
    """Raised when Last-Event-ID and Mcp-Session-Id are used inconsistently."""
    pass


class StreamEventLog:
    def __init__(self):
        # session_id -> stream_id -> list of (event_id, payload)
        self._events = {}
        # session_id -> list of pending server-to-client requests
        self._pending = {}

    def register_session(self, session_id):
        self._events.setdefault(session_id, {})
        self._pending.setdefault(session_id, [])

    def append(self, session_id, stream_id, event_id, payload):
        self._events[session_id].setdefault(stream_id, []).append((event_id, payload))

    def queue_server_request(self, session_id, request):
        self._pending[session_id].append(request)

    def resume(self, session_id, stream_id, last_event_id):
        if session_id not in self._events:
            raise ResumptionBoundaryError(
                f"unknown session {session_id!r}; cannot resume stream"
            )

        if last_event_id is not None:
            # Stream resumption: replay only, do not touch pending requests.
            streams = self._events[session_id]
            if stream_id not in streams:
                raise ResumptionBoundaryError(
                    f"Last-Event-ID {last_event_id!r} presented for unknown stream "
                    f"{stream_id!r} in session {session_id!r}; refusing global replay"
                )
            log = streams[stream_id]
            ids = [eid for eid, _ in log]
            if last_event_id not in ids:
                raise ResumptionBoundaryError(
                    f"Last-Event-ID {last_event_id!r} not valid for stream "
                    f"{stream_id!r} in session {session_id!r}"
                )
            idx = ids.index(last_event_id)
            return {"replay": [p for _, p in log[idx + 1:]], "reemit": []}

        # Session resumption on a new stream: re-emit pending requests, no replay.
        reemit = list(self._pending[session_id])
        self._pending[session_id] = []
        return {"replay": [], "reemit": reemit}


def handle_reconnect(log, headers):
    session_id = headers.get("Mcp-Session-Id")
    last_event_id = headers.get("Last-Event-ID")
    stream_id = headers.get("X-Stream-Id")

    if last_event_id is not None and session_id is None:
        raise ResumptionBoundaryError(
            "Last-Event-ID supplied without Mcp-Session-Id; event IDs are "
            "stream-scoped and cannot be interpreted without a session"
        )

    if session_id is None:
        raise ResumptionBoundaryError(
            "no Mcp-Session-Id; client must establish a session before resuming"
        )

    return log.resume(session_id, stream_id, last_event_id)