Skip to content
9 changes: 9 additions & 0 deletions src/omnibase_core/constants/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -141,9 +141,13 @@
TOPIC_RUNTIME_EVENTS,
TOPIC_RUNTIME_INTENTS,
TOPIC_TYPE_COMMANDS,
TOPIC_TYPE_DLQ,
TOPIC_TYPE_EVENTS,
TOPIC_TYPE_INTENTS,
TOPIC_TYPE_SNAPSHOTS,
get_token_to_topic_type,
get_topic_type_to_token,
get_valid_topic_suffix_kinds,
topic_name,
)

Expand Down Expand Up @@ -250,9 +254,14 @@
"validate_capabilities",
# Topic taxonomy constants
"TOPIC_TYPE_COMMANDS",
"TOPIC_TYPE_DLQ",
"TOPIC_TYPE_EVENTS",
"TOPIC_TYPE_INTENTS",
"TOPIC_TYPE_SNAPSHOTS",
# Token mapping accessors (OMN-1537)
"get_token_to_topic_type",
"get_topic_type_to_token",
"get_valid_topic_suffix_kinds",
"DOMAIN_REGISTRATION",
"DOMAIN_DISCOVERY",
"DOMAIN_RUNTIME",
Expand Down
122 changes: 120 additions & 2 deletions src/omnibase_core/constants/constants_topic_taxonomy.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,25 +11,138 @@
- events: Immutable event logs
- intents: Coordination intents
- snapshots: Compacted state snapshots
- dlq: Dead letter queue (failed messages)

Thread Safety:
All constants in this module are immutable strings and thread-safe.
The topic_name() function is pure and thread-safe.

Cross-reference:
For topic kind tokens used in suffix validation (TOPIC_KIND_CMD, etc.),
see omnibase_core.models.validation.model_topic_suffix_parts.
This module defines TOPIC_TYPE_* semantic values (e.g., "commands"),
while model_topic_suffix_parts defines TOPIC_KIND_* tokens (e.g., "cmd").
"""

from __future__ import annotations

import functools
import re
from typing import TYPE_CHECKING

if TYPE_CHECKING:
from omnibase_core.enums.enum_topic_taxonomy import EnumTopicType

# Topic Type Suffixes
# Topic Type Suffixes (semantic values used internally)
TOPIC_TYPE_COMMANDS = "commands"
TOPIC_TYPE_DLQ = "dlq"
TOPIC_TYPE_EVENTS = "events"
TOPIC_TYPE_INTENTS = "intents"
TOPIC_TYPE_SNAPSHOTS = "snapshots"

# Valid topic types set for validation
_VALID_TOPIC_TYPES = frozenset(
{TOPIC_TYPE_COMMANDS, TOPIC_TYPE_EVENTS, TOPIC_TYPE_INTENTS, TOPIC_TYPE_SNAPSHOTS}
{
TOPIC_TYPE_COMMANDS,
TOPIC_TYPE_DLQ,
TOPIC_TYPE_EVENTS,
TOPIC_TYPE_INTENTS,
TOPIC_TYPE_SNAPSHOTS,
}
)


# Token-to-Type Mappings
# Maps short tokens (cmd, evt, dlq, intent, snapshot) used in topic suffixes
# to their corresponding EnumTopicType semantic values (commands, events, etc.)
def _build_token_mappings() -> tuple[
dict[str, EnumTopicType], dict[EnumTopicType, str], frozenset[str]
]:
"""
Build token-to-enum and enum-to-token mappings.

Deferred import to avoid circular dependency at module load time.
Called lazily when mappings are first accessed.

Returns:
Tuple of (token_to_type, type_to_token, valid_suffix_kinds).
"""
from omnibase_core.enums.enum_topic_taxonomy import EnumTopicType

token_to_type: dict[str, EnumTopicType] = {
"cmd": EnumTopicType.COMMANDS,
"dlq": EnumTopicType.DLQ,
"evt": EnumTopicType.EVENTS,
"intent": EnumTopicType.INTENTS,
"snapshot": EnumTopicType.SNAPSHOTS,
}
type_to_token: dict[EnumTopicType, str] = {v: k for k, v in token_to_type.items()}
valid_kinds: frozenset[str] = frozenset(token_to_type.keys())
Comment on lines +72 to +80

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical

🧩 Analysis chain

🏁 Script executed:

# First, find and examine the constants file mentioned
cat -n src/omnibase_core/constants/constants_topic_taxonomy.py | head -100

Repository: OmniNode-ai/omnibase_core

Length of output: 3705


🏁 Script executed:

# Search for validator constants and rules
rg -n "TOPIC_KIND_|VALID_TOPIC_KINDS" src/omnibase_core --type py

Repository: OmniNode-ai/omnibase_core

Length of output: 2183


🏁 Script executed:

# Search for validator suffix rules and kind validation
rg -n "kind not in|cmd.*evt.*dlq|snapshot" src/omnibase_core/validation/ --type py -A 3

Repository: OmniNode-ai/omnibase_core

Length of output: 2336


🏁 Script executed:

# Find ModelTopicSuffixParts definition
rg -n "class ModelTopicSuffixParts|EnumTopicType" src/omnibase_core/models/ --type py -B 2 -A 8

Repository: OmniNode-ai/omnibase_core

Length of output: 21823


Remove the "intent" token from token_to_type mapping.

The token_to_type dictionary includes "intent", but the validator and ModelTopicSuffixParts only allow cmd, evt, dlq, and snapshot as valid topic suffix tokens. The validator regex explicitly restricts kind to (cmd|evt|dlq|snapshot), and the centralized VALID_TOPIC_KINDS in model_topic_suffix_parts.py omits "intent". This mismatch means get_valid_topic_suffix_kinds() will incorrectly include "intent" even though it fails validation. Remove "intent" from the token map to align with the validator's canonical definition.

Proposed fix
    token_to_type: dict[str, EnumTopicType] = {
        "cmd": EnumTopicType.COMMANDS,
        "dlq": EnumTopicType.DLQ,
        "evt": EnumTopicType.EVENTS,
-       "intent": EnumTopicType.INTENTS,
        "snapshot": EnumTopicType.SNAPSHOTS,
    }
🤖 Prompt for AI Agents
In `@src/omnibase_core/constants/constants_topic_taxonomy.py` around lines 62 -
70, Remove the inconsistent "intent" entry from the token_to_type mapping so the
canonical topic-kind set matches the validator and ModelTopicSuffixParts; update
token_to_type (the dict mapping token_to_type) to only include "cmd", "dlq",
"evt", and "snapshot", regenerate type_to_token (which is built from
token_to_type) and valid_kinds so they reflect the same four tokens, ensuring
EnumTopicType values referenced remain consistent and
get_valid_topic_suffix_kinds() no longer exposes "intent".

return token_to_type, type_to_token, valid_kinds


@functools.lru_cache(maxsize=1)
def _get_token_mappings() -> tuple[
dict[str, EnumTopicType], dict[EnumTopicType, str], frozenset[str]
]:
"""Get or initialize the token mappings (cached via lru_cache)."""
return _build_token_mappings()


def get_token_to_topic_type() -> dict[str, EnumTopicType]:
"""
Get mapping from topic suffix tokens to EnumTopicType values.

Topic suffix tokens are the abbreviated forms used in topic names
(e.g., 'cmd', 'evt', 'dlq'), while EnumTopicType values are the
semantic internal representations (e.g., 'commands', 'events', 'dlq').

Returns:
Dict mapping tokens to enum values.

Example:
>>> mapping = get_token_to_topic_type()
>>> mapping["evt"]
<EnumTopicType.EVENTS: 'events'>
"""
# Return defensive copy to prevent mutation of cached mapping
return dict(_get_token_mappings()[0])


def get_topic_type_to_token() -> dict[EnumTopicType, str]:
"""
Get reverse mapping from EnumTopicType values to tokens.

Used for composing topic suffixes from enum values.

Returns:
Dict mapping enum values to tokens.

Example:
>>> from omnibase_core.enums import EnumTopicType
>>> mapping = get_topic_type_to_token()
>>> mapping[EnumTopicType.COMMANDS]
'cmd'
"""
# Return defensive copy to prevent mutation of cached mapping
return dict(_get_token_mappings()[1])


def get_valid_topic_suffix_kinds() -> frozenset[str]:
"""
Get the set of valid topic suffix kind tokens.

Returns:
Frozenset of valid tokens ('cmd', 'evt', 'dlq', 'intent', 'snapshot').

Example:
>>> kinds = get_valid_topic_suffix_kinds()
>>> "evt" in kinds
True
"""
return _get_token_mappings()[2]


# Domain validation pattern: lowercase alphanumeric with hyphens, starting with letter,
# cannot end with hyphen (must end with letter or digit, or be single letter)
_DOMAIN_PATTERN = re.compile(r"^[a-z][a-z0-9-]*[a-z0-9]$|^[a-z]$")
Expand Down Expand Up @@ -120,9 +233,14 @@ def topic_name(domain: str, topic_type: str) -> str:
__all__ = [
# Type suffixes
"TOPIC_TYPE_COMMANDS",
"TOPIC_TYPE_DLQ",
"TOPIC_TYPE_EVENTS",
"TOPIC_TYPE_INTENTS",
"TOPIC_TYPE_SNAPSHOTS",
# Token mapping accessors (lazy-loaded to avoid circular imports)
"get_token_to_topic_type",
"get_topic_type_to_token",
"get_valid_topic_suffix_kinds",
# Domains
"DOMAIN_REGISTRATION",
"DOMAIN_DISCOVERY",
Expand Down
4 changes: 3 additions & 1 deletion src/omnibase_core/enums/enum_topic_taxonomy.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,14 +19,16 @@ class EnumTopicType(StrValueHelper, str, Enum):
"""
Valid topic types per ONEX topic taxonomy.

Defines the four standard topic categories used across all ONEX domains:
Defines the standard topic categories used across all ONEX domains:
- COMMANDS: Request/action topics (imperative, exactly-once semantics)
- EVENTS: Immutable event logs (append-only, time-based retention)
- INTENTS: Intent coordination topics (workflow orchestration)
- SNAPSHOTS: State snapshots (compacted, key-based retention)
- DLQ: Dead letter queue topics (failed message storage)
"""

COMMANDS = "commands"
DLQ = "dlq"
EVENTS = "events"
INTENTS = "intents"
SNAPSHOTS = "snapshots"
Expand Down
2 changes: 2 additions & 0 deletions src/omnibase_core/models/contracts/subcontracts/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,7 @@
from .model_statistical_computation import ModelStatisticalComputation
from .model_synchronization_point import ModelSynchronizationPoint
from .model_tool_execution_subcontract import ModelToolExecutionSubcontract
from .model_topic_meta import ModelTopicMeta
from .model_validation_subcontract import ModelValidationSubcontract
from .model_validator_rule import ModelValidatorRule
from .model_validator_subcontract import ModelValidatorSubcontract
Expand Down Expand Up @@ -202,6 +203,7 @@
"ModelEventPersistence",
"ModelEventRouting",
"ModelEventTransformation",
"ModelTopicMeta",
# FSM subcontracts and components
"ModelFSMSubcontract",
"ModelFSMOperation",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,12 +18,13 @@

from typing import ClassVar

from pydantic import BaseModel, ConfigDict, Field, model_validator
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator

from omnibase_core.constants import KAFKA_REQUEST_TIMEOUT_MS
from omnibase_core.enums.enum_core_error_code import EnumCoreErrorCode
from omnibase_core.models.common.model_error_context import ModelErrorContext
from omnibase_core.models.common.model_schema_value import ModelSchemaValue
from omnibase_core.models.contracts.subcontracts.model_topic_meta import ModelTopicMeta
from omnibase_core.models.errors.model_onex_error import ModelOnexError
from omnibase_core.models.primitives.model_semver import ModelSemVer

Expand Down Expand Up @@ -199,6 +200,41 @@ class ModelEventBusSubcontract(BaseModel):
description="Default event patterns if no other patterns can be determined",
)

# Topic-based routing configuration (OMN-1537)
publish_topics: list[str] = Field(
default_factory=list,
description="Topic suffixes this node publishes to. Format: onex.{kind}.{producer}.{event-name}.v{n}",
)

subscribe_topics: list[str] = Field(
default_factory=list,
description="Topic suffixes this node subscribes to. Format: onex.{kind}.{producer}.{event-name}.v{n}",
)

# Extension path for future schema_ref support
publish_topic_metadata: dict[str, ModelTopicMeta] | None = Field(
default=None,
description="Optional metadata per publish topic suffix (keyed by suffix string)",
)

subscribe_topic_metadata: dict[str, ModelTopicMeta] | None = Field(
default=None,
description="Optional metadata per subscribe topic suffix (keyed by suffix string)",
)

@field_validator("publish_topics", "subscribe_topics", mode="after")
@classmethod
def validate_topic_suffixes(cls, topics: list[str]) -> list[str]:
"""Validate each topic suffix against ONEX naming convention."""
# Import here to avoid circular import at module load time
from omnibase_core.validation import validate_topic_suffix

for topic in topics:
result = validate_topic_suffix(topic)
if not result.is_valid:
raise ValueError(f"Invalid topic suffix '{topic}': {result.error}")
return topics
Comment on lines +225 to +236

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Use ModelOnexError for topic suffix validation failures.
This is ONEX-specific contract validation; using ValueError makes error handling inconsistent with the rest of this model. Consider raising ModelOnexError with EnumCoreErrorCode.VALIDATION_ERROR and context. As per coding guidelines, ONEX-specific validation errors should use ModelOnexError.

Example adjustment
         for topic in topics:
             result = validate_topic_suffix(topic)
             if not result.is_valid:
-                raise ValueError(f"Invalid topic suffix '{topic}': {result.error}")
+                raise ModelOnexError(
+                    message=f"Invalid topic suffix '{topic}': {result.error}",
+                    error_code=EnumCoreErrorCode.VALIDATION_ERROR,
+                    details=ModelErrorContext.with_context(
+                        {
+                            "error_type": ModelSchemaValue.from_value("valueerror"),
+                            "validation_context": ModelSchemaValue.from_value(
+                                "topic_suffix_validation"
+                            ),
+                            "topic": ModelSchemaValue.from_value(topic),
+                        },
+                    ),
+                )
         return topics
🤖 Prompt for AI Agents
In
`@src/omnibase_core/models/contracts/subcontracts/model_event_bus_subcontract.py`
around lines 225 - 236, Replace the ValueError raise in the
validate_topic_suffixes class validator with a ModelOnexError using
EnumCoreErrorCode.VALIDATION_ERROR: import ModelOnexError and EnumCoreErrorCode
at top of the function (or module) and when validate_topic_suffix(topic) returns
not is_valid raise ModelOnexError(EnumCoreErrorCode.VALIDATION_ERROR,
context={"field":"publish_topics/subscribe_topics","invalid_value":topic,"detail":result.error})
so ONEX-specific consumers can handle the validation failure consistently; keep
the loop and return topics as-is.


@model_validator(mode="after")
def validate_event_bus_configuration(self) -> "ModelEventBusSubcontract":
"""Validate event bus configuration fields after model construction."""
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,37 @@
"""
Topic Metadata Model.

Model for topic subscription/publication metadata in ONEX event bus contracts.
Provides extension point for schema_ref and description per topic.
"""

from pydantic import BaseModel, ConfigDict, Field

__all__ = ["ModelTopicMeta"]


class ModelTopicMeta(BaseModel):
"""
Metadata for a topic subscription/publication.

Provides optional metadata for topics declared in publish_topics
and subscribe_topics fields, keyed by the topic suffix string.
This is an extension point for future schema validation and
documentation generation.
"""

schema_ref: str | None = Field(
default=None,
description="Reference to the Pydantic model for this topic's payload",
)

description: str | None = Field(
default=None,
description="Human-readable description of this topic's purpose",
)

model_config = ConfigDict(
frozen=True,
extra="forbid",
from_attributes=True,
)
19 changes: 19 additions & 0 deletions src/omnibase_core/models/events/model_topic_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -139,6 +139,25 @@ def snapshots_default(cls) -> "ModelTopicConfig":
replication_factor=1,
)

@classmethod
def dlq_default(cls) -> "ModelTopicConfig":
"""
Default configuration for dead letter queue (DLQ) topics.

DLQ topics store failed messages for investigation and retry.
Default retention is 30 days per ONEX standard (same as events for audit).

Note:
This model is immutable (frozen) and thread-safe after instantiation.
"""
return cls(
topic_type=EnumTopicType.DLQ,
cleanup_policy=EnumCleanupPolicy.DELETE,
retention_ms=2592000000, # 30 days per ONEX standard
partitions=3,
replication_factor=1,
)


__all__ = [
"ModelTopicConfig",
Expand Down
4 changes: 3 additions & 1 deletion src/omnibase_core/models/events/model_topic_manifest.py
Original file line number Diff line number Diff line change
Expand Up @@ -117,6 +117,7 @@ def registration_domain(cls) -> "ModelTopicManifest":
domain="registration",
topics={
EnumTopicType.COMMANDS: ModelTopicConfig.commands_default(),
EnumTopicType.DLQ: ModelTopicConfig.dlq_default(),
EnumTopicType.EVENTS: ModelTopicConfig.events_default(),
EnumTopicType.INTENTS: ModelTopicConfig.intents_default(),
EnumTopicType.SNAPSHOTS: ModelTopicConfig.snapshots_default(),
Expand All @@ -128,7 +129,7 @@ def create_standard_manifest(cls, domain: str) -> "ModelTopicManifest":
"""
Create a standard topic manifest for any domain.

Creates all four topic types with default configurations.
Creates all five topic types with default configurations.

Args:
domain: Domain name (lowercase, alphanumeric with hyphens)
Expand All @@ -140,6 +141,7 @@ def create_standard_manifest(cls, domain: str) -> "ModelTopicManifest":
domain=domain,
topics={
EnumTopicType.COMMANDS: ModelTopicConfig.commands_default(),
EnumTopicType.DLQ: ModelTopicConfig.dlq_default(),
EnumTopicType.EVENTS: ModelTopicConfig.events_default(),
EnumTopicType.INTENTS: ModelTopicConfig.intents_default(),
EnumTopicType.SNAPSHOTS: ModelTopicConfig.snapshots_default(),
Expand Down
21 changes: 21 additions & 0 deletions src/omnibase_core/models/validation/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,18 @@
from .model_lint_statistics import ModelLintStatistics
from .model_migration_conflict_union import ModelMigrationConflictUnion
from .model_shape_validation_result import ModelShapeValidationResult

# Topic suffix validation models (OMN-1537)
from .model_topic_suffix_parts import (
TOPIC_KIND_CMD,
TOPIC_KIND_DLQ,
TOPIC_KIND_EVT,
TOPIC_KIND_INTENT,
TOPIC_KIND_SNAPSHOT,
VALID_TOPIC_KINDS,
ModelTopicSuffixParts,
)
from .model_topic_validation_result import ModelTopicValidationResult
from .model_unique_name_result import ModelUniqueNameResult
from .model_validation_base import ModelValidationBase
from .model_validation_container import ModelValidationContainer
Expand Down Expand Up @@ -54,6 +66,15 @@
"ModelExecutionShape",
"ModelExecutionShapeValidation",
"ModelShapeValidationResult",
# Topic suffix validation models (OMN-1537)
"ModelTopicSuffixParts",
"ModelTopicValidationResult",
"TOPIC_KIND_CMD",
"TOPIC_KIND_DLQ",
"TOPIC_KIND_EVT",
"TOPIC_KIND_INTENT",
"TOPIC_KIND_SNAPSHOT",
"VALID_TOPIC_KINDS",
# Workflow validation models (OMN-176)
"ModelCycleDetectionResult",
"ModelDependencyValidationResult",
Expand Down
Loading