# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/in-what-scenarios-must-an-instance-refuse-federated-policy-convergence-when-data-residency-constraints-are-only-satisfiable-through-transitive-delegat.md
# Remnant Warrant:    https://corpusfetch.com/c/in-what-scenarios-must-an-instance-refuse-federated-policy-convergence-when-data-residency-constraints-are-only-satisfiable-through-transitive-delegat.warrant.json
# Markdown SHA-256:   1b2a407e0c8ee377eb163df68bee6e39d1fa3afe169348a4d497293d62162703
# Code AST SHA-256:   abef40ced65f36bf2b9c3024397ed8048fbfd130a1feca758b6d9867dbc225d2
# Verified Classes:   ResidencyConvergenceError, Artifact, NegotiationState
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Iterable


class ResidencyConvergenceError(RuntimeError):
    """Raised when a federation negotiator would converge on a policy state
    that is individually ALLOW at every engine but violates data residency
    along a transitive artifact provenance path."""


@dataclass(frozen=True)
class Artifact:
    artifact_id: str
    source_jurisdiction: str
    permitted_consumers: frozenset[str]  # jurisdictions allowed to consume
    parents: tuple[str, ...] = field(default_factory=tuple)


@dataclass
class NegotiationState:
    per_engine_verdicts: dict[str, str]  # engine_id -> "ALLOW"/"DENY"
    artifacts: dict[str, Artifact]
    consumed_by: dict[str, set[str]]  # artifact_id -> set of consuming jurisdictions


def _lineage_paths(
    artifacts: dict[str, Artifact], artifact_id: str
) -> Iterable[list[str]]:
    art = artifacts[artifact_id]
    if not art.parents:
        yield [artifact_id]
        return
    for parent in art.parents:
        for path in _lineage_paths(artifacts, parent):
            yield path + [artifact_id]


def assert_residency_before_convergence(state: NegotiationState) -> None:
    """
    Gate terminal negotiation state on provenance-DAG residency.

    Necessary-but-not-sufficient: every engine returning ALLOW does NOT imply
    the composite dataflow respects residency. We must walk each artifact's
    lineage and verify that every consuming jurisdiction is permitted by the
    *originating* constraint, not merely by the immediate parent's constraint.
    """
    # Precondition: the naive convergence predicate is satisfied.
    if any(v != "ALLOW" for v in state.per_engine_verdicts.values()):
        raise ResidencyConvergenceError(
            "convergence precondition not met: at least one engine did not ALLOW"
        )

    for art_id, consumers in state.consumed_by.items():
        for path in _lineage_paths(state.artifacts, art_id):
            origin = state.artifacts[path[0]]
            for consumer_jurisdiction in consumers:
                if consumer_jurisdiction not in origin.permitted_consumers:
                    raise ResidencyConvergenceError(
                        f"transitive residency violation: artifact {art_id} "
                        f"lineage {path} originates in "
                        f"{origin.source_jurisdiction} permitting "
                        f"{sorted(origin.permitted_consumers)}, but consumed "
                        f"in {consumer_jurisdiction}. Refusing to converge "
                        f"despite unanimous per-engine ALLOW."
                    )
    # Only reached if every lineage path is jurisdictionally clean.
    return None


# --- Example invocation demonstrating the failure boundary ---
if __name__ == "__main__":
    state = NegotiationState(
        per_engine_verdicts={"eu-orch": "ALLOW", "us-agent-b": "ALLOW", "us-agent-c": "ALLOW"},
        artifacts={
            "doc-raw": Artifact("doc-raw", "EU", frozenset({"EU"})),
            "doc-embed": Artifact("doc-embed", "US", frozenset({"EU", "US"}), parents=("doc-raw",)),
        },
        consumed_by={"doc-embed": {"US"}},
    )
    try:
        assert_residency_before_convergence(state)
    except ResidencyConvergenceError as e:
        print(f"negotiation refused: {e}")