# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/where-must-an-instance-draw-the-line-between-sse-resumption-and-session-re-establishment-in-mcp-http-native-streamable-transport.md
# Remnant Warrant:    https://corpusfetch.com/c/where-must-an-instance-draw-the-line-between-sse-resumption-and-session-re-establishment-in-mcp-http-native-streamable-transport.warrant.json
# Markdown SHA-256:   a0dcbcbea256d8e80ed69a3fbcd35a0e0e132f3fa5c1e25d23a7a891969c96fa
# Code AST SHA-256:   8379601e2f2fd887394862e5d8ed1f450613c26824e3a901c0718b3d887a3511
# Verified Classes:   ResumptionBoundaryError
# Defensive Guards:   3 fail-closed invariants (zero external deps)

class ResumptionBoundaryError(RuntimeError):
    """Raised when a resumption request cannot be honored without fabricating continuity."""
    pass


def handle_resumption(session_store, event_buffer, session_id, last_event_id):
    session = session_store.get(session_id)
    if session is None:
        # Session missing: refuse. Do NOT open a stream with a fresh id namespace.
        raise ResumptionBoundaryError(
            f"session {session_id!r} not found; client must re-initialize"
        )

    incarnation = session.incarnation
    cursor = event_buffer.parse_cursor(last_event_id)
    if cursor is None:
        raise ResumptionBoundaryError(
            f"unparseable Last-Event-ID {last_event_id!r}"
        )

    if cursor.incarnation != incarnation:
        # Cursor belongs to a dead incarnation. Same id string, different session.
        raise ResumptionBoundaryError(
            f"cursor incarnation {cursor.incarnation!r} != live {incarnation!r}; "
            "refusing to replay across incarnation boundary"
        )

    if not event_buffer.covers(incarnation, cursor.seq):
        # Session is valid but the gap is unrecoverable. Must signal resync,
        # never silently restart the namespace.
        return ResyncStream(
            session_id=session_id,
            incarnation=incarnation,
            earliest_available=event_buffer.earliest_seq(incarnation),
            requested=cursor.seq,
        )

    return ReplayStream(
        session_id=session_id,
        incarnation=incarnation,
        events=event_buffer.since(incarnation, cursor.seq),
    )