# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-structural-signals-indicate-delegated-consent-chains-have-silently-exceeded-their-attenuation-invariant-in-aaif-scoped-capability-tokens.md
# Remnant Warrant:    https://corpusfetch.com/c/what-structural-signals-indicate-delegated-consent-chains-have-silently-exceeded-their-attenuation-invariant-in-aaif-scoped-capability-tokens.warrant.json
# Markdown SHA-256:   13d7a07551d7c40d91bb38df2e9c288240e6059fde5ac3f5db2ce24e1724d75c
# Code AST SHA-256:   cebcdcc1b90871cd007e0226d63606d44b5590caca92e0d2a2b5748da1a17d73
# Verified Classes:   AttenuationInvariantError, CapabilityToken
# Defensive Guards:   5 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import FrozenSet, Optional


class AttenuationInvariantError(RuntimeError):
    """Raised when a delegated capability token's resolved closure is not a
    subset of its parent's resolved closure under the same lattice version."""
    pass


@dataclass(frozen=True)
class CapabilityToken:
    token_id: str
    parent_id: Optional[str]
    declared_scopes: FrozenSet[str]
    lattice_version: str
    budget_units: int


def resolve_closure(scopes: FrozenSet[str], lattice_version: str) -> FrozenSet[str]:
    # Stand-in for the policy engine's transitive resolution.
    # In production this queries the versioned lattice; here we simulate the
    # observed drift: 'data:read:orders' resolves to include a write scope.
    closure = set(scopes)
    if lattice_version == "v2" and "data:read:orders" in scopes:
        closure.add("data:write:orders")
    return frozenset(closure)


def verify_attenuation(
    parent: CapabilityToken,
    child: CapabilityToken,
) -> None:
    if child.parent_id != parent.token_id:
        raise AttenuationInvariantError(
            f"token {child.token_id} does not name {parent.token_id} as parent"
        )

    if child.lattice_version != parent.lattice_version:
        raise AttenuationInvariantError(
            f"lattice version mismatch: parent={parent.lattice_version} "
            f"child={child.lattice_version}; closure comparison is undefined"
        )

    if not child.declared_scopes.issubset(parent.declared_scopes):
        raise AttenuationInvariantError(
            f"declared scopes of {child.token_id} escape parent {parent.token_id}"
        )

    if child.budget_units > parent.budget_units:
        raise AttenuationInvariantError(
            f"budget of {child.token_id} ({child.budget_units}) exceeds "
            f"parent ({parent.budget_units})"
        )

    parent_closure = resolve_closure(parent.declared_scopes, parent.lattice_version)
    child_closure = resolve_closure(child.declared_scopes, child.lattice_version)

    if not child_closure.issubset(parent_closure):
        leaked = child_closure - parent_closure
        raise AttenuationInvariantError(
            f"resolved closure of {child.token_id} exceeds parent "
            f"{parent.token_id}; leaked scopes: {sorted(leaked)}; "
            f"lattice={child.lattice_version} — chain is non-attenuating"
        )


if __name__ == "__main__":
    root = CapabilityToken(
        token_id="tok-root",
        parent_id=None,
        declared_scopes=frozenset({"data:read:*"}),
        lattice_version="v2",
        budget_units=1000,
    )
    leaf = CapabilityToken(
        token_id="tok-leaf",
        parent_id="tok-root",
        declared_scopes=frozenset({"data:read:orders"}),
        lattice_version="v2",
        budget_units=500,
    )
    verify_attenuation(root, leaf)