# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/in-what-scenarios-must-an-instance-treat-mcp-http-native-transport-session-resumption-as-unsafe-after-a-mid-stream-last-event-id-gap.md
# Remnant Warrant:    https://corpusfetch.com/c/in-what-scenarios-must-an-instance-treat-mcp-http-native-transport-session-resumption-as-unsafe-after-a-mid-stream-last-event-id-gap.warrant.json
# Markdown SHA-256:   e1986f357b6fa220255cf4298435c23ace949d494701fc98ce30d1c03b8947de
# Code AST SHA-256:   a6f485a14272c5ab710838a1a9fb86dc18981e1b07cb94a933d671377b2fc2c4
# Verified Classes:   McpResumptionGapError
# Defensive Guards:   4 fail-closed invariants (zero external deps)

class McpResumptionGapError(RuntimeError):
    """Raised when an MCP HTTP-native stream resumes a session but cannot
    prove event-ID continuity, so incremental delivery is untrustworthy."""
    pass


def validate_resumption(
    session_id: str | None,
    presented_last_event_id: str | None,
    first_event_id_after_reconnect: str | None,
    server_replay_marker: str | None = None,
) -> None:
    """Validate that a reconnected MCP stream honored the requested replay point.

    Raises McpResumptionGapError if the session is valid but event continuity
    cannot be proven; caller must fall back to full state re-derivation.
    """
    if session_id is None:
        raise McpResumptionGapError("no Mcp-Session-Id; cannot resume, start new session")

    if presented_last_event_id is None:
        # Fresh stream, nothing to replay; continuity is vacuously fine.
        return

    if server_replay_marker is not None:
        # Server explicitly signaled where replay begins; trust it.
        return

    if first_event_id_after_reconnect is None:
        raise McpResumptionGapError(
            f"session {session_id}: stream opened but no event ID observed; "
            "cannot verify continuity"
        )

    expected_successor = _increment_event_id(presented_last_event_id)
    if first_event_id_after_reconnect != expected_successor:
        raise McpResumptionGapError(
            f"session {session_id}: presented Last-Event-ID {presented_last_event_id}, "
            f"expected successor {expected_successor}, got {first_event_id_after_reconnect}; "
            "replay buffer likely evicted the gap — re-derive state"
        )


def _increment_event_id(event_id: str) -> str:
    # Event IDs are opaque strings; only the numeric-suffix convention is
    # validated here. Non-conforming IDs must be treated as unverifiable.
    prefix, _, suffix = event_id.rpartition("-")
    if not suffix.isdigit():
        raise McpResumptionGapError(
            f"event ID {event_id!r} not in verifiable numeric-suffix form"
        )
    return f"{prefix}-{int(suffix) + 1:06d}"