# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-treat-mcp-streamable-http-session-reattachment-as-a-new-logical-stream-rather-than-a-continuation.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-treat-mcp-streamable-http-session-reattachment-as-a-new-logical-stream-rather-than-a-continuation.warrant.json
# Markdown SHA-256:   f29adad583849e6e4aaf820ae231f3a496a18ccf60525f4894d8e3a4787e7a81
# Code AST SHA-256:   aec9e5a5aa8fc56c0b59584d912ee2b651d916d2393edf4f818715e45bc5f8ba
# Verified Classes:   InvalidResumptionError, StreamResumptionGuard
# Defensive Guards:   3 fail-closed invariants (zero external deps)

class InvalidResumptionError(RuntimeError):
    """Raised when a client attempts to resume a stream that is not resumable."""


class StreamResumptionGuard:
    """
    Guards the boundary between session reattachment and stream continuation.

    An event ID is a cursor into a stream, not proof the stream is resumable.
    Resumption is valid only if the request has not reached a terminal state
    and the server still holds the referenced cursor in a resumable stream.
    """

    def __init__(self):
        self._terminal_requests = set()
        self._last_event_id_by_request = {}

    def observe_event(self, request_id: str, event_id: str, terminal: bool) -> None:
        self._last_event_id_by_request[request_id] = event_id
        if terminal:
            self._terminal_requests.add(request_id)

    def build_resume_headers(self, request_id: str, session_id: str) -> dict:
        if request_id in self._terminal_requests:
            # The stream ended because the work completed. Replaying the
            # cursor would duplicate a terminal event under a new stream
            # identity. Do not attach Last-Event-ID.
            raise InvalidResumptionError(
                f"request {request_id!r} is terminal; refusing to resume via "
                f"Last-Event-ID (session {session_id!r}). Open a new request "
                f"with idempotency protection instead."
            )
        event_id = self._last_event_id_by_request.get(request_id)
        if event_id is None:
            raise InvalidResumptionError(
                f"request {request_id!r} has no observed event cursor; "
                f"cannot resume without a Last-Event-ID."
            )
        return {
            "Mcp-Session-Id": session_id,
            "Last-Event-ID": event_id,
        }

    def interpret_resume_response(
        self, request_id: str, status: int, body_empty: bool
    ) -> str:
        if status in (404, 409):
            return "buffer_evicted_retry_from_scratch"
        if status == 200 and body_empty:
            # Empty replay is NOT proof that no events were missed.
            if request_id in self._terminal_requests:
                return "terminal_already_observed_no_action"
            return "ambiguous_empty_replay_escalate_to_retry"
        if status == 200:
            return "continuation_valid"
        raise InvalidResumptionError(
            f"unexpected resume status {status} for request {request_id!r}"
        )