From b22827bf321b49351ce28b676ceb5878037d412f Mon Sep 17 00:00:00 2001 From: Jonah Gray Date: Tue, 27 Jan 2026 14:17:18 -0500 Subject: [PATCH 1/6] feat(OMN-1547): Replace hardcoded topics with validated suffix constants - Add platform topic suffix constants module with import-time validation - Add build_full_topic() composition helper with env/namespace/suffix validation - Update 7 platform files to use suffix constants instead of hardcoded strings - Remove hardcoded omniclaude domain topics from config_consumer.py - Consolidate introspection topic definitions (model = canonical, mixin = consumer) - Add 30 unit tests for suffix validation and topic composition --- .../mixins/mixin_node_introspection.py | 14 +- .../discovery/model_introspection_config.py | 27 ++-- .../projection/model_snapshot_topic_config.py | 5 +- .../models/model_runtime_scheduler_config.py | 7 +- .../transition_notification_publisher.py | 5 +- .../services/mcp/service_mcp_tool_sync.py | 3 +- .../services/session/config_consumer.py | 33 +++- src/omnibase_infra/topics/__init__.py | 43 ++++++ .../topics/platform_topic_suffixes.py | 141 ++++++++++++++++++ .../topics/util_topic_composition.py | 60 ++++++++ tests/unit/topics/__init__.py | 1 + .../topics/test_platform_topic_suffixes.py | 116 ++++++++++++++ .../topics/test_util_topic_composition.py | 107 +++++++++++++ 13 files changed, 531 insertions(+), 31 deletions(-) create mode 100644 src/omnibase_infra/topics/__init__.py create mode 100644 src/omnibase_infra/topics/platform_topic_suffixes.py create mode 100644 src/omnibase_infra/topics/util_topic_composition.py create mode 100644 tests/unit/topics/__init__.py create mode 100644 tests/unit/topics/test_platform_topic_suffixes.py create mode 100644 tests/unit/topics/test_util_topic_composition.py diff --git a/src/omnibase_infra/mixins/mixin_node_introspection.py b/src/omnibase_infra/mixins/mixin_node_introspection.py index d3dcf3c2ff..5e05224c8a 100644 --- a/src/omnibase_infra/mixins/mixin_node_introspection.py +++ b/src/omnibase_infra/mixins/mixin_node_introspection.py @@ -237,10 +237,16 @@ 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" +# Event topic constants - imported from canonical source +from omnibase_infra.models.discovery.model_introspection_config import ( + DEFAULT_HEARTBEAT_TOPIC as HEARTBEAT_TOPIC, +) +from omnibase_infra.models.discovery.model_introspection_config import ( + DEFAULT_INTROSPECTION_TOPIC as INTROSPECTION_TOPIC, +) +from omnibase_infra.models.discovery.model_introspection_config import ( + DEFAULT_REQUEST_INTROSPECTION_TOPIC as REQUEST_INTROSPECTION_TOPIC, +) # Performance threshold constants (in milliseconds) PERF_THRESHOLD_GET_CAPABILITIES_MS = 50.0 diff --git a/src/omnibase_infra/models/discovery/model_introspection_config.py b/src/omnibase_infra/models/discovery/model_introspection_config.py index 06afea161b..b1576bef01 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 @@ -290,9 +295,9 @@ 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", }, { "node_id": "550e8400-e29b-41d4-a716-446655440001", @@ -302,9 +307,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": "onex.evt.platform.node-introspection.v1", + "heartbeat_topic": "onex.evt.platform.node-heartbeat.v1", + "request_introspection_topic": "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 a7520500b8..46238f078b 100644 --- a/src/omnibase_infra/services/mcp/service_mcp_tool_sync.py +++ b/src/omnibase_infra/services/mcp/service_mcp_tool_sync.py @@ -29,6 +29,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,7 +78,7 @@ class ServiceMCPToolSync: """ # Topic for node registration events - TOPIC = "node.registration.v1" + TOPIC = SUFFIX_NODE_REGISTRATION GROUP_ID = "mcp-tool-sync" # MCP tag constants 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..0cd940da03 --- /dev/null +++ b/src/omnibase_infra/topics/__init__.py @@ -0,0 +1,43 @@ +"""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 list 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 ( + 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 list + "ALL_PLATFORM_SUFFIXES", + # Topic composition utilities + "build_full_topic", + "TopicCompositionError", +] 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..298455c559 --- /dev/null +++ b/src/omnibase_infra/topics/platform_topic_suffixes.py @@ -0,0 +1,141 @@ +"""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.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 LIST +# ============================================================================= + +ALL_PLATFORM_SUFFIXES: list[str] = [ + SUFFIX_NODE_REGISTRATION, + SUFFIX_NODE_INTROSPECTION, + SUFFIX_NODE_HEARTBEAT, + SUFFIX_REQUEST_INTROSPECTION, + SUFFIX_FSM_STATE_TRANSITIONS, + SUFFIX_RUNTIME_TICK, + SUFFIX_REGISTRATION_SNAPSHOTS, +] +"""Complete list of all platform-reserved topic suffixes. + +Use this list 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: + ValueError: 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 ValueError( + 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..07fe31b4d6 --- /dev/null +++ b/src/omnibase_infra/topics/util_topic_composition.py @@ -0,0 +1,60 @@ +"""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.validation import validate_topic_suffix +from omnibase_core.validation.validator_topic_suffix import ENV_PREFIXES + + +class TopicCompositionError(ValueError): + """Raised when topic composition fails due to invalid components.""" + + +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") + 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 + + 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 + if not namespace: + raise TopicCompositionError("Namespace cannot be empty") + if not namespace.replace("-", "").replace("_", "").isalnum(): + raise TopicCompositionError( + f"Invalid namespace '{namespace}'. " + "Must contain only alphanumeric characters, hyphens, and underscores" + ) + + # 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/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..e3ba1d4b52 --- /dev/null +++ b/tests/unit/topics/test_platform_topic_suffixes.py @@ -0,0 +1,116 @@ +"""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_list(self) -> None: + """ALL_PLATFORM_SUFFIXES should be a list.""" + assert isinstance(ALL_PLATFORM_SUFFIXES, list) + + 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}" 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..b25ae777b0 --- /dev/null +++ b/tests/unit/topics/test_util_topic_composition.py @@ -0,0 +1,107 @@ +"""Tests for topic composition utilities.""" + +import pytest + +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_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 + + +class TestTopicCompositionError: + """Tests for TopicCompositionError exception class.""" + + def test_is_value_error_subclass(self) -> None: + """TopicCompositionError should be a ValueError subclass.""" + assert issubclass(TopicCompositionError, ValueError) + + 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_value_error(self) -> None: + """TopicCompositionError should be catchable as ValueError.""" + try: + build_full_topic("invalid", "omnibase", SUFFIX_NODE_INTROSPECTION) + except ValueError: + pass # Should be caught + else: + pytest.fail("Expected ValueError to be raised") From d9b698fb183e618544b8a83a5ce03c944f4af1d2 Mon Sep 17 00:00:00 2001 From: Jonah Gray Date: Tue, 27 Jan 2026 14:40:28 -0500 Subject: [PATCH 2/6] fix(OMN-1547): Address PR review feedback - Move imports to top of mixin_node_introspection.py (PEP 8) - Change ALL_PLATFORM_SUFFIXES from list to immutable tuple - Add edge case tests for numeric namespace validation --- .../mixins/mixin_node_introspection.py | 21 ++++++++-------- src/omnibase_infra/topics/__init__.py | 4 ++-- .../topics/platform_topic_suffixes.py | 10 ++++---- .../topics/test_platform_topic_suffixes.py | 6 ++--- .../topics/test_util_topic_composition.py | 24 +++++++++++++++++++ 5 files changed, 45 insertions(+), 20 deletions(-) diff --git a/src/omnibase_infra/mixins/mixin_node_introspection.py b/src/omnibase_infra/mixins/mixin_node_introspection.py index 5e05224c8a..39dba7bda8 100644 --- a/src/omnibase_infra/mixins/mixin_node_introspection.py +++ b/src/omnibase_infra/mixins/mixin_node_introspection.py @@ -217,6 +217,13 @@ async def shutdown(self): ModelIntrospectionConfig, ModelIntrospectionTaskConfig, ) + +# Event topic constants - imported from canonical source +from omnibase_infra.models.discovery.model_introspection_config import ( + DEFAULT_HEARTBEAT_TOPIC, + DEFAULT_INTROSPECTION_TOPIC, + DEFAULT_REQUEST_INTROSPECTION_TOPIC, +) from omnibase_infra.models.discovery.model_introspection_performance_metrics import ( ModelIntrospectionPerformanceMetrics, ) @@ -237,16 +244,10 @@ async def shutdown(self): logger = logging.getLogger(__name__) -# Event topic constants - imported from canonical source -from omnibase_infra.models.discovery.model_introspection_config import ( - DEFAULT_HEARTBEAT_TOPIC as HEARTBEAT_TOPIC, -) -from omnibase_infra.models.discovery.model_introspection_config import ( - DEFAULT_INTROSPECTION_TOPIC as INTROSPECTION_TOPIC, -) -from omnibase_infra.models.discovery.model_introspection_config import ( - DEFAULT_REQUEST_INTROSPECTION_TOPIC as REQUEST_INTROSPECTION_TOPIC, -) +# Module-level topic aliases (imported at top from canonical source) +HEARTBEAT_TOPIC = DEFAULT_HEARTBEAT_TOPIC +INTROSPECTION_TOPIC = DEFAULT_INTROSPECTION_TOPIC +REQUEST_INTROSPECTION_TOPIC = DEFAULT_REQUEST_INTROSPECTION_TOPIC # Performance threshold constants (in milliseconds) PERF_THRESHOLD_GET_CAPABILITIES_MS = 50.0 diff --git a/src/omnibase_infra/topics/__init__.py b/src/omnibase_infra/topics/__init__.py index 0cd940da03..97966688b0 100644 --- a/src/omnibase_infra/topics/__init__.py +++ b/src/omnibase_infra/topics/__init__.py @@ -6,7 +6,7 @@ Exports: Platform topic suffix constants (e.g., SUFFIX_NODE_REGISTRATION) - ALL_PLATFORM_SUFFIXES: Complete list of all platform-reserved suffixes + 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 """ @@ -35,7 +35,7 @@ "SUFFIX_FSM_STATE_TRANSITIONS", "SUFFIX_RUNTIME_TICK", "SUFFIX_REGISTRATION_SNAPSHOTS", - # Aggregate list + # Aggregate tuple "ALL_PLATFORM_SUFFIXES", # Topic composition utilities "build_full_topic", diff --git a/src/omnibase_infra/topics/platform_topic_suffixes.py b/src/omnibase_infra/topics/platform_topic_suffixes.py index 298455c559..da5d6c7a04 100644 --- a/src/omnibase_infra/topics/platform_topic_suffixes.py +++ b/src/omnibase_infra/topics/platform_topic_suffixes.py @@ -97,10 +97,10 @@ """ # ============================================================================= -# AGGREGATE LIST +# AGGREGATE TUPLE # ============================================================================= -ALL_PLATFORM_SUFFIXES: list[str] = [ +ALL_PLATFORM_SUFFIXES: tuple[str, ...] = ( SUFFIX_NODE_REGISTRATION, SUFFIX_NODE_INTROSPECTION, SUFFIX_NODE_HEARTBEAT, @@ -108,10 +108,10 @@ SUFFIX_FSM_STATE_TRANSITIONS, SUFFIX_RUNTIME_TICK, SUFFIX_REGISTRATION_SNAPSHOTS, -] -"""Complete list of all platform-reserved topic suffixes. +) +"""Complete tuple of all platform-reserved topic suffixes. -Use this list for: +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 diff --git a/tests/unit/topics/test_platform_topic_suffixes.py b/tests/unit/topics/test_platform_topic_suffixes.py index e3ba1d4b52..40274ca917 100644 --- a/tests/unit/topics/test_platform_topic_suffixes.py +++ b/tests/unit/topics/test_platform_topic_suffixes.py @@ -100,9 +100,9 @@ def test_registration_snapshots_suffix_format(self) -> None: == "onex.snapshot.platform.registration-snapshots.v1" ) - def test_all_platform_suffixes_is_list(self) -> None: - """ALL_PLATFORM_SUFFIXES should be a list.""" - assert isinstance(ALL_PLATFORM_SUFFIXES, list) + 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.""" diff --git a/tests/unit/topics/test_util_topic_composition.py b/tests/unit/topics/test_util_topic_composition.py index b25ae777b0..9d298416c6 100644 --- a/tests/unit/topics/test_util_topic_composition.py +++ b/tests/unit/topics/test_util_topic_composition.py @@ -59,6 +59,30 @@ def test_namespace_with_spaces_raises(self) -> None: 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) From 6d29348944b691d7ad288f602ff2eabffb2eacd3 Mon Sep 17 00:00:00 2001 From: Jonah Gray Date: Tue, 27 Jan 2026 14:47:34 -0500 Subject: [PATCH 3/6] docs(OMN-1547): Add clarifying comments for PR review feedback - Add namespace validation rules to build_full_topic() docstring - Document that numeric-only namespaces are intentionally valid - Mark mixin topic aliases as DEPRECATED with migration guidance --- .../mixins/mixin_node_introspection.py | 4 +++- src/omnibase_infra/topics/util_topic_composition.py | 12 +++++++++++- 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/src/omnibase_infra/mixins/mixin_node_introspection.py b/src/omnibase_infra/mixins/mixin_node_introspection.py index 39dba7bda8..ac608a3099 100644 --- a/src/omnibase_infra/mixins/mixin_node_introspection.py +++ b/src/omnibase_infra/mixins/mixin_node_introspection.py @@ -244,7 +244,9 @@ async def shutdown(self): logger = logging.getLogger(__name__) -# Module-level topic aliases (imported at top from canonical source) +# DEPRECATED: Module-level topic aliases for backward compatibility. +# New code should import directly from omnibase_infra.topics instead. +# These aliases may be removed in a future version. HEARTBEAT_TOPIC = DEFAULT_HEARTBEAT_TOPIC INTROSPECTION_TOPIC = DEFAULT_INTROSPECTION_TOPIC REQUEST_INTROSPECTION_TOPIC = DEFAULT_REQUEST_INTROSPECTION_TOPIC diff --git a/src/omnibase_infra/topics/util_topic_composition.py b/src/omnibase_infra/topics/util_topic_composition.py index 07fe31b4d6..b4aeab3cc7 100644 --- a/src/omnibase_infra/topics/util_topic_composition.py +++ b/src/omnibase_infra/topics/util_topic_composition.py @@ -17,7 +17,7 @@ def build_full_topic(env: str, namespace: str, suffix: str) -> str: Args: env: Environment prefix (e.g., "dev", "staging", "prod", "test", "local") - namespace: Namespace/tenant identifier (e.g., "omnibase", "myapp") + namespace: Namespace/tenant identifier (e.g., "omnibase", "myapp", "tenant-123") suffix: Validated topic suffix (e.g., "onex.evt.platform.node-introspection.v1") Returns: @@ -28,6 +28,12 @@ def build_full_topic(env: str, namespace: str, suffix: str) -> str: 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' @@ -43,6 +49,10 @@ def build_full_topic(env: str, namespace: str, suffix: str) -> str: ) # 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 not namespace.replace("-", "").replace("_", "").isalnum(): From 32d1f57cb5462c9c7aabb25ea0606633beca1b28 Mon Sep 17 00:00:00 2001 From: Jonah Gray Date: Tue, 27 Jan 2026 15:18:01 -0500 Subject: [PATCH 4/6] fix(OMN-1547): Address PR review issues - Remove deprecated topic aliases per no-backwards-compat policy - Restore JSON schema example variation for documentation clarity - Add edge case tests for topic composition (unicode, long names, case) --- .../mixins/mixin_node_introspection.py | 17 --------- .../discovery/model_introspection_config.py | 9 +++-- .../topics/test_util_topic_composition.py | 35 +++++++++++++++++++ 3 files changed, 41 insertions(+), 20 deletions(-) diff --git a/src/omnibase_infra/mixins/mixin_node_introspection.py b/src/omnibase_infra/mixins/mixin_node_introspection.py index ac608a3099..bcc9ad67b9 100644 --- a/src/omnibase_infra/mixins/mixin_node_introspection.py +++ b/src/omnibase_infra/mixins/mixin_node_introspection.py @@ -217,13 +217,6 @@ async def shutdown(self): ModelIntrospectionConfig, ModelIntrospectionTaskConfig, ) - -# Event topic constants - imported from canonical source -from omnibase_infra.models.discovery.model_introspection_config import ( - DEFAULT_HEARTBEAT_TOPIC, - DEFAULT_INTROSPECTION_TOPIC, - DEFAULT_REQUEST_INTROSPECTION_TOPIC, -) from omnibase_infra.models.discovery.model_introspection_performance_metrics import ( ModelIntrospectionPerformanceMetrics, ) @@ -244,13 +237,6 @@ async def shutdown(self): logger = logging.getLogger(__name__) -# DEPRECATED: Module-level topic aliases for backward compatibility. -# New code should import directly from omnibase_infra.topics instead. -# These aliases may be removed in a future version. -HEARTBEAT_TOPIC = DEFAULT_HEARTBEAT_TOPIC -INTROSPECTION_TOPIC = DEFAULT_INTROSPECTION_TOPIC -REQUEST_INTROSPECTION_TOPIC = DEFAULT_REQUEST_INTROSPECTION_TOPIC - # Performance threshold constants (in milliseconds) PERF_THRESHOLD_GET_CAPABILITIES_MS = 50.0 PERF_THRESHOLD_DISCOVER_CAPABILITIES_MS = 30.0 @@ -2494,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 b1576bef01..eb85519e79 100644 --- a/src/omnibase_infra/models/discovery/model_introspection_config.py +++ b/src/omnibase_infra/models/discovery/model_introspection_config.py @@ -287,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", @@ -299,6 +300,8 @@ def validate_topic_name(cls, v: str) -> str: "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", @@ -307,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.evt.platform.node-introspection.v1", - "heartbeat_topic": "onex.evt.platform.node-heartbeat.v1", - "request_introspection_topic": "onex.cmd.platform.request-introspection.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/tests/unit/topics/test_util_topic_composition.py b/tests/unit/topics/test_util_topic_composition.py index 9d298416c6..82635419be 100644 --- a/tests/unit/topics/test_util_topic_composition.py +++ b/tests/unit/topics/test_util_topic_composition.py @@ -107,6 +107,41 @@ def test_error_message_for_invalid_suffix(self) -> None: 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_very_long_namespace(self) -> None: + """Very long namespace should still be valid if characters are valid.""" + long_namespace = "a" * 100 + topic = build_full_topic("dev", long_namespace, SUFFIX_NODE_INTROSPECTION) + assert f"dev.{long_namespace}." in topic + + 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(self) -> None: + """Namespace with mixed case should be valid.""" + topic = build_full_topic("dev", "MyApp", SUFFIX_NODE_INTROSPECTION) + assert topic == f"dev.MyApp.{SUFFIX_NODE_INTROSPECTION}" + class TestTopicCompositionError: """Tests for TopicCompositionError exception class.""" From cf30bfec8a6a1cadfdae62091084bc46342a0856 Mon Sep 17 00:00:00 2001 From: Jonah Gray Date: Tue, 27 Jan 2026 15:34:45 -0500 Subject: [PATCH 5/6] fix(OMN-1547): Add namespace max length validation and export test - Add MAX_NAMESPACE_LENGTH constant (100 chars) to prevent overly long topics - Add validation in build_full_topic() with clear error message - Export MAX_NAMESPACE_LENGTH from topics package for user access - Add test_all_suffix_constants_exported to catch missing exports - Update namespace length tests to verify new validation behavior --- src/omnibase_infra/topics/__init__.py | 2 ++ .../topics/util_topic_composition.py | 13 +++++++++ .../topics/test_platform_topic_suffixes.py | 17 +++++++++++ .../topics/test_util_topic_composition.py | 28 +++++++++++++++---- 4 files changed, 55 insertions(+), 5 deletions(-) diff --git a/src/omnibase_infra/topics/__init__.py b/src/omnibase_infra/topics/__init__.py index 97966688b0..b84f1c0eb5 100644 --- a/src/omnibase_infra/topics/__init__.py +++ b/src/omnibase_infra/topics/__init__.py @@ -22,6 +22,7 @@ SUFFIX_RUNTIME_TICK, ) from omnibase_infra.topics.util_topic_composition import ( + MAX_NAMESPACE_LENGTH, TopicCompositionError, build_full_topic, ) @@ -40,4 +41,5 @@ # Topic composition utilities "build_full_topic", "TopicCompositionError", + "MAX_NAMESPACE_LENGTH", ] diff --git a/src/omnibase_infra/topics/util_topic_composition.py b/src/omnibase_infra/topics/util_topic_composition.py index b4aeab3cc7..6b5751fe85 100644 --- a/src/omnibase_infra/topics/util_topic_composition.py +++ b/src/omnibase_infra/topics/util_topic_composition.py @@ -7,6 +7,14 @@ 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(ValueError): """Raised when topic composition fails due to invalid components.""" @@ -55,6 +63,11 @@ def build_full_topic(env: str, namespace: str, suffix: str) -> str: # 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}'. " diff --git a/tests/unit/topics/test_platform_topic_suffixes.py b/tests/unit/topics/test_platform_topic_suffixes.py index 40274ca917..9ef835076c 100644 --- a/tests/unit/topics/test_platform_topic_suffixes.py +++ b/tests/unit/topics/test_platform_topic_suffixes.py @@ -114,3 +114,20 @@ def test_suffixes_use_platform_producer(self) -> None: 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 index 82635419be..70320bd86c 100644 --- a/tests/unit/topics/test_util_topic_composition.py +++ b/tests/unit/topics/test_util_topic_composition.py @@ -126,11 +126,29 @@ def test_namespace_with_unicode_is_valid(self) -> None: topic = build_full_topic("dev", "my-äpp", SUFFIX_NODE_INTROSPECTION) assert topic == f"dev.my-äpp.{SUFFIX_NODE_INTROSPECTION}" - def test_very_long_namespace(self) -> None: - """Very long namespace should still be valid if characters are valid.""" - long_namespace = "a" * 100 - topic = build_full_topic("dev", long_namespace, SUFFIX_NODE_INTROSPECTION) - assert f"dev.{long_namespace}." in topic + 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.""" From ea67f9d3fa235f729d1f6ef1b5980cc5dd586051 Mon Sep 17 00:00:00 2001 From: Jonah Gray Date: Tue, 27 Jan 2026 16:12:43 -0500 Subject: [PATCH 6/6] fix(OMN-1547): Update test to expect fail-fast behavior for session consumer The PR intentionally removed hardcoded domain topics from ConfigSessionConsumer and added fail-fast validation requiring explicit topic configuration. This updates the test suite to: - Add test_consumer_requires_explicit_topics: verifies ProtocolConfigurationError is raised when no topics are provided (fail-fast behavior) - Add test_consumer_uses_explicit_config: verifies consumer works correctly when topics are explicitly configured - Remove test_consumer_uses_default_config: no longer valid since empty defaults are now rejected This fixes CI failure caused by the intentional behavioral change in config_consumer.py. --- tests/services/session/test_consumer.py | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) 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."""