Repository navigation
feat(OMN-1613): Add event bus topic storage to registry for dynamic topic routing - #199
Merged
jonahgabriel merged 14 commits intoJan 27, 2026
Conversation
…opic routing Enable registry-driven topic routing by storing event_bus configuration (subscribe_topics, publish_topics) as part of node introspection data in Consul KV. Key changes: - Add ModelEventBusTopicEntry and ModelNodeEventBusConfig models - Add event_bus field to ModelNodeIntrospectionEvent - Add _extract_event_bus_config() to MixinNodeIntrospection - Add MixinConsulTopicIndex for topic index management in Consul KV - Add get_subscribers_for_topic() to RuntimeHostProcess - Update RegistrationReducer to pass through event_bus Design decisions: - Topics stored as environment-qualified strings (e.g., dev.onex.evt...) - Consul KV is the authoritative routing index - Index updates are idempotent (safe for restarts/retries) - Metadata is tooling-only; routing uses only topic strings - Fail-fast on unresolved placeholders Test coverage: - 88 unit tests across 6 test files - 33 integration tests across 2 test files
📝 WalkthroughWalkthroughAdds event-bus models and introspection-based topic resolution, a Consul KV-backed topic→subscriber index mixin integrated into service registration, a runtime API to query subscribers, and extensive unit/integration tests for extraction, storage, idempotency, and queries. Changes
Sequence Diagram(s)sequenceDiagram
participant Node as Node Introspection
participant Contract as Contract Spec
participant Extract as _extract_event_bus_config()
participant Reducer as RegistrationReducer
participant Handler as HandlerConsul (Service + TopicIndex)
participant ConsulKV as Consul KV
Node->>Contract: read event_bus subcontract
Node->>Extract: resolve topics with env_prefix
Extract-->>Node: ModelNodeEventBusConfig
Node->>Reducer: introspection event (includes event_bus)
Reducer->>Handler: build ModelPayloadConsulRegister (event_bus_config)
Handler->>ConsulKV: _update_topic_index() (get previous subscribers)
ConsulKV-->>Handler: previous subscriber lists
Handler->>ConsulKV: add/remove subscriber entries (onex/topics/{topic}/subscribers)
Handler->>ConsulKV: _store_node_event_bus() (per-node keys)
sequenceDiagram
participant App as Caller
participant Runtime as RuntimeHostProcess
participant ConsulHandler as Consul Handler
participant ConsulKV as Consul KV
App->>Runtime: get_subscribers_for_topic(topic)
Runtime->>ConsulHandler: GET onex/topics/{topic}/subscribers
ConsulHandler->>ConsulKV: retrieve key
ConsulKV-->>ConsulHandler: JSON array or missing
ConsulHandler-->>Runtime: raw payload or None/errors
Runtime->>Runtime: parse JSON, validate UUIDs
Runtime-->>App: list[UUID] or []
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 3✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🤖 Fix all issues with AI agents
In `@src/omnibase_infra/handlers/mixins/mixin_consul_service.py`:
- Around line 153-209: The _parse_event_bus_config function should validate
topic entries and accept a correlation_id and a JsonType payload: change the
signature to accept event_bus_data: JsonType and correlation_id: str, and inside
both subscribe and publish loops validate that entry is a dict and that
entry.get("topic") is a non-empty string (not coerced via str(...)); if invalid,
raise a typed validation error (e.g., PayloadValidationError or ValueError if no
custom error exists) including the correlation_id and a clear message, and only
construct ModelEventBusTopicEntry when topic is valid; ensure the same checks
are applied for both subscribe_topics and publish_topics before returning the
ModelNodeEventBusConfig.
- Around line 316-336: The handler currently ignores cases where payload
contains event_bus_config but node_id is missing/invalid; change the logic in
the block that reads payload.get("event_bus_config") and payload.get("node_id")
to fail fast by raising a RuntimeHostError with context when event_bus_config is
a dict but node_id is not a non-empty string. Specifically, before calling
self._parse_event_bus_config, validate node_id and if invalid raise
RuntimeHostError including correlation_id and a clear message about the
missing/invalid node_id so misconfigured registrations are surfaced rather than
silently dropped; keep subsequent calls to _parse_event_bus_config,
_update_topic_index, and _store_node_event_bus unchanged when node_id is valid.
In `@src/omnibase_infra/handlers/mixins/mixin_consul_topic_index.py`:
- Around line 313-317: JSON parsing of Consul values using json.loads (e.g.,
when reading previous_result into old_topics around previous_key and wherever
json.loads is used in _update_topic_index and related methods) can raise
JSONDecodeError/TypeError and currently bubbles as a raw ValueError; wrap every
json.loads call in a try/except that catches
ValueError/TypeError/JSONDecodeError and re-raise as InfraConsulError including
the consul key (e.g., previous_key or the specific consul_key variable) and the
correlation_id, preserving exception chaining (raise InfraConsulError(...,
consul_key=consul_key, correlation_id=correlation_id) from e); apply this
pattern to the json.loads usages referenced (around
previous_key/event_bus.subscribe_topic_strings and the blocks at the other noted
locations) so the error contract uses OnexError subclasses and retains trace
context.
In `@src/omnibase_infra/mixins/mixin_node_introspection.py`:
- Around line 1444-1453: Replace the plain ValueError re-raise with the
project's structured error type: catch the ValueError from
self._extract_event_bus_config and raise ProtocolConfigurationError (a subclass
of OnexError) instead, preserving the original message and chaining the original
exception via "from e"; also add (or update) the import for
ProtocolConfigurationError at the top of the module so the symbol is available.
In `@tests/integration/registration/test_introspection_event_bus_e2e.py`:
- Around line 681-704: The v1 registration currently calls
handler._store_node_event_bus before handler._update_topic_index which reverses
the intended sequence used for delta computation; change the v1 block to call
handler._update_topic_index(node_id=unique_node_id, event_bus=config_v1,
correlation_id=correlation_id) first and then handler._store_node_event_bus(...)
so it matches the v2 ordering (update index FIRST, then store SECOND) and aligns
with the comment about reading previous state.
🧹 Nitpick comments (1)
tests/integration/registration/test_introspection_event_bus_e2e.py (1)
342-351: Consider using specific exception type for clarity.The test catches generic
Exceptionfor the frozen model mutation check. Usingpydantic.ValidationErrorwould make the test more precise and self-documenting.Proposed refinement
+from pydantic import ValidationError + def test_event_bus_config_immutability(self) -> None: """ModelNodeEventBusConfig is frozen (immutable).""" config = ModelNodeEventBusConfig( subscribe_topics=[ModelEventBusTopicEntry(topic="dev.test.v1")], ) # Attempting to modify should raise - with pytest.raises(Exception): # ValidationError for frozen model + with pytest.raises(ValidationError): config.subscribe_topics = [] # type: ignore[misc]
…ting Code quality improvements from PR review: - Add warning logs for empty/missing topic entries in _parse_event_bus_config - Add specific exception handlers (InfraConsulError, InfraTimeoutError, InfraUnavailableError) before broad Exception catch in get_subscribers_for_topic - Document non-atomic read-modify-write limitation in _update_topic_index with production recommendations (Consul transactions, optimistic locking)
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In `@src/omnibase_infra/runtime/service_runtime_host_process.py`:
- Around line 2297-2331: The current list comprehension [UUID(nid) for nid in
node_ids_raw if isinstance(nid, str)] will raise ValueError on the first invalid
UUID and cause all subscribers to be dropped; replace it with an explicit loop
over node_ids_raw (checking isinstance(nid, str)), attempt to convert each nid
with UUID(nid) inside a try/except ValueError, append successful conversions to
a new list and skip (and optionally logger.warning) invalid entries so valid
UUIDs are preserved, then return that filtered list (also remove or adjust the
outer ValueError handler that currently returns [] for any conversion error).
Use the existing variables node_ids_raw, topic, correlation_id, and logger to
provide contextual warnings for skipped invalid IDs.
pytest 8+ no longer supports defining pytest_plugins in non-top-level conftest files. The directive now raises an error instead of a warning. Changes: - Move pytest_plugins from tests/integration/registration/conftest.py to tests/conftest.py (root level) - Update docstrings to document fixture availability This fixes the "2 errors" seen in all CI test splits: ERROR collecting tests/integration/registration Defining 'pytest_plugins' in a non-top-level conftest is no longer supported Reference: https://docs.pytest.org/en/stable/deprecations.html#pytest-plugins-in-non-top-level-conftest-files
…egistration pytest_plugins cannot be used for conftest.py files that are already in the test directory tree - pytest automatically discovers them during collection, causing "Plugin already registered under a different name" errors. Solution: Use explicit Python imports + re-exports to share fixtures from sibling conftest directories. Changes: - Remove pytest_plugins from tests/conftest.py - Add explicit fixture imports in tests/integration/registration/conftest.py This pattern is the recommended approach for sharing fixtures across sibling test directories without causing plugin registration conflicts.
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In `@tests/integration/registration/conftest.py`:
- Around line 5-6: The module docstring in
tests/integration/registration/conftest.py incorrectly mentions Kafka fixtures;
update the top-level docstring text to reflect the actual fixtures imported
(e.g., Graph, Qdrant, Vault, HTTP, Consul, PostgreSQL) and remove any reference
to Kafka so the description matches the imported fixtures used in this conftest.
…dation - Add topic validation in _parse_event_bus_config with RuntimeHostError - Fail-fast when event_bus_config present without valid node_id - Wrap json.loads() calls with InfraConsulError for proper error context - Replace ValueError with ProtocolConfigurationError in introspection - Fix UUID parsing to skip invalid entries instead of dropping all - Fix test operation order (update index FIRST, then store) - Update conftest docstring to reflect actual fixtures
- Update test_raises_on_unresolved_placeholder to expect ProtocolConfigurationError instead of ValueError (aligns with ONEX error hierarchy) - Add documentation explaining why inner _client null checks in closures are NOT redundant (type narrowing for mypy + race condition protection in async context)
…lid UUIDs Update test_get_subscribers_for_topic_mixed_valid_invalid_uuids to match the implementation's actual behavior: invalid UUIDs are skipped and logged, valid ones are returned. This graceful degradation is preferred in distributed systems where one corrupt entry shouldn't break routing.
- Add .strip() validation to reject whitespace-only topic strings in event_bus_config parsing (subscribe_topics and publish_topics) - Include consul_key attribute in InfraConsulError logging for better debugging in get_subscribers_for_topic method Addresses PR review feedback for improved input validation and observability.
There was a problem hiding this comment.
Actionable comments posted: 0
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/omnibase_infra/handlers/mixins/mixin_consul_service.py (1)
395-463: Add event bus cleanup to deregistration flow.The
_deregister_servicemethod only deregisters the service from Consul but doesn't clean up event bus config (onex/nodes/{node_id}/event_bus/) or remove the node from topic subscriber indices (onex/topics/{topic}/subscribers). The_register_servicemethod explicitly performs both operations during registration, creating an asymmetry. The_remove_subscriber_from_topicmethod exists and is used elsewhere; verify whether deregistration should invoke it and any related cleanup methods to maintain consistency with the registration flow.
🧹 Nitpick comments (4)
src/omnibase_infra/runtime/service_runtime_host_process.py (1)
2332-2382: Includecorrelation_idin exception handler logs for traceability.The
correlation_idis created at line 2259 and used in some log statements within the try block (lines 2304, 2324-2326), but it's not included in any of the exception handler logextradicts. This reduces observability when debugging Consul-related failures in production.As per coding guidelines, correlation IDs should be included in all error context.
♻️ Proposed fix
except json.JSONDecodeError as e: logger.warning( "Failed to parse topic subscriber JSON", extra={ "topic": topic, "error": str(e), + "correlation_id": str(correlation_id), }, ) return [] except InfraConsulError as e: logger.warning( "Consul error querying topic subscribers", extra={ "topic": topic, "error": str(e), "error_type": "InfraConsulError", "consul_key": getattr(e, "consul_key", None), + "correlation_id": str(correlation_id), }, ) return [] except InfraTimeoutError as e: logger.warning( "Timeout querying topic subscribers", extra={ "topic": topic, "error": str(e), "error_type": "InfraTimeoutError", + "correlation_id": str(correlation_id), }, ) return [] except InfraUnavailableError as e: logger.warning( "Service unavailable for topic subscriber query", extra={ "topic": topic, "error": str(e), "error_type": "InfraUnavailableError", + "correlation_id": str(correlation_id), }, ) return [] except Exception as e: # Graceful degradation - Consul unavailable is not fatal logger.warning( "Failed to query topic subscribers from Consul", extra={ "topic": topic, "error": str(e), "error_type": type(e).__name__, + "correlation_id": str(correlation_id), }, ) return []src/omnibase_infra/handlers/mixins/mixin_consul_service.py (3)
196-198: Consider validatingmessage_categorytype consistency.The
str()coercion onmessage_categorysilently converts non-string values (e.g.,123→"123",None→"None"if no default). While the default"EVENT"handles the missing-key case, an unexpected type likeintwould be coerced without warning.If strict type enforcement is desired, consider adding a type check similar to the
event_typeanddescriptionfields.♻️ Suggested approach
- message_category=str( - entry.get("message_category", "EVENT") - ), + message_category=entry.get("message_category") + if isinstance(entry.get("message_category"), str) + else "EVENT",
228-230: Samemessage_categorycoercion applies here.The same
str()coercion pattern formessage_categoryis used in thepublish_topicsparsing. If the optional refactor is applied tosubscribe_topics, apply it consistently here as well.
374-380: Non-atomic index update and storage sequence.The ordering is correctly documented (update index first to compute delta from old topics, then store new config). However, if
_store_node_event_busfails after_update_topic_indexsucceeds, the reverse index could reference a config that wasn't persisted.Given the PR mentions idempotent index updates, re-registration should recover from this state. Just flagging for awareness—if stronger consistency is needed later, consider wrapping both operations in a Consul transaction or implementing compensating logic.
jonahgabriel
commented
Jan 27, 2026
jonahgabriel
left a comment
jonahgabriel
left a comment
Collaborator
Author
There was a problem hiding this comment.
Code Review
Found 3 issues requiring attention:
- CLAUDE.md Violation: Direct
ModelInfraErrorContext()constructor usage (4 instances) - Bug: Whitespace-padded topics pass validation but are stored unstripped (2 instances)
- Security Concern: Topic strings used in Consul KV paths without character validation
See inline comments for details.
| operation="consul.kv_get_raw", | ||
| target_name="consul_handler", | ||
| correlation_id=correlation_id, | ||
| ) |
Collaborator
Author
There was a problem hiding this comment.
CLAUDE.md Violation: Direct ModelInfraErrorContext() constructor usage.
CLAUDE.md states:
MANDATORY: Use
with_correlation()factory method for error context creation.
This occurs 4 times in this file (lines 125, 145, 201, 221). The file correctly uses .with_correlation() elsewhere (lines 343, 418, 473, 521), showing inconsistent application.
Suggested change
| ) | |
| context = ModelInfraErrorContext.with_correlation( | |
| correlation_id=correlation_id, | |
| transport_type=EnumInfraTransportType.CONSUL, | |
| operation="consul.kv_get_raw", | |
| target_name="consul_handler", | |
| ) |
Additional fixes needed at lines 145, 201, 221 - same pattern.
| ) | ||
| subscribe_topics.append( | ||
| ModelEventBusTopicEntry( | ||
| topic=raw_topic, |
Collaborator
Author
There was a problem hiding this comment.
Bug: Whitespace-padded topics stored without stripping causes routing failures.
Root cause: Line 179 validates using not raw_topic.strip() to reject whitespace-only strings, but stores the unstripped raw_topic here.
Impact: A topic like " dev.topic.v1 " passes validation but gets stored with whitespace. Later lookups for "dev.topic.v1" fail because Consul KV stores the key as onex/topics/ dev.topic.v1 /subscribers.
Suggested change
| topic=raw_topic, | |
| topic=raw_topic.strip(), |
Same fix needed at line 224 for publish_topics.
| concurrency notes. Accepted MVP limitation. | ||
| """ | ||
| # NOTE: Non-atomic read-modify-write. See _update_topic_index for details. | ||
| key = f"onex/topics/{topic}/subscribers" |
Collaborator
Author
There was a problem hiding this comment.
Security Concern: Topic strings used in Consul KV paths without character validation.
Topics are interpolated directly into KV paths:
key = f"onex/topics/{topic}/subscribers"A topic containing slashes (e.g., foo/bar) or other special characters could:
- Create unexpected key hierarchy
- Potentially bypass ACL prefix matching rules
- Cause key parsing issues
Note: The codebase has robust topic validation in event_bus_kafka.py (lines 1420-1479) using regex ^[a-zA-Z0-9._-]+$, but this is NOT called in the Consul topic index flow.
Fix src/omnibase_infra/handlers/mixins/mixin_consul_topic_index.py: Add topic format validation before using in KV paths. Either reuse the Kafka validation pattern (regex ^[a-zA-Z0-9._-]+$) or add a @field_validator to ModelEventBusTopicEntry.topic to enforce this at the model level.
…use error context factory - Strip whitespace from topic strings before storage to prevent routing mismatches - Add topic format validation (alphanumeric, periods, underscores, hyphens only) to prevent Consul KV path traversal via slashes or special characters - Replace direct ModelInfraErrorContext() calls with .with_correlation() factory method per CLAUDE.md requirements (4 instances)
…or handling - Add _validate_topic_entry method to mixin_consul_service.py for consolidated topic validation (format, whitespace, message_category) - Replace plain ValueError with ProtocolConfigurationError in mixin_node_introspection.py for structured error reporting - Add correlation_id to all exception handler logs for traceability - Add fail-fast check when event_bus_config present without valid node_id - Update conftest.py docstring with comprehensive fixture documentation - Document operation order consistency in test_introspection_event_bus_e2e.py
…r messages
The topic validation was checking format (alphanumeric, dots, hyphens,
underscores) before checking for unresolved placeholders like {env}.
Since curly braces fail the format regex, users got a confusing "Invalid
topic suffix format" error instead of the more helpful "Unresolved
placeholder in topic" message.
Changes:
- Move placeholder detection BEFORE format validation in resolve_topic()
- Raise ValueError from _extract_event_bus_config for placeholders
- Wrap ValueError in ProtocolConfigurationError in get_introspection_data
with "Event bus extraction failed" message for consistent error handling
This restores the expected test behavior where placeholder errors are
caught early with a specific, actionable error message.
…ent-bus-topic-storage-to-registry-for-dynamic-topic
…lper After merging main (OMN-1602), RuntimeHostProcess now requires service_name and node_name in config. Update test fixtures to use the make_runtime_config() helper from tests/conftest.py.
…dule Extract duplicate topic validation regex to `constants_topic_patterns.py`: - Create new `TOPIC_NAME_PATTERN` constant in dependency-free module - Remove inline `import re` from `mixin_node_introspection.py` - Update all topic index mixins to use shared constant - Prevents circular imports and ensures consistent validation This addresses PR #199 code review feedback to consolidate duplicate regex patterns and improve code quality.
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/omnibase_infra/mixins/mixin_node_introspection.py (1)
1680-1686: Prevent UnboundLocalError in exception logging.If
get_introspection_data()raises beforefinal_correlation_idis assigned, Line 1684 will throwUnboundLocalError, masking the original error. Initialize the correlation id before thetry.🐛 Suggested fix
- async with self.track_operation("publish_introspection"): + final_correlation_id = correlation_id or uuid4() + async with self.track_operation("publish_introspection"): try: # Get introspection data event = await self.get_introspection_data() # Create publish event with updated reason and correlation_id # Use model_copy for clean field updates (Pydantic v2) - final_correlation_id = correlation_id or uuid4() publish_event = event.model_copy( update={ "reason": reason_enum, "correlation_id": final_correlation_id, } )
🤖 Fix all issues with AI agents
In `@src/omnibase_infra/handlers/mixins/mixin_consul_topic_index.py`:
- Around line 108-139: In _validate_topic_format, guard against non-string topic
inputs so TOPIC_NAME_PATTERN.match(topic) cannot raise a raw TypeError: add an
explicit type check (isinstance(topic, str)) at the top of the method and, if it
fails, construct the same ModelInfraErrorContext (use correlation_id,
EnumInfraTransportType.CONSUL, operation="validate_topic_format",
target_name="consul_handler") and raise ProtocolConfigurationError with
parameter="topic" and value=topic and a clear message that topic must be a
string; otherwise proceed to use TOPIC_NAME_PATTERN.match(topic) as before so
all failures surface as OnexError subclasses.
In `@src/omnibase_infra/mixins/mixin_node_introspection.py`:
- Around line 1044-1057: The helper resolve_topic currently raises a plain
ValueError on unresolved placeholders; change that to raise
ProtocolConfigurationError (an OnexError subclass) with a clear message and
context so callers don't need to catch/translate ValueError. Locate
resolve_topic in mixin_node_introspection.py and replace the ValueError raise
with ProtocolConfigurationError(...) including the offending suffix in the
message (and any relevant metadata if the ProtocolConfigurationError supports
it), ensuring imports/reference to ProtocolConfigurationError are used instead
of ValueError.
Comment on lines
+108
to
+139
| def _validate_topic_format(self, topic: str, correlation_id: UUID) -> None: | ||
| """Validate topic format for safe use in Consul KV paths. | ||
|
|
||
| Topics are interpolated into Consul KV paths like `onex/topics/{topic}/subscribers`. | ||
| Invalid characters (especially slashes) could create unexpected key hierarchies | ||
| or bypass ACL prefix matching. | ||
|
|
||
| Args: | ||
| topic: The topic string to validate. | ||
| correlation_id: Correlation ID for error context. | ||
|
|
||
| Raises: | ||
| ProtocolConfigurationError: If topic contains invalid characters. | ||
| Valid characters: alphanumeric (a-zA-Z0-9), periods (.), underscores (_), | ||
| and hyphens (-). | ||
| """ | ||
| if not TOPIC_NAME_PATTERN.match(topic): | ||
| context = ModelInfraErrorContext.with_correlation( | ||
| correlation_id=correlation_id, | ||
| transport_type=EnumInfraTransportType.CONSUL, | ||
| operation="validate_topic_format", | ||
| target_name="consul_handler", | ||
| ) | ||
| raise ProtocolConfigurationError( | ||
| f"Topic name '{topic}' contains invalid characters. " | ||
| "Only alphanumeric characters, periods (.), underscores (_), " | ||
| "and hyphens (-) are allowed. Slashes and other special " | ||
| "characters are prohibited to prevent Consul KV path traversal.", | ||
| context=context, | ||
| parameter="topic", | ||
| value=topic, | ||
| ) |
There was a problem hiding this comment.
Guard non-string topics to avoid raw TypeError.
If corrupted KV data yields a non-string topic, TOPIC_NAME_PATTERN.match() will raise TypeError and bypass the OnexError contract. A small type guard keeps failures typed and traceable. As per coding guidelines, only OnexError subclasses should surface from these paths.
🐛 Proposed fix
def _validate_topic_format(self, topic: str, correlation_id: UUID) -> None:
"""Validate topic format for safe use in Consul KV paths.
@@
- if not TOPIC_NAME_PATTERN.match(topic):
+ if not isinstance(topic, str):
+ context = ModelInfraErrorContext.with_correlation(
+ correlation_id=correlation_id,
+ transport_type=EnumInfraTransportType.CONSUL,
+ operation="validate_topic_format",
+ target_name="consul_handler",
+ )
+ raise ProtocolConfigurationError(
+ f"Topic must be a string, got {type(topic).__name__}",
+ context=context,
+ parameter="topic",
+ value=str(topic),
+ )
+ if not TOPIC_NAME_PATTERN.match(topic):
context = ModelInfraErrorContext.with_correlation(
correlation_id=correlation_id,
transport_type=EnumInfraTransportType.CONSUL,
operation="validate_topic_format",
target_name="consul_handler",
)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| def _validate_topic_format(self, topic: str, correlation_id: UUID) -> None: | |
| """Validate topic format for safe use in Consul KV paths. | |
| Topics are interpolated into Consul KV paths like `onex/topics/{topic}/subscribers`. | |
| Invalid characters (especially slashes) could create unexpected key hierarchies | |
| or bypass ACL prefix matching. | |
| Args: | |
| topic: The topic string to validate. | |
| correlation_id: Correlation ID for error context. | |
| Raises: | |
| ProtocolConfigurationError: If topic contains invalid characters. | |
| Valid characters: alphanumeric (a-zA-Z0-9), periods (.), underscores (_), | |
| and hyphens (-). | |
| """ | |
| if not TOPIC_NAME_PATTERN.match(topic): | |
| context = ModelInfraErrorContext.with_correlation( | |
| correlation_id=correlation_id, | |
| transport_type=EnumInfraTransportType.CONSUL, | |
| operation="validate_topic_format", | |
| target_name="consul_handler", | |
| ) | |
| raise ProtocolConfigurationError( | |
| f"Topic name '{topic}' contains invalid characters. " | |
| "Only alphanumeric characters, periods (.), underscores (_), " | |
| "and hyphens (-) are allowed. Slashes and other special " | |
| "characters are prohibited to prevent Consul KV path traversal.", | |
| context=context, | |
| parameter="topic", | |
| value=topic, | |
| ) | |
| def _validate_topic_format(self, topic: str, correlation_id: UUID) -> None: | |
| """Validate topic format for safe use in Consul KV paths. | |
| Topics are interpolated into Consul KV paths like `onex/topics/{topic}/subscribers`. | |
| Invalid characters (especially slashes) could create unexpected key hierarchies | |
| or bypass ACL prefix matching. | |
| Args: | |
| topic: The topic string to validate. | |
| correlation_id: Correlation ID for error context. | |
| Raises: | |
| ProtocolConfigurationError: If topic contains invalid characters. | |
| Valid characters: alphanumeric (a-zA-Z0-9), periods (.), underscores (_), | |
| and hyphens (-). | |
| """ | |
| try: | |
| if not TOPIC_NAME_PATTERN.match(topic): | |
| context = ModelInfraErrorContext.with_correlation( | |
| correlation_id=correlation_id, | |
| transport_type=EnumInfraTransportType.CONSUL, | |
| operation="validate_topic_format", | |
| target_name="consul_handler", | |
| ) | |
| raise ProtocolConfigurationError( | |
| f"Topic name '{topic}' contains invalid characters. " | |
| "Only alphanumeric characters, periods (.), underscores (_), " | |
| "and hyphens (-) are allowed. Slashes and other special " | |
| "characters are prohibited to prevent Consul KV path traversal.", | |
| context=context, | |
| parameter="topic", | |
| value=topic, | |
| ) | |
| except TypeError as e: | |
| context = ModelInfraErrorContext.with_correlation( | |
| correlation_id=correlation_id, | |
| transport_type=EnumInfraTransportType.CONSUL, | |
| operation="validate_topic_format", | |
| target_name="consul_handler", | |
| ) | |
| raise ProtocolConfigurationError( | |
| f"Topic must be a string, got {type(topic).__name__}", | |
| context=context, | |
| parameter="topic", | |
| value=str(topic), | |
| ) from e |
🤖 Prompt for AI Agents
In `@src/omnibase_infra/handlers/mixins/mixin_consul_topic_index.py` around lines
108 - 139, In _validate_topic_format, guard against non-string topic inputs so
TOPIC_NAME_PATTERN.match(topic) cannot raise a raw TypeError: add an explicit
type check (isinstance(topic, str)) at the top of the method and, if it fails,
construct the same ModelInfraErrorContext (use correlation_id,
EnumInfraTransportType.CONSUL, operation="validate_topic_format",
target_name="consul_handler") and raise ProtocolConfigurationError with
parameter="topic" and value=topic and a clear message that topic must be a
string; otherwise proceed to use TOPIC_NAME_PATTERN.match(topic) as before so
all failures surface as OnexError subclasses.
Comment on lines
+1044
to
+1057
| def resolve_topic(suffix: str) -> str: | ||
| """Resolve topic suffix to full topic with env prefix.""" | ||
| # Full topic format: {env}.{suffix} | ||
| # Strip whitespace from suffix to handle YAML formatting artifacts | ||
| suffix = suffix.strip() | ||
|
|
||
| # Fail-fast: check for unresolved placeholders BEFORE format validation | ||
| # This provides more helpful error messages when placeholders aren't resolved | ||
| if "{" in suffix or "}" in suffix: | ||
| raise ValueError( | ||
| f"Unresolved placeholder in topic: '{suffix}'. " | ||
| "Ensure all placeholders like {env} or {namespace} are resolved " | ||
| "before topic validation." | ||
| ) |
There was a problem hiding this comment.
Use ProtocolConfigurationError for unresolved placeholders.
Line 1052 raises a plain ValueError, which can leak outside this helper and violates the “OnexError-only” rule. Raise ProtocolConfigurationError with context instead (and the caller no longer needs to wrap ValueError). As per coding guidelines, only OnexError subclasses should be raised.
♻️ Suggested fix
- if "{" in suffix or "}" in suffix:
- raise ValueError(
- f"Unresolved placeholder in topic: '{suffix}'. "
- "Ensure all placeholders like {env} or {namespace} are resolved "
- "before topic validation."
- )
+ if "{" in suffix or "}" in suffix:
+ context = ModelInfraErrorContext.with_correlation(
+ transport_type=EnumInfraTransportType.RUNTIME,
+ operation="_extract_event_bus_config",
+ target_name=suffix,
+ )
+ raise ProtocolConfigurationError(
+ f"Unresolved placeholder in topic: '{suffix}'. "
+ "Ensure all placeholders like {env} or {namespace} are resolved "
+ "before topic validation.",
+ context=context,
+ parameter="topic_suffix",
+ )📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| def resolve_topic(suffix: str) -> str: | |
| """Resolve topic suffix to full topic with env prefix.""" | |
| # Full topic format: {env}.{suffix} | |
| # Strip whitespace from suffix to handle YAML formatting artifacts | |
| suffix = suffix.strip() | |
| # Fail-fast: check for unresolved placeholders BEFORE format validation | |
| # This provides more helpful error messages when placeholders aren't resolved | |
| if "{" in suffix or "}" in suffix: | |
| raise ValueError( | |
| f"Unresolved placeholder in topic: '{suffix}'. " | |
| "Ensure all placeholders like {env} or {namespace} are resolved " | |
| "before topic validation." | |
| ) | |
| def resolve_topic(suffix: str) -> str: | |
| """Resolve topic suffix to full topic with env prefix.""" | |
| # Full topic format: {env}.{suffix} | |
| # Strip whitespace from suffix to handle YAML formatting artifacts | |
| suffix = suffix.strip() | |
| # Fail-fast: check for unresolved placeholders BEFORE format validation | |
| # This provides more helpful error messages when placeholders aren't resolved | |
| if "{" in suffix or "}" in suffix: | |
| context = ModelInfraErrorContext.with_correlation( | |
| transport_type=EnumInfraTransportType.RUNTIME, | |
| operation="_extract_event_bus_config", | |
| target_name=suffix, | |
| ) | |
| raise ProtocolConfigurationError( | |
| f"Unresolved placeholder in topic: '{suffix}'. " | |
| "Ensure all placeholders like {env} or {namespace} are resolved " | |
| "before topic validation.", | |
| context=context, | |
| parameter="topic_suffix", | |
| ) |
🤖 Prompt for AI Agents
In `@src/omnibase_infra/mixins/mixin_node_introspection.py` around lines 1044 -
1057, The helper resolve_topic currently raises a plain ValueError on unresolved
placeholders; change that to raise ProtocolConfigurationError (an OnexError
subclass) with a clear message and context so callers don't need to
catch/translate ValueError. Locate resolve_topic in mixin_node_introspection.py
and replace the ValueError raise with ProtocolConfigurationError(...) including
the offending suffix in the message (and any relevant metadata if the
ProtocolConfigurationError supports it), ensuring imports/reference to
ProtocolConfigurationError are used instead of ValueError.
7 tasks done
jonahgabriel
deleted the
jonah/omn-1613-add-event-bus-topic-storage-to-registry-for-dynamic-topic
branch
April 4, 2026 02:06
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Enable registry-driven topic routing by storing event_bus configuration (subscribe_topics, publish_topics) as part of node introspection data in Consul KV.
event_busfield toModelNodeIntrospectionEventcontaining resolved, environment-qualified topic stringsonex/topics/{topic}/subscribers) for topic → node_id lookupget_subscribers_for_topic()toRuntimeHostProcessfor dynamic routing queriesKey Changes
Models
ModelEventBusTopicEntry- Single topic entry with optional tooling metadataModelNodeEventBusConfig- Container for subscribe/publish topic lists withsubscribe_topic_strings/publish_topic_stringspropertiesIntrospection
_extract_event_bus_config()inMixinNodeIntrospection- Extracts topics from contract and resolves env prefix{env},{namespace})Consul Storage
MixinConsulTopicIndex- New mixin for topic index managementRuntime
get_subscribers_for_topic(topic: str) -> list[UUID]- Query Consul for nodes subscribing to a topicDesign Decisions (LOCKED)
dev.onex.evt.intent-classified.v1)Test Plan
event_busonex/nodes/{node_id}/event_bus/,onex/topics/{topic}/subscribers)get_subscribers_for_topic()returns correct node UUIDsRelated
Summary by CodeRabbit
New Features
Tests
✏️ Tip: You can customize this high-level summary in your review settings.