# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-refuse-a-delegated-capability-token-that-is-structurally-valid-but-temporally-incoherent-across-federation-boundaries.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-refuse-a-delegated-capability-token-that-is-structurally-valid-but-temporally-incoherent-across-federation-boundaries.warrant.json
# Markdown SHA-256:   6ed624f7d28fa168912444683761abb603b7181725973a85bcb60eed44208e3b
# Code AST SHA-256:   d09e9c74ea84fde8a7df4e569b3f06864c6b3b07dfc0412e145acb9f6d9780fb
# Verified Classes:   TemporalIncoherenceError, Hop, ConsumerView
# Defensive Guards:   5 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from typing import Sequence


class TemporalIncoherenceError(RuntimeError):
    """Raised when a delegated capability token is structurally valid but
    temporally incoherent across federation boundaries: no single instant
    exists at which all hops agree the grant was live and un-revoked."""

    def __init__(self, reason: str, chain_id: str, detail: dict):
        super().__init__(f"[{chain_id}] {reason}: {detail}")
        self.reason = reason
        self.chain_id = chain_id
        self.detail = detail


@dataclass(frozen=True)
class Hop:
    org: str
    not_before: float          # issuer-clock seconds
    not_after: float           # issuer-clock seconds
    countersign_time: float    # this hop's clock at countersign
    policy_eval_time: float    # this hop's clock at policy evaluation
    snapshot_version: int      # causal version of policy snapshot used


@dataclass(frozen=True)
class ConsumerView:
    now: float                        # consumer clock
    revocation_watermark_version: int # causal version of issuer revocation feed seen
    parent_issuance_version: int      # causal version of parent grant issuance
    max_replication_lag: float        # seconds, measured
    max_clock_skew: float             # seconds, declared for federation


def assert_temporal_coherence(
    chain_id: str,
    hops: Sequence[Hop],
    view: ConsumerView,
) -> None:
    """Enforce the coherence invariant before honoring a delegated token.

    Refuses (raises) when the token's apparent validity rests on a stale
    causal watermark or on clock skew larger than the remaining margin.
    """
    if not hops:
        raise TemporalIncoherenceError(
            "empty delegation chain", chain_id, {"hops": 0}
        )

    # Condition 4: consumer's revocation watermark must be causally newer
    # than the parent grant's issuance, else absence of revocation is unprovable.
    if view.revocation_watermark_version < view.parent_issuance_version:
        raise TemporalIncoherenceError(
            "stale revocation watermark",
            chain_id,
            {
                "watermark_version": view.revocation_watermark_version,
                "required_version": view.parent_issuance_version,
            },
        )

    # Condition 5: minimum remaining validity margin vs. max clock skew.
    margins = [h.not_after - view.now for h in hops]
    min_margin = min(margins)
    if min_margin <= view.max_clock_skew:
        raise TemporalIncoherenceError(
            "token within clock-skew shadow",
            chain_id,
            {
                "min_remaining_margin": min_margin,
                "max_clock_skew": view.max_clock_skew,
            },
        )

    # Cross-hop causal ordering: each hop's policy evaluation must not
    # precede the previous hop's countersign by more than skew budget.
    for prev, cur in zip(hops, hops[1:]):
        delta = cur.policy_eval_time - prev.countersign_time
        if delta < -view.max_clock_skew:
            raise TemporalIncoherenceError(
                "cross-hop causal inversion",
                chain_id,
                {
                    "prev_org": prev.org,
                    "cur_org": cur.org,
                    "delta_seconds": delta,
                    "skew_budget": view.max_clock_skew,
                },
            )
        if cur.snapshot_version < prev.snapshot_version:
            raise TemporalIncoherenceError(
                "attenuation against stale parent snapshot",
                chain_id,
                {
                    "cur_org": cur.org,
                    "cur_snapshot_version": cur.snapshot_version,
                    "parent_snapshot_version": prev.snapshot_version,
                },
            )

    # All checks passed: the token is coherent and may be honored.
    return None