# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-treat-an-mcp-sse-upgrade-as-a-fresh-session-rather-than-a-resumption.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-treat-an-mcp-sse-upgrade-as-a-fresh-session-rather-than-a-resumption.warrant.json
# Markdown SHA-256:   106247f05f661c1f6bb7a6f0e8fd0665f8be5c5c01fc22d13811348279342da7
# Code AST SHA-256:   56ec4d1e3b367b7b3d677215a8d99d4b43321130c9de96f40ebca67ad7912c71
# Verified Classes:   ResumptionInvariantError, SessionRegistry
# Defensive Guards:   2 fail-closed invariants (zero external deps)

class ResumptionInvariantError(RuntimeError):
    """Raised when an SSE upgrade would violate session resumption invariants."""
    pass


class SessionRegistry:
    def __init__(self, retention_window: int = 82):
        # session_id -> {"head": int, "oldest_retained": int, "terminated": bool}
        self._sessions: dict[str, dict] = {}
        self._retention_window = retention_window

    def register(self, session_id: str, head: int) -> None:
        self._sessions[session_id] = {
            "head": head,
            "oldest_retained": head,
            "terminated": False,
        }

    def terminate(self, session_id: str) -> None:
        if session_id in self._sessions:
            self._sessions[session_id]["terminated"] = True

    def advance(self, session_id: str, new_head: int) -> None:
        s = self._sessions[session_id]
        s["head"] = new_head
        s["oldest_retained"] = max(s["oldest_retained"], new_head - self._retention_window)

    def decide_upgrade(self, session_id: str, last_event_id: int) -> str:
        """Return 'resume' or 'fresh'. Raise on ambiguous-but-silent cases."""
        s = self._sessions.get(session_id)
        if s is None:
            return "fresh"
        if s["terminated"]:
            # Token present but session terminated: must not resume.
            return "fresh"
        if last_event_id < s["oldest_retained"]:
            # Event evicted from retention window.
            raise ResumptionInvariantError(
                f"session {session_id}: requested Last-Event-ID {last_event_id} "
                f"predates oldest_retained {s['oldest_retained']}; "
                f"resumption is not valid, upgrade must be rejected or reset"
            )
        if last_event_id > s["head"]:
            raise ResumptionInvariantError(
                f"session {session_id}: requested Last-Event-ID {last_event_id} "
                f"exceeds head {s['head']}; client state is inconsistent"
            )
        return "resume"