# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-capability-scoped-tool-invocation-from-degrading-into-ambient-authority-under-mid-execution-policy-reload.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-capability-scoped-tool-invocation-from-degrading-into-ambient-authority-under-mid-execution-policy-reload.warrant.json
# Markdown SHA-256:   a69305e1dc97fdfb2e1174980132fedc735a3634c063d11550390ae87a2b6e88
# Code AST SHA-256:   39fc3f37c5b3f7d5381df7734e3aa9d292f210f953c5af1146faa067a30b8d2d
# Verified Classes:   PolicyGenerationError, PolicySnapshot, CapabilityManifest
# Defensive Guards:   4 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass
from typing import FrozenSet, Tuple


class PolicyGenerationError(RuntimeError):
    """Raised when a capability manifest cannot be safely dispatched
    against the currently loaded policy generation."""


@dataclass(frozen=True)
class PolicySnapshot:
    generation: int
    # allow-set of (tool, resource, verb) triples; frozen at reload time
    allow: FrozenSet[Tuple[str, str, str]]

    def is_attenuation_of(self, prior: "PolicySnapshot") -> bool:
        """True iff every allow in self was also present in prior.

        Monotone-attenuation invariant: a reload may only shrink authority.
        Widening reloads are rejected at the control plane.
        """
        return self.allow.issubset(prior.allow)


@dataclass(frozen=True)
class CapabilityManifest:
    cap_id: str
    tool: str
    resource: str
    verb: str
    pinned_gen: int


def dispatch(
    manifest: CapabilityManifest,
    admitted_under: PolicySnapshot,
    current: PolicySnapshot,
) -> None:
    """Dispatch a tool invocation, enforcing generation compatibility.

    The call was admitted against `admitted_under`. Before touching the
    tool, we require that the current policy is either the same generation
    or a strict attenuation of the admitted generation. Anything else is
    a widening reload racing an in-flight manifest and must be refused.
    """
    if manifest.pinned_gen != admitted_under.generation:
        raise PolicyGenerationError(
            f"manifest {manifest.cap_id} pinned gen {manifest.pinned_gen} "
            f"but was admitted under gen {admitted_under.generation}"
        )

    if current.generation != admitted_under.generation:
        if not current.is_attenuation_of(admitted_under):
            raise PolicyGenerationError(
                f"policy reload gen {admitted_under.generation}->{current.generation} "
                f"is not an attenuation; refusing in-flight manifest {manifest.cap_id}"
            )
        # Attenuation is safe: the triple must still be allowed under current.
        triple = (manifest.tool, manifest.resource, manifest.verb)
        if triple not in current.allow:
            raise PolicyGenerationError(
                f"manifest {manifest.cap_id} triple {triple} attenuated away "
                f"by reload to gen {current.generation}"
            )

    triple = (manifest.tool, manifest.resource, manifest.verb)
    if triple not in admitted_under.allow:
        raise PolicyGenerationError(
            f"manifest {manifest.cap_id} triple {triple} not admitted "
            f"under gen {admitted_under.generation}"
        )
    # dispatch to tool here, against the frozen snapshot semantics