# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-happens-when-an-instance-resumes-an-mcp-http-native-transport-stream-against-a-migrated-session.md
# Remnant Warrant:    https://corpusfetch.com/c/what-happens-when-an-instance-resumes-an-mcp-http-native-transport-stream-against-a-migrated-session.warrant.json
# Markdown SHA-256:   ee615332c1762ebc8b24f715420d3e777fd8e9aa5cbba16a721c1e4fbb943de8
# Code AST SHA-256:   31789e97dd281cd8e8a7075ddf8aea7e8b2a7dcb81f0e0622af151618e158046
# Verified Classes:   StreamResumeInvariantError, EventCursor, SessionRecord
# Defensive Guards:   2 fail-closed invariants (zero external deps)

class StreamResumeInvariantError(RuntimeError):
    """Raised when a resume request cannot satisfy the gap-free replay invariant."""
    pass


class EventCursor:
    def __init__(self, producer_epoch: str, last_event_id: int):
        self.producer_epoch = producer_epoch
        self.last_event_id = last_event_id


class SessionRecord:
    def __init__(self, session_id: str, producer_epoch: str, buffer_epochs: set[str]):
        self.session_id = session_id
        self.producer_epoch = producer_epoch      # instance that owns the live stream
        self.buffer_epochs = buffer_epochs        # epochs whose events we can actually replay


def resume_stream(
    session: SessionRecord,
    cursor: EventCursor,
    local_instance_epoch: str,
    shared_event_log: bool,
) -> int:
    """
    Decide whether a Last-Event-ID resume can be honored.

    Returns the event ID to replay from. Raises if the invariant cannot hold.
    """
    if cursor.producer_epoch not in session.buffer_epochs:
        # We have no events from the producer that emitted this cursor.
        # Replaying from 0 or from the current cursor both violate the invariant.
        raise StreamResumeInvariantError(
            f"session {session.session_id}: cursor epoch {cursor.producer_epoch!r} "
            f"not in replayable set {sorted(session.buffer_epochs)}; "
            f"resume is unanswerable, client must re-establish session"
        )

    if cursor.producer_epoch != local_instance_epoch and not shared_event_log:
        # Foreign producer, local-only buffer: we cannot guarantee gap-free replay.
        raise StreamResumeInvariantError(
            f"session {session.session_id}: cursor owned by {cursor.producer_epoch!r}, "
            f"local instance is {local_instance_epoch!r} with no shared event log"
        )

    # Safe: either same producer with local buffer, or shared log with global order.
    return cursor.last_event_id + 1


def handle_resume(session: SessionRecord, cursor: EventCursor,
                  local_instance_epoch: str, shared_event_log: bool) -> str:
    try:
        next_id = resume_stream(session, cursor, local_instance_epoch, shared_event_log)
    except StreamResumeInvariantError as exc:
        # Emit an explicit, distinguishable failure — never a silent fresh stream.
        return f"event: error\ndata: {{\"code\":\"stream_cursor_unrecoverable\",\"detail\":{exc!r}}}\n\n"
    return f"event: resume\ndata: {{\"replay_from\":{next_id}}}\n\n"