# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-of-a-federated-tool-call-dependency-graph-deadlock-silently-when-every-cycle-it-contains-is-already-marked-recoverable.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-of-a-federated-tool-call-dependency-graph-deadlock-silently-when-every-cycle-it-contains-is-already-marked-recoverable.warrant.json
# Markdown SHA-256:   2b59a3e53406e6d222d4a90ddda67571ce883aa9f6b8d5ab63b6c099925b29a5
# Code AST SHA-256:   3ef48b2534327c58d76c21db4de1fa3b84e3c786802909e239abdf0d5c7b8dc2
# Verified Classes:   RecoveryInterlockError, RecoveryCoordinator
# Defensive Guards:   2 fail-closed invariants (zero external deps)

class RecoveryInterlockError(RuntimeError):
    """Raised when a local detector would dispatch an in-band recovery RPC
    that could form a cycle in the federated recovery subgraph."""
    pass


class RecoveryCoordinator:
    def __init__(self, domain_id: str, serialized: bool):
        self.domain_id = domain_id
        self.serialized = serialized
        self.in_flight = set()  # (src_agent, dst_agent) recovery edges

    def propose_recovery(self, cycle: list[tuple[str, str]]) -> bool:
        if not self.serialized:
            raise RecoveryInterlockError(
                f"domain {self.domain_id}: autonomous in-band recovery is unsafe; "
                f"recovery subgraph is not serialized and cycle {cycle} may interlock "
                f"with peer-domain recovery actions"
            )
        for edge in cycle:
            if edge in self.in_flight:
                raise RecoveryInterlockError(
                    f"domain {self.domain_id}: recovery edge {edge} already in flight; "
                    f"dispatching would create a recovery-subgraph cycle"
                )
        # Coordinator serializes: mark edges, then dispatch.
        self.in_flight.update(cycle)
        return True

    def release(self, cycle: list[tuple[str, str]]) -> None:
        self.in_flight.difference_update(cycle)