# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-happens-when-an-instance-revokes-an-agent-identity-that-is-still-mid-handshake-across-a-federated-trust-boundary.md
# Remnant Warrant:    https://corpusfetch.com/c/what-happens-when-an-instance-revokes-an-agent-identity-that-is-still-mid-handshake-across-a-federated-trust-boundary.warrant.json
# Markdown SHA-256:   02119e952ad8068a1cbf05068f9aaf7c5bb0bf3da4da83e79ebfcc0206b5acc5
# Code AST SHA-256:   b45eab4af88e5f54c1c172485755c19f549b318325437323b5ce539c6d033545
# Verified Classes:   RevocationAcceptanceViolation, RevocationRecord, EdgeState
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import time
from dataclasses import dataclass, field


class RevocationAcceptanceViolation(RuntimeError):
    """Raised when a relying party is honoring an identity past its acceptance bound."""


@dataclass
class RevocationRecord:
    subject: str
    revision: int
    issued_at: float
    force_invalidate: bool = False


@dataclass
class EdgeState:
    # subject -> (applied_revision, applied_at)
    applied: dict = field(default_factory=dict)
    # subject -> (issuing_revision, session_expires_at)
    sessions: dict = field(default_factory=dict)

    def apply(self, rec: RevocationRecord) -> None:
        prior = self.applied.get(rec.subject)
        # Dedup must key on revision, never on (subject, reason, issued_at).
        if prior is not None and prior[0] >= rec.revision and not rec.force_invalidate:
            return  # duplicate — correctly ignored
        self.applied[rec.subject] = (rec.revision, time.time())

    def admit(self, subject: str, now: float, r_max: float) -> None:
        rec = self.applied.get(subject)
        if rec is None:
            return
        applied_rev, applied_at = rec
        if now - applied_at > r_max:
            raise RevocationAcceptanceViolation(
                f"{subject} admitted {now - applied_at:.1f}s after revocation "
                f"revision {applied_rev}; R_max={r_max}s exceeded"
            )

    def audit_session(self, subject: str, now: float, s_max: float,
                      current_revision: int) -> None:
        sess = self.sessions.get(subject)
        if sess is None:
            return
        issuing_rev, expires_at = sess
        if issuing_rev < current_revision and now > expires_at - s_max:
            raise RevocationAcceptanceViolation(
                f"{subject} session issued at revision {issuing_rev} still live "
                f"against current revision {current_revision}; S_max={s_max}s exceeded"
            )


def enforce_revocation_boundary(edge: EdgeState, subject: str, now: float,
                                r_max: float, s_max: float,
                                current_revision: int) -> None:
    edge.admit(subject, now, r_max)
    edge.audit_session(subject, now, s_max, current_revision)