# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-silent-capability-escalation-across-federated-a2a-trust-boundaries.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-silent-capability-escalation-across-federated-a2a-trust-boundaries.warrant.json
# Markdown SHA-256:   b9f9b29345d60bb7b679e41022a3f5913cae5ad1168d580232a9d3908753650e
# Code AST SHA-256:   94098b279666b8e0e1a2f920bce1d207466d7a0ee96161e0cc783004ca236cce
# Verified Classes:   GrantMonotonicityError, Grant, Attestation
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass, field
from typing import FrozenSet


class GrantMonotonicityError(RuntimeError):
    """Raised when a federation hop would widen effective capability scope."""


@dataclass(frozen=True)
class Grant:
    capability: str
    scope: FrozenSet[tuple[str, str]]  # e.g. frozenset({("tenant", "acme")})

    def covers(self, other: "Grant") -> bool:
        # A grant covers another iff same capability and scope is a superset.
        return self.capability == other.capability and self.scope.issuperset(other.scope)


@dataclass(frozen=True)
class Attestation:
    origin_org: str
    subject: str
    grants: FrozenSet[Grant]
    grant_digest: str


def verify_hop(
    origin: Attestation,
    hop_effective: FrozenSet[Grant],
    hop_name: str,
) -> None:
    """Enforce grant monotonicity: every effective grant at this hop must be
    covered by some origin grant. Widenings fail closed."""
    for eff in hop_effective:
        if not any(og.covers(eff) for og in origin.grants):
            raise GrantMonotonicityError(
                f"hop {hop_name!r}: effective grant {eff.capability} "
                f"scope={sorted(eff.scope)} is not covered by origin "
                f"{origin.origin_org!r} grant digest {origin.grant_digest}; "
                f"refusing to widen (fail closed)"
            )


def negotiate(
    origin: Attestation,
    broker_effective: FrozenSet[Grant],
    remote_effective: FrozenSet[Grant],
) -> FrozenSet[Grant]:
    verify_hop(origin, broker_effective, "broker")
    verify_hop(origin, remote_effective, "remote")
    return remote_effective


def cache_key(origin: Attestation, remote_org: str, policy_version: str) -> tuple:
    # Scope and origin grant digest MUST be part of the key.
    return (
        origin.origin_org,
        remote_org,
        origin.grant_digest,
        policy_version,
        tuple(sorted((g.capability, tuple(sorted(g.scope))) for g in origin.grants)),
    )