# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-replay-divergence-between-a-resumable-mcp-stream-and-its-originating-session-state.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-replay-divergence-between-a-resumable-mcp-stream-and-its-originating-session-state.warrant.json
# Markdown SHA-256:   532be0fc4bc09cdbdcbe28f80974520877cc2a786c0818146dc9db0bda4b54b4
# Code AST SHA-256:   5f7411da843edf825812178a8761849549f0dc3ef014600faddabb6d9db8e455
# Verified Classes:   StreamResumeDivergence, SessionEpoch
# Defensive Guards:   4 fail-closed invariants (zero external deps)

import json
from dataclasses import dataclass
from typing import Optional


class StreamResumeDivergence(RuntimeError):
    """Raised when a resumed MCP stream cannot be proven to belong to the same session epoch."""
    pass


@dataclass
class SessionEpoch:
    session_id: str
    epoch: int  # server-assigned, incremented on every session (re-)creation


def validate_resumed_event(
    observed_epoch: Optional[SessionEpoch],
    first_resumed_event: dict,
    last_seen_event_id: int,
) -> None:
    """
    Call on the FIRST event received after a Last-Event-ID reconnect.
    Raises StreamResumeDivergence if the resume cannot be trusted.
    """
    if observed_epoch is None:
        # No epoch captured at initialize: resumption across a session-lifetime
        # boundary is unprovable. Refuse rather than risk a silent splice.
        raise StreamResumeDivergence(
            "no session epoch captured at initialize; "
            "cannot validate resumed stream across session boundary"
        )

    event_epoch = first_resumed_event.get("_meta", {}).get("session_epoch")
    if event_epoch is None:
        raise StreamResumeDivergence(
            f"resumed event {first_resumed_event.get('id')} carries no session_epoch; "
            "server does not support epoch-tagged resumption"
        )

    if event_epoch != observed_epoch.epoch:
        raise StreamResumeDivergence(
            f"epoch mismatch on resume: expected {observed_epoch.epoch}, "
            f"got {event_epoch}; session was re-created while stream was in flight"
        )

    event_id = int(first_resumed_event["id"])
    if event_id != last_seen_event_id + 1:
        raise StreamResumeDivergence(
            f"event id gap on resume: last seen {last_seen_event_id}, "
            f"first resumed {event_id}; buffer was evicted or reordered"
        )


def resume_or_restart(
    observed_epoch: Optional[SessionEpoch],
    first_resumed_event: dict,
    last_seen_event_id: int,
) -> str:
    try:
        validate_resumed_event(observed_epoch, first_resumed_event, last_seen_event_id)
        return "resume"
    except StreamResumeDivergence:
        # Loud, recoverable: re-initialize and re-issue original request.
        return "restart"