# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-silent-cardinality-inflation-in-cross-org-trace-aggregation-when-k-anonymity-buckets-are-merged.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-silent-cardinality-inflation-in-cross-org-trace-aggregation-when-k-anonymity-buckets-are-merged.warrant.json
# Markdown SHA-256:   a022ff62b2225f1f46d78b51f8b70b341b02a1a3d88716b573449d470fb67a1b
# Code AST SHA-256:   7a7cb13f7ffa2cac517ab1aab4f1870a0bfd364051999fbb507d1aa9bf5ca119
# Verified Classes:   ReconciliationFault
# Defensive Guards:   2 fail-closed invariants (zero external deps)

class ReconciliationFault(RuntimeError):
    """Raised when cross-org bucket reconciliation produces a cardinality mismatch
    that cannot be attributed to a declared lossy merge policy."""
    pass


RESERVED_UNMAPPED = "__unmapped__"


def reconcile_keys(
    contributor_keys: dict[str, set[str]],
    closure_table: dict[str, str],
    declared_cardinality: dict[str, int],
    expected_reduction: dict[str, int] | None = None,
) -> dict[str, set[str]]:
    """
    contributor_keys: org_id -> set of raw dimension keys emitted by that org
    closure_table: raw_key -> global_key (must be total over all raw keys)
    declared_cardinality: dimension -> number of logical keys the federation claims
    expected_reduction: dimension -> expected (declared - observed) under a documented lossy policy

    Returns global_key -> set of org_ids contributing to it.
    Raises ReconciliationFault on any mismatch that is not explained by expected_reduction.
    """
    if expected_reduction is None:
        expected_reduction = {}

    reconciled: dict[str, set[str]] = {}
    unmapped_by_org: dict[str, set[str]] = {}

    for org_id, raw_keys in contributor_keys.items():
        for raw in raw_keys:
            global_key = closure_table.get(raw)
            if global_key is None:
                # Total function requirement violated: do NOT pass through.
                unmapped_by_org.setdefault(org_id, set()).add(raw)
                global_key = RESERVED_UNMAPPED
            reconciled.setdefault(global_key, set()).add(org_id)

    if RESERVED_UNMAPPED in reconciled:
        raise ReconciliationFault(
            f"reconciliation is not total; unmapped keys by org: "
            f"{ {o: sorted(k) for o, k in unmapped_by_org.items()} }"
        )

    for dim, declared in declared_cardinality.items():
        observed = len(reconciled)
        expected = declared - expected_reduction.get(dim, 0)
        if observed != expected:
            kind = "inflation (unmerged siblings)" if observed > expected else "over-merge"
            raise ReconciliationFault(
                f"dimension {dim!r}: declared={declared} expected={expected} "
                f"observed={observed} -> {kind}; refusing to publish"
            )

    return reconciled