# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-stale-authorization-from-surviving-agent-identity-revocation-in-long-lived-sessions.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-stale-authorization-from-surviving-agent-identity-revocation-in-long-lived-sessions.warrant.json
# Markdown SHA-256:   605cb711d2b81f4adc1ee4a3cf606bc3c1c62af95807a2549789632c4283213b
# Code AST SHA-256:   115ebd2d808d7aba53476753b98567577fbadca4902ed3d30192ab1f0c657b5b
# Verified Classes:   StaleAuthorizationError, Decision, RevocationAuthority, AuthorizationCache
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from typing import Dict, Optional


class StaleAuthorizationError(RuntimeError):
    """Raised when an authorization decision's provenance epoch predates
    the authority's current revocation epoch for any principal in the chain."""


@dataclass(frozen=True)
class Decision:
    principal: str
    resource: str
    scope: str
    epoch: int          # revocation epoch observed at decision time
    allow: bool


class RevocationAuthority:
    def __init__(self) -> None:
        self._epochs: Dict[str, int] = {}

    def current_epoch(self, principal: str) -> int:
        return self._epochs.get(principal, 0)

    def revoke(self, principal: str) -> None:
        # Monotonic: epochs are never reused across principal re-creation.
        self._epochs[principal] = self.current_epoch(principal) + 1


class AuthorizationCache:
    """Cache-with-lease that MUST NOT renew its lease on a hit.

    A hit is served only if the decision's epoch is still current for every
    principal in the chain; otherwise the entry is evicted and the caller
    must re-authorize against the authority.
    """

    def __init__(self, authority: RevocationAuthority, lease_seconds: int) -> None:
        self._authority = authority
        self._lease = lease_seconds
        self._entries: Dict[tuple, Decision] = {}

    def get(self, principal: str, resource: str, scope: str) -> Optional[Decision]:
        key = (principal, resource, scope)
        decision = self._entries.get(key)
        if decision is None:
            return None
        # Epoch check is mandatory; a cache key without epoch is a defect.
        if decision.epoch < self._authority.current_epoch(principal):
            del self._entries[key]
            return None
        return decision

    def put(self, decision: Decision) -> None:
        self._entries[(decision.principal, decision.resource, decision.scope)] = decision


def authorize(
    authority: RevocationAuthority,
    cache: AuthorizationCache,
    principal: str,
    resource: str,
    scope: str,
) -> Decision:
    cached = cache.get(principal, resource, scope)
    if cached is not None:
        # Re-assert the invariant on the fast path; never trust the hit alone.
        if cached.epoch < authority.current_epoch(principal):
            raise StaleAuthorizationError(
                f"stale allow for principal={principal!r} resource={resource!r} "
                f"scope={scope!r}: decision epoch {cached.epoch} < "
                f"current epoch {authority.current_epoch(principal)}"
            )
        return cached

    # Slow path: consult the authority and stamp the decision with its epoch.
    decision = Decision(
        principal=principal,
        resource=resource,
        scope=scope,
        epoch=authority.current_epoch(principal),
        allow=True,
    )
    cache.put(decision)
    return decision


def enforce_live_session(
    authority: RevocationAuthority,
    cache: AuthorizationCache,
    principal: str,
    resource: str,
    scope: str,
) -> None:
    decision = authorize(authority, cache, principal, resource, scope)
    if not decision.allow:
        raise StaleAuthorizationError(
            f"authorization denied for principal={principal!r}"
        )
    if decision.epoch != authority.current_epoch(principal):
        # Defensive: catches any path that returned a decision without
        # consulting the authority's current epoch.
        raise StaleAuthorizationError(
            f"decision epoch {decision.epoch} != current epoch "
            f"{authority.current_epoch(principal)} for principal={principal!r}"
        )