# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-bound-aaif-delegated-authorization-token-fail-silently-when-an-autonomous-agent-s-identity-is-re-attested-mid-token-lifetime.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-bound-aaif-delegated-authorization-token-fail-silently-when-an-autonomous-agent-s-identity-is-re-attested-mid-token-lifetime.warrant.json
# Markdown SHA-256:   d188047c42866f0e1a60ffae6bdd827f41daacb0a37f5da124a69317bf7b5799
# Code AST SHA-256:   2a06cf8e6119727adbf717b3b7c4cd8be6cb2cb45882dfbe39795254dc365593
# Verified Classes:   BindingDigestMismatch, DelegatedToken, LiveAttestation
# Defensive Guards:   4 fail-closed invariants (zero external deps)

import hashlib
from dataclasses import dataclass


class BindingDigestMismatch(RuntimeError):
    """Raised when a delegated token's recorded attestation digest no longer
    matches the presenting instance's current attestation. This is a hard
    authority-boundary violation under measurement-equality semantics; it must
    never be downgraded to a warning, because the token would otherwise
    silently authorize a code image it was not minted for."""


@dataclass(frozen=True)
class DelegatedToken:
    jti: str
    instance_id: str
    attestation_digest: str  # recorded at mint time
    grants: frozenset[str]
    exp: int


@dataclass(frozen=True)
class LiveAttestation:
    instance_id: str
    attestation_digest: str  # current, post-re-attestation
    measured_modules: frozenset[str]


def validate_binding(
    token: DelegatedToken,
    live: LiveAttestation,
    *,
    semantics: str = "measurement_equality",
) -> None:
    """Enforce the AAIF instance-binding invariant at delegated exchange.

    semantics='measurement_equality' -> digest divergence is fatal (default).
    semantics='identity_continuity'   -> digest divergence is advisory only;
                                         caller accepts the documented risk.
    """
    if semantics not in ("measurement_equality", "identity_continuity"):
        raise ValueError(f"unknown binding semantics: {semantics!r}")

    if token.instance_id != live.instance_id:
        raise BindingDigestMismatch(
            f"instance_id mismatch for jti={token.jti}: "
            f"token={token.instance_id!r} live={live.instance_id!r}"
        )

    if token.attestation_digest != live.attestation_digest:
        # The token was minted against a measurement that no longer exists.
        # Under measurement_equality this is a hard failure: re-issue required.
        if semantics == "measurement_equality":
            raise BindingDigestMismatch(
                f"attestation digest rotated mid-token-lifetime for jti={token.jti} "
                f"instance={token.instance_id}: "
                f"minted={token.attestation_digest} live={live.attestation_digest}; "
                f"re-issue the delegated token before exercising grants "
                f"{sorted(token.grants)}"
            )
        # identity_continuity: proceed, but the caller has explicitly opted in.


def exchange(token: DelegatedToken, live: LiveAttestation, resource: str) -> frozenset[str]:
    validate_binding(token, live)  # measurement_equality by default
    if resource not in token.grants:
        raise PermissionError(f"resource {resource!r} not in grants for jti={token.jti}")
    return token.grants


def _digest(modules: frozenset[str]) -> str:
    h = hashlib.sha256()
    for m in sorted(modules):
        h.update(m.encode())
    return h.hexdigest()


if __name__ == "__main__":
    original = frozenset({"core", "http_client"})
    patched = frozenset({"core", "http_client", "tls_patched", "plugin_write"})

    token = DelegatedToken(
        jti="tok-7f3a",
        instance_id="agent-worker-42",
        attestation_digest=_digest(original),
        grants=frozenset({"dataset_a:read"}),
        exp=1_700_000_000,
    )
    live = LiveAttestation(
        instance_id="agent-worker-42",
        attestation_digest=_digest(patched),
        measured_modules=patched,
    )

    # Under default measurement_equality semantics this raises, forcing
    # re-issuance rather than silently authorizing the patched module set.
    exchange(token, live, "dataset_a:read")