# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-mcp-federation-gateway-refuse-a-delegated-capability-token-that-is-structurally-valid-but-semantically-over.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-mcp-federation-gateway-refuse-a-delegated-capability-token-that-is-structurally-valid-but-semantically-over.warrant.json
# Markdown SHA-256:   69aee58b15e5e31bd4b8a73f5bd67f5c11e5f52a37254fecc29eb233cb397e83
# Code AST SHA-256:   6202e681f97ee86cd869ce7eb848b6df00f6815362c85c2fd8598a6bbe8f7684
# Verified Classes:   DelegationScopeEscalation, FederationContext
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from dataclasses import dataclass


class DelegationScopeEscalation(RuntimeError):
    """Raised when a delegated MCP capability token is structurally valid
    but its terminal scope exceeds the permitted intersection."""


@dataclass(frozen=True)
class FederationContext:
    originating_principal_scope: frozenset[str]
    federation_allowed_capabilities: frozenset[str]
    target_server_surface: frozenset[str]
    grant_cache_fresh: bool


def authorize_delegated_call(
    requested_scope: frozenset[str],
    ctx: FederationContext,
) -> frozenset[str]:
    # Temporal invariant: never serve a grant that predates a narrowing event.
    if not ctx.grant_cache_fresh:
        raise DelegationScopeEscalation(
            "cached federation grant is stale relative to upstream scope narrowing; "
            "re-evaluate before dispatch"
        )

    # The invariant is the INTERSECTION of all three authorities, not their union.
    permitted = (
        ctx.originating_principal_scope
        & ctx.federation_allowed_capabilities
        & ctx.target_server_surface
    )

    excess = requested_scope - permitted
    if excess:
        # Fail closed. Do not silently trim to `permitted`: a partial grant
        # masks the escalation and produces harder-to-audit partial execution.
        raise DelegationScopeEscalation(
            f"delegated scope escalation across MCP federation: "
            f"requested={sorted(requested_scope)} "
            f"permitted_intersection={sorted(permitted)} "
            f"excess={sorted(excess)}"
        )

    if not ctx.originating_principal_scope:
        raise DelegationScopeEscalation(
            "originating principal scope unresolvable; cannot prove non-escalation"
        )

    return requested_scope