# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-trust-broker-replay-in-cross-organization-multi-agent-federation.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-trust-broker-replay-in-cross-organization-multi-agent-federation.warrant.json
# Markdown SHA-256:   ffd7d949354cf1bde9e38a65923ead68b4365212a99828775b98730cb1fec580
# Code AST SHA-256:   318e3fd6a8e4b260b85d15ef46d04a83d334bbdbc65b5121da982e19d05f6987
# Verified Classes:   DelegationEdgeReplay, DepthViolation, EdgeKey
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from typing import Optional


class DelegationEdgeReplay(RuntimeError):
    """Raised when a delegation edge is presented more than once."""
    pass


class DepthViolation(RuntimeError):
    """Raised when delegation depth exceeds policy before replay check."""
    pass


@dataclass(frozen=True)
class EdgeKey:
    iss: str
    sub: str
    aud: str
    parent_jti: Optional[str]
    depth: int


def admit_assertion(assertion: dict, consumed: set, max_depth: int) -> EdgeKey:
    depth = assertion.get("delegation_depth", 0)
    if depth > max_depth:
        # Reject before consuming: a rejected edge must not burn the slot.
        raise DepthViolation(
            f"depth {depth} exceeds policy {max_depth} for {assertion.get('aud')}"
        )

    parent = assertion.get("parent_jti")
    key = EdgeKey(
        iss=assertion["iss"],
        sub=assertion["sub"],
        aud=assertion["aud"],
        parent_jti=parent,
        depth=depth,
    )

    # A child is admissible only if its parent edge was itself admitted.
    if parent is not None:
        parent_key = EdgeKey(
            iss=assertion["iss"],
            sub=assertion["sub"],
            aud=assertion["aud"],
            parent_jti=None,
            depth=depth - 1,
        )
        if parent_key not in consumed:
            raise DelegationEdgeReplay(
                f"parent edge for jti={parent} was never admitted"
            )

    if key in consumed:
        raise DelegationEdgeReplay(
            f"edge replay: {key} already consumed"
        )

    consumed.add(key)
    return key