diff --git a/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py b/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py index ceaf09b527..298148348c 100644 --- a/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py +++ b/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py @@ -1359,18 +1359,33 @@ def _materialize_known_handler_dependencies( ) if value is not None } - if ( - "dispatch_port" in constructor_params - and event_bus is not None - and requires_delegation_port - ): - from omnibase_infra.runtime.service_delegation_dispatch_port import ( - RuntimeDelegationDispatchPort, - ) + if "dispatch_port" in constructor_params and requires_delegation_port: + # When container is available, inject ContainerBackedDelegationDispatchPort + # which lazy-resolves the DirectBridgeDelegationDispatchPort registered by + # PluginDelegation at start_consumers() time. This avoids the asyncio stall + # that occurs when the Kafka-backed RuntimeDelegationDispatchPort subscribes + # to delegation-completed and awaits while the delegation orchestrator runs + # in the same event loop. + if container is not None: + from omnibase_infra.runtime.service_delegation_dispatch_port import ( + ContainerBackedDelegationDispatchPort, + RuntimeDelegationDispatchPort, + ) - available["dispatch_port"] = RuntimeDelegationDispatchPort( - cast("ProtocolPatternBBrokerTransport", event_bus) - ) + available["dispatch_port"] = ContainerBackedDelegationDispatchPort( + container=container, + fallback_event_bus=cast( + "ProtocolPatternBBrokerTransport | None", event_bus + ), + ) + elif event_bus is not None: + from omnibase_infra.runtime.service_delegation_dispatch_port import ( + RuntimeDelegationDispatchPort, + ) + + available["dispatch_port"] = RuntimeDelegationDispatchPort( + cast("ProtocolPatternBBrokerTransport", event_bus) + ) if not required_params <= set(available): return materialized_explicit_dependencies diff --git a/src/omnibase_infra/runtime/protocols/protocol_delegation_dispatch_port.py b/src/omnibase_infra/runtime/protocols/protocol_delegation_dispatch_port.py new file mode 100644 index 0000000000..4278a7762e --- /dev/null +++ b/src/omnibase_infra/runtime/protocols/protocol_delegation_dispatch_port.py @@ -0,0 +1,32 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Protocol interface for delegation dispatch ports. + +Defines the structural interface that both RuntimeDelegationDispatchPort and +ContainerBackedDelegationDispatchPort conform to, enabling handler_wiring to +inject either implementation without a concrete dependency. +""" + +from __future__ import annotations + +from typing import Protocol +from uuid import UUID + + +class ProtocolDelegationDispatchPort(Protocol): + async def dispatch( + self, + *, + prompt: str, + task_type: str, + correlation_id: UUID, + max_tokens: int, + source_file_path: str | None, + source_session_id: str | None, + wait: bool, + quality_contract_mode: str, + acceptance_criteria: tuple[str, ...], + ) -> dict[str, object]: ... + + +__all__ = ["ProtocolDelegationDispatchPort"] diff --git a/src/omnibase_infra/runtime/service_delegation_dispatch_port.py b/src/omnibase_infra/runtime/service_delegation_dispatch_port.py index c1965c9d53..cfd17e92c5 100644 --- a/src/omnibase_infra/runtime/service_delegation_dispatch_port.py +++ b/src/omnibase_infra/runtime/service_delegation_dispatch_port.py @@ -18,6 +18,9 @@ from omnibase_infra.runtime.models.model_pattern_b_broker_config import ( ModelPatternBBrokerConfig, ) +from omnibase_infra.runtime.protocols.protocol_delegation_dispatch_port import ( + ProtocolDelegationDispatchPort, +) from omnibase_infra.runtime.runtime_local_ingress import ( RuntimeLocalIngressRoute, discover_runtime_local_ingress_routes, @@ -210,4 +213,82 @@ async def dispatch( ) -__all__ = ["RuntimeDelegationDispatchPort"] +class ContainerBackedDelegationDispatchPort: + """Delegation dispatch port that lazy-resolves a bridge-backed port from the container. + + At construction time the bridge singleton may not yet exist (plugins register it + during start_consumers(), which runs after auto-wiring). This port holds a reference + to the container and resolves the ``DirectBridgeDelegationDispatchPort`` on first + dispatch, falling back to the Kafka-backed ``RuntimeDelegationDispatchPort`` if + the bridge port is not yet registered. + """ + + _BRIDGE_PORT_KEY = "DirectBridgeDelegationDispatchPort" + + def __init__( + self, + *, + container: object, + fallback_event_bus: ProtocolPatternBBrokerTransport | None = None, + ) -> None: + self._container = container + self._fallback_event_bus = fallback_event_bus + self._resolved: ProtocolDelegationDispatchPort | None = None + + async def _resolve(self) -> ProtocolDelegationDispatchPort: + if self._resolved is not None: + return self._resolved + registry = getattr(self._container, "service_registry", None) + if registry is not None: + try: + port: ProtocolDelegationDispatchPort = await registry.resolve_service( + self._BRIDGE_PORT_KEY + ) + self._resolved = port + return port + except Exception: # noqa: BLE001 — fallback-ok: bridge not yet registered, use Kafka port + pass + if self._fallback_event_bus is not None: + fallback: ProtocolDelegationDispatchPort = RuntimeDelegationDispatchPort( + self._fallback_event_bus + ) + self._resolved = fallback + return fallback + raise RuntimeError( + "ContainerBackedDelegationDispatchPort: no bridge port in container and " + "no fallback event_bus provided" + ) + + async def dispatch( + self, + *, + prompt: str, + task_type: str, + correlation_id: UUID, + max_tokens: int, + source_file_path: str | None, + source_session_id: str | None, + wait: bool, + output_schema_key: str | None = None, + quality_contract_mode: str = "extend_task_class", + acceptance_criteria: tuple[str, ...] = (), + ) -> dict[str, object]: + port = await self._resolve() + return await port.dispatch( + prompt=prompt, + task_type=task_type, + correlation_id=correlation_id, + max_tokens=max_tokens, + source_file_path=source_file_path, + source_session_id=source_session_id, + wait=wait, + quality_contract_mode=quality_contract_mode, + acceptance_criteria=acceptance_criteria, + ) + + +__all__ = [ + "ContainerBackedDelegationDispatchPort", + "ProtocolDelegationDispatchPort", + "RuntimeDelegationDispatchPort", +] diff --git a/tests/unit/contracts/test_protocol_ownership.py b/tests/unit/contracts/test_protocol_ownership.py index 339280698b..e996c941b0 100644 --- a/tests/unit/contracts/test_protocol_ownership.py +++ b/tests/unit/contracts/test_protocol_ownership.py @@ -122,6 +122,7 @@ "ProtocolSecretResolver": "runtime/config_discovery/models/protocol_secret_resolver.py", "ProtocolSecretResolverMetrics": "runtime/secret_resolver.py", "ProtocolHandleable": "runtime/auto_wiring/handler_wiring.py", # [RUNTIME] OMN-7656 auto-wiring dispatch + "ProtocolDelegationDispatchPort": "runtime/protocols/protocol_delegation_dispatch_port.py", # [RUNTIME] OMN-E0 delegation dispatch port interface — infra-internal, narrows dispatch surface for handler injection "ProtocolLocalRuntimeBus": "protocols/protocol_local_runtime_bus.py", # [RUNTIME] OMN-11570 local runtime bus adapter boundary "ProtocolLocalRuntimeCallableTarget": "protocols/protocol_local_runtime_callable_target.py", # [RUNTIME] OMN-11570 local handler invocation boundary "ProtocolLocalRuntimeDumpModel": "protocols/protocol_local_runtime_dump_model.py", # [RUNTIME] OMN-11570 local result serialization boundary