# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-a-torn-sse-event-boundary-when-an-mcp-http-native-transport-resumes-a-session-mid-stream.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-a-torn-sse-event-boundary-when-an-mcp-http-native-transport-resumes-a-session-mid-stream.warrant.json
# Markdown SHA-256:   1c2f039994a1c54279d691a9323eba05fae9105db740e453c4059b5dc8095fa2
# Code AST SHA-256:   84d14100d36b8d23e9e9c5ff13a856a4c21752691cf3ebc678b8ac8bdd2debae
# Verified Classes:   TornStreamBoundaryError, ResumableStreamGuard
# Defensive Guards:   5 fail-closed invariants (zero external deps)

import json
from typing import Optional


class TornStreamBoundaryError(RuntimeError):
    """Raised when a resumed MCP SSE stream cannot be safely continued."""


class ResumableStreamGuard:
    def __init__(self) -> None:
        self.last_parsed_id: Optional[int] = None
        self.partial_data: str = ""

    def on_event(self, event_id: int, data: str) -> dict:
        # A resumed stream must begin exactly after the last fully parsed event.
        if self.last_parsed_id is not None and event_id <= self.last_parsed_id:
            raise TornStreamBoundaryError(
                f"replayed event id={event_id} <= last_parsed_id={self.last_parsed_id}; "
                "server restarted stream without acknowledging loss"
            )
        if self.partial_data:
            # Never append a resumed event onto an unflushed partial buffer.
            raise TornStreamBoundaryError(
                f"partial event buffer present ({len(self.partial_data)} bytes) "
                f"at event id={event_id}; refusing to append"
            )
        try:
            message = json.loads(data)
        except json.JSONDecodeError as exc:
            # Hold the partial payload; do not advance last_parsed_id.
            self.partial_data = data
            raise TornStreamBoundaryError(
                f"event id={event_id} data is not complete JSON: {exc.msg}"
            ) from exc
        self.last_parsed_id = event_id
        return message

    def on_reconnect(self, requested_last_event_id: Optional[int],
                     server_echoed_id: Optional[int]) -> None:
        if self.partial_data:
            raise TornStreamBoundaryError(
                "reconnect while holding partial event; must renegotiate session"
            )
        if requested_last_event_id is not None and server_echoed_id != requested_last_event_id:
            raise TornStreamBoundaryError(
                f"server did not honor Last-Event-ID "
                f"(requested={requested_last_event_id}, echoed={server_echoed_id}); "
                "resumption contract broken"
            )