butler/gateway/gateway_contracts.py

Rama main · réplica de lectura del repositorio original.

"""Gateway wiring for :class:`butler.contracts` OwnerGate + BridgeAccess (R1 续)."""

from __future__ import annotations

from typing import Any, cast


def register_gateway_contracts() -> None:
    """Wire contracts registries to live gateway owner/bridge helpers (idempotent)."""
    from butler.contracts.gateway_registry import (
        get_bridge_access,
        get_owner_gate,
        set_bridge_access,
        set_owner_gate,
    )

    if get_owner_gate() is not None and get_bridge_access() is not None:
        return

    from butler.gateway.completion_notify import try_push_workflow_failure
    from butler.gateway.outbound_bridge import get_gateway_bridge_optional
    from butler.gateway.owner_gate import is_gateway_owner, owner_required_message

    class _GatewayOwnerGate:
        def is_gateway_owner(
            self,
            *,
            platform: str,
            external_id: str | None = None,
            session_key: str = "",
        ) -> bool:
            return bool(
                is_gateway_owner(
                    platform=platform,
                    external_id=external_id,
                    session_key=session_key,
                )
            )

        def owner_required_message(self) -> str:
            return cast(str, owner_required_message())

    class _GatewayBridgeAccess:
        def get_optional_bridge(self) -> Any:
            return get_gateway_bridge_optional()

        def try_push_workflow_failure(
            self,
            workflow_name: str,
            error: Exception | str,
            *,
            session_key: str = "",
        ) -> bool:
            bridge = self.get_optional_bridge()
            return bool(
                try_push_workflow_failure(
                    bridge,
                    workflow_name,
                    error,
                    session_key=session_key,
                )
            )

    if get_owner_gate() is None:
        set_owner_gate(_GatewayOwnerGate())
    if get_bridge_access() is None:
        set_bridge_access(_GatewayBridgeAccess())

    from butler.contracts.inbound_idempotency_registry import (
        get_inbound_idempotency_port,
        set_inbound_idempotency_port,
    )

    if get_inbound_idempotency_port() is None:

        class _GatewayInboundIdempotencyPort:
            def reset_session(self, session_id: str) -> None:
                from butler.resilience.inbound_idempotency import reset_session

                reset_session(session_id)

        set_inbound_idempotency_port(_GatewayInboundIdempotencyPort())


__all__ = ["register_gateway_contracts"]