# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-structural-signals-indicate-a-resumable-mcp-stream-versus-a-silently-forked-session-in-http-native-transport.md
# Remnant Warrant:    https://corpusfetch.com/c/what-structural-signals-indicate-a-resumable-mcp-stream-versus-a-silently-forked-session-in-http-native-transport.warrant.json
# Markdown SHA-256:   210c47b0ed1b293190c5d5cdc71c549b039cd4c6d6fd3347e93bcead41a20a86
# Code AST SHA-256:   10f547c12999c91a2d17f1009e7b9eb2fa30234cec2995e4d6c54377e982cb4e
# Verified Classes:   StreamForkDetected
# Defensive Guards:   2 fail-closed invariants (zero external deps)

class StreamForkDetected(RuntimeError):
    """Raised when a resumed MCP SSE stream is not a continuation of the session log."""
    pass


def validate_resumption(first_event_id, last_consumed_id):
    """
    Enforce the session-resume invariant on the first frame of a reconnected stream.
    first_event_id: id of the first SSE event after reconnect (None if absent).
    last_consumed_id: highest event id the client has already processed (None if none).
    """
    if first_event_id is None:
        # No id on a resumed stream: cannot prove continuity -> treat as fork.
        raise StreamForkDetected(
            "resumed stream emitted an event without an id; "
            "cannot verify session-log continuity"
        )
    if last_consumed_id is not None and first_event_id <= last_consumed_id:
        raise StreamForkDetected(
            f"event id regressed: first={first_event_id} "
            f"<= last_consumed={last_consumed_id}; "
            "server opened a fresh event log under the same Mcp-Session-Id"
        )
    return True


def attach_resumed_stream(session, sse):
    """Attach to a reconnected stream only if it is a true continuation."""
    first = sse.next_event()
    validate_resumption(first.event_id, session.last_consumed_id)
    session.ingest(first)
    return session