From 81d02eec4f56e101030023d4c45a7fa05ab61848 Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Mon, 25 May 2026 18:57:47 -0400 Subject: [PATCH 1/2] feat(OMN-E0): inject ContainerBackedDelegationDispatchPort at auto-wiring time Replaces direct RuntimeDelegationDispatchPort injection with ContainerBackedDelegationDispatchPort, which lazy-resolves the in-process DirectBridgeDelegationDispatchPort from the DI container at first dispatch() call (after PluginDelegation.start_consumers() has registered it). Falls back to RuntimeDelegationDispatchPort when no container or bridge port is available. --- .../runtime/auto_wiring/handler_wiring.py | 32 +++++-- .../service_delegation_dispatch_port.py | 89 ++++++++++++++++++- 2 files changed, 113 insertions(+), 8 deletions(-) diff --git a/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py b/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py index 57b2de7d54..47b1fae2b8 100644 --- a/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py +++ b/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py @@ -1339,16 +1339,34 @@ def _materialize_known_handler_dependencies( } 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, - ) + # 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/service_delegation_dispatch_port.py b/src/omnibase_infra/runtime/service_delegation_dispatch_port.py index c1965c9d53..a95168a17c 100644 --- a/src/omnibase_infra/runtime/service_delegation_dispatch_port.py +++ b/src/omnibase_infra/runtime/service_delegation_dispatch_port.py @@ -7,6 +7,7 @@ from collections.abc import Mapping, Sequence from dataclasses import dataclass from datetime import UTC, datetime +from typing import Protocol from uuid import UUID from omnibase_core.models.dispatch.model_dispatch_bus_command import ( @@ -210,4 +211,90 @@ async def dispatch( ) -__all__ = ["RuntimeDelegationDispatchPort"] +class _ProtocolDispatchPort(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]: ... + + +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: _ProtocolDispatchPort | None = None + + async def _resolve(self) -> _ProtocolDispatchPort: + if self._resolved is not None: + return self._resolved + registry = getattr(self._container, "service_registry", None) + if registry is not None: + try: + port = 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 = 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( # type: ignore[union-attr] + 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", "RuntimeDelegationDispatchPort"] From 324a104e48650eec42bc9c63f005828e72c87458 Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Mon, 25 May 2026 22:04:45 -0400 Subject: [PATCH 2/2] fix(OMN-E0): resolve validator violations in delegation dispatch port MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Extract _ProtocolDispatchPort to runtime/protocols/protocol_delegation_dispatch_port.py, renamed to ProtocolDelegationDispatchPort (PascalCase, no leading underscore) - Removes mixed models+protocols in service_delegation_dispatch_port.py (architecture gate) - Eliminates redundant # type: ignore[union-attr] — mypy is clean against typed Protocol - Add ProtocolDelegationDispatchPort to KNOWN_INFRA_PROTOCOLS allowlist ([RUNTIME] category) - Run ruff format on handler_wiring.py - Stamp schema fingerprint (69 migration files) --- docker/migrations/schema_fingerprint.sha256 | 2 +- .../runtime/auto_wiring/handler_wiring.py | 5 +-- .../protocol_delegation_dispatch_port.py | 32 +++++++++++++++ .../service_delegation_dispatch_port.py | 40 ++++++++----------- .../unit/contracts/test_protocol_ownership.py | 1 + 5 files changed, 52 insertions(+), 28 deletions(-) create mode 100644 src/omnibase_infra/runtime/protocols/protocol_delegation_dispatch_port.py diff --git a/docker/migrations/schema_fingerprint.sha256 b/docker/migrations/schema_fingerprint.sha256 index 91ab462f13..4a40e8b1ce 100644 --- a/docker/migrations/schema_fingerprint.sha256 +++ b/docker/migrations/schema_fingerprint.sha256 @@ -2,5 +2,5 @@ # Regenerate: python scripts/check_schema_fingerprint.py stamp # Verify: python scripts/check_schema_fingerprint.py verify sha256:6f4891be982289898e5cf47ded9921e301136041df1a2859a3e5f4487532d46f -generated_at: 2026-05-25T23:08:52Z +generated_at: 2026-05-26T02:17:19Z migration_file_count: 69 diff --git a/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py b/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py index 47b1fae2b8..997448991c 100644 --- a/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py +++ b/src/omnibase_infra/runtime/auto_wiring/handler_wiring.py @@ -1337,10 +1337,7 @@ def _materialize_known_handler_dependencies( ) if value is not None } - if ( - "dispatch_port" in constructor_params - and requires_delegation_port - ): + 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 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 a95168a17c..cfd17e92c5 100644 --- a/src/omnibase_infra/runtime/service_delegation_dispatch_port.py +++ b/src/omnibase_infra/runtime/service_delegation_dispatch_port.py @@ -7,7 +7,6 @@ from collections.abc import Mapping, Sequence from dataclasses import dataclass from datetime import UTC, datetime -from typing import Protocol from uuid import UUID from omnibase_core.models.dispatch.model_dispatch_bus_command import ( @@ -19,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, @@ -211,22 +213,6 @@ async def dispatch( ) -class _ProtocolDispatchPort(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]: ... - - class ContainerBackedDelegationDispatchPort: """Delegation dispatch port that lazy-resolves a bridge-backed port from the container. @@ -247,21 +233,25 @@ def __init__( ) -> None: self._container = container self._fallback_event_bus = fallback_event_bus - self._resolved: _ProtocolDispatchPort | None = None + self._resolved: ProtocolDelegationDispatchPort | None = None - async def _resolve(self) -> _ProtocolDispatchPort: + 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 = await registry.resolve_service(self._BRIDGE_PORT_KEY) + 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 = RuntimeDelegationDispatchPort(self._fallback_event_bus) + fallback: ProtocolDelegationDispatchPort = RuntimeDelegationDispatchPort( + self._fallback_event_bus + ) self._resolved = fallback return fallback raise RuntimeError( @@ -284,7 +274,7 @@ async def dispatch( acceptance_criteria: tuple[str, ...] = (), ) -> dict[str, object]: port = await self._resolve() - return await port.dispatch( # type: ignore[union-attr] + return await port.dispatch( prompt=prompt, task_type=task_type, correlation_id=correlation_id, @@ -297,4 +287,8 @@ async def dispatch( ) -__all__ = ["ContainerBackedDelegationDispatchPort", "RuntimeDelegationDispatchPort"] +__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