# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-refuse-to-propagate-a-parent-span-context-across-a-federation-boundary.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-refuse-to-propagate-a-parent-span-context-across-a-federation-boundary.warrant.json
# Markdown SHA-256:   1a758cf35b0ddaca16ca981b34994d0da61a99569dec65c1b5ab9e51a40927d1
# Code AST SHA-256:   3d43b68947b7c1c7832cce1a6583d88af5a7fa5cea3304c93b92b43f6b4a9dbf
# Verified Classes:   UntrustedFederationContextError, SpanContext, TrustPolicy
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from __future__ import annotations

import time
from dataclasses import dataclass
from typing import Optional


class UntrustedFederationContextError(RuntimeError):
    """Raised when a parent span context cannot be propagated as authoritative."""


@dataclass(frozen=True)
class SpanContext:
    trace_id: str
    span_id: str
    federation_id: str
    sampled: bool
    baggage: dict[str, str]
    created_at: float  # epoch seconds
    signed: bool


@dataclass(frozen=True)
class TrustPolicy:
    trusted_peers: frozenset[str]
    allowed_baggage_keys: frozenset[str]
    max_context_age_s: float = 300.0


def decide_propagation(
    ctx: SpanContext,
    local_federation: str,
    policy: TrustPolicy,
    now: Optional[float] = None,
) -> dict:
    """
    Decide how to handle an incoming parent span context at a boundary.

    Returns a dict describing the resulting local span's linkage:
      - mode: "parent" | "link" | "root"
      - parent_span_id: str | None
      - trace_id: str | None
      - baggage: filtered baggage
      - sampled: local sampling decision (never inherited across boundary)
    """
    now = time.time() if now is None else now
    cross_boundary = ctx.federation_id != local_federation

    # Intra-domain: trust is implicit, propagate as parent.
    if not cross_boundary:
        return {
            "mode": "parent",
            "parent_span_id": ctx.span_id,
            "trace_id": ctx.trace_id,
            "baggage": dict(ctx.baggage),
            "sampled": ctx.sampled,
        }

    # Cross-boundary: authority must be verified.
    if ctx.federation_id not in policy.trusted_peers:
        raise UntrustedFederationContextError(
            f"refusing parent propagation: federation {ctx.federation_id!r} "
            f"is not a trusted peer of {local_federation!r}"
        )

    if not ctx.signed:
        raise UntrustedFederationContextError(
            f"refusing parent propagation: context from {ctx.federation_id!r} "
            f"is unsigned; possession of traceparent is not authority"
        )

    age = now - ctx.created_at
    if age > policy.max_context_age_s:
        # Stale: correlate via link, but never adopt as parent.
        return {
            "mode": "link",
            "parent_span_id": None,
            "trace_id": ctx.trace_id,
            "baggage": {},
            "sampled": None,  # local sampler decides
            "link_reason": f"stale_context age={age:.0f}s",
        }

    # Fresh, signed, trusted peer: correlate, but do NOT inherit authority.
    filtered_baggage = {
        k: v
        for k, v in ctx.baggage.items()
        if k in policy.allowed_baggage_keys
    }
    return {
        "mode": "link",
        "parent_span_id": None,  # federated edge is a link, not a parent
        "trace_id": ctx.trace_id,
        "baggage": filtered_baggage,
        "sampled": None,  # local sampling policy re-decides
        "link_reason": "cross_federation_correlation",
    }


def ingest_remote_context(
    ctx: SpanContext,
    local_federation: str,
    policy: TrustPolicy,
) -> dict:
    """Entry point used by the span processor at a federation boundary."""
    decision = decide_propagation(ctx, local_federation, policy)
    if decision["mode"] == "link" and decision.get("sampled") is None:
        # Local sampler runs here; placeholder for policy hook.
        decision["sampled"] = True
    return decision


# Example of an explicit refusal path used by a stricter caller:
def require_trusted_parent(ctx: SpanContext, local_federation: str, policy: TrustPolicy) -> SpanContext:
    if ctx.federation_id != local_federation and ctx.federation_id not in policy.trusted_peers:
        raise UntrustedFederationContextError(
            f"cannot adopt parent from untrusted federation {ctx.federation_id!r}"
        )
    return ctx