# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-treat-a-peer-directory-s-capability-grant-as-advisory-until-policy-negotiation-converges.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-treat-a-peer-directory-s-capability-grant-as-advisory-until-policy-negotiation-converges.warrant.json
# Markdown SHA-256:   4a9be027893e1daf15704702b6e31e1b8f8d4b296231a87f08b822b5209c88fc
# Code AST SHA-256:   35a93b232d62c3bf376d0a9e7a2181dfa27c4e09d72425be5e23b87667e5a87c
# Verified Classes:   FederationGrantNotConverged
# Defensive Guards:   4 fail-closed invariants (zero external deps)

class FederationGrantNotConverged(RuntimeError):
    """Raised when a directory-published capability grant is exercised before
    the negotiated policy bundle for the (caller, callee, capability) triple
    has reached CONVERGED with a matching policy_epoch."""
    pass


CONVERGED = "CONVERGED"


def dispatch_federated(peer, capability, directory_entry, cached_bundle):
    entry_epoch = directory_entry.get("policy_epoch")
    if entry_epoch is None:
        # Directory does not publish policy state; grant is advisory at best.
        raise FederationGrantNotConverged(
            f"peer {peer} capability {capability}: directory entry carries no "
            f"policy_epoch; cannot establish convergence"
        )

    if cached_bundle is None:
        raise FederationGrantNotConverged(
            f"peer {peer} capability {capability}: no cached policy bundle; "
            f"negotiation has not been attempted"
        )

    bundle_epoch = cached_bundle.get("policy_epoch")
    bundle_state = cached_bundle.get("state")

    if bundle_epoch != entry_epoch:
        # Either behind (negotiation lag) or ahead (directory replicated first).
        # Both directions are unsafe: only exact match is actionable.
        raise FederationGrantNotConverged(
            f"peer {peer} capability {capability}: epoch mismatch "
            f"(directory={entry_epoch!r}, bundle={bundle_epoch!r}); "
            f"grant is advisory until epochs align"
        )

    if bundle_state != CONVERGED:
        raise FederationGrantNotConverged(
            f"peer {peer} capability {capability}: bundle epoch matches but "
            f"state is {bundle_state!r}, not {CONVERGED!r}"
        )

    return _invoke(peer, capability)


def _invoke(peer, capability):
    # Actual transport call; omitted.
    return {"peer": peer, "capability": capability, "status": "dispatched"}