# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-structural-signals-indicate-a-trust-anchor-exchange-has-silently-degraded-into-unilateral-policy-acceptance-in-cross-org-multi-agent-federation.md
# Remnant Warrant:    https://corpusfetch.com/c/what-structural-signals-indicate-a-trust-anchor-exchange-has-silently-degraded-into-unilateral-policy-acceptance-in-cross-org-multi-agent-federation.warrant.json
# Markdown SHA-256:   02cee7b5bb4c2c2f8d36dcdbe397b8902d41e5c7b93fdbded22d1c10e383a739
# Code AST SHA-256:   b98ce2a3af5a7aed97030bc9c201a154b38d7e86289947f4b6d4d215aca70218
# Verified Classes:   ReconciliationCollapseError, PolicyClass, Decision
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from enum import Enum


class ReconciliationCollapseError(RuntimeError):
    """Raised when trust-anchor reconciliation has silently collapsed to
    unilateral policy acceptance on a deny-bearing policy class."""
    pass


class PolicyClass(Enum):
    PERMISSIVE = "permissive"
    DENY_BEARING = "deny_bearing"


@dataclass(frozen=True)
class Decision:
    request_id: str
    provenance: str          # e.g. "A&B" (intersection) or "A" (single-org)
    policy_class: PolicyClass
    orgs_configured: int     # number of orgs whose policy sources are live


def assert_reconciliation_intact(
    decisions: list[Decision],
    *,
    single_org_rate_threshold: float = 0.20,
) -> None:
    """Validate that reconciliation has not collapsed to unilateral acceptance.

    Raises ReconciliationCollapseError on either:
      (1) any single-org-provenance decision on a DENY_BEARING class, or
      (2) a single-org-provenance rate above threshold over the window.
    """
    if not decisions:
        return

    single_org = [d for d in decisions if "&" not in d.provenance]

    for d in single_org:
        if d.policy_class is PolicyClass.DENY_BEARING:
            raise ReconciliationCollapseError(
                f"request {d.request_id}: provenance={d.provenance!r} on "
                f"DENY_BEARING class while {d.orgs_configured} orgs configured "
                f"-- unilateral acceptance of a deny constraint"
            )

    rate = len(single_org) / len(decisions)
    if rate > single_org_rate_threshold:
        raise ReconciliationCollapseError(
            f"single-org provenance rate {rate:.3f} exceeds "
            f"{single_org_rate_threshold:.3f} over {len(decisions)} decisions "
            f"-- reconciliation drift, force hard re-anchor"
        )