# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-structural-signals-indicate-per-session-isolation-failure-in-http-native-mcp-server-multiplexing.md
# Remnant Warrant:    https://corpusfetch.com/c/what-structural-signals-indicate-per-session-isolation-failure-in-http-native-mcp-server-multiplexing.warrant.json
# Markdown SHA-256:   5de74dd2923aa39fdffa39c9b363af6aeda40afa99faaf80e5d849f318cf0675
# Code AST SHA-256:   f01877c405be6ccf6e250a35eb4a415e4f0afdb0656bcfea7ca40dbcab3f7b5e
# Verified Classes:   SessionIsolationViolation, SessionRegistry
# Defensive Guards:   5 fail-closed invariants (zero external deps)

class SessionIsolationViolation(RuntimeError):
    """Raised when a request observes state owned by a different MCP session."""
    pass


class SessionRegistry:
    def __init__(self):
        self._sessions = {}          # session_id -> immutable capability snapshot
        self._progress = {}          # (session_id, progress_token) -> stream
        self._inflight = {}          # (session_id, request_id) -> cancel handle
        self._subscriptions = {}     # (session_id, uri) -> subscriber

    def resolve(self, header_session_id, advertised_capabilities):
        if header_session_id is None:
            # Never fall back to a global "current" session.
            raise SessionIsolationViolation(
                "request arrived without Mcp-Session-Id; refusing global fallback"
            )
        snap = self._sessions.get(header_session_id)
        if snap is None:
            raise SessionIsolationViolation(
                f"unknown session {header_session_id!r}; issue a fresh initialize"
            )
        if snap != frozenset(advertised_capabilities):
            raise SessionIsolationViolation(
                f"capability drift in session {header_session_id!r}: "
                f"negotiated={sorted(snap)} presented={sorted(advertised_capabilities)}"
            )
        return header_session_id

    def register_progress(self, session_id, progress_token, stream):
        key = (session_id, progress_token)
        if key in self._progress:
            raise SessionIsolationViolation(
                f"duplicate progress token {progress_token!r} within session {session_id!r}"
            )
        self._progress[key] = stream
        return key

    def cancel(self, session_id, request_id):
        key = (session_id, request_id)
        handle = self._inflight.get(key)
        if handle is None:
            raise SessionIsolationViolation(
                f"cancel for unknown (session, request) pair {key!r}; "
                "refusing to cancel across session boundaries"
            )
        handle.cancel()