# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-treat-an-mcp-last-event-id-resumption-as-a-new-stream-rather-than-a-continuation-of-the-prior-sse-session.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-treat-an-mcp-last-event-id-resumption-as-a-new-stream-rather-than-a-continuation-of-the-prior-sse-session.warrant.json
# Markdown SHA-256:   417c2924027f1d8ba787af7eb43f407cfffe5c7fc2aa00229ea089abd449d234
# Code AST SHA-256:   d86f2fbf58900507bfc3d8605deb3d5c6d89c243c7f2dc69aa44a42942ddb237
# Verified Classes:   ResumeContinuityError, StreamIncarnation
# Defensive Guards:   4 fail-closed invariants (zero external deps)

class ResumeContinuityError(RuntimeError):
    """Raised when a Last-Event-ID resumption cannot be honored as a continuation."""
    pass


class StreamIncarnation:
    def __init__(self, stream_id: str, owner_replica: str, head_id: int):
        self.stream_id = stream_id
        self.owner_replica = owner_replica
        self.head_id = head_id
        # IDs actually present in the authoritative log for this incarnation.
        self.present_ids: set[int] = set()


def resume_stream(
    session_active: bool,
    last_event_id: int,
    incarnation: StreamIncarnation | None,
    current_replica: str,
) -> list[int]:
    """Return the events to replay, or raise if the cursor is not a valid continuation."""
    if not session_active:
        raise ResumeContinuityError(
            f"session inactive; Last-Event-ID={last_event_id} is a new stream, not a resume"
        )

    if incarnation is None:
        raise ResumeContinuityError(
            f"no stream incarnation owns Last-Event-ID={last_event_id}; refusing silent replay"
        )

    if incarnation.owner_replica != current_replica:
        raise ResumeContinuityError(
            f"Last-Event-ID={last_event_id} owned by replica "
            f"{incarnation.owner_replica!r}, not {current_replica!r}; "
            "redirect to owner or treat as new stream"
        )

    # Contiguity check: every ID from last_event_id+1 .. head must be present.
    expected = set(range(last_event_id + 1, incarnation.head_id + 1))
    missing = expected - incarnation.present_ids
    if missing:
        raise ResumeContinuityError(
            f"Last-Event-ID={last_event_id} has a gap; missing {sorted(missing)} "
            "between cursor and head — replaying would be lossy, treat as new stream"
        )

    return sorted(expected)