# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-mcp-http-native-instance-invalidate-a-resumption-token-after-last-event-id-replay.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-mcp-http-native-instance-invalidate-a-resumption-token-after-last-event-id-replay.warrant.json
# Markdown SHA-256:   dc345932b96eaa45f4d9048fdc8454091cb7eca2033b98e19d98d41187af486d
# Code AST SHA-256:   79bcc4900c07dc22182b37053514dee885890eff58db26e5cfdae93d2582ae2a
# Verified Classes:   ResumptionTokenError, Session
# Defensive Guards:   4 fail-closed invariants (zero external deps)

class ResumptionTokenError(RuntimeError):
    """Raised when a Last-Event-ID replay request violates the replay invariant."""
    pass


class Session:
    def __init__(self, token: str, generation: int, replay_floor: int, high_water_mark: int):
        self.token = token
        self.generation = generation
        self.replay_floor = replay_floor
        self.high_water_mark = high_water_mark
        self.evicted = False


def authorize_replay(
    session: Session | None,
    requested_generation: int,
    last_event_id: int,
) -> None:
    if session is None or session.evicted:
        raise ResumptionTokenError(
            f"session token unknown or evicted; client must re-initialize"
        )
    if requested_generation != session.generation:
        raise ResumptionTokenError(
            f"generation mismatch: token generation {requested_generation} "
            f"!= live generation {session.generation}; refusing cross-generation replay"
        )
    if last_event_id < session.replay_floor:
        raise ResumptionTokenError(
            f"Last-Event-ID {last_event_id} below replay floor {session.replay_floor}; "
            f"events compacted, force re-initialize"
        )
    if last_event_id > session.high_water_mark:
        raise ResumptionTokenError(
            f"Last-Event-ID {last_event_id} ahead of high-water mark "
            f"{session.high_water_mark}; refusing to clamp"
        )
    # valid: replay events (last_event_id, high_water_mark] under this generation