# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-streamable-session-resumption-from-silently-forking-an-mcp-session.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-streamable-session-resumption-from-silently-forking-an-mcp-session.warrant.json
# Markdown SHA-256:   d51c3f4123c8c754ee347ed73e73c0d20a1953b24515627d6425c4a4e3dbacb3
# Code AST SHA-256:   b352a4fc0df201728b62e104a85a9a334a2c7954b76c4099981f243b51f1b9fc
# Verified Classes:   StreamResumptionError, StreamState, Session
# Defensive Guards:   5 fail-closed invariants (zero external deps)

import time
from dataclasses import dataclass, field
from typing import Dict, Optional


class StreamResumptionError(RuntimeError):
    """Raised when a resume cannot be proven gap-free on the same stream identity."""
    pass


@dataclass
class StreamState:
    stream_id: str
    high_water: int = -1          # last event id emitted on this stream
    fenced: bool = False
    last_emit_ts: float = 0.0


@dataclass
class Session:
    session_id: str
    streams: Dict[str, StreamState] = field(default_factory=dict)
    active_stream_id: Optional[str] = None
    grace_seconds: float = 30.0


def resume_stream(
    session: Session,
    resumption_token: Optional[str],
    last_event_id: Optional[int],
    now: Optional[float] = None,
) -> StreamState:
    """
    Honor a resume only if token and Last-Event-ID co-validate to the same,
    non-fenced stream and the server has not advanced past the client's cursor.
    Otherwise raise; never silently mint a new stream.
    """
    now = time.time() if now is None else now

    if resumption_token is None or last_event_id is None:
        # Cannot prove identity or gap-freedom -> require a new session.
        raise StreamResumptionError(
            "resume requires both resumption_token and Last-Event-ID; "
            "treat reconnect as a new session request"
        )

    stream = session.streams.get(resumption_token)
    if stream is None:
        raise StreamResumptionError(
            f"unknown stream identity {resumption_token!r}; refusing to fork session"
        )

    if stream.fenced:
        raise StreamResumptionError(
            f"stream {resumption_token!r} already fenced; a new stream would fork the session"
        )

    if last_event_id > stream.high_water:
        raise StreamResumptionError(
            f"client cursor {last_event_id} ahead of server high-water "
            f"{stream.high_water} on stream {resumption_token!r}"
        )

    # Fence any other live stream so a late background emit cannot land
    # on a stream the client has abandoned.
    for sid, other in session.streams.items():
        if sid != resumption_token and not other.fenced:
            if now - other.last_emit_ts <= session.grace_seconds:
                # Orphaned stream still inside grace window: do not resume
                # into a state where two streams are live.
                raise StreamResumptionError(
                    f"orphaned stream {sid!r} still live within grace window; "
                    "resume would create two active streams"
                )
            other.fenced = True

    session.active_stream_id = resumption_token
    return stream