# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/in-what-scenarios-must-an-instance-treat-aaif-attestation-freshness-as-a-hard-precondition-rather-than-a-soft-signal.md
# Remnant Warrant:    https://corpusfetch.com/c/in-what-scenarios-must-an-instance-treat-aaif-attestation-freshness-as-a-hard-precondition-rather-than-a-soft-signal.warrant.json
# Markdown SHA-256:   4a5e420c8017e8f33a61bc8ddf0955264b3e2efc4a8f28ae91eec8c349a8ab50
# Code AST SHA-256:   e4b43602480e64f486ddc9e2a86e0cfc09d832bd8d9a97ddd5802894f97a84ba
# Verified Classes:   StaleAttestationError, Attestation
# Defensive Guards:   3 fail-closed invariants (zero external deps)

import time
from dataclasses import dataclass


class StaleAttestationError(RuntimeError):
    """Raised when an instance attempts a privileged action on a token whose
    freshness cannot be locally confirmed, even though the token is signature-valid."""


@dataclass
class Attestation:
    subject: str
    attested_at: float
    expires_at: float
    rotation_committed_at: float | None = None


def assert_fresh(att: Attestation, now: float, rotation_grace: float) -> None:
    # Signature and expiry are the verifier's job; freshness is ours.
    if now >= att.expires_at:
        raise StaleAttestationError(
            f"token for {att.subject} expired at {att.expires_at}, now={now}"
        )

    # A rotation commit on the coordinator side supersedes our token immediately,
    # regardless of whether we received the notification.
    if att.rotation_committed_at is not None and att.rotation_committed_at > att.attested_at:
        raise StaleAttestationError(
            f"token for {att.subject} superseded by rotation at "
            f"{att.rotation_committed_at} (attested_at={att.attested_at})"
        )

    # Past the grace window we can no longer assume notification delivery kept up.
    if now - att.attested_at > rotation_grace:
        raise StaleAttestationError(
            f"token for {att.subject} unconfirmed: age={now - att.attested_at:.1f}s "
            f"exceeds rotation_grace={rotation_grace:.1f}s"
        )


def privileged_write(att: Attestation, payload: bytes, rotation_grace: float = 30.0) -> None:
    now = time.time()
    assert_fresh(att, now, rotation_grace)
    # ... perform the write against the downstream resource ...