diff --git a/src/omnibase_infra/runtime/service_dispatch_result_applier.py b/src/omnibase_infra/runtime/service_dispatch_result_applier.py index 95fa75a6e4..834ec1142d 100644 --- a/src/omnibase_infra/runtime/service_dispatch_result_applier.py +++ b/src/omnibase_infra/runtime/service_dispatch_result_applier.py @@ -133,6 +133,7 @@ def __init__( intent_executor: IntentExecutor | None = None, clock: Callable[[], datetime] | None = None, projection_effect: ProtocolProjectionEffect | None = None, + topic_router: dict[str, str] | None = None, ) -> None: """Initialize the dispatch result applier. @@ -148,12 +149,21 @@ def __init__( (OMN-2510). When provided, its ``execute()`` is called with each ``ModelProjectionIntent`` in the dispatch result. Kafka publish is skipped if ``execute()`` raises. + topic_router: Optional mapping of Python event class names to their + declared Kafka topics (e.g. ``{"ModelNodeRegistrationAccepted": + "onex.evt.platform.node-registration-accepted.v1"}``). When + provided, each output event is published to its per-type topic + instead of the single ``output_topic`` fallback. Events whose + class name is not in the map fall back to ``output_topic``. + Build this map with ``build_topic_router_from_contract()`` + (OMN-4881). """ self._event_bus = event_bus self._output_topic = output_topic self._intent_executor = intent_executor self._clock = clock or (lambda: datetime.now(UTC)) self._projection_effect = projection_effect + self._topic_router: dict[str, str] = topic_router or {} def _resolve_partition_key(self, event: BaseModel) -> bytes | None: """Extract partition key from event model for per-entity ordering. @@ -423,15 +433,18 @@ async def apply( str(effective_correlation_id), ) + resolved_topic = self._topic_router.get( + type(output_event).__name__, self._output_topic + ) await self._event_bus.publish_envelope( envelope=output_envelope, - topic=self._output_topic, + topic=resolved_topic, key=partition_key, ) logger.info( "Published output event to %s (correlation_id=%s)", - self._output_topic, + resolved_topic, str(effective_correlation_id), extra={ "output_event_type": type(output_event).__name__, diff --git a/tests/unit/runtime/test_service_dispatch_result_applier.py b/tests/unit/runtime/test_service_dispatch_result_applier.py index 933e3050cd..9a3ba5c068 100644 --- a/tests/unit/runtime/test_service_dispatch_result_applier.py +++ b/tests/unit/runtime/test_service_dispatch_result_applier.py @@ -164,9 +164,10 @@ async def _track_publish(*args: object, **kwargs: object) -> None: ) await applier.apply(result) - assert call_order == ["execute_all", "publish_envelope"], ( - f"Expected intents before events, got: {call_order}" - ) + assert call_order == [ + "execute_all", + "publish_envelope", + ], f"Expected intents before events, got: {call_order}" # --------------------------------------------------------------------------- @@ -382,3 +383,86 @@ async def test_correlation_id_always_present(self) -> None: ) assert bare_result.correlation_id is not None assert isinstance(bare_result.correlation_id, UUID) + + +# --------------------------------------------------------------------------- +# topic_router tests (OMN-4881) +# --------------------------------------------------------------------------- + +import uuid as _uuid +from datetime import timedelta + +from omnibase_infra.models.registration.events.model_node_registration_accepted import ( + ModelNodeRegistrationAccepted, +) + + +def _make_accepted_event() -> ModelNodeRegistrationAccepted: + now = datetime.now(UTC) + return ModelNodeRegistrationAccepted( + entity_id=_uuid.uuid4(), + node_id=_uuid.uuid4(), + correlation_id=_uuid.uuid4(), + causation_id=_uuid.uuid4(), + emitted_at=now, + ack_deadline=now + timedelta(seconds=90), + ) + + +def _make_result_with(events: list) -> ModelDispatchResult: # type: ignore[type-arg] + return ModelDispatchResult( + status=EnumDispatchStatus.SUCCESS, + topic="onex.evt.platform.node-introspection.v1", + started_at=datetime.now(UTC), + output_events=events, + ) + + +@pytest.mark.asyncio +@pytest.mark.unit +async def test_topic_router_routes_known_event_to_declared_topic() -> None: + """Router overrides output_topic for a known event type.""" + bus = AsyncMock() + router = { + "ModelNodeRegistrationAccepted": "onex.evt.platform.node-registration-accepted.v1" + } + applier = DispatchResultApplier( + event_bus=bus, + output_topic="responses", + topic_router=router, + ) + result = _make_result_with([_make_accepted_event()]) + await applier.apply(result) + published_topic = bus.publish_envelope.call_args.kwargs["topic"] + assert published_topic == "onex.evt.platform.node-registration-accepted.v1" + + +@pytest.mark.asyncio +@pytest.mark.unit +async def test_topic_router_falls_back_for_unknown_event_type() -> None: + """Router falls back to output_topic for event types not in the map.""" + bus = AsyncMock() + applier = DispatchResultApplier( + event_bus=bus, + output_topic="responses", + topic_router={"ModelSomeOtherClass": "onex.evt.platform.other.v1"}, + ) + result = _make_result_with([_make_accepted_event()]) + await applier.apply(result) + published_topic = bus.publish_envelope.call_args.kwargs["topic"] + assert published_topic == "responses" + + +@pytest.mark.asyncio +@pytest.mark.unit +async def test_no_topic_router_uses_output_topic() -> None: + """Backward compat: no router → all events go to output_topic.""" + bus = AsyncMock() + applier = DispatchResultApplier( + event_bus=bus, + output_topic="responses", + ) + result = _make_result_with([_make_accepted_event()]) + await applier.apply(result) + published_topic = bus.publish_envelope.call_args.kwargs["topic"] + assert published_topic == "responses"