# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-happens-when-an-instance-drops-aaif-attestation-headers-during-chained-agent-to-agent-delegation.md
# Remnant Warrant:    https://corpusfetch.com/c/what-happens-when-an-instance-drops-aaif-attestation-headers-during-chained-agent-to-agent-delegation.warrant.json
# Markdown SHA-256:   d5facd471e0853236200d83389eee76a6b4d6869825cd61ebe62d0f476064acb
# Code AST SHA-256:   88ddbe15e5f5cfa2f46b1361464a0062c4804dd95781929b4ddcde0d26d43bbc
# Verified Classes:   AttestationOmissionError, DelegationRequest, ReceiverPolicy
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Optional


class AttestationOmissionError(RuntimeError):
    """Raised when a delegation hop forwards a request without an AAIF attestation
    that the receiver requires, or when a receiver would silently fall through to a
    broader default principal on absence."""


@dataclass
class DelegationRequest:
    chain_id: str
    subject: str                 # originating principal
    attestation: Optional[str]   # X-AAIF-Attestation, may be None if dropped
    body: dict = field(default_factory=dict)


@dataclass
class ReceiverPolicy:
    # Paths explicitly allowed to run without attestation (must be enumerated).
    unauthenticated_paths: frozenset[str] = frozenset()
    # Scope of the fall-through service principal, if any.
    fallthrough_scope: frozenset[str] = frozenset()


def forward_hop(req: DelegationRequest, outbound_header_allowlist: frozenset[str]) -> DelegationRequest:
    """Simulate a reconstructing intermediate. If the attestation header is not in
    the allowlist, it is silently dropped -- the defect surface."""
    attestation = req.attestation if "X-AAIF-Attestation" in outbound_header_allowlist else None
    return DelegationRequest(
        chain_id=req.chain_id,
        subject=req.subject,
        attestation=attestation,
        body=dict(req.body),
    )


def authorize(req: DelegationRequest, path: str, policy: ReceiverPolicy,
              required_scope: frozenset[str]) -> str:
    """Return the acting principal. Raise on any silent-elevation condition."""
    if req.attestation is None:
        if path in policy.unauthenticated_paths:
            return "anonymous-enumerated"
        # Absence on a non-enumerated path is a policy event, never a fall-through
        # to a broader principal. Fail closed and loudly.
        if policy.fallthrough_scope >= required_scope and required_scope:
            raise AttestationOmissionError(
                f"chain {req.chain_id}: attestation absent on path {path!r}; "
                f"fall-through principal scope {sorted(policy.fallthrough_scope)} "
                f"would satisfy required scope {sorted(required_scope)} -- "
                f"silent privilege elevation from subject {req.subject!r}"
            )
        raise AttestationOmissionError(
            f"chain {req.chain_id}: attestation absent on non-enumerated path {path!r}"
        )
    # Attestation present: authorize as the attested subject (validation elided).
    return req.subject


def run_chain() -> None:
    req = DelegationRequest(chain_id="c-42", subject="agent:user-7",
                            attestation="aaif.v1.sig...", body={"op": "write"})
    # Intermediate drops the attestation because it is not in the allowlist.
    forwarded = forward_hop(req, outbound_header_allowlist=frozenset({"content-type"}))
    policy = ReceiverPolicy(fallthrough_scope=frozenset({"read", "write", "admin"}))
    # This raises: absence would have been admitted as the broad service principal.
    authorize(forwarded, path="/resource/write", policy=policy,
              required_scope=frozenset({"write"}))


if __name__ == "__main__":
    run_chain()