# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-handle-stale-revocation-state-when-federated-trust-domains-disagree-on-the-revocation-epoch.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-handle-stale-revocation-state-when-federated-trust-domains-disagree-on-the-revocation-epoch.warrant.json
# Markdown SHA-256:   2cb0aedb4bfb95d6efe007dbec8d145fa3ceb02aeeb7ad39f16057cb204005e0
# Code AST SHA-256:   798fa4067c9e219cb907ba7be49fe7bf5f664a3174298d9df193166d7a0eb2b9
# Verified Classes:   StaleEpochError, RevokedError, HighWaterStore
# Defensive Guards:   3 fail-closed invariants (zero external deps)

import time
from dataclasses import dataclass, field


class StaleEpochError(RuntimeError):
    """Raised when a presented revocation epoch is older than the locally
    persisted high-water mark for its issuer. Distinct from RevokedError:
    the agent is unverifiable, not known-revoked."""
    pass


class RevokedError(PermissionError):
    pass


@dataclass
class HighWaterStore:
    # issuer_id -> highest epoch ever observed and durably persisted
    marks: dict = field(default_factory=dict)

    def observe(self, issuer_id: str, epoch: int) -> None:
        prev = self.marks.get(issuer_id, -1)
        if epoch > prev:
            self.marks[issuer_id] = epoch

    def mark_for(self, issuer_id: str) -> int:
        return self.marks.get(issuer_id, -1)


def admit_assertion(
    store: HighWaterStore,
    issuer_id: str,
    assertion_epoch: int,
    revoked_at_epoch,
    max_skew_epochs: int = 0,
    now_epoch: int = 0,
):
    """Admission gate enforcing the per-issuer revocation high-water mark.

    revoked_at_epoch: issuer's epoch at which the agent was revoked,
    or None if not revoked as of the assertion's epoch.
    """
    if assertion_epoch > now_epoch + max_skew_epochs:
        raise StaleEpochError(
            f"epoch {assertion_epoch} for issuer {issuer_id} exceeds "
            f"now {now_epoch} by more than max_skew {max_skew_epochs}; malformed"
        )

    hw = store.mark_for(issuer_id)
    if assertion_epoch < hw:
        # Out-of-order or replayed record; do not regress, do not admit silently.
        raise StaleEpochError(
            f"assertion epoch {assertion_epoch} for issuer {issuer_id} is older "
            f"than persisted high-water mark {hw}; re-fetch from issuer"
        )

    if revoked_at_epoch is not None and revoked_at_epoch <= assertion_epoch:
        store.observe(issuer_id, revoked_at_epoch)
        raise RevokedError(
            f"agent revoked at epoch {revoked_at_epoch} (issuer {issuer_id}); "
            f"assertion at epoch {assertion_epoch}"
        )

    store.observe(issuer_id, assertion_epoch)
    return True