# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-structural-signals-indicate-capability-negotiation-has-silently-desynchronized-across-federated-trust-brokers.md
# Remnant Warrant:    https://corpusfetch.com/c/what-structural-signals-indicate-capability-negotiation-has-silently-desynchronized-across-federated-trust-brokers.warrant.json
# Markdown SHA-256:   c1bc324a545bb57df1ecd61889558d7226a68e2b730d5dc11988a45e3f89a450
# Code AST SHA-256:   408d14a701ca6b0f01527baf7cad1daa3d8036d5d144366b947818c4a32e9337
# Verified Classes:   FederationDesyncError, CapabilityView
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from typing import Optional


class FederationDesyncError(RuntimeError):
    """Raised when cross-domain capability views diverge in a misroute-capable way."""


@dataclass(frozen=True)
class CapabilityView:
    capability_id: str
    issuer: str
    epoch: int
    scope_hash: str
    delegation_depth: int
    trust_anchor_id: str
    anchor_superseded: bool  # issuer has a newer anchor; this one is in grace overlap


def assert_capability_agreement(
    router_view: CapabilityView,
    issuer_view: CapabilityView,
    capability_is_sensitive: bool,
) -> None:
    """
    Enforce the conjunctive desync gate. Raises FederationDesyncError when the
    router's cached view is causally capable of producing a misroute for a
    depth- or scope-sensitive capability.
    """
    if (router_view.capability_id, router_view.issuer) != (
        issuer_view.capability_id,
        issuer_view.issuer,
    ):
        raise FederationDesyncError(
            f"view pair mismatch: {router_view} vs {issuer_view}"
        )

    if not capability_is_sensitive:
        # Benign divergence on insensitive capabilities is tolerated.
        return

    equal_epoch_scope_divergence = (
        router_view.epoch == issuer_view.epoch
        and router_view.scope_hash != issuer_view.scope_hash
    )
    depth_asymmetry = router_view.delegation_depth < issuer_view.delegation_depth
    stale_anchor = (
        router_view.trust_anchor_id != issuer_view.trust_anchor_id
        and router_view.anchor_superseded
    )

    if equal_epoch_scope_divergence or depth_asymmetry or stale_anchor:
        reasons = []
        if equal_epoch_scope_divergence:
            reasons.append("equal-epoch scope divergence")
        if depth_asymmetry:
            reasons.append(
                f"depth asymmetry router={router_view.delegation_depth} "
                f"issuer={issuer_view.delegation_depth}"
            )
        if stale_anchor:
            reasons.append("stale trust anchor within overlap window")
        raise FederationDesyncError(
            f"capability {router_view.capability_id!r} from {router_view.issuer!r} "
            f"desynchronized: {'; '.join(reasons)}"
        )


def route_task(
    capability_id: str,
    issuer: str,
    router_view: CapabilityView,
    issuer_view: Optional[CapabilityView],
    capability_is_sensitive: bool,
) -> str:
    if issuer_view is None:
        # No cross-domain view available: refuse to route sensitive capabilities
        # on cached authorization alone.
        if capability_is_sensitive:
            raise FederationDesyncError(
                f"cannot route sensitive capability {capability_id!r}: "
                "issuer view unavailable, cached authorization unverified"
            )
        return "routed_on_cache"

    assert_capability_agreement(router_view, issuer_view, capability_is_sensitive)
    return "routed_on_agreed_view"