diff --git a/src/omnibase_infra/mixins/mixin_node_introspection.py b/src/omnibase_infra/mixins/mixin_node_introspection.py index d3dcf3c2ff..bcc9ad67b9 100644 --- a/src/omnibase_infra/mixins/mixin_node_introspection.py +++ b/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", diff --git a/src/omnibase_infra/models/discovery/model_introspection_config.py b/src/omnibase_infra/models/discovery/model_introspection_config.py index 06afea161b..409f55fa3f 100644 --- a/src/omnibase_infra/models/discovery/model_introspection_config.py +++ b/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", }, ] }, diff --git a/src/omnibase_infra/models/projection/model_snapshot_topic_config.py b/src/omnibase_infra/models/projection/model_snapshot_topic_config.py index 054266624b..05eced987c 100644 --- a/src/omnibase_infra/models/projection/model_snapshot_topic_config.py +++ b/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", diff --git a/src/omnibase_infra/runtime/models/model_runtime_scheduler_config.py b/src/omnibase_infra/runtime/models/model_runtime_scheduler_config.py index 17febfbaba..e189d5b174 100644 --- a/src/omnibase_infra/runtime/models/model_runtime_scheduler_config.py +++ b/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, diff --git a/src/omnibase_infra/runtime/transition_notification_publisher.py b/src/omnibase_infra/runtime/transition_notification_publisher.py index eac47bffc9..e1f4bc8af7 100644 --- a/src/omnibase_infra/runtime/transition_notification_publisher.py +++ b/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 diff --git a/src/omnibase_infra/services/mcp/service_mcp_tool_sync.py b/src/omnibase_infra/services/mcp/service_mcp_tool_sync.py index 732acc6684..3910ef84f3 100644 --- a/src/omnibase_infra/services/mcp/service_mcp_tool_sync.py +++ b/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" diff --git a/src/omnibase_infra/services/session/config_consumer.py b/src/omnibase_infra/services/session/config_consumer.py index 49bf917d2d..a0638fe978 100644 --- a/src/omnibase_infra/services/session/config_consumer.py +++ b/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, diff --git a/src/omnibase_infra/topics/__init__.py b/src/omnibase_infra/topics/__init__.py new file mode 100644 index 0000000000..b84f1c0eb5 --- /dev/null +++ b/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", +] diff --git a/src/omnibase_infra/topics/platform_topic_suffixes.py b/src/omnibase_infra/topics/platform_topic_suffixes.py new file mode 100644 index 0000000000..f380295738 --- /dev/null +++ b/src/omnibase_infra/topics/platform_topic_suffixes.py @@ -0,0 +1,140 @@ +"""Platform-reserved topic suffixes for ONEX infrastructure. + +WARNING: These are platform-reserved suffixes. Domain services must NOT +import from this module. Domain topics should be defined in domain contracts. + +Topic Suffix Format: + onex....v + + Structure: + - onex: Required prefix for all ONEX topics + - kind: Message category (evt, cmd, intent, snapshot, dlq) + - producer: Service/module that produces the message + - event-name: Descriptive name using kebab-case + - version: Semantic version (v1, v2, etc.) + + Kinds: + evt - Event topics (state changes, notifications) + cmd - Command topics (requests for action) + intent - Intent topics (internal workflow coordination) + snapshot - Snapshot topics (periodic state snapshots) + dlq - Dead letter queue topics + + Examples: + onex.evt.platform.node-registration.v1 + onex.cmd.platform.request-introspection.v1 + onex.intent.platform.runtime-tick.v1 + +Usage: + from omnibase_infra.topics import SUFFIX_NODE_REGISTRATION + + # Compose full topic with tenant/namespace prefix + full_topic = f"{tenant}.{namespace}.{SUFFIX_NODE_REGISTRATION}" + +See Also: + omnibase_core.validation.validate_topic_suffix - Validation function + omnibase_core.validation.compose_full_topic - Topic composition utility +""" + +from omnibase_core.errors import OnexError +from omnibase_core.validation import validate_topic_suffix + +# ============================================================================= +# PLATFORM-RESERVED TOPIC SUFFIXES +# ============================================================================= + +# Node lifecycle events +SUFFIX_NODE_REGISTRATION: str = "onex.evt.platform.node-registration.v1" +"""Topic suffix for node registration events. + +Published when a node registers with the runtime. Contains node metadata, +capabilities, and health check configuration. +""" + +SUFFIX_NODE_INTROSPECTION: str = "onex.evt.platform.node-introspection.v1" +"""Topic suffix for node introspection events. + +Published when a node responds to an introspection request. Contains node +capabilities, supported operations, and current state. +""" + +SUFFIX_NODE_HEARTBEAT: str = "onex.evt.platform.node-heartbeat.v1" +"""Topic suffix for node heartbeat events. + +Published periodically by nodes to indicate liveness. Contains timestamp, +resource usage metrics, and health status. +""" + +# Command topics +SUFFIX_REQUEST_INTROSPECTION: str = "onex.cmd.platform.request-introspection.v1" +"""Topic suffix for introspection request commands. + +Published to request introspection from a specific node or all nodes. +Nodes respond on the SUFFIX_NODE_INTROSPECTION topic. +""" + +# FSM and state management +SUFFIX_FSM_STATE_TRANSITIONS: str = "onex.evt.platform.fsm-state-transitions.v1" +"""Topic suffix for FSM state transition events. + +Published when a node's finite state machine transitions between states. +Contains previous state, new state, trigger event, and transition metadata. +""" + +# Runtime coordination +SUFFIX_RUNTIME_TICK: str = "onex.intent.platform.runtime-tick.v1" +"""Topic suffix for runtime tick intents. + +Internal topic for runtime orchestration. Triggers periodic tasks like +heartbeat collection, health checks, and scheduled workflows. +""" + +# Registration snapshots +SUFFIX_REGISTRATION_SNAPSHOTS: str = "onex.snapshot.platform.registration-snapshots.v1" +"""Topic suffix for registration snapshot events. + +Published periodically with aggregated registration state. Used for +dashboard displays and monitoring systems. +""" + +# ============================================================================= +# AGGREGATE TUPLE +# ============================================================================= + +ALL_PLATFORM_SUFFIXES: tuple[str, ...] = ( + SUFFIX_NODE_REGISTRATION, + SUFFIX_NODE_INTROSPECTION, + SUFFIX_NODE_HEARTBEAT, + SUFFIX_REQUEST_INTROSPECTION, + SUFFIX_FSM_STATE_TRANSITIONS, + SUFFIX_RUNTIME_TICK, + SUFFIX_REGISTRATION_SNAPSHOTS, +) +"""Complete tuple of all platform-reserved topic suffixes. + +Use this tuple for: + - Validating that domain topics don't conflict with platform topics + - Iterating over all platform topics for subscription setup + - Documentation and discovery +""" + +# ============================================================================= +# IMPORT-TIME VALIDATION +# ============================================================================= + + +def _validate_all_suffixes() -> None: + """Validate all suffixes at import time to fail fast on invalid format. + + Raises: + OnexError: If any suffix fails validation with details about which + suffix failed and why. + """ + for suffix in ALL_PLATFORM_SUFFIXES: + result = validate_topic_suffix(suffix) + if not result.is_valid: + raise OnexError(f"Invalid platform topic suffix '{suffix}': {result.error}") + + +# Run validation at import time +_validate_all_suffixes() diff --git a/src/omnibase_infra/topics/util_topic_composition.py b/src/omnibase_infra/topics/util_topic_composition.py new file mode 100644 index 0000000000..39ad20e073 --- /dev/null +++ b/src/omnibase_infra/topics/util_topic_composition.py @@ -0,0 +1,95 @@ +"""Topic composition utilities for ONEX infrastructure. + +IMPORTANT: build_full_topic() is the ONLY supported way to compose +Kafka topics in omnibase_infra. Direct string concatenation is prohibited. +""" + +from omnibase_core.errors import OnexError +from omnibase_core.validation import validate_topic_suffix +from omnibase_core.validation.validator_topic_suffix import ENV_PREFIXES + +MAX_NAMESPACE_LENGTH = 100 +"""Maximum allowed namespace length. + +Kafka topics have a 249 character limit. With env prefix (~10 chars), +separators (2 dots), and suffix (~50 chars), namespace should be limited +to ensure total topic length stays within bounds. +""" + + +class TopicCompositionError(OnexError): + """Raised when topic composition fails due to invalid components. + + Extends OnexError to follow ONEX error handling conventions. + """ + + +def build_full_topic(env: str, namespace: str, suffix: str) -> str: + """Build full topic from components with validation. + + Args: + env: Environment prefix (e.g., "dev", "staging", "prod", "test", "local") + namespace: Namespace/tenant identifier (e.g., "omnibase", "myapp", "tenant-123") + suffix: Validated topic suffix (e.g., "onex.evt.platform.node-introspection.v1") + + Returns: + Full topic string: {env}.{namespace}.{suffix} + + Raises: + TopicCompositionError: If env is not a valid environment prefix + TopicCompositionError: If namespace is empty or contains invalid characters + TopicCompositionError: If suffix doesn't match ONEX topic format + + Namespace Validation Rules: + - Must not be empty + - Allowed characters: alphanumeric (a-z, A-Z, 0-9), hyphens (-), underscores (_) + - Numeric-only namespaces ARE valid (e.g., "12345") to support numeric tenant IDs + - Invalid: spaces, dots, special characters + + Example: + >>> build_full_topic("dev", "omnibase", "onex.evt.platform.node-introspection.v1") + 'dev.omnibase.onex.evt.platform.node-introspection.v1' + + >>> build_full_topic("prod", "myapp", "onex.cmd.platform.request-introspection.v1") + 'prod.myapp.onex.cmd.platform.request-introspection.v1' + """ + # Validate environment prefix + if env not in ENV_PREFIXES: + raise TopicCompositionError( + f"Invalid environment prefix '{env}'. " + f"Must be one of: {', '.join(sorted(ENV_PREFIXES))}" + ) + + # Validate namespace + # Allowed characters: alphanumeric (a-z, A-Z, 0-9), hyphens, underscores + # Note: Numeric-only namespaces (e.g., "12345") ARE valid because isalnum() + # returns True for digit-only strings. This is intentional to support + # numeric tenant/organization IDs as namespaces. + if not namespace: + raise TopicCompositionError("Namespace cannot be empty") + if len(namespace) > MAX_NAMESPACE_LENGTH: + raise TopicCompositionError( + f"Namespace exceeds maximum length of {MAX_NAMESPACE_LENGTH} characters. " + f"Got {len(namespace)} characters." + ) + if not namespace.replace("-", "").replace("_", "").isalnum(): + raise TopicCompositionError( + f"Invalid namespace '{namespace}'. " + "Must contain only alphanumeric characters, hyphens, and underscores" + ) + + # Enforce lowercase to ensure composed topics are valid + # (ONEX topic suffixes are lowercase, so namespaces should match) + if namespace != namespace.lower(): + raise TopicCompositionError( + f"Namespace must be lowercase: '{namespace}'. " + "Use lowercase to ensure consistent topic naming." + ) + + # Validate suffix using omnibase_core validation + result = validate_topic_suffix(suffix) + if not result.is_valid: + raise TopicCompositionError(f"Invalid topic suffix '{suffix}': {result.error}") + + # Compose full topic + return f"{env}.{namespace}.{suffix}" diff --git a/tests/services/session/test_consumer.py b/tests/services/session/test_consumer.py index a8b9e29044..929bbb4917 100644 --- a/tests/services/session/test_consumer.py +++ b/tests/services/session/test_consumer.py @@ -25,6 +25,7 @@ import pytest +from omnibase_infra.errors import ProtocolConfigurationError from omnibase_infra.services.session import ( ConfigSessionConsumer, ConsumerMetrics, @@ -134,12 +135,23 @@ def test_consumer_initializes_with_config( assert consumer.circuit_state == EnumCircuitState.CLOSED assert consumer.consumer_id.startswith("session-consumer-") - def test_consumer_uses_default_config(self, aggregator: MockAggregator) -> None: - """Consumer should work with default config.""" - config = ConfigSessionConsumer() + def test_consumer_requires_explicit_topics(self) -> None: + """Consumer config should fail-fast when no topics are configured. + + This validates the intentional change in OMN-1547: topics must now be + explicitly configured to prevent silent misconfiguration with + environment-specific domain topics. + """ + with pytest.raises(ProtocolConfigurationError, match="No topics configured"): + ConfigSessionConsumer() + + def test_consumer_uses_explicit_config(self, aggregator: MockAggregator) -> None: + """Consumer should work with explicitly configured topics.""" + config = ConfigSessionConsumer(topics=["test.session.events.v1"]) consumer = SessionEventConsumer(config=config, aggregator=aggregator) assert consumer._config.group_id == "omnibase-infra-session-consumer" + assert consumer._config.topics == ["test.session.events.v1"] def test_consumer_has_metrics(self, consumer: SessionEventConsumer) -> None: """Consumer should have metrics instance.""" diff --git a/tests/unit/mixins/test_mixin_node_introspection.py b/tests/unit/mixins/test_mixin_node_introspection.py index 07a75f137c..e5c89c2dd4 100644 --- a/tests/unit/mixins/test_mixin_node_introspection.py +++ b/tests/unit/mixins/test_mixin_node_introspection.py @@ -66,6 +66,7 @@ ModelNodeHeartbeatEvent, ModelNodeIntrospectionEvent, ) +from omnibase_infra.topics import SUFFIX_NODE_HEARTBEAT, SUFFIX_NODE_INTROSPECTION # CI environments may be slower - apply multiplier for performance thresholds _CI_MODE: bool = os.environ.get("CI", "false").lower() == "true" @@ -726,7 +727,7 @@ async def test_publish_introspection_succeeds_with_event_bus( assert len(event_bus.published_envelopes) == 1 envelope, topic = event_bus.published_envelopes[0] - assert topic == "node.introspection" + assert topic == SUFFIX_NODE_INTROSPECTION assert isinstance(envelope, ModelNodeIntrospectionEvent) async def test_publish_introspection_with_correlation_id( @@ -3420,7 +3421,7 @@ async def test_heartbeat_publishes_events(self) -> None: # All events should be on heartbeat topic for envelope, topic in event_bus.published_envelopes: - assert topic == "node.heartbeat" + assert topic == SUFFIX_NODE_HEARTBEAT assert isinstance(envelope, ModelNodeHeartbeatEvent) finally: await node.stop_introspection_tasks() diff --git a/tests/unit/models/projection/test_model_snapshot_topic_config.py b/tests/unit/models/projection/test_model_snapshot_topic_config.py index 1f444e6905..c887ad8c7f 100644 --- a/tests/unit/models/projection/test_model_snapshot_topic_config.py +++ b/tests/unit/models/projection/test_model_snapshot_topic_config.py @@ -20,15 +20,16 @@ from omnibase_infra.errors import ProtocolConfigurationError from omnibase_infra.models.projection import ModelSnapshotTopicConfig +from omnibase_infra.topics import SUFFIX_REGISTRATION_SNAPSHOTS class TestModelSnapshotTopicConfigDefaults: """Tests for default configuration values.""" def test_default_topic(self) -> None: - """Test default topic follows ONEX convention.""" + """Test default topic uses ONEX platform suffix constant.""" config = ModelSnapshotTopicConfig.default() - assert config.topic == "onex.registration.snapshots" + assert config.topic == SUFFIX_REGISTRATION_SNAPSHOTS def test_default_cleanup_policy_is_compact(self) -> None: """Test that cleanup_policy defaults to compact.""" diff --git a/tests/unit/runtime/test_runtime_scheduler.py b/tests/unit/runtime/test_runtime_scheduler.py index 53b4185be7..a0995e82c8 100644 --- a/tests/unit/runtime/test_runtime_scheduler.py +++ b/tests/unit/runtime/test_runtime_scheduler.py @@ -48,6 +48,7 @@ ModelRuntimeTick, ) from omnibase_infra.runtime.runtime_scheduler import RuntimeScheduler +from omnibase_infra.topics import SUFFIX_RUNTIME_TICK # ============================================================================ # Fixtures @@ -246,7 +247,7 @@ def test_default_values(self) -> None: assert config.tick_interval_ms == 1000 assert config.scheduler_id == "runtime-scheduler-default" - assert config.tick_topic == "runtime.tick.v1" + assert config.tick_topic == SUFFIX_RUNTIME_TICK assert config.persist_sequence_number is True assert config.sequence_number_key == "runtime_scheduler_sequence" assert config.max_tick_jitter_ms == 100 @@ -1081,7 +1082,7 @@ def test_config_default_method(self) -> None: assert config.tick_interval_ms == 1000 assert config.scheduler_id == "runtime-scheduler-default" - assert config.tick_topic == "runtime.tick.v1" + assert config.tick_topic == SUFFIX_RUNTIME_TICK def test_config_jitter_validation(self) -> None: """Test max_tick_jitter_ms validation.""" diff --git a/tests/unit/topics/__init__.py b/tests/unit/topics/__init__.py new file mode 100644 index 0000000000..147e588836 --- /dev/null +++ b/tests/unit/topics/__init__.py @@ -0,0 +1 @@ +"""Unit tests for omnibase_infra.topics module.""" diff --git a/tests/unit/topics/test_platform_topic_suffixes.py b/tests/unit/topics/test_platform_topic_suffixes.py new file mode 100644 index 0000000000..9ef835076c --- /dev/null +++ b/tests/unit/topics/test_platform_topic_suffixes.py @@ -0,0 +1,133 @@ +"""Tests for platform topic suffix constants.""" + +import pytest + +from omnibase_core.validation import validate_topic_suffix +from omnibase_infra.topics 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, +) + + +class TestPlatformTopicSuffixes: + """Tests for platform-reserved topic suffix constants.""" + + def test_all_platform_suffixes_are_valid(self) -> None: + """Every platform suffix constant must pass validation.""" + for suffix in ALL_PLATFORM_SUFFIXES: + result = validate_topic_suffix(suffix) + assert result.is_valid, f"Invalid suffix: {suffix} - {result.error}" + + def test_all_platform_suffixes_list_is_complete(self) -> None: + """ALL_PLATFORM_SUFFIXES should contain all defined constants.""" + expected_suffixes = { + SUFFIX_NODE_REGISTRATION, + SUFFIX_NODE_INTROSPECTION, + SUFFIX_NODE_HEARTBEAT, + SUFFIX_REQUEST_INTROSPECTION, + SUFFIX_FSM_STATE_TRANSITIONS, + SUFFIX_RUNTIME_TICK, + SUFFIX_REGISTRATION_SNAPSHOTS, + } + assert set(ALL_PLATFORM_SUFFIXES) == expected_suffixes + + def test_suffixes_follow_onex_format(self) -> None: + """All suffixes should follow onex.{kind}.{producer}.{event}.v{n} format.""" + for suffix in ALL_PLATFORM_SUFFIXES: + assert suffix.startswith("onex."), ( + f"Suffix must start with 'onex.': {suffix}" + ) + parts = suffix.split(".") + assert len(parts) == 5, f"Suffix must have 5 parts: {suffix}" + assert parts[-1].startswith("v"), f"Suffix must end with version: {suffix}" + + def test_suffix_constants_are_strings(self) -> None: + """All suffix constants should be strings.""" + assert isinstance(SUFFIX_NODE_REGISTRATION, str) + assert isinstance(SUFFIX_NODE_INTROSPECTION, str) + assert isinstance(SUFFIX_NODE_HEARTBEAT, str) + assert isinstance(SUFFIX_REQUEST_INTROSPECTION, str) + assert isinstance(SUFFIX_FSM_STATE_TRANSITIONS, str) + assert isinstance(SUFFIX_RUNTIME_TICK, str) + assert isinstance(SUFFIX_REGISTRATION_SNAPSHOTS, str) + + def test_suffix_kinds_are_valid(self) -> None: + """All suffixes should use valid message kinds.""" + valid_kinds = {"evt", "cmd", "intent", "snapshot", "dlq"} + for suffix in ALL_PLATFORM_SUFFIXES: + parts = suffix.split(".") + kind = parts[1] + assert kind in valid_kinds, f"Invalid kind '{kind}' in suffix: {suffix}" + + def test_node_registration_suffix_format(self) -> None: + """Node registration suffix should have correct format.""" + assert SUFFIX_NODE_REGISTRATION == "onex.evt.platform.node-registration.v1" + + def test_node_introspection_suffix_format(self) -> None: + """Node introspection suffix should have correct format.""" + assert SUFFIX_NODE_INTROSPECTION == "onex.evt.platform.node-introspection.v1" + + def test_node_heartbeat_suffix_format(self) -> None: + """Node heartbeat suffix should have correct format.""" + assert SUFFIX_NODE_HEARTBEAT == "onex.evt.platform.node-heartbeat.v1" + + def test_request_introspection_suffix_format(self) -> None: + """Request introspection suffix should have correct format.""" + assert ( + SUFFIX_REQUEST_INTROSPECTION == "onex.cmd.platform.request-introspection.v1" + ) + + def test_fsm_state_transitions_suffix_format(self) -> None: + """FSM state transitions suffix should have correct format.""" + assert ( + SUFFIX_FSM_STATE_TRANSITIONS == "onex.evt.platform.fsm-state-transitions.v1" + ) + + def test_runtime_tick_suffix_format(self) -> None: + """Runtime tick suffix should have correct format.""" + assert SUFFIX_RUNTIME_TICK == "onex.intent.platform.runtime-tick.v1" + + def test_registration_snapshots_suffix_format(self) -> None: + """Registration snapshots suffix should have correct format.""" + assert ( + SUFFIX_REGISTRATION_SNAPSHOTS + == "onex.snapshot.platform.registration-snapshots.v1" + ) + + def test_all_platform_suffixes_is_tuple(self) -> None: + """ALL_PLATFORM_SUFFIXES should be an immutable tuple.""" + assert isinstance(ALL_PLATFORM_SUFFIXES, tuple) + + def test_no_duplicate_suffixes(self) -> None: + """ALL_PLATFORM_SUFFIXES should not contain duplicates.""" + assert len(ALL_PLATFORM_SUFFIXES) == len(set(ALL_PLATFORM_SUFFIXES)) + + def test_suffixes_use_platform_producer(self) -> None: + """All platform suffixes should use 'platform' as producer.""" + for suffix in ALL_PLATFORM_SUFFIXES: + parts = suffix.split(".") + producer = parts[2] + assert producer == "platform", f"Expected 'platform' producer in: {suffix}" + + def test_all_suffix_constants_exported(self) -> None: + """All SUFFIX_* constants should be exported from topics package.""" + from omnibase_infra import topics + from omnibase_infra.topics import platform_topic_suffixes + + # Find all SUFFIX_* constants in the module + suffix_constants = [ + name for name in dir(platform_topic_suffixes) if name.startswith("SUFFIX_") + ] + + # Verify each is exported from the package + for name in suffix_constants: + assert hasattr(topics, name), ( + f"SUFFIX constant '{name}' not exported from omnibase_infra.topics. " + f"Add it to __all__ in topics/__init__.py" + ) diff --git a/tests/unit/topics/test_util_topic_composition.py b/tests/unit/topics/test_util_topic_composition.py new file mode 100644 index 0000000000..bd731ea0fc --- /dev/null +++ b/tests/unit/topics/test_util_topic_composition.py @@ -0,0 +1,190 @@ +"""Tests for topic composition utilities.""" + +import pytest + +from omnibase_core.errors import OnexError +from omnibase_infra.topics import ( + SUFFIX_NODE_INTROSPECTION, + SUFFIX_NODE_REGISTRATION, + TopicCompositionError, + build_full_topic, +) + + +class TestBuildFullTopic: + """Tests for build_full_topic() helper.""" + + def test_build_full_topic_format(self) -> None: + """Full topic should be {env}.{namespace}.{suffix}.""" + topic = build_full_topic("dev", "omnibase", SUFFIX_NODE_INTROSPECTION) + assert topic == f"dev.omnibase.{SUFFIX_NODE_INTROSPECTION}" + + def test_build_full_topic_with_different_envs(self) -> None: + """Should work with all valid environment prefixes.""" + for env in ["dev", "staging", "prod", "test", "local"]: + topic = build_full_topic(env, "myapp", SUFFIX_NODE_INTROSPECTION) + assert topic.startswith(f"{env}.myapp.") + + def test_invalid_suffix_raises(self) -> None: + """Invalid suffix should raise TopicCompositionError.""" + with pytest.raises(TopicCompositionError): + build_full_topic("dev", "omnibase", "bad.format") + + def test_invalid_env_raises(self) -> None: + """Invalid environment should raise TopicCompositionError.""" + with pytest.raises(TopicCompositionError): + build_full_topic("invalid-env", "omnibase", SUFFIX_NODE_INTROSPECTION) + + def test_empty_namespace_raises(self) -> None: + """Empty namespace should raise TopicCompositionError.""" + with pytest.raises(TopicCompositionError): + build_full_topic("dev", "", SUFFIX_NODE_INTROSPECTION) + + def test_namespace_with_hyphen(self) -> None: + """Namespace with hyphen should be valid.""" + topic = build_full_topic("dev", "my-app", SUFFIX_NODE_INTROSPECTION) + assert topic == f"dev.my-app.{SUFFIX_NODE_INTROSPECTION}" + + def test_namespace_with_underscore(self) -> None: + """Namespace with underscore should be valid.""" + topic = build_full_topic("dev", "my_app", SUFFIX_NODE_INTROSPECTION) + assert topic == f"dev.my_app.{SUFFIX_NODE_INTROSPECTION}" + + def test_namespace_with_invalid_chars_raises(self) -> None: + """Namespace with invalid characters should raise TopicCompositionError.""" + with pytest.raises(TopicCompositionError): + build_full_topic("dev", "my.app", SUFFIX_NODE_INTROSPECTION) + + def test_namespace_with_spaces_raises(self) -> None: + """Namespace with spaces should raise TopicCompositionError.""" + with pytest.raises(TopicCompositionError): + build_full_topic("dev", "my app", SUFFIX_NODE_INTROSPECTION) + + def test_namespace_starting_with_number(self) -> None: + """Namespace starting with number is valid. + + Documents current behavior: namespaces like "123app" are accepted + because isalnum() returns True for strings containing only letters + and digits, regardless of position. + + Note: If this behavior should change (e.g., to require namespaces + start with a letter like Python identifiers), update the validation + in build_full_topic() and change this test to expect + TopicCompositionError. + """ + topic = build_full_topic("dev", "123app", SUFFIX_NODE_INTROSPECTION) + assert topic == f"dev.123app.{SUFFIX_NODE_INTROSPECTION}" + + def test_namespace_all_numbers(self) -> None: + """Namespace with only numbers is valid. + + Documents current behavior: pure numeric namespaces like "12345" + are accepted because isalnum() returns True for digit-only strings. + """ + topic = build_full_topic("dev", "12345", SUFFIX_NODE_INTROSPECTION) + assert topic == f"dev.12345.{SUFFIX_NODE_INTROSPECTION}" + + def test_build_full_topic_with_different_suffixes(self) -> None: + """Should work with different valid suffixes.""" + topic1 = build_full_topic("prod", "omnibase", SUFFIX_NODE_INTROSPECTION) + topic2 = build_full_topic("prod", "omnibase", SUFFIX_NODE_REGISTRATION) + assert topic1 != topic2 + assert topic1.endswith(SUFFIX_NODE_INTROSPECTION) + assert topic2.endswith(SUFFIX_NODE_REGISTRATION) + + def test_error_message_contains_valid_envs(self) -> None: + """Error message for invalid env should list valid options.""" + with pytest.raises(TopicCompositionError) as exc_info: + build_full_topic("invalid", "omnibase", SUFFIX_NODE_INTROSPECTION) + error_message = str(exc_info.value) + assert "dev" in error_message + assert "prod" in error_message + assert "staging" in error_message + + def test_error_message_for_invalid_suffix(self) -> None: + """Error message for invalid suffix should explain the issue.""" + with pytest.raises(TopicCompositionError) as exc_info: + build_full_topic("dev", "omnibase", "not-a-valid-suffix") + error_message = str(exc_info.value) + assert "not-a-valid-suffix" in error_message + + def test_env_case_sensitivity(self) -> None: + """Environment prefix is case-sensitive - 'Dev' should fail.""" + with pytest.raises(TopicCompositionError): + build_full_topic("Dev", "omnibase", SUFFIX_NODE_INTROSPECTION) + + def test_namespace_with_unicode_is_valid(self) -> None: + """Namespace with unicode letters is valid. + + Documents current behavior: Python's str.isalnum() returns True for + Unicode letters (not just ASCII), so namespaces like "my-äpp" are + accepted. This is intentional as it enables internationalized namespaces. + + Note: If ASCII-only namespaces are required in the future, update + the validation in build_full_topic() to use str.isascii() check and + change this test to expect TopicCompositionError. + """ + topic = build_full_topic("dev", "my-äpp", SUFFIX_NODE_INTROSPECTION) + assert topic == f"dev.my-äpp.{SUFFIX_NODE_INTROSPECTION}" + + def test_namespace_at_max_length(self) -> None: + """Namespace at exactly MAX_NAMESPACE_LENGTH should be valid.""" + from omnibase_infra.topics.util_topic_composition import MAX_NAMESPACE_LENGTH + + max_namespace = "a" * MAX_NAMESPACE_LENGTH + topic = build_full_topic("dev", max_namespace, SUFFIX_NODE_INTROSPECTION) + assert f"dev.{max_namespace}." in topic + + def test_namespace_exceeds_max_length_raises(self) -> None: + """Namespace exceeding MAX_NAMESPACE_LENGTH should raise TopicCompositionError.""" + from omnibase_infra.topics.util_topic_composition import MAX_NAMESPACE_LENGTH + + too_long_namespace = "a" * (MAX_NAMESPACE_LENGTH + 1) + with pytest.raises(TopicCompositionError) as exc_info: + build_full_topic("dev", too_long_namespace, SUFFIX_NODE_INTROSPECTION) + assert "maximum length" in str(exc_info.value).lower() + assert str(MAX_NAMESPACE_LENGTH) in str(exc_info.value) + + def test_very_long_namespace_raises(self) -> None: + """Very long namespace (100+ chars) should raise TopicCompositionError.""" + long_namespace = "a" * 150 + with pytest.raises(TopicCompositionError): + build_full_topic("dev", long_namespace, SUFFIX_NODE_INTROSPECTION) + + def test_namespace_with_consecutive_hyphens(self) -> None: + """Namespace with consecutive hyphens should be valid.""" + topic = build_full_topic("dev", "my--app", SUFFIX_NODE_INTROSPECTION) + assert topic == f"dev.my--app.{SUFFIX_NODE_INTROSPECTION}" + + def test_namespace_with_mixed_case_raises(self) -> None: + """Namespace with mixed case should raise TopicCompositionError. + + Namespaces must be lowercase to ensure consistent topic naming + since ONEX topic suffixes are lowercase. + """ + with pytest.raises(TopicCompositionError) as exc_info: + build_full_topic("dev", "MyApp", SUFFIX_NODE_INTROSPECTION) + assert "lowercase" in str(exc_info.value).lower() + + +class TestTopicCompositionError: + """Tests for TopicCompositionError exception class.""" + + def test_is_onex_error_subclass(self) -> None: + """TopicCompositionError should be an OnexError subclass.""" + assert issubclass(TopicCompositionError, OnexError) + + def test_error_can_be_raised_with_message(self) -> None: + """TopicCompositionError should accept a message.""" + with pytest.raises(TopicCompositionError) as exc_info: + raise TopicCompositionError("Custom error message") + assert "Custom error message" in str(exc_info.value) + + def test_error_can_be_caught_as_onex_error(self) -> None: + """TopicCompositionError should be catchable as OnexError.""" + try: + build_full_topic("invalid", "omnibase", SUFFIX_NODE_INTROSPECTION) + except OnexError: + pass # Should be caught + else: + pytest.fail("Expected OnexError to be raised")