# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-reconcile-trace-correlation-when-federated-agent-hops-redact-parent-span-identifiers.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-reconcile-trace-correlation-when-federated-agent-hops-redact-parent-span-identifiers.warrant.json
# Markdown SHA-256:   2deebfa817a36920eb726ddf2081d538ce533351b02e80a81584e2df5809b80a
# Code AST SHA-256:   c58e2ded6a7a16c45ada75e196fe8c682e8199d358a52364fe4c303f6cb8a178
# Verified Classes:   FederatedCorrelationError
# Defensive Guards:   3 fail-closed invariants (zero external deps)

class FederatedCorrelationError(RuntimeError):
    """Raised when a federated trace's parent chain cannot be resolved
    under the redaction policy in force at span emission time."""
    pass


def assemble_federated_trace(spans, policy_registry):
    """
    spans: list of dicts with keys:
        trace_id, span_id, parent_span_id, policy_version, emitted_at
    policy_registry: maps policy_version -> dict with
        'redaction_salt' (str or None), 'rotation_window_s' (int)

    Returns a dict trace_id -> list of root span_ids, or raises
    FederatedCorrelationError if any span's parent is unresolvable
    under a *stable* policy within the trace.
    """
    by_trace = {}
    for s in spans:
        by_trace.setdefault(s["trace_id"], []).append(s)

    assembled = {}
    for trace_id, trace_spans in by_trace.items():
        present_ids = {s["span_id"] for s in trace_spans}
        versions = {s["policy_version"] for s in trace_spans}

        # A single trace must be joinable under one policy instance.
        # Multiple policy versions imply a salt-boundary crossing.
        if len(versions) > 1:
            raise FederatedCorrelationError(
                f"trace {trace_id} spans policy versions {sorted(versions)}; "
                f"parent IDs hashed under different salts cannot be joined"
            )

        version = next(iter(versions))
        policy = policy_registry.get(version)
        if policy is None:
            raise FederatedCorrelationError(
                f"trace {trace_id} references unknown policy version {version}"
            )

        roots = []
        for s in trace_spans:
            pid = s["parent_span_id"]
            if pid is None:
                roots.append(s["span_id"])
                continue
            if pid in present_ids:
                continue
            # Parent absent. Is that expected (external, never exported)
            # or anomalous (redacted but should have been joinable)?
            if policy["redaction_salt"] is None:
                # No redaction in force: a missing parent is a real gap.
                raise FederatedCorrelationError(
                    f"trace {trace_id} span {s['span_id']} has unresolvable "
                    f"parent {pid} under non-redacting policy {version}"
                )
            # Redaction in force: parent may be legitimately external.
            # Emit as explicit federated orphan, do not silently re-root.
            s["federated_orphan"] = True
            roots.append(s["span_id"])

        assembled[trace_id] = roots

    return assembled