From a7d2b2a5c89992945bc409cbdd2c9e58d4f3ae86 Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Mon, 25 May 2026 14:33:32 -0400 Subject: [PATCH 1/2] fix(OMN-12116): set event_type on output envelopes in dispatch result applier MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Without this fix, DispatchResultApplier published output envelopes with event_type=None. Multi-step FSM orchestrators register dispatchers under the topic-derived alias (e.g. 'omnimarket.swarm-endpoint-health-completed') but received envelopes with event_type=None, causing every response event to be routed to DLQ. Fix: derive event_type from the resolved output topic using the same ONEX convention used by EventBusSubcontractWiring._derive_event_type_from_topic (strip 'onex.{kind}.' prefix and '.v{n}' suffix → '{producer}.{event-name}'). Non-ONEX fallback topics leave event_type as None (backwards-compatible). Adds TestEventTypeDerivedFromTopic with 6 cases covering: static derivation, cmd-kind topics, non-ONEX fallback, topic_router path, output_topic_map path, and non-ONEX topic returning None. --- .../service_dispatch_result_applier.py | 48 +++++++++ .../test_service_dispatch_result_applier.py | 102 ++++++++++++++++++ 2 files changed, 150 insertions(+) diff --git a/src/omnibase_infra/runtime/service_dispatch_result_applier.py b/src/omnibase_infra/runtime/service_dispatch_result_applier.py index 4502a2719e..fa4537d9fe 100644 --- a/src/omnibase_infra/runtime/service_dispatch_result_applier.py +++ b/src/omnibase_infra/runtime/service_dispatch_result_applier.py @@ -296,6 +296,38 @@ def _allowed_output_topics(self) -> set[str]: *self._allowed_output_topic_set, } + @staticmethod + def _derive_event_type_from_topic(topic: str) -> str | None: + """Derive the event_type routing key from an ONEX topic name. + + ONEX topics follow the convention:: + + onex.{kind}.{producer}.{event-name}.v{n} + + This method extracts ``{producer}.{event-name}`` as a dot-path routing + key suitable for ``ModelEventEnvelope.event_type``, which is the alias + format used by dispatcher registration (e.g. + ``omnimarket.swarm-endpoint-health-completed``). + + Args: + topic: Full topic name following ONEX naming convention + (e.g., ``'onex.evt.omnimarket.swarm-endpoint-health-completed.v1'``). + + Returns: + Derived event_type as ``'{producer}.{event-name}'`` + (e.g., ``'omnimarket.swarm-endpoint-health-completed'``), or ``None`` + if the topic does not follow the expected ONEX format. + + .. versionadded:: 0.41.0 (OMN-12116) + """ + parts = topic.split(".") + if len(parts) >= 5 and parts[0] == "onex": + # onex.{kind}.{producer}.{event-name}.v{n} + producer = parts[2] + event_name = parts[3] + return f"{producer}.{event_name}" + return None + def _execute_projection( self, intent: ModelProjectionIntent, @@ -548,6 +580,22 @@ async def apply( self._resolve_mapped_output_topic(output_event), ) ) + + # OMN-12116: Derive event_type from the resolved topic so + # multi-step FSM orchestrators can match the dispatcher + # registration alias (e.g. 'omnimarket.swarm-endpoint-health-completed'). + # Without this, the envelope arrives with event_type=None + # and the orchestrator's dispatcher — registered under the + # topic-derived alias — cannot find a matching handler, + # causing the response event to be routed to DLQ. + derived_event_type = self._derive_event_type_from_topic( + resolved_topic + ) + if derived_event_type is not None: + output_envelope = output_envelope.model_copy( + update={"event_type": derived_event_type} + ) + await self._event_bus.publish_envelope( envelope=output_envelope, topic=resolved_topic, diff --git a/tests/unit/runtime/test_service_dispatch_result_applier.py b/tests/unit/runtime/test_service_dispatch_result_applier.py index 5610568786..d1f4f97cc5 100644 --- a/tests/unit/runtime/test_service_dispatch_result_applier.py +++ b/tests/unit/runtime/test_service_dispatch_result_applier.py @@ -464,3 +464,105 @@ async def test_no_topic_router_uses_output_topic() -> None: await applier.apply(result) published_topic = bus.publish_envelope.call_args.kwargs["topic"] assert published_topic == "responses" + + +# --------------------------------------------------------------------------- +# event_type derivation tests (OMN-12116) +# --------------------------------------------------------------------------- + + +class TestEventTypeDerivedFromTopic: + """Verify that event_type is set on output envelopes from the resolved topic. + + Without this fix multi-step FSM orchestrators DLQ response events because + the dispatcher is registered under the topic-derived alias + (e.g. 'omnimarket.swarm-endpoint-health-completed') but the envelope + arrived with event_type=None. + """ + + def test_derive_event_type_from_onex_topic(self) -> None: + """Topic alias derives to '{producer}.{event-name}'.""" + result = DispatchResultApplier._derive_event_type_from_topic( + "onex.evt.omnimarket.swarm-endpoint-health-completed.v1" + ) + assert result == "omnimarket.swarm-endpoint-health-completed" + + def test_derive_event_type_from_cmd_topic(self) -> None: + """Works for cmd-kind topics too.""" + result = DispatchResultApplier._derive_event_type_from_topic( + "onex.cmd.platform.node-registration.v2" + ) + assert result == "platform.node-registration" + + def test_derive_event_type_non_onex_topic_returns_none(self) -> None: + """Non-ONEX topic format returns None without crashing.""" + assert DispatchResultApplier._derive_event_type_from_topic("responses") is None + assert DispatchResultApplier._derive_event_type_from_topic("") is None + assert DispatchResultApplier._derive_event_type_from_topic("a.b.c") is None + + @pytest.mark.asyncio + async def test_event_type_set_on_published_envelope_from_onex_topic( + self, + ) -> None: + """OMN-12116: output envelope must carry the topic-derived event_type. + + When the resolved topic is an ONEX-format topic, the published envelope + must have event_type set to '{producer}.{event-name}' so multi-step FSM + orchestrators can match it to the dispatcher registration alias. + """ + bus = AsyncMock() + event = _StubEvent(value="health-result") + router = { + "_StubEvent": "onex.evt.omnimarket.swarm-endpoint-health-completed.v1" + } + result = _make_result(output_events=[event]) + + applier = DispatchResultApplier( + event_bus=bus, + output_topic="responses", + topic_router=router, + ) + await applier.apply(result) + + published_envelope = bus.publish_envelope.call_args.kwargs["envelope"] + assert ( + published_envelope.event_type + == "omnimarket.swarm-endpoint-health-completed" + ) + + @pytest.mark.asyncio + async def test_event_type_none_for_non_onex_fallback_topic(self) -> None: + """When resolved topic is not ONEX format, event_type stays None.""" + bus = AsyncMock() + event = _StubEvent(value="generic") + result = _make_result(output_events=[event]) + + applier = DispatchResultApplier( + event_bus=bus, + output_topic="generic-responses", + ) + await applier.apply(result) + + published_envelope = bus.publish_envelope.call_args.kwargs["envelope"] + # Non-ONEX topic → no derivation possible → event_type stays None. + assert published_envelope.event_type is None + + @pytest.mark.asyncio + async def test_event_type_set_via_output_topic_map(self) -> None: + """event_type derives from topic resolved via output_topic_map.""" + bus = AsyncMock() + event = _StubEvent(value="mapped") + output_topic_map = { + "_StubEvent": "onex.evt.omnimarket.swarm-dispatch-completed.v1" + } + result = _make_result(output_events=[event]) + + applier = DispatchResultApplier( + event_bus=bus, + output_topic="responses", + output_topic_map=output_topic_map, + ) + await applier.apply(result) + + published_envelope = bus.publish_envelope.call_args.kwargs["envelope"] + assert published_envelope.event_type == "omnimarket.swarm-dispatch-completed" From 4e6d16dc39359ef28eca3a8c22f987a435a3bce0 Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Mon, 25 May 2026 14:42:24 -0400 Subject: [PATCH 2/2] fix(OMN-12116): remove duplicate delegate skill topic constants --- src/omnibase_infra/event_bus/topic_constants.py | 14 -------------- 1 file changed, 14 deletions(-) diff --git a/src/omnibase_infra/event_bus/topic_constants.py b/src/omnibase_infra/event_bus/topic_constants.py index 167a4abf3f..b9dc610a97 100644 --- a/src/omnibase_infra/event_bus/topic_constants.py +++ b/src/omnibase_infra/event_bus/topic_constants.py @@ -525,20 +525,6 @@ def derive_dlq_topic_for_event_type( TOPIC_DELEGATE_SKILL_FAILED: Final[str] = "onex.evt.omnimarket.delegate-skill-failed.v1" """Event topic published by node_delegate_skill_orchestrator on skill dispatch failure.""" - -# --------------------------------------------------------------------------- -# Delegate-Skill Orchestrator Topics (OMN-11996) -# --------------------------------------------------------------------------- - -TOPIC_DELEGATE_SKILL_COMPLETED: Final[str] = ( - "onex.evt.omnimarket.delegate-skill-completed.v1" -) -"""Terminal success event from node_delegate_skill_orchestrator.""" - -TOPIC_DELEGATE_SKILL_FAILED: Final[str] = "onex.evt.omnimarket.delegate-skill-failed.v1" -"""Terminal failure event from node_delegate_skill_orchestrator.""" - - __all__ = [ "TOPIC_DELEGATE_SKILL_COMPLETED", "TOPIC_DELEGATE_SKILL_FAILED",