# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-revocation-from-being-a-linearizable-operation-across-federated-agent-identity-graphs.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-revocation-from-being-a-linearizable-operation-across-federated-agent-identity-graphs.warrant.json
# Markdown SHA-256:   13c3ba381eefbf370072427a0ffeaca551ef567a44ea1298f80548e05f357ae5
# Code AST SHA-256:   ca1bfb0d2bc016c0c25560d59d32eb62b68bf0840a6fee9e3eec06f52f35ad12
# Verified Classes:   RevocationGapError, StatusSnapshot, AdmissionRequest
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import time
from dataclasses import dataclass


class RevocationGapError(RuntimeError):
    """Raised when an admission decision cannot be causally ordered after the issuer's revocation state."""


@dataclass(frozen=True)
class StatusSnapshot:
    credential_id: str
    epoch: int
    fetched_at: float
    revoked: bool


@dataclass(frozen=True)
class AdmissionRequest:
    credential_id: str
    chain_root_id: str
    now: float


def admit(
    req: AdmissionRequest,
    snapshot: StatusSnapshot,
    chain_root_snapshot: StatusSnapshot,
    max_epoch_lag: int,
    issuer_epoch: int,
) -> bool:
    # Invariant (i): snapshot must be within bounded epoch lag of issuer.
    if issuer_epoch - snapshot.epoch > max_epoch_lag:
        raise RevocationGapError(
            f"credential {req.credential_id}: snapshot epoch {snapshot.epoch} "
            f"lags issuer epoch {issuer_epoch} by more than {max_epoch_lag}; "
            f"cannot order admission after revocation state"
        )
    # Invariant (ii): delegation chain root must be re-checked at admission.
    if issuer_epoch - chain_root_snapshot.epoch > max_epoch_lag:
        raise RevocationGapError(
            f"credential {req.credential_id}: chain root {req.chain_root_id} "
            f"status epoch {chain_root_snapshot.epoch} exceeds lag bound"
        )
    if chain_root_snapshot.revoked:
        return False
    if snapshot.revoked:
        return False
    return True


def reconcile_gap_window(
    admissions: list[tuple[AdmissionRequest, StatusSnapshot]],
    revocations: dict[str, float],
) -> list[AdmissionRequest]:
    """Return admissions whose relied-upon epoch was superseded by a revocation prior to admission time."""
    suspect = []
    for req, snap in admissions:
        rev_time = revocations.get(req.credential_id)
        if rev_time is not None and rev_time < req.now and not snap.revoked:
            suspect.append(req)
    return suspect