# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-abandon-sse-resumption-and-re-establish-a-fresh-mcp-session-rather-than-replaying-from-last-event-id.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-abandon-sse-resumption-and-re-establish-a-fresh-mcp-session-rather-than-replaying-from-last-event-id.warrant.json
# Markdown SHA-256:   9da3787dd6f06dd834cb390f027e2dc4f9f8704cb4460a84241fe7180f1f7277
# Code AST SHA-256:   326d5796d6d7f111851ab2900ac711a9f486ac973b884d0cb299f4b25fd1f61a
# Verified Classes:   SessionResumeRejected, ResumeRequest, SessionRecord
# Defensive Guards:   5 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass


class SessionResumeRejected(RuntimeError):
    """Raised when a resumable-SSE resume cannot be honored safely.

    Signals the client to abandon Last-Event-ID replay, re-run the MCP
    initialize handshake, and re-dispatch only idempotent tool calls.
    """


@dataclass(frozen=True)
class ResumeRequest:
    session_id: str
    last_event_id: int
    resuming_instance: str
    originating_instance: str
    capability_hash: str


@dataclass(frozen=True)
class SessionRecord:
    session_id: str
    owner_instance: str
    capability_hash: str
    event_head: int
    event_ids: frozenset[int]
    resolvable_request_ids: frozenset[str]


def authorize_resume(
    req: ResumeRequest,
    record: SessionRecord | None,
    replayed_response_request_ids: frozenset[str],
) -> None:
    """Permit replay only if ownership, continuity, and context all hold.

    Raises SessionResumeRejected on any violation; never returns a
    partial success, because a partial replay is indistinguishable from
    a completed stream at the transport layer.
    """
    if record is None or record.session_id != req.session_id:
        raise SessionResumeRejected(
            f"unknown session {req.session_id!r} on instance "
            f"{req.resuming_instance!r}; fresh initialize required"
        )

    if record.owner_instance != req.resuming_instance:
        # Stateless server: a different instance cannot prove context.
        raise SessionResumeRejected(
            f"session {req.session_id!r} owned by {record.owner_instance!r}, "
            f"resume attempted on {req.resuming_instance!r}; "
            "stateless server must not replay across instances"
        )

    expected = set(range(req.last_event_id + 1, record.event_head + 1))
    missing = expected - record.event_ids
    if missing:
        raise SessionResumeRejected(
            f"event-log gap for session {req.session_id!r}: "
            f"missing ids {sorted(missing)[:5]}...; replay would be lossy"
        )

    orphaned = replayed_response_request_ids - record.resolvable_request_ids
    if orphaned:
        raise SessionResumeRejected(
            f"replayed responses reference unresolvable requests "
            f"{sorted(orphaned)}; dependent tool calls would be invalid"
        )

    if record.capability_hash != req.capability_hash:
        raise SessionResumeRejected(
            "negotiated capability hash mismatch; replayed events were "
            "produced under a different initialize context"
        )
    # All three invariants hold; replay is safe.