# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-reject-an-mcp-sse-last-event-id-resumption-that-arrives-on-a-different-tcp-connection-than-the-one-that-opened.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-reject-an-mcp-sse-last-event-id-resumption-that-arrives-on-a-different-tcp-connection-than-the-one-that-opened.warrant.json
# Markdown SHA-256:   1413457f20c3d8749ec962d39e2f7dda38f413eff1aeb702038ec9b733c41a01
# Code AST SHA-256:   af7672307da5236a1a531252d8afb80328eb648ade42293c87e496051c2c9a0b
# Verified Classes:   ResumptionRejectedError
# Defensive Guards:   6 fail-closed invariants (zero external deps)

class ResumptionRejectedError(RuntimeError):
    """Raised when an MCP SSE Last-Event-ID resumption cannot be honored
    with causal consistency against session state."""


def validate_resumption(
    session,
    last_event_id: int,
    new_uplink_id: str,
    *,
    fence_grace_seconds: float = 5.0,
) -> None:
    """Validate a resumption request before attaching the new uplink.

    Raises ResumptionRejectedError if the replay would be causally
    inconsistent with in-flight server-initiated requests, or if the
    prior uplink cannot be fenced. Caller must then invalidate the
    session and force the client to re-initialize.
    """
    if session is None or not session.is_live:
        raise ResumptionRejectedError("session token does not resolve to a live session")

    # 1. Fence the prior uplink before attaching the new writer.
    prior = session.active_uplink
    if prior is not None and prior.id != new_uplink_id:
        if not prior.fence(timeout=fence_grace_seconds):
            raise ResumptionRejectedError(
                f"prior uplink {prior.id} could not be fenced within "
                f"{fence_grace_seconds}s; refusing to race two writers"
            )

    # 2. Replay window must be fully retained.
    oldest, newest = session.event_buffer.retained_range()
    if last_event_id < oldest - 1:
        raise ResumptionRejectedError(
            f"Last-Event-ID {last_event_id} older than retained buffer (oldest={oldest})"
        )
    if last_event_id > newest:
        raise ResumptionRejectedError(
            f"Last-Event-ID {last_event_id} newer than last emitted ({newest})"
        )

    # 3. Replay window must not contain already-answered server-initiated requests.
    for event in session.event_buffer.range(last_event_id + 1, newest + 1):
        req_id = event.server_request_id
        if req_id is not None and session.requests[req_id].answered:
            raise ResumptionRejectedError(
                f"replay window contains response to already-answered request {req_id}"
            )

    # 4. Unanswered in-flight requests must be re-issuable or explicitly failed.
    for req_id, req in session.requests.items():
        if not req.answered and not req.reissuable:
            raise ResumptionRejectedError(
                f"in-flight request {req_id} is not reissuable; "
                "cannot resume without orphaning it"
            )

    # All conditions hold: caller may attach new_uplink_id and replay.