# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-cross-organizational-causal-inversion-when-clock-skew-exceeds-trace-propagation-latency.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-cross-organizational-causal-inversion-when-clock-skew-exceeds-trace-propagation-latency.warrant.json
# Markdown SHA-256:   d213c077a71f228359bf74fbd5f0cdfe4867355c824299d03147694be021a734
# Code AST SHA-256:   5b59ab93538e79a72fa8acf3f6f61c69a7ec1abe56188fcd8132b44851251ace
# Verified Classes:   CausalInversionError, PairObservation
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import time
from dataclasses import dataclass


class CausalInversionError(RuntimeError):
    """Raised when wall-clock ordering cannot be trusted across an org boundary."""


@dataclass(frozen=True)
class PairObservation:
    org_x: str
    org_y: str
    max_clock_offset_s: float   # from active probing, not peer-reported
    max_jitter_s: float         # measurement noise on the offset estimate
    min_propagation_s: float    # smallest observed inter-org span latency
    observed_at: float          # unix ts of the observation window


MAX_OBSERVATION_AGE_S = 300.0


def causal_ordering_is_valid(obs: PairObservation, now: float | None = None) -> bool:
    """True only if observed offset+jitter is strictly below observed min latency."""
    now = time.time() if now is None else now
    if now - obs.observed_at > MAX_OBSERVATION_AGE_S:
        # Stale observation: cannot certify validity.
        return False
    return (obs.max_clock_offset_s + obs.max_jitter_s) < obs.min_propagation_s


def assert_happens_before(
    parent_end_s: float,
    child_start_s: float,
    obs: PairObservation,
    now: float | None = None,
) -> None:
    """
    Assert that parent span (org_x) happened-before child span (org_y) using
    wall-clock timestamps. Raises if the boundary condition is violated, because
    in that regime the comparison is meaningless and may be inverted.
    """
    if not causal_ordering_is_valid(obs, now=now):
        raise CausalInversionError(
            f"cannot assert happens-before {obs.org_x}->{obs.org_y}: "
            f"offset+jitter={obs.max_clock_offset_s + obs.max_jitter_s:.3f}s "
            f">= min_propagation={obs.min_propagation_s:.3f}s "
            f"(or observation stale); downgrade to correlation-only."
        )
    if not (parent_end_s <= child_start_s):
        raise CausalInversionError(
            f"timestamp inversion {obs.org_x}->{obs.org_y}: "
            f"parent_end={parent_end_s:.6f} > child_start={child_start_s:.6f} "
            f"despite valid boundary; treat as true causal violation."
        )