# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-federation-policy-drift-when-cross-org-data-residency-negotiation-silently-degrades-tool-access-scope.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-federation-policy-drift-when-cross-org-data-residency-negotiation-silently-degrades-tool-access-scope.warrant.json
# Markdown SHA-256:   f9bfa01d90940f66dd64c85c94e751b7a7c5b2fad01efebbc7d9bada5ecc3694
# Code AST SHA-256:   84aff1b939fe7f7e82523d4d47e056cf498efacea31aafef8a71c6edb2ad7d6c
# Verified Classes:   FederationClosureError, ToolManifest, FederationBundle
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Any

ALL_JURISDICTIONS = frozenset({"*"})

class FederationClosureError(RuntimeError):
    """Raised when an accepted federation bundle has a tool whose argument
    schema closure escapes the accepted residency set."""
    pass

@dataclass
class ToolManifest:
    name: str
    schema: dict[str, Any]

@dataclass
class FederationBundle:
    residency: frozenset[str]
    tools: list[ToolManifest]
    peer_org: str

def _schema_closure(schema: dict[str, Any]) -> frozenset[str]:
    """Return the set of jurisdiction tags reachable by any argument path.
    Open schemas are treated as reaching ALL jurisdictions."""
    if schema.get("additionalProperties", False) is True:
        return ALL_JURISDICTIONS
    if schema.get("type") == "object" and "properties" not in schema:
        return ALL_JURISDICTIONS
    tags: set[str] = set()
    for prop in schema.get("properties", {}).values():
        if "x-jurisdiction" in prop:
            tags.add(prop["x-jurisdiction"])
        if prop.get("type") in ("object", "array"):
            nested = prop.get("items", prop)
            tags |= _schema_closure(nested)
    return frozenset(tags) if tags else frozenset()

def validate_bundle(bundle: FederationBundle) -> None:
    for tool in bundle.tools:
        closure = _schema_closure(tool.schema)
        if closure == ALL_JURISDICTIONS:
            raise FederationClosureError(
                f"peer={bundle.peer_org} tool={tool.name}: open schema "
                f"reaches ALL jurisdictions; residency={sorted(bundle.residency)} "
                f"cannot contain it"
            )
        if not closure <= bundle.residency:
            raise FederationClosureError(
                f"peer={bundle.peer_org} tool={tool.name}: closure="
                f"{sorted(closure)} escapes residency={sorted(bundle.residency)}"
            )

def accept_bundle(bundle: FederationBundle) -> str:
    try:
        validate_bundle(bundle)
    except FederationClosureError as exc:
        return f"degraded: {exc}"
    return "active"