# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-treat-mcp-streamable-http-session-negotiation-as-a-resumability-contract-rather-than-a-connection-handshake.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-treat-mcp-streamable-http-session-negotiation-as-a-resumability-contract-rather-than-a-connection-handshake.warrant.json
# Markdown SHA-256:   80f0b5dede43db057926e532bce9cd5e86d6eeeec8f27eb4ab81fc2698d8678b
# Code AST SHA-256:   6b7ad6e95eda7dd8caa334851910ab6b9d1122580deb3b7ee03968e95fba0384
# Verified Classes:   SessionResumptionViolation, ResumeToken
# Defensive Guards:   6 fail-closed invariants (zero external deps)

import hashlib
from dataclasses import dataclass


class SessionResumptionViolation(RuntimeError):
    """Raised when a stream resume would violate the session/event co-issuance contract."""


@dataclass(frozen=True)
class ResumeToken:
    session_id: str
    last_event_id: str

    def fingerprint(self) -> str:
        h = hashlib.sha256()
        h.update(self.session_id.encode("utf-8"))
        h.update(b"\x00")
        h.update(self.last_event_id.encode("utf-8"))
        return h.hexdigest()


def validate_resume(
    persisted: ResumeToken,
    response_session_id: str | None,
    first_event_id: str | None,
    status_code: int,
) -> ResumeToken:
    # Terminal session loss must never be papered over with a fresh session.
    if status_code == 404:
        raise SessionResumptionViolation(
            f"session {persisted.session_id!r} is gone; invalidate world model, "
            "reconcile in-flight tool calls, do not auto-retry with a new session"
        )

    if status_code != 200:
        raise SessionResumptionViolation(
            f"unexpected status {status_code} on resume of {persisted.session_id!r}"
        )

    # 200 with a different (or absent) session id is the exact shape of replay corruption.
    if response_session_id is None:
        raise SessionResumptionViolation(
            "resume response carried no Mcp-Session-Id; cannot bind event namespace"
        )
    if response_session_id != persisted.session_id:
        raise SessionResumptionViolation(
            f"resume of {persisted.session_id!r} returned session "
            f"{response_session_id!r}; refusing to merge epochs"
        )

    if first_event_id is None:
        raise SessionResumptionViolation(
            "resumed stream produced no event id; cannot prove continuity"
        )
    if first_event_id == persisted.last_event_id:
        raise SessionResumptionViolation(
            f"resumed stream replayed last event {first_event_id!r}; "
            "would duplicate a side effect already acted on"
        )

    # Caller must independently confirm first_event_id orders strictly after
    # persisted.last_event_id within this session before committing.
    return ResumeToken(session_id=response_session_id, last_event_id=first_event_id)