feat(runtime): add topic_router to DispatchResultApplier for per-event-type topic routing

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__,

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"