feat(OMN-1547): Replace hardcoded topics with validated suffix constants

src/omnibase_infra/mixins/mixin_node_introspection.py

@@ -237,11 +237,6 @@ async def shutdown(self):
 
 logger = logging.getLogger(__name__)
 
-# Event topic constants
-INTROSPECTION_TOPIC = "node.introspection"
-HEARTBEAT_TOPIC = "node.heartbeat"
-REQUEST_INTROSPECTION_TOPIC = "node.request_introspection"
-
 # Performance threshold constants (in milliseconds)
 PERF_THRESHOLD_GET_CAPABILITIES_MS = 50.0
 PERF_THRESHOLD_DISCOVER_CAPABILITIES_MS = 30.0
@@ -2485,13 +2480,10 @@ async def get_active_operations_count(self) -> int:
 
 
 __all__ = [
-    "HEARTBEAT_TOPIC",
-    "INTROSPECTION_TOPIC",
     "PERF_THRESHOLD_CACHE_HIT_MS",
     "PERF_THRESHOLD_DISCOVER_CAPABILITIES_MS",
     "PERF_THRESHOLD_GET_CAPABILITIES_MS",
     "PERF_THRESHOLD_GET_INTROSPECTION_DATA_MS",
-    "REQUEST_INTROSPECTION_TOPIC",
     "DiscoveredCapabilitiesCacheDict",  # TypedDict for cached discovered capabilities
     "IntrospectionCacheDict",
     "MixinNodeIntrospection",

src/omnibase_infra/models/discovery/model_introspection_config.py

@@ -28,6 +28,11 @@
 
 from omnibase_core.enums import EnumNodeKind
 from omnibase_core.models.contracts import ModelContractBase
+from omnibase_infra.topics import (
+    SUFFIX_NODE_HEARTBEAT,
+    SUFFIX_NODE_INTROSPECTION,
+    SUFFIX_REQUEST_INTROSPECTION,
+)
 
 if TYPE_CHECKING:
     from omnibase_core.protocols.event_bus.protocol_event_bus import ProtocolEventBus
@@ -42,11 +47,11 @@
 
 logger = logging.getLogger(__name__)
 
-# Default topic constants following ONEX legacy conventions
-# Note: These use the legacy "node." prefix for backward compatibility
-DEFAULT_INTROSPECTION_TOPIC = "node.introspection"
-DEFAULT_HEARTBEAT_TOPIC = "node.heartbeat"
-DEFAULT_REQUEST_INTROSPECTION_TOPIC = "node.request_introspection"
+# Default topic constants using ONEX platform suffix constants
+# These are the canonical source for introspection-related topic defaults
+DEFAULT_INTROSPECTION_TOPIC = SUFFIX_NODE_INTROSPECTION
+DEFAULT_HEARTBEAT_TOPIC = SUFFIX_NODE_HEARTBEAT
+DEFAULT_REQUEST_INTROSPECTION_TOPIC = SUFFIX_REQUEST_INTROSPECTION
 
 # Topic validation patterns
 # Matches valid topic characters: lowercase alphanumeric, dots, hyphens, underscores
@@ -79,14 +84,14 @@ class ModelIntrospectionConfig(BaseModel):
             discovery. Methods starting with these prefixes are filtered out.
             If None, uses MixinNodeIntrospection.DEFAULT_EXCLUDE_PREFIXES.
         introspection_topic: Topic for publishing introspection events.
-            Defaults to "node.introspection". ONEX topics (onex.*) require
-            version suffix (.v1, .v2, etc.).
+            Defaults to SUFFIX_NODE_INTROSPECTION (onex.evt.platform.node-introspection.v1).
+            ONEX topics (onex.*) require version suffix (.v1, .v2, etc.).
         heartbeat_topic: Topic for publishing heartbeat events.
-            Defaults to "node.heartbeat". ONEX topics (onex.*) require
-            version suffix (.v1, .v2, etc.).
+            Defaults to SUFFIX_NODE_HEARTBEAT (onex.evt.platform.node-heartbeat.v1).
+            ONEX topics (onex.*) require version suffix (.v1, .v2, etc.).
         request_introspection_topic: Topic for receiving introspection requests.
-            Defaults to "node.request_introspection". ONEX topics (onex.*)
-            require version suffix (.v1, .v2, etc.).
+            Defaults to SUFFIX_REQUEST_INTROSPECTION (onex.cmd.platform.request-introspection.v1).
+            ONEX topics (onex.*) require version suffix (.v1, .v2, etc.).
         contract: Optional typed contract model for capability extraction.
             When provided, MixinNodeIntrospection extracts contract_capabilities
             using ContractCapabilityExtractor. None for legacy nodes.
@@ -282,6 +287,7 @@ def validate_topic_name(cls, v: str) -> str:
         arbitrary_types_allowed=True,  # Allow arbitrary types for event_bus
         json_schema_extra={
             "examples": [
+                # First example: Default topic configuration using suffix constants
                 {
                     "node_id": "550e8400-e29b-41d4-a716-446655440000",
                     "node_type": "EFFECT",
@@ -290,10 +296,12 @@ def validate_topic_name(cls, v: str) -> str:
                     "cache_ttl": 300.0,
                     "operation_keywords": None,
                     "exclude_prefixes": None,
-                    "introspection_topic": "node.introspection",
-                    "heartbeat_topic": "node.heartbeat",
-                    "request_introspection_topic": "node.request_introspection",
+                    "introspection_topic": "onex.evt.platform.node-introspection.v1",
+                    "heartbeat_topic": "onex.evt.platform.node-heartbeat.v1",
+                    "request_introspection_topic": "onex.cmd.platform.request-introspection.v1",
                 },
+                # Second example: Custom prefixed topics for environment isolation
+                # Demonstrates adding env/namespace prefix to suffix constants
                 {
                     "node_id": "550e8400-e29b-41d4-a716-446655440001",
                     "node_type": "COMPUTE",
@@ -302,9 +310,9 @@ def validate_topic_name(cls, v: str) -> str:
                     "cache_ttl": 120.0,
                     "operation_keywords": ["process", "transform", "analyze"],
                     "exclude_prefixes": ["_internal", "_private"],
-                    "introspection_topic": "onex.node.introspection.published.v1",
-                    "heartbeat_topic": "onex.node.heartbeat.published.v1",
-                    "request_introspection_topic": "onex.registry.introspection.requested.v1",
+                    "introspection_topic": "prod.myapp.onex.evt.platform.node-introspection.v1",
+                    "heartbeat_topic": "prod.myapp.onex.evt.platform.node-heartbeat.v1",
+                    "request_introspection_topic": "prod.myapp.onex.cmd.platform.request-introspection.v1",
                 },
             ]
         },

src/omnibase_infra/models/projection/model_snapshot_topic_config.py

@@ -58,6 +58,7 @@
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
 from omnibase_infra.enums import EnumInfraTransportType
+from omnibase_infra.topics import SUFFIX_REGISTRATION_SNAPSHOTS
 
 if TYPE_CHECKING:
     from omnibase_infra.errors.error_infra import ProtocolConfigurationError
@@ -154,7 +155,7 @@ class ModelSnapshotTopicConfig(BaseModel):
 
     # Topic identity
     topic: str = Field(
-        default="onex.registration.snapshots",
+        default=SUFFIX_REGISTRATION_SNAPSHOTS,
         min_length=1,
         max_length=255,
         description="Full Kafka topic name for registration snapshots",
@@ -442,7 +443,7 @@ def default(cls) -> ModelSnapshotTopicConfig:
             Default configuration instance with environment overrides
         """
         base_config = cls(
-            topic="onex.registration.snapshots",
+            topic=SUFFIX_REGISTRATION_SNAPSHOTS,
             partition_count=12,
             replication_factor=3,
             cleanup_policy="compact",

src/omnibase_infra/runtime/models/model_runtime_scheduler_config.py

@@ -28,7 +28,7 @@
             Example: "runtime-scheduler-prod-1"
 
         ONEX_RUNTIME_SCHEDULER_TICK_TOPIC: Kafka topic for publishing ticks
-            Default: "runtime.tick.v1"
+            Default: SUFFIX_RUNTIME_TICK from omnibase_infra.topics
             Example: "prod.runtime.tick.v1"
 
     Restart-Safety Settings:
@@ -106,6 +106,7 @@
 
 from omnibase_infra.enums import EnumInfraTransportType
 from omnibase_infra.errors import ModelInfraErrorContext, ProtocolConfigurationError
+from omnibase_infra.topics import SUFFIX_RUNTIME_TICK
 
 logger = logging.getLogger(__name__)
 
@@ -167,7 +168,7 @@ class ModelRuntimeSchedulerConfig(BaseModel):
         max_length=255,
     )
     tick_topic: str = Field(
-        default="runtime.tick.v1",
+        default=SUFFIX_RUNTIME_TICK,
         description="Kafka topic for publishing tick events",
         min_length=1,
         max_length=255,
@@ -604,7 +605,7 @@ def default(cls) -> ModelRuntimeSchedulerConfig:
         base_config = cls(
             tick_interval_ms=1000,
             scheduler_id="runtime-scheduler-default",
-            tick_topic="runtime.tick.v1",
+            tick_topic=SUFFIX_RUNTIME_TICK,
             persist_sequence_number=True,
             sequence_number_key="runtime_scheduler_sequence",
             max_tick_jitter_ms=100,

src/omnibase_infra/runtime/transition_notification_publisher.py

@@ -49,7 +49,7 @@
     # Initialize publisher with event bus
     publisher = TransitionNotificationPublisher(
         event_bus=kafka_event_bus,
-        topic="onex.fsm.state.transitions.v1",
+        topic=SUFFIX_FSM_STATE_TRANSITIONS,
     )
 
     # Publish single notification
@@ -110,6 +110,7 @@
 from omnibase_infra.runtime.models.model_transition_notification_publisher_metrics import (
     ModelTransitionNotificationPublisherMetrics,
 )
+from omnibase_infra.topics import SUFFIX_FSM_STATE_TRANSITIONS
 from omnibase_infra.utils.util_error_sanitization import sanitize_error_string
 
 if TYPE_CHECKING:
@@ -754,7 +755,7 @@ def _verify_protocol_compliance() -> None:  # pragma: no cover
     publisher: ProtocolTransitionNotificationPublisher = (
         TransitionNotificationPublisher(
             event_bus=bus,
-            topic="onex.fsm.state.transitions.v1",
+            topic=SUFFIX_FSM_STATE_TRANSITIONS,
         )
     )
     # Use the variable to silence unused warnings

src/omnibase_infra/services/mcp/service_mcp_tool_sync.py

@@ -8,7 +8,7 @@
 - Deregistration: Removed orchestrators are removed from tool registry
 - Idempotency: Duplicate/out-of-order events are handled correctly
 
-Event Topic: node.registration.v1
+Event Topic: Uses SUFFIX_NODE_REGISTRATION (onex.evt.platform.node-registration.v1)
 Event Types:
     - registered: New node registered → upsert tool
     - updated: Node updated → upsert tool
@@ -30,6 +30,7 @@
 from omnibase_infra.models.mcp.model_mcp_tool_definition import (
     ModelMCPToolDefinition,
 )
+from omnibase_infra.topics import SUFFIX_NODE_REGISTRATION
 
 if TYPE_CHECKING:
     from omnibase_infra.event_bus.event_bus_kafka import EventBusKafka
@@ -77,8 +78,8 @@ class ServiceMCPToolSync:
         >>> await sync.stop()
     """
 
-    # Topic for node registration events
-    TOPIC = "node.registration.v1"
+    # Topic for node registration events (uses platform suffix constant)
+    TOPIC = SUFFIX_NODE_REGISTRATION
 
     # MCP tag constants
     TAG_MCP_ENABLED = "mcp-enabled"

src/omnibase_infra/services/session/config_consumer.py

@@ -8,6 +8,7 @@
 from __future__ import annotations
 
 import logging
+from typing import Self
 
 from pydantic import Field, model_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
@@ -42,13 +43,8 @@ class ConfigSessionConsumer(BaseSettings):
 
     # Topics to subscribe
     topics: list[str] = Field(
-        default=[
-            "dev.omniclaude.session.started.v1",
-            "dev.omniclaude.session.ended.v1",
-            "dev.omniclaude.prompt.submitted.v1",
-            "dev.omniclaude.tool.executed.v1",
-        ],
-        description="Kafka topics to consume",
+        default_factory=list,
+        description="Kafka topics to consume. Must be explicitly configured via environment or discovery.",
     )
 
     # Consumer behavior
@@ -96,7 +92,28 @@ class ConfigSessionConsumer(BaseSettings):
     )
 
     @model_validator(mode="after")
-    def validate_timing_relationships(self) -> ConfigSessionConsumer:
+    def validate_topic_configuration(self) -> Self:
+        """Ensure topics are explicitly configured.
+
+        Fails fast if no topics provided, preventing silent misconfiguration.
+
+        Returns:
+            Self if validation passes.
+
+        Raises:
+            ProtocolConfigurationError: If no topics are configured.
+        """
+        if not self.topics:
+            from omnibase_infra.errors import ProtocolConfigurationError
+
+            raise ProtocolConfigurationError(
+                "No topics configured for session consumer. "
+                "Provide explicit 'topics' via configuration or environment variable."
+            )
+        return self
+
+    @model_validator(mode="after")
+    def validate_timing_relationships(self) -> Self:
         """Validate timing relationships between configuration values.
 
         Warns if circuit breaker timeout is very short relative to batch processing,

src/omnibase_infra/topics/__init__.py

@@ -0,0 +1,45 @@
+"""ONEX Infrastructure Topic Constants.
+
+This module provides platform-reserved topic suffix constants for ONEX infrastructure
+components. Domain services should NOT import from this module - domain topics should
+be defined in domain contracts.
+
+Exports:
+    Platform topic suffix constants (e.g., SUFFIX_NODE_REGISTRATION)
+    ALL_PLATFORM_SUFFIXES: Complete tuple of all platform-reserved suffixes
+    build_full_topic: Compose full topic from env, namespace, and suffix
+    TopicCompositionError: Error raised when topic composition fails
+"""
+
+from omnibase_infra.topics.platform_topic_suffixes import (
+    ALL_PLATFORM_SUFFIXES,
+    SUFFIX_FSM_STATE_TRANSITIONS,
+    SUFFIX_NODE_HEARTBEAT,
+    SUFFIX_NODE_INTROSPECTION,
+    SUFFIX_NODE_REGISTRATION,
+    SUFFIX_REGISTRATION_SNAPSHOTS,
+    SUFFIX_REQUEST_INTROSPECTION,
+    SUFFIX_RUNTIME_TICK,
+)
+from omnibase_infra.topics.util_topic_composition import (
+    MAX_NAMESPACE_LENGTH,
+    TopicCompositionError,
+    build_full_topic,
+)
+
+__all__: list[str] = [
+    # Individual suffix constants
+    "SUFFIX_NODE_REGISTRATION",
+    "SUFFIX_NODE_INTROSPECTION",
+    "SUFFIX_NODE_HEARTBEAT",
+    "SUFFIX_REQUEST_INTROSPECTION",
+    "SUFFIX_FSM_STATE_TRANSITIONS",
+    "SUFFIX_RUNTIME_TICK",
+    "SUFFIX_REGISTRATION_SNAPSHOTS",
+    # Aggregate tuple
+    "ALL_PLATFORM_SUFFIXES",
+    # Topic composition utilities
+    "build_full_topic",
+    "TopicCompositionError",
+    "MAX_NAMESPACE_LENGTH",
+]