# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-happens-when-an-instance-caches-its-mcp-allowlist-across-a-policy-reload-stale-allowlist-admission-windows-in-tool-invocation-enforcement.md
# Remnant Warrant:    https://corpusfetch.com/c/what-happens-when-an-instance-caches-its-mcp-allowlist-across-a-policy-reload-stale-allowlist-admission-windows-in-tool-invocation-enforcement.warrant.json
# Markdown SHA-256:   733e5c254d0036b5b434c3edca77f8193a7a4dd2284d3391fba8ea6945dd43f2
# Code AST SHA-256:   311b0af90532bf21355d81267c3db8eff4da12eaab5977e8a633d3b0d86e038d
# Verified Classes:   StaleAllowlistAdmissionError, Session, Policy, Host
# Defensive Guards:   3 fail-closed invariants (zero external deps)

"""Straddling-session detector for MCP allowlist enforcement.

Models the invariant: no active session may admit a tool under a
superseded policy generation beyond the revocation SLA.
"""

from dataclasses import dataclass, field
from typing import Dict, Set


class StaleAllowlistAdmissionError(RuntimeError):
    """Raised when a session admits a tool under a superseded policy generation."""


@dataclass
class Session:
    session_id: str
    bound_generation: int
    bound_allowlist: Set[str]


@dataclass
class Policy:
    generation: int
    allowed_tools: Set[str]


@dataclass
class Host:
    current: Policy
    sessions: Dict[str, Session] = field(default_factory=dict)

    def initialize_session(self, session_id: str) -> Session:
        s = Session(
            session_id=session_id,
            bound_generation=self.current.generation,
            bound_allowlist=set(self.current.allowed_tools),
        )
        self.sessions[session_id] = s
        return s

    def reload_policy(self, new_policy: Policy) -> None:
        # Atomic swap. Note: does NOT touch bound sessions.
        self.current = new_policy

    def call_tool(self, session_id: str, tool: str) -> None:
        s = self.sessions[session_id]
        if tool not in s.bound_allowlist:
            raise PermissionError(f"{tool} not in bound allowlist for {session_id}")
        # The boundary failure: admission is checked against the BOUND
        # generation, which may be superseded by self.current.
        if s.bound_generation != self.current.generation:
            if tool not in self.current.allowed_tools:
                raise StaleAllowlistAdmissionError(
                    f"session {session_id} admitted {tool!r} under superseded "
                    f"generation {s.bound_generation}; current generation "
                    f"{self.current.generation} prohibits it"
                )

    def straddling_sessions(self) -> int:
        gen = self.current.generation
        return sum(1 for s in self.sessions.values() if s.bound_generation != gen)


def audit_revocation(host: Host, revoked_tool: str, sla_exceeded: bool) -> None:
    """Assert the operational invariant after a revocation."""
    if sla_exceeded and host.straddling_sessions() > 0:
        raise StaleAllowlistAdmissionError(
            f"revocation of {revoked_tool!r} not effective within SLA: "
            f"{host.straddling_sessions()} session(s) still bound to a "
            f"superseded generation"
        )


if __name__ == "__main__":
    host = Host(current=Policy(generation=417, allowed_tools={"search", "shell.exec"}))
    host.initialize_session("sess-a")
    host.reload_policy(Policy(generation=418, allowed_tools={"search"}))

    # A post-reload session is correctly restricted.
    host.initialize_session("sess-b")
    try:
        host.call_tool("sess-b", "shell.exec")
    except PermissionError:
        print("sess-b correctly denied")

    # The straddling session reveals the gap.
    try:
        host.call_tool("sess-a", "shell.exec")
    except StaleAllowlistAdmissionError as e:
        print(f"detected: {e}")

    audit_revocation(host, "shell.exec", sla_exceeded=True)