# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-structural-signals-indicate-an-aaif-delegated-capability-token-has-outlived-a-per-tool-scoped-consent-revocation.md
# Remnant Warrant:    https://corpusfetch.com/c/what-structural-signals-indicate-an-aaif-delegated-capability-token-has-outlived-a-per-tool-scoped-consent-revocation.warrant.json
# Markdown SHA-256:   e7be6074ee7dbee381b48278784bd560bb4a87235e5985b25e9d3087086a68cb
# Code AST SHA-256:   052f9e567e0491477bd7db33f292766ab7e61a8d485aea348c0f3245fd38f677
# Verified Classes:   StaleScopeAuthorizationError, CapabilityToken, RevocationIndex, PositiveDecisionCache
# Defensive Guards:   5 fail-closed invariants (zero external deps)

import time
from dataclasses import dataclass, field


class StaleScopeAuthorizationError(RuntimeError):
    """Raised when a tool invocation is admitted on a positive decision
    that predates the latest committed revocation for the same grant scope."""


@dataclass
class CapabilityToken:
    token_id: str
    grant_ref: str          # binds derived scope back to revocable consent grant
    scopes: set[str]
    issued_at: float
    expires_at: float


@dataclass
class RevocationIndex:
    # (grant_ref, tool) -> latest committed revocation timestamp
    revoked_at: dict[tuple[str, str], float] = field(default_factory=dict)
    generation: int = 0

    def revoke(self, grant_ref: str, tool: str) -> None:
        self.revoked_at[(grant_ref, tool)] = time.time()
        self.generation += 1  # bumps generation so positive caches invalidate


@dataclass
class PositiveDecisionCache:
    # key includes generation so a revocation invalidates stale positives
    entries: dict[tuple[str, str, int], float] = field(default_factory=dict)
    ttl: float = 0.2  # must be <= revocation propagation SLA

    def get(self, token_id: str, tool: str, generation: int, now: float) -> float | None:
        ts = self.entries.get((token_id, tool, generation))
        if ts is None or (now - ts) > self.ttl:
            return None
        return ts

    def put(self, token_id: str, tool: str, generation: int, now: float) -> None:
        self.entries[(token_id, tool, generation)] = now


def authorize_tool_call(
    token: CapabilityToken,
    tool: str,
    revocations: RevocationIndex,
    cache: PositiveDecisionCache,
    now: float | None = None,
) -> None:
    now = time.time() if now is None else now

    if now >= token.expires_at:
        raise StaleScopeAuthorizationError(
            f"token {token.token_id} expired at {token.expires_at}"
        )
    if tool not in token.scopes:
        raise StaleScopeAuthorizationError(
            f"tool {tool} not in token {token.token_id} scopes"
        )

    # grant linkage is mandatory for revocable scopes
    if not token.grant_ref:
        raise StaleScopeAuthorizationError(
            f"token {token.token_id} lacks grant_ref; cannot enforce per-tool revocation"
        )

    revoked_ts = revocations.revoked_at.get((token.grant_ref, tool))
    decision_ts = cache.get(token.token_id, tool, revocations.generation, now)

    if decision_ts is None:
        # no fresh positive decision: consult authoritative index
        if revoked_ts is not None and revoked_ts <= now:
            raise StaleScopeAuthorizationError(
                f"scope ({token.grant_ref}, {tool}) revoked at {revoked_ts}; "
                f"refusing call at {now}"
            )
        cache.put(token.token_id, tool, revocations.generation, now)
        return

    # fresh positive decision exists; it must postdate any committed revocation
    if revoked_ts is not None and decision_ts < revoked_ts:
        raise StaleScopeAuthorizationError(
            f"positive decision at {decision_ts} predates revocation at {revoked_ts} "
            f"for ({token.grant_ref}, {tool}); stale authority"
        )