# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-privilege-re-expansion-when-attenuating-tool-invocation-capabilities-across-delegated-agent-calls.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-privilege-re-expansion-when-attenuating-tool-invocation-capabilities-across-delegated-agent-calls.warrant.json
# Markdown SHA-256:   8801fe72f118f816b38640f631553dac2575c9343cdd452d8960db08cf40fc83
# Code AST SHA-256:   99c2b63f37a7936a4df81bdd4b60ba684264e864a3876957fee5ee5d82c7a69e
# Verified Classes:   PrivilegeReexpansionError, Capability, ToolRegistry
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import FrozenSet

ALL_EFFECTS = frozenset({"*"})


class PrivilegeReexpansionError(RuntimeError):
    """Raised when a delegated capability set's effect closure exceeds the parent's."""


@dataclass(frozen=True)
class Capability:
    kind: str          # "read", "write", "invoke"
    target: str        # path glob or tool name


@dataclass
class ToolRegistry:
    # declared maximal effect of each invokable tool; missing => ALL_EFFECTS (fail closed)
    effects: dict = field(default_factory=dict)

    def effect_closure(self, caps: FrozenSet[Capability]) -> FrozenSet[Capability]:
        closure = set(caps)
        frontier = [c for c in caps if c.kind == "invoke"]
        while frontier:
            cap = frontier.pop()
            declared = self.effects.get(cap.target, ALL_EFFECTS)
            if declared is ALL_EFFECTS:
                # cannot bound this tool: treat as granting everything
                return frozenset({Capability("*", "*")})
            for eff in declared:
                if eff not in closure:
                    closure.add(eff)
                    if eff.kind == "invoke":
                        frontier.append(eff)
        return frozenset(closure)


def attenuate(
    parent: FrozenSet[Capability],
    child: FrozenSet[Capability],
    registry: ToolRegistry,
) -> FrozenSet[Capability]:
    """Return the child set iff its effect closure is contained in the parent's.

    Attenuation is checked on the closure, not per-token, because an invoke
    token composes with its arguments to reconstruct read/write rights the
    parent may not hold. Per-category checks pass exactly the bad cases.
    """
    parent_closure = registry.effect_closure(parent)
    child_closure = registry.effect_closure(child)

    if parent_closure == frozenset({Capability("*", "*")}):
        # parent itself unbounded; cannot attenuate meaningfully
        raise PrivilegeReexpansionError(
            "parent effect closure is unbounded (undeclared tool); refusing delegation"
        )

    if not child_closure.issubset(parent_closure):
        excess = child_closure - parent_closure
        raise PrivilegeReexpansionError(
            f"delegation re-expands privilege; excess effects: "
            f"{sorted((c.kind, c.target) for c in excess)}"
        )
    return child


def check_redelegation(
    root: FrozenSet[Capability],
    grandchild: FrozenSet[Capability],
    registry: ToolRegistry,
) -> FrozenSet[Capability]:
    """Re-delegation must attenuate against the ROOT floor, not the immediate delegator.

    Checking against the immediate delegator lets a chain of individually-valid
    subset steps walk the set back up toward the root and beyond.
    """
    return attenuate(root, grandchild, registry)