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
14 changes: 0 additions & 14 deletions src/omnibase_infra/event_bus/topic_constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
48 changes: 48 additions & 0 deletions src/omnibase_infra/runtime/service_dispatch_result_applier.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down
102 changes: 102 additions & 0 deletions tests/unit/runtime/test_service_dispatch_result_applier.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Loading