Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 26 additions & 11 deletions src/omnibase_infra/runtime/auto_wiring/handler_wiring.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
Original file line number Diff line number Diff line change
@@ -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]: ...

Check notice

Code scanning / CodeQL

Statement has no effect Note

This statement has no effect.


__all__ = ["ProtocolDelegationDispatchPort"]
83 changes: 82 additions & 1 deletion src/omnibase_infra/runtime/service_delegation_dispatch_port.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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",
]
1 change: 1 addition & 0 deletions tests/unit/contracts/test_protocol_ownership.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading