# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-distinguish-a-resumable-sse-stream-from-a-silently-forked-session-in-mcp-http-native-transport.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-distinguish-a-resumable-sse-stream-from-a-silently-forked-session-in-mcp-http-native-transport.warrant.json
# Markdown SHA-256:   7c421c29a328e865a7a591215cb6e5ac804c00a4fc18483ee59f159e5254ddb2
# Code AST SHA-256:   2b0379acccba6df6d82c03ed35bd47cc575267d0bdca323d202ce8cd5750537d
# Verified Classes:   SessionForkError, ResumableStream
# Defensive Guards:   4 fail-closed invariants (zero external deps)

class SessionForkError(RuntimeError):
    """Raised when a resume would silently fork a session instead of continuing it."""
    pass


class ResumableStream:
    def __init__(self, session_id: str, incarnation: str):
        self.session_id = session_id
        self.incarnation = incarnation
        self.last_event_seq = -1
        self.seen = set()  # keys are (session_id, incarnation, event_seq)

    def resume(self, server_session_id: str, server_incarnation: str,
               last_event_id: int, server_log_has: bool) -> None:
        if server_session_id != self.session_id:
            # Server minted a replacement under a different ID: fork.
            raise SessionForkError(
                f"resume targeted {server_session_id!r} but client holds "
                f"{self.session_id!r}; refusing to adopt forked session"
            )
        if server_incarnation != self.incarnation:
            raise SessionForkError(
                f"session {self.session_id!r} reincarnated "
                f"({self.incarnation!r} -> {server_incarnation!r}); "
                f"prior event log is gone, resume would replay stale IDs"
            )
        if not server_log_has:
            raise SessionForkError(
                f"Last-Event-ID {last_event_id} absent from server log for "
                f"session {self.session_id!r}; replay-from-start would be "
                f"dedup-dropped by client"
            )
        if last_event_id < self.last_event_seq:
            raise SessionForkError(
                f"resume event {last_event_id} precedes last-seen "
                f"{self.last_event_seq} on same session; sequence regression "
                f"indicates silent fork"
            )

    def accept_event(self, event_seq: int, payload: dict) -> bool:
        key = (self.session_id, self.incarnation, event_seq)
        if key in self.seen:
            return False
        self.seen.add(key)
        self.last_event_seq = max(self.last_event_seq, event_seq)
        return True