From 83e1bd0f3491ac3070a8c2778c7a52de58a11aab Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Thu, 19 Mar 2026 09:29:47 -0400 Subject: [PATCH 1/4] feat(kafka): wire session_timeout_ms and heartbeat_interval_ms to all consumers [OMN-5445] Eliminate implicit reliance on aiokafka's aggressive 10s default session timeout across all AIOKafkaConsumer instances in the platform. The new platform defaults (30s session / 10s heartbeat) prevent rebalance storms caused by UnknownMemberIdError during transient processing delays. Three-surface fix: - ModelKafkaEventBusConfig: new fields with env var overrides (KAFKA_SESSION_TIMEOUT_MS, KAFKA_HEARTBEAT_INTERVAL_MS) and advisory heartbeat/session ratio validator - EventBusKafka: wire session/heartbeat/reconnect kwargs into consumer constructor; wire reconnect backoff kwargs into both producer constructors (previously configured but never passed) - 6 standalone consumers: add fields to BaseSettings configs and wire into AIOKafkaConsumer constructors - 3 ephemeral consumers: hardcode 30000/10000 defaults (no config surface) Includes 50 new unit tests covering config defaults, env overrides, field constraints, advisory validator, and constructor kwarg wiring. --- .../event_bus/event_bus_kafka.py | 8 + .../config/model_kafka_event_bus_config.py | 51 ++++ .../snapshot_publisher_registration.py | 2 + .../runtime/request_response_wiring.py | 2 + .../observability/agent_actions/config.py | 20 ++ .../observability/agent_actions/consumer.py | 2 + .../observability/context_audit/config.py | 20 ++ .../observability/context_audit/consumer.py | 2 + .../injection_effectiveness/config.py | 20 ++ .../injection_effectiveness/consumer.py | 2 + .../llm_cost_aggregation/config.py | 20 ++ .../llm_cost_aggregation/consumer.py | 2 + .../observability/skill_lifecycle/config.py | 20 ++ .../observability/skill_lifecycle/consumer.py | 2 + .../services/session/config_consumer.py | 21 ++ .../services/session/consumer.py | 2 + .../tui/consumers/consumer_status.py | 2 + .../validation/validation_exemptions.yaml | 17 ++ .../test_kafka_config_session_timeout.py | 186 +++++++++++++++ .../event_bus/test_kafka_timeout_kwargs.py | 222 ++++++++++++++++++ .../services/test_consumer_session_timeout.py | 96 ++++++++ 21 files changed, 719 insertions(+) create mode 100644 tests/unit/event_bus/test_kafka_config_session_timeout.py create mode 100644 tests/unit/event_bus/test_kafka_timeout_kwargs.py create mode 100644 tests/unit/services/test_consumer_session_timeout.py diff --git a/src/omnibase_infra/event_bus/event_bus_kafka.py b/src/omnibase_infra/event_bus/event_bus_kafka.py index 3481ec5582..f84c5075cf 100644 --- a/src/omnibase_infra/event_bus/event_bus_kafka.py +++ b/src/omnibase_infra/event_bus/event_bus_kafka.py @@ -639,6 +639,8 @@ async def start(self) -> None: bootstrap_servers=self._bootstrap_servers, acks=self._config.acks_aiokafka, enable_idempotence=self._config.enable_idempotence, + reconnect_backoff_ms=self._config.reconnect_backoff_ms, + reconnect_backoff_max_ms=self._config.reconnect_backoff_max_ms, **self._build_auth_kwargs(), ) @@ -955,6 +957,8 @@ async def _ensure_producer(self, correlation_id: UUID) -> None: bootstrap_servers=self._bootstrap_servers, acks=self._config.acks_aiokafka, enable_idempotence=self._config.enable_idempotence, + reconnect_backoff_ms=self._config.reconnect_backoff_ms, + reconnect_backoff_max_ms=self._config.reconnect_backoff_max_ms, **self._build_auth_kwargs(), ) @@ -1555,6 +1559,10 @@ async def _start_consumer_for_topic(self, topic: str, group_id: str) -> None: group_id=effective_group_id, auto_offset_reset=self._config.auto_offset_reset, enable_auto_commit=self._config.enable_auto_commit, + session_timeout_ms=self._config.session_timeout_ms, + heartbeat_interval_ms=self._config.heartbeat_interval_ms, + reconnect_backoff_ms=self._config.reconnect_backoff_ms, + reconnect_backoff_max_ms=self._config.reconnect_backoff_max_ms, **self._build_auth_kwargs(), ) diff --git a/src/omnibase_infra/event_bus/models/config/model_kafka_event_bus_config.py b/src/omnibase_infra/event_bus/models/config/model_kafka_event_bus_config.py index 613f3f0615..0f0436601a 100644 --- a/src/omnibase_infra/event_bus/models/config/model_kafka_event_bus_config.py +++ b/src/omnibase_infra/event_bus/models/config/model_kafka_event_bus_config.py @@ -367,6 +367,29 @@ class ModelKafkaEventBusConfig(BaseModel): ge=0, ) + # Session timeout configuration (OMN-5445) + session_timeout_ms: int = Field( + default=30000, + description=( + "Session timeout in milliseconds. Consumer is removed from the group " + "if no heartbeat is received within this interval. Override via " + "KAFKA_SESSION_TIMEOUT_MS. Default raised from aiokafka's 10s to 30s " + "to prevent rebalance storms." + ), + ge=6000, + le=300000, + ) + heartbeat_interval_ms: int = Field( + default=10000, + description=( + "Heartbeat interval in milliseconds. Kafka recommends <= session_timeout_ms / 3. " + "Override via KAFKA_HEARTBEAT_INTERVAL_MS. Override users are responsible " + "for coherent combinations with session_timeout_ms." + ), + ge=1000, + le=100000, + ) + @model_validator(mode="after") def validate_reconnect_backoff(self) -> ModelKafkaEventBusConfig: """Validate that reconnect_backoff_max_ms >= reconnect_backoff_ms. @@ -392,6 +415,28 @@ def validate_reconnect_backoff(self) -> ModelKafkaEventBusConfig: ) return self + @model_validator(mode="after") + def validate_heartbeat_session_ratio(self) -> ModelKafkaEventBusConfig: + """Advisory: warn if heartbeat_interval_ms > session_timeout_ms / 3. + + Kafka recommends heartbeat <= session_timeout / 3 to allow enough + missed heartbeats before the session expires. This is advisory only + (logs a warning) — it does not raise. + + Returns: + Self after validation + """ + if self.heartbeat_interval_ms > self.session_timeout_ms / 3: + logger.warning( + "heartbeat_interval_ms (%d) exceeds session_timeout_ms / 3 (%d). " + "Kafka recommends heartbeat_interval_ms <= session_timeout_ms / 3 " + "to prevent unnecessary rebalances. Current session_timeout_ms=%d.", + self.heartbeat_interval_ms, + self.session_timeout_ms // 3, + self.session_timeout_ms, + ) + return self + @model_validator(mode="after") def validate_auth_config(self) -> ModelKafkaEventBusConfig: """Validate authentication configuration consistency. @@ -690,6 +735,8 @@ def apply_environment_overrides(self) -> ModelKafkaEventBusConfig: - KAFKA_SSL_CA_FILE -> ssl_ca_file - KAFKA_RECONNECT_BACKOFF_MS -> reconnect_backoff_ms - KAFKA_RECONNECT_BACKOFF_MAX_MS -> reconnect_backoff_max_ms + - KAFKA_SESSION_TIMEOUT_MS -> session_timeout_ms + - KAFKA_HEARTBEAT_INTERVAL_MS -> heartbeat_interval_ms Returns: New configuration instance with environment overrides applied @@ -719,6 +766,8 @@ def apply_environment_overrides(self) -> ModelKafkaEventBusConfig: "KAFKA_SSL_CA_FILE": "ssl_ca_file", "KAFKA_RECONNECT_BACKOFF_MS": "reconnect_backoff_ms", "KAFKA_RECONNECT_BACKOFF_MAX_MS": "reconnect_backoff_max_ms", + "KAFKA_SESSION_TIMEOUT_MS": "session_timeout_ms", + "KAFKA_HEARTBEAT_INTERVAL_MS": "heartbeat_interval_ms", } # Integer fields for type conversion @@ -728,6 +777,8 @@ def apply_environment_overrides(self) -> ModelKafkaEventBusConfig: "circuit_breaker_threshold", "reconnect_backoff_ms", "reconnect_backoff_max_ms", + "session_timeout_ms", + "heartbeat_interval_ms", } # Float fields for type conversion diff --git a/src/omnibase_infra/projectors/snapshot_publisher_registration.py b/src/omnibase_infra/projectors/snapshot_publisher_registration.py index 359039a8fc..1b6b5bf3b2 100644 --- a/src/omnibase_infra/projectors/snapshot_publisher_registration.py +++ b/src/omnibase_infra/projectors/snapshot_publisher_registration.py @@ -910,6 +910,8 @@ async def _load_cache_from_topic(self, correlation_id: UUID) -> None: group_id=consumer_group, auto_offset_reset="earliest", enable_auto_commit=False, + session_timeout_ms=30000, + heartbeat_interval_ms=10000, ) try: diff --git a/src/omnibase_infra/runtime/request_response_wiring.py b/src/omnibase_infra/runtime/request_response_wiring.py index 3bafc38af4..77ba859dd2 100644 --- a/src/omnibase_infra/runtime/request_response_wiring.py +++ b/src/omnibase_infra/runtime/request_response_wiring.py @@ -362,6 +362,8 @@ async def _wire_instance(self, instance: ModelRequestResponseInstance) -> None: group_id=consumer_group, auto_offset_reset=instance.auto_offset_reset, enable_auto_commit=True, + session_timeout_ms=30000, + heartbeat_interval_ms=10000, ) rr_instance.consumer = consumer diff --git a/src/omnibase_infra/services/observability/agent_actions/config.py b/src/omnibase_infra/services/observability/agent_actions/config.py index a586366fc3..1b98e654c3 100644 --- a/src/omnibase_infra/services/observability/agent_actions/config.py +++ b/src/omnibase_infra/services/observability/agent_actions/config.py @@ -76,6 +76,26 @@ class ConfigAgentActionsConsumer(BaseSettings): description="Disable auto-commit for at-least-once delivery", ) + # Session timeout (OMN-5445) + session_timeout_ms: int = Field( + default=30000, + ge=6000, + le=300000, + description=( + "Session timeout in milliseconds. Default raised from aiokafka's 10s " + "to 30s to prevent rebalance storms." + ), + ) + heartbeat_interval_ms: int = Field( + default=10000, + ge=1000, + le=100000, + description=( + "Heartbeat interval in milliseconds. " + "Kafka recommends <= session_timeout_ms / 3." + ), + ) + # PostgreSQL connection postgres_dsn: str = Field( description=( diff --git a/src/omnibase_infra/services/observability/agent_actions/consumer.py b/src/omnibase_infra/services/observability/agent_actions/consumer.py index 404821432f..689bb7cb55 100644 --- a/src/omnibase_infra/services/observability/agent_actions/consumer.py +++ b/src/omnibase_infra/services/observability/agent_actions/consumer.py @@ -697,6 +697,8 @@ async def start(self) -> None: auto_offset_reset=self._config.auto_offset_reset, enable_auto_commit=False, # Manual commits for at-least-once max_poll_records=self._config.batch_size, + session_timeout_ms=self._config.session_timeout_ms, + heartbeat_interval_ms=self._config.heartbeat_interval_ms, ) await self._consumer.start() diff --git a/src/omnibase_infra/services/observability/context_audit/config.py b/src/omnibase_infra/services/observability/context_audit/config.py index 489928117c..8df96e8140 100644 --- a/src/omnibase_infra/services/observability/context_audit/config.py +++ b/src/omnibase_infra/services/observability/context_audit/config.py @@ -78,6 +78,26 @@ class ConfigContextAuditConsumer(BaseSettings): description="Disable auto-commit for at-least-once delivery", ) + # Session timeout (OMN-5445) + session_timeout_ms: int = Field( + default=30000, + ge=6000, + le=300000, + description=( + "Session timeout in milliseconds. Default raised from aiokafka's 10s " + "to 30s to prevent rebalance storms." + ), + ) + heartbeat_interval_ms: int = Field( + default=10000, + ge=1000, + le=100000, + description=( + "Heartbeat interval in milliseconds. " + "Kafka recommends <= session_timeout_ms / 3." + ), + ) + # PostgreSQL connection postgres_dsn: str = Field( description=( diff --git a/src/omnibase_infra/services/observability/context_audit/consumer.py b/src/omnibase_infra/services/observability/context_audit/consumer.py index c22ff7d4ea..ead268dc8c 100644 --- a/src/omnibase_infra/services/observability/context_audit/consumer.py +++ b/src/omnibase_infra/services/observability/context_audit/consumer.py @@ -323,6 +323,8 @@ async def start(self) -> None: group_id=self.config.kafka_group_id, auto_offset_reset=self.config.auto_offset_reset, enable_auto_commit=self.config.enable_auto_commit, + session_timeout_ms=self.config.session_timeout_ms, + heartbeat_interval_ms=self.config.heartbeat_interval_ms, value_deserializer=lambda v: v, ) await self._consumer.start() diff --git a/src/omnibase_infra/services/observability/injection_effectiveness/config.py b/src/omnibase_infra/services/observability/injection_effectiveness/config.py index 8dbeba6fcd..c8673c8817 100644 --- a/src/omnibase_infra/services/observability/injection_effectiveness/config.py +++ b/src/omnibase_infra/services/observability/injection_effectiveness/config.py @@ -123,6 +123,26 @@ class ConfigInjectionEffectivenessConsumer(BaseSettings): description="Disable auto-commit for at-least-once delivery", ) + # Session timeout (OMN-5445) + session_timeout_ms: int = Field( + default=30000, + ge=6000, + le=300000, + description=( + "Session timeout in milliseconds. Default raised from aiokafka's 10s " + "to 30s to prevent rebalance storms." + ), + ) + heartbeat_interval_ms: int = Field( + default=10000, + ge=1000, + le=100000, + description=( + "Heartbeat interval in milliseconds. " + "Kafka recommends <= session_timeout_ms / 3." + ), + ) + # PostgreSQL connection postgres_dsn: str = Field( description=( diff --git a/src/omnibase_infra/services/observability/injection_effectiveness/consumer.py b/src/omnibase_infra/services/observability/injection_effectiveness/consumer.py index 865a151a60..a2b8b0cb35 100644 --- a/src/omnibase_infra/services/observability/injection_effectiveness/consumer.py +++ b/src/omnibase_infra/services/observability/injection_effectiveness/consumer.py @@ -516,6 +516,8 @@ async def start(self) -> None: auto_offset_reset=self._config.auto_offset_reset, enable_auto_commit=False, # Manual commits for at-least-once max_poll_records=self._config.batch_size, + session_timeout_ms=self._config.session_timeout_ms, + heartbeat_interval_ms=self._config.heartbeat_interval_ms, ) await self._consumer.start() diff --git a/src/omnibase_infra/services/observability/llm_cost_aggregation/config.py b/src/omnibase_infra/services/observability/llm_cost_aggregation/config.py index fe0631891d..7412d872fd 100644 --- a/src/omnibase_infra/services/observability/llm_cost_aggregation/config.py +++ b/src/omnibase_infra/services/observability/llm_cost_aggregation/config.py @@ -72,6 +72,26 @@ class ConfigLlmCostAggregation(BaseSettings): ), ) + # Session timeout (OMN-5445) + session_timeout_ms: int = Field( + default=30000, + ge=6000, + le=300000, + description=( + "Session timeout in milliseconds. Default raised from aiokafka's 10s " + "to 30s to prevent rebalance storms." + ), + ) + heartbeat_interval_ms: int = Field( + default=10000, + ge=1000, + le=100000, + description=( + "Heartbeat interval in milliseconds. " + "Kafka recommends <= session_timeout_ms / 3." + ), + ) + # PostgreSQL connection postgres_dsn: str = Field( ..., diff --git a/src/omnibase_infra/services/observability/llm_cost_aggregation/consumer.py b/src/omnibase_infra/services/observability/llm_cost_aggregation/consumer.py index 87af5b7741..cb17d5b3fd 100644 --- a/src/omnibase_infra/services/observability/llm_cost_aggregation/consumer.py +++ b/src/omnibase_infra/services/observability/llm_cost_aggregation/consumer.py @@ -474,6 +474,8 @@ async def start(self) -> None: auto_offset_reset=self._config.auto_offset_reset, enable_auto_commit=False, max_poll_records=self._config.batch_size, + session_timeout_ms=self._config.session_timeout_ms, + heartbeat_interval_ms=self._config.heartbeat_interval_ms, ) await self._consumer.start() diff --git a/src/omnibase_infra/services/observability/skill_lifecycle/config.py b/src/omnibase_infra/services/observability/skill_lifecycle/config.py index 45131f4608..bab543e11c 100644 --- a/src/omnibase_infra/services/observability/skill_lifecycle/config.py +++ b/src/omnibase_infra/services/observability/skill_lifecycle/config.py @@ -66,6 +66,26 @@ class ConfigSkillLifecycleConsumer(BaseSettings): description="Disable auto-commit for at-least-once delivery", ) + # Session timeout (OMN-5445) + session_timeout_ms: int = Field( + default=30000, + ge=6000, + le=300000, + description=( + "Session timeout in milliseconds. Default raised from aiokafka's 10s " + "to 30s to prevent rebalance storms." + ), + ) + heartbeat_interval_ms: int = Field( + default=10000, + ge=1000, + le=100000, + description=( + "Heartbeat interval in milliseconds. " + "Kafka recommends <= session_timeout_ms / 3." + ), + ) + # PostgreSQL connection postgres_dsn: str = Field( description=( diff --git a/src/omnibase_infra/services/observability/skill_lifecycle/consumer.py b/src/omnibase_infra/services/observability/skill_lifecycle/consumer.py index 488a3f7323..0cb32bf7b9 100644 --- a/src/omnibase_infra/services/observability/skill_lifecycle/consumer.py +++ b/src/omnibase_infra/services/observability/skill_lifecycle/consumer.py @@ -348,6 +348,8 @@ async def start(self) -> None: group_id=self.config.kafka_group_id, auto_offset_reset=self.config.auto_offset_reset, enable_auto_commit=self.config.enable_auto_commit, + session_timeout_ms=self.config.session_timeout_ms, + heartbeat_interval_ms=self.config.heartbeat_interval_ms, value_deserializer=lambda v: v, ) await self._consumer.start() diff --git a/src/omnibase_infra/services/session/config_consumer.py b/src/omnibase_infra/services/session/config_consumer.py index f321fbad5f..d7da70923a 100644 --- a/src/omnibase_infra/services/session/config_consumer.py +++ b/src/omnibase_infra/services/session/config_consumer.py @@ -59,6 +59,27 @@ class ConfigSessionConsumer(BaseSettings): default=False, description="Disable auto-commit for at-least-once delivery", ) + + # Session timeout (OMN-5445) + session_timeout_ms: int = Field( + default=30000, + ge=6000, + le=300000, + description=( + "Session timeout in milliseconds. Default raised from aiokafka's 10s " + "to 30s to prevent rebalance storms." + ), + ) + heartbeat_interval_ms: int = Field( + default=10000, + ge=1000, + le=100000, + description=( + "Heartbeat interval in milliseconds. " + "Kafka recommends <= session_timeout_ms / 3." + ), + ) + max_poll_records: int = Field( default=100, ge=1, diff --git a/src/omnibase_infra/services/session/consumer.py b/src/omnibase_infra/services/session/consumer.py index 75a6e661f8..0a31ea6af7 100644 --- a/src/omnibase_infra/services/session/consumer.py +++ b/src/omnibase_infra/services/session/consumer.py @@ -368,6 +368,8 @@ async def start(self) -> None: auto_offset_reset=self._config.auto_offset_reset, enable_auto_commit=False, # Manual commits for at-least-once max_poll_records=self._config.max_poll_records, + session_timeout_ms=self._config.session_timeout_ms, + heartbeat_interval_ms=self._config.heartbeat_interval_ms, ) await self._consumer.start() diff --git a/src/omnibase_infra/tui/consumers/consumer_status.py b/src/omnibase_infra/tui/consumers/consumer_status.py index 2ba30fe669..5b37647ee8 100644 --- a/src/omnibase_infra/tui/consumers/consumer_status.py +++ b/src/omnibase_infra/tui/consumers/consumer_status.py @@ -113,6 +113,8 @@ async def consume_all(app: object) -> None: value_deserializer=lambda v: v, # raw bytes; we JSON-decode below request_timeout_ms=5000, connections_max_idle_ms=10000, + session_timeout_ms=30000, + heartbeat_interval_ms=10000, ) await consumer.start() logger.info("TUI consumer started. Topics: %s.", _ALL_TOPICS) diff --git a/src/omnibase_infra/validation/validation_exemptions.yaml b/src/omnibase_infra/validation/validation_exemptions.yaml index aebb4effa6..b583bf5559 100644 --- a/src/omnibase_infra/validation/validation_exemptions.yaml +++ b/src/omnibase_infra/validation/validation_exemptions.yaml @@ -143,6 +143,23 @@ pattern_exemptions: - OMN-1519 PR description ticket: OMN-1519 # ========================================================================== + # ModelKafkaEventBusConfig Exemptions (OMN-5445) + # ========================================================================== + # Config model with Pydantic model_validators for cross-field validation + # (reconnect backoff, heartbeat/session ratio, auth config, bootstrap servers, + # environment, instance_id) plus factory methods (default, from_yaml, + # apply_environment_overrides) and computed properties (acks_aiokafka, + # get_dlq_topic). All methods are cohesive config-related concerns. + - file_pattern: 'model_kafka_event_bus_config\.py' + class_pattern: "Class 'ModelKafkaEventBusConfig'" + violation_pattern: 'has \d+ methods' + reason: > + Pydantic config model with model_validators for cross-field validation (reconnect backoff, heartbeat/session ratio, auth, bootstrap servers), factory methods (default, from_yaml, apply_environment_overrides), and computed properties. All methods are cohesive config concerns. + + documentation: + - OMN-5445 (Kafka session timeout fix) + ticket: OMN-5445 + # ========================================================================== # EventBusKafka Exemptions (OMN-934, OMN-1305) # ========================================================================== # Event bus pattern requires many methods for lifecycle (start/stop/health), diff --git a/tests/unit/event_bus/test_kafka_config_session_timeout.py b/tests/unit/event_bus/test_kafka_config_session_timeout.py new file mode 100644 index 0000000000..173cba591c --- /dev/null +++ b/tests/unit/event_bus/test_kafka_config_session_timeout.py @@ -0,0 +1,186 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Unit tests for ModelKafkaEventBusConfig session timeout fields (OMN-5445). + +Tests the session_timeout_ms and heartbeat_interval_ms fields added to +ModelKafkaEventBusConfig, including defaults, env var overrides, advisory +validator, and field constraints. +""" + +from __future__ import annotations + +import logging + +import pytest +from pydantic import ValidationError + +from omnibase_infra.event_bus.models.config import ModelKafkaEventBusConfig + + +class TestSessionTimeoutDefaults: + """Test default values for session timeout fields.""" + + @pytest.mark.unit + def test_default_session_timeout_ms(self) -> None: + """Default session_timeout_ms should be 30000 ms.""" + config = ModelKafkaEventBusConfig() + assert config.session_timeout_ms == 30000 + + @pytest.mark.unit + def test_default_heartbeat_interval_ms(self) -> None: + """Default heartbeat_interval_ms should be 10000 ms.""" + config = ModelKafkaEventBusConfig() + assert config.heartbeat_interval_ms == 10000 + + @pytest.mark.unit + def test_defaults_satisfy_kafka_ratio_recommendation(self) -> None: + """Default heartbeat (10s) should be <= session_timeout (30s) / 3.""" + config = ModelKafkaEventBusConfig() + assert config.heartbeat_interval_ms <= config.session_timeout_ms / 3 + + +class TestSessionTimeoutCustomValues: + """Test custom values can be set via constructor.""" + + @pytest.mark.unit + def test_custom_session_timeout_ms(self) -> None: + """Should accept a custom session_timeout_ms value.""" + config = ModelKafkaEventBusConfig(session_timeout_ms=60000) + assert config.session_timeout_ms == 60000 + + @pytest.mark.unit + def test_custom_heartbeat_interval_ms(self) -> None: + """Should accept a custom heartbeat_interval_ms value.""" + config = ModelKafkaEventBusConfig(heartbeat_interval_ms=5000) + assert config.heartbeat_interval_ms == 5000 + + +class TestSessionTimeoutConstraints: + """Test field constraints (ge/le bounds).""" + + @pytest.mark.unit + def test_session_timeout_ms_below_minimum_raises(self) -> None: + """session_timeout_ms below 6000 should raise ValidationError.""" + with pytest.raises(ValidationError): + ModelKafkaEventBusConfig(session_timeout_ms=5000) + + @pytest.mark.unit + def test_session_timeout_ms_above_maximum_raises(self) -> None: + """session_timeout_ms above 300000 should raise ValidationError.""" + with pytest.raises(ValidationError): + ModelKafkaEventBusConfig(session_timeout_ms=400000) + + @pytest.mark.unit + def test_heartbeat_interval_ms_below_minimum_raises(self) -> None: + """heartbeat_interval_ms below 1000 should raise ValidationError.""" + with pytest.raises(ValidationError): + ModelKafkaEventBusConfig(heartbeat_interval_ms=500) + + @pytest.mark.unit + def test_heartbeat_interval_ms_above_maximum_raises(self) -> None: + """heartbeat_interval_ms above 100000 should raise ValidationError.""" + with pytest.raises(ValidationError): + ModelKafkaEventBusConfig(heartbeat_interval_ms=200000) + + @pytest.mark.unit + def test_session_timeout_ms_at_minimum(self) -> None: + """session_timeout_ms at boundary 6000 should be accepted.""" + config = ModelKafkaEventBusConfig(session_timeout_ms=6000) + assert config.session_timeout_ms == 6000 + + @pytest.mark.unit + def test_session_timeout_ms_at_maximum(self) -> None: + """session_timeout_ms at boundary 300000 should be accepted.""" + config = ModelKafkaEventBusConfig(session_timeout_ms=300000) + assert config.session_timeout_ms == 300000 + + +class TestHeartbeatSessionRatioValidator: + """Test advisory validator for heartbeat/session ratio.""" + + @pytest.mark.unit + def test_no_warning_when_ratio_is_valid( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """No warning when heartbeat <= session_timeout / 3.""" + with caplog.at_level(logging.WARNING): + ModelKafkaEventBusConfig( + session_timeout_ms=30000, + heartbeat_interval_ms=10000, + ) + assert "heartbeat_interval_ms" not in caplog.text + + @pytest.mark.unit + def test_warning_when_heartbeat_exceeds_ratio( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """Warning when heartbeat > session_timeout / 3.""" + with caplog.at_level(logging.WARNING): + config = ModelKafkaEventBusConfig( + session_timeout_ms=30000, + heartbeat_interval_ms=15000, + ) + assert "heartbeat_interval_ms" in caplog.text + assert "exceeds" in caplog.text + # Validator is advisory: config is still created + assert config.session_timeout_ms == 30000 + assert config.heartbeat_interval_ms == 15000 + + @pytest.mark.unit + def test_no_error_raised_for_bad_ratio(self) -> None: + """Advisory validator must not raise, even with bad ratio.""" + config = ModelKafkaEventBusConfig( + session_timeout_ms=30000, + heartbeat_interval_ms=20000, + ) + assert config.heartbeat_interval_ms == 20000 + + +class TestSessionTimeoutEnvOverrides: + """Test environment variable overrides for session timeout fields.""" + + @pytest.mark.unit + def test_kafka_session_timeout_ms_env_override( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """KAFKA_SESSION_TIMEOUT_MS env var should override session_timeout_ms.""" + monkeypatch.setenv("KAFKA_SESSION_TIMEOUT_MS", "45000") + config = ModelKafkaEventBusConfig.default() + assert config.session_timeout_ms == 45000 + + @pytest.mark.unit + def test_kafka_heartbeat_interval_ms_env_override( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """KAFKA_HEARTBEAT_INTERVAL_MS env var should override heartbeat_interval_ms.""" + monkeypatch.setenv("KAFKA_HEARTBEAT_INTERVAL_MS", "5000") + config = ModelKafkaEventBusConfig.default() + assert config.heartbeat_interval_ms == 5000 + + +class TestSessionTimeoutDefaultFactory: + """Test default() factory includes session timeout fields.""" + + @pytest.mark.unit + def test_default_factory_includes_session_timeout_ms(self) -> None: + """default() factory should return expected session_timeout_ms default.""" + config = ModelKafkaEventBusConfig.default() + assert config.session_timeout_ms == 30000 + + @pytest.mark.unit + def test_default_factory_includes_heartbeat_interval_ms(self) -> None: + """default() factory should return expected heartbeat_interval_ms default.""" + config = ModelKafkaEventBusConfig.default() + assert config.heartbeat_interval_ms == 10000 + + +class TestSessionTimeoutJsonSchema: + """Test session timeout fields appear in model JSON schema.""" + + @pytest.mark.unit + def test_fields_appear_in_json_schema(self) -> None: + """session_timeout_ms and heartbeat_interval_ms must appear in JSON schema.""" + schema = ModelKafkaEventBusConfig.model_json_schema() + properties = schema.get("properties", {}) + assert "session_timeout_ms" in properties + assert "heartbeat_interval_ms" in properties diff --git a/tests/unit/event_bus/test_kafka_timeout_kwargs.py b/tests/unit/event_bus/test_kafka_timeout_kwargs.py new file mode 100644 index 0000000000..d98a971250 --- /dev/null +++ b/tests/unit/event_bus/test_kafka_timeout_kwargs.py @@ -0,0 +1,222 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Unit tests for EventBusKafka consumer/producer session timeout and reconnect kwargs (OMN-5445). + +Verifies that session_timeout_ms, heartbeat_interval_ms, reconnect_backoff_ms, and +reconnect_backoff_max_ms are correctly passed from ModelKafkaEventBusConfig through +to the AIOKafkaConsumer and AIOKafkaProducer constructors. +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from omnibase_infra.event_bus.event_bus_kafka import EventBusKafka +from omnibase_infra.event_bus.models.config import ModelKafkaEventBusConfig + + +@pytest.fixture +def kafka_config() -> ModelKafkaEventBusConfig: + """Create a config with known session/heartbeat/reconnect values.""" + return ModelKafkaEventBusConfig( + session_timeout_ms=45000, + heartbeat_interval_ms=15000, + reconnect_backoff_ms=3000, + reconnect_backoff_max_ms=60000, + ) + + +@pytest.fixture +def bus(kafka_config: ModelKafkaEventBusConfig) -> EventBusKafka: + """Create EventBusKafka instance with test config.""" + return EventBusKafka(config=kafka_config) + + +class TestConsumerKwargs: + """Test that AIOKafkaConsumer receives session/heartbeat/reconnect kwargs.""" + + @pytest.mark.unit + @pytest.mark.asyncio + async def test_consumer_receives_session_timeout_ms( + self, bus: EventBusKafka, kafka_config: ModelKafkaEventBusConfig + ) -> None: + """Consumer constructor must receive session_timeout_ms from config.""" + mock_consumer = MagicMock() + mock_consumer.start = AsyncMock() + + with patch( + "omnibase_infra.event_bus.event_bus_kafka.AIOKafkaProducer" + ) as mock_producer_cls: + mock_producer = MagicMock() + mock_producer.start = AsyncMock() + mock_producer.stop = AsyncMock() + mock_producer_cls.return_value = mock_producer + await bus.start() + + with patch( + "omnibase_infra.event_bus.event_bus_kafka.AIOKafkaConsumer", + return_value=mock_consumer, + ) as mock_consumer_cls: + await bus.subscribe( + "test-topic", on_message=AsyncMock(), group_id="test-group" + ) + call_kwargs = mock_consumer_cls.call_args + assert call_kwargs.kwargs["session_timeout_ms"] == 45000 + + @pytest.mark.unit + @pytest.mark.asyncio + async def test_consumer_receives_heartbeat_interval_ms( + self, bus: EventBusKafka, kafka_config: ModelKafkaEventBusConfig + ) -> None: + """Consumer constructor must receive heartbeat_interval_ms from config.""" + mock_consumer = MagicMock() + mock_consumer.start = AsyncMock() + + with patch( + "omnibase_infra.event_bus.event_bus_kafka.AIOKafkaProducer" + ) as mock_producer_cls: + mock_producer = MagicMock() + mock_producer.start = AsyncMock() + mock_producer.stop = AsyncMock() + mock_producer_cls.return_value = mock_producer + await bus.start() + + with patch( + "omnibase_infra.event_bus.event_bus_kafka.AIOKafkaConsumer", + return_value=mock_consumer, + ) as mock_consumer_cls: + await bus.subscribe( + "test-topic", on_message=AsyncMock(), group_id="test-group" + ) + call_kwargs = mock_consumer_cls.call_args + assert call_kwargs.kwargs["heartbeat_interval_ms"] == 15000 + + @pytest.mark.unit + @pytest.mark.asyncio + async def test_consumer_receives_reconnect_backoff_ms( + self, bus: EventBusKafka, kafka_config: ModelKafkaEventBusConfig + ) -> None: + """Consumer constructor must receive reconnect_backoff_ms from config.""" + mock_consumer = MagicMock() + mock_consumer.start = AsyncMock() + + with patch( + "omnibase_infra.event_bus.event_bus_kafka.AIOKafkaProducer" + ) as mock_producer_cls: + mock_producer = MagicMock() + mock_producer.start = AsyncMock() + mock_producer.stop = AsyncMock() + mock_producer_cls.return_value = mock_producer + await bus.start() + + with patch( + "omnibase_infra.event_bus.event_bus_kafka.AIOKafkaConsumer", + return_value=mock_consumer, + ) as mock_consumer_cls: + await bus.subscribe( + "test-topic", on_message=AsyncMock(), group_id="test-group" + ) + call_kwargs = mock_consumer_cls.call_args + assert call_kwargs.kwargs["reconnect_backoff_ms"] == 3000 + + @pytest.mark.unit + @pytest.mark.asyncio + async def test_consumer_receives_reconnect_backoff_max_ms( + self, bus: EventBusKafka, kafka_config: ModelKafkaEventBusConfig + ) -> None: + """Consumer constructor must receive reconnect_backoff_max_ms from config.""" + mock_consumer = MagicMock() + mock_consumer.start = AsyncMock() + + with patch( + "omnibase_infra.event_bus.event_bus_kafka.AIOKafkaProducer" + ) as mock_producer_cls: + mock_producer = MagicMock() + mock_producer.start = AsyncMock() + mock_producer.stop = AsyncMock() + mock_producer_cls.return_value = mock_producer + await bus.start() + + with patch( + "omnibase_infra.event_bus.event_bus_kafka.AIOKafkaConsumer", + return_value=mock_consumer, + ) as mock_consumer_cls: + await bus.subscribe( + "test-topic", on_message=AsyncMock(), group_id="test-group" + ) + call_kwargs = mock_consumer_cls.call_args + assert call_kwargs.kwargs["reconnect_backoff_max_ms"] == 60000 + + +class TestProducerKwargs: + """Test that AIOKafkaProducer receives reconnect backoff kwargs.""" + + @pytest.mark.unit + @pytest.mark.asyncio + async def test_producer_start_receives_reconnect_backoff_ms( + self, bus: EventBusKafka + ) -> None: + """Producer constructor in start() must receive reconnect_backoff_ms.""" + with patch( + "omnibase_infra.event_bus.event_bus_kafka.AIOKafkaProducer" + ) as mock_producer_cls: + mock_producer = MagicMock() + mock_producer.start = AsyncMock() + mock_producer.stop = AsyncMock() + mock_producer_cls.return_value = mock_producer + await bus.start() + call_kwargs = mock_producer_cls.call_args + assert call_kwargs.kwargs["reconnect_backoff_ms"] == 3000 + + @pytest.mark.unit + @pytest.mark.asyncio + async def test_producer_start_receives_reconnect_backoff_max_ms( + self, bus: EventBusKafka + ) -> None: + """Producer constructor in start() must receive reconnect_backoff_max_ms.""" + with patch( + "omnibase_infra.event_bus.event_bus_kafka.AIOKafkaProducer" + ) as mock_producer_cls: + mock_producer = MagicMock() + mock_producer.start = AsyncMock() + mock_producer.stop = AsyncMock() + mock_producer_cls.return_value = mock_producer + await bus.start() + call_kwargs = mock_producer_cls.call_args + assert call_kwargs.kwargs["reconnect_backoff_max_ms"] == 60000 + + +class TestDefaultValues: + """Test that default config values are correctly wired through.""" + + @pytest.mark.unit + @pytest.mark.asyncio + async def test_default_session_timeout_wired_to_consumer(self) -> None: + """Default 30000ms session timeout must reach consumer constructor.""" + config = ModelKafkaEventBusConfig() + bus = EventBusKafka(config=config) + + mock_consumer = MagicMock() + mock_consumer.start = AsyncMock() + + with patch( + "omnibase_infra.event_bus.event_bus_kafka.AIOKafkaProducer" + ) as mock_producer_cls: + mock_producer = MagicMock() + mock_producer.start = AsyncMock() + mock_producer.stop = AsyncMock() + mock_producer_cls.return_value = mock_producer + await bus.start() + + with patch( + "omnibase_infra.event_bus.event_bus_kafka.AIOKafkaConsumer", + return_value=mock_consumer, + ) as mock_consumer_cls: + await bus.subscribe( + "test-topic", on_message=AsyncMock(), group_id="test-group" + ) + call_kwargs = mock_consumer_cls.call_args + assert call_kwargs.kwargs["session_timeout_ms"] == 30000 + assert call_kwargs.kwargs["heartbeat_interval_ms"] == 10000 diff --git a/tests/unit/services/test_consumer_session_timeout.py b/tests/unit/services/test_consumer_session_timeout.py new file mode 100644 index 0000000000..4d6c137ba2 --- /dev/null +++ b/tests/unit/services/test_consumer_session_timeout.py @@ -0,0 +1,96 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Unit tests for standalone consumer config session timeout fields (OMN-5445). + +Verifies that all 6 standalone consumer configs have session_timeout_ms and +heartbeat_interval_ms fields with correct defaults (30000/10000). +""" + +from __future__ import annotations + +import pytest + +from omnibase_infra.services.observability.agent_actions.config import ( + ConfigAgentActionsConsumer, +) +from omnibase_infra.services.observability.context_audit.config import ( + ConfigContextAuditConsumer, +) +from omnibase_infra.services.observability.injection_effectiveness.config import ( + ConfigInjectionEffectivenessConsumer, +) +from omnibase_infra.services.observability.llm_cost_aggregation.config import ( + ConfigLlmCostAggregation, +) +from omnibase_infra.services.observability.skill_lifecycle.config import ( + ConfigSkillLifecycleConsumer, +) +from omnibase_infra.services.session.config_consumer import ConfigSessionConsumer + +# All standalone consumer config classes with their names for parametrized tests +_STANDALONE_CONFIGS = [ + ConfigSessionConsumer, + ConfigAgentActionsConsumer, + ConfigInjectionEffectivenessConsumer, + ConfigLlmCostAggregation, + ConfigSkillLifecycleConsumer, + ConfigContextAuditConsumer, +] + + +class TestStandaloneConsumerConfigDefaults: + """Test that all standalone consumer configs have session timeout fields.""" + + @pytest.mark.unit + @pytest.mark.parametrize( + "config_cls", + _STANDALONE_CONFIGS, + ids=[c.__name__ for c in _STANDALONE_CONFIGS], + ) + def test_has_session_timeout_ms_field(self, config_cls: type) -> None: + """Config class must have a session_timeout_ms field.""" + fields = config_cls.model_fields + assert "session_timeout_ms" in fields, ( + f"{config_cls.__name__} missing session_timeout_ms field" + ) + + @pytest.mark.unit + @pytest.mark.parametrize( + "config_cls", + _STANDALONE_CONFIGS, + ids=[c.__name__ for c in _STANDALONE_CONFIGS], + ) + def test_has_heartbeat_interval_ms_field(self, config_cls: type) -> None: + """Config class must have a heartbeat_interval_ms field.""" + fields = config_cls.model_fields + assert "heartbeat_interval_ms" in fields, ( + f"{config_cls.__name__} missing heartbeat_interval_ms field" + ) + + @pytest.mark.unit + @pytest.mark.parametrize( + "config_cls", + _STANDALONE_CONFIGS, + ids=[c.__name__ for c in _STANDALONE_CONFIGS], + ) + def test_session_timeout_ms_default_is_30000(self, config_cls: type) -> None: + """session_timeout_ms default must be 30000.""" + field_info = config_cls.model_fields["session_timeout_ms"] + assert field_info.default == 30000, ( + f"{config_cls.__name__}.session_timeout_ms default is " + f"{field_info.default}, expected 30000" + ) + + @pytest.mark.unit + @pytest.mark.parametrize( + "config_cls", + _STANDALONE_CONFIGS, + ids=[c.__name__ for c in _STANDALONE_CONFIGS], + ) + def test_heartbeat_interval_ms_default_is_10000(self, config_cls: type) -> None: + """heartbeat_interval_ms default must be 10000.""" + field_info = config_cls.model_fields["heartbeat_interval_ms"] + assert field_info.default == 10000, ( + f"{config_cls.__name__}.heartbeat_interval_ms default is " + f"{field_info.default}, expected 10000" + ) From 950283ca6fb7f15a9f46f702163d224ba369a936 Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Thu, 19 Mar 2026 10:13:23 -0400 Subject: [PATCH 2/4] fix(arch): replace raw DLQ topic literals with canonical constants [OMN-5445] Moves two raw topic literal strings in observability consumer config files to the canonical constants in platform_topic_suffixes.py, fixing the Arch Invariants (OMN-3343) CI check that blocked this PR. - agent_actions/config.py: import and use SUFFIX_OMNICLAUDE_AGENT_ACTIONS_DLQ - skill_lifecycle/config.py: import and use new SUFFIX_OMNICLAUDE_SKILL_LIFECYCLE_DLQ - platform_topic_suffixes.py: add SUFFIX_OMNICLAUDE_SKILL_LIFECYCLE_DLQ constant and include it in _OMNICLAUDE_OBSERVABILITY_DLQ_TOPIC_SUFFIXES for provisioning - topic_literal_baseline.txt: update grandfathered line numbers after import insertion; remove now-fixed DLQ entries (lines 131/134 agent_actions, 124/126 skill_lifecycle) - check_contract_topic_parity.py: add skill-lifecycle-dlq to legacy allowlist (mirrors pattern for agent-actions-dlq and agent-observability-dlq) --- scripts/check_contract_topic_parity.py | 1 + scripts/validation/topic_literal_baseline.txt | 23 ++++++------------- .../observability/agent_actions/config.py | 6 ++++- .../observability/skill_lifecycle/config.py | 6 ++++- .../topics/platform_topic_suffixes.py | 15 ++++++++++++ 5 files changed, 33 insertions(+), 18 deletions(-) diff --git a/scripts/check_contract_topic_parity.py b/scripts/check_contract_topic_parity.py index 0190d22460..5775f39f30 100644 --- a/scripts/check_contract_topic_parity.py +++ b/scripts/check_contract_topic_parity.py @@ -258,6 +258,7 @@ # --- omniclaude special/non-skill topics (need contract.yaml in omniclaude) --- "onex.evt.omniclaude.agent-actions-dlq.v1": "DLQ topic, no contract.yaml yet; needs omniclaude node contract | owner: jonah | expiry: 2026-06-01", "onex.evt.omniclaude.agent-observability-dlq.v1": "DLQ topic, no contract.yaml yet; needs omniclaude node contract | owner: jonah | expiry: 2026-06-01", + "onex.evt.omniclaude.skill-lifecycle-dlq.v1": "DLQ topic for skill-lifecycle consumer (OMN-5445); provisioned in platform_topic_suffixes; needs omniclaude node contract | owner: jonah | expiry: 2026-06-01", "onex.evt.omniclaude.audit-compression-triggered.v1": "context audit topic [OMN-5240]; produced by omniclaude, needs contract.yaml | owner: jonah | expiry: 2026-06-01", "onex.evt.omniclaude.audit-context-budget-exceeded.v1": "context audit topic [OMN-5240]; produced by omniclaude, needs contract.yaml | owner: jonah | expiry: 2026-06-01", "onex.evt.omniclaude.audit-dispatch-validated.v1": "context audit topic [OMN-5240]; produced by omniclaude, needs contract.yaml | owner: jonah | expiry: 2026-06-01", diff --git a/scripts/validation/topic_literal_baseline.txt b/scripts/validation/topic_literal_baseline.txt index 2c209c46cf..0e308f1baf 100644 --- a/scripts/validation/topic_literal_baseline.txt +++ b/scripts/validation/topic_literal_baseline.txt @@ -2,7 +2,7 @@ # Pre-existing violations grandfathered in on 2026-03-02 (updated 2026-03-10 for SPDX line shifts). # Format: : # DO NOT add new entries here — fix violations instead. -# Total entries: 131 +# Total entries: 122 src/omnibase_infra/cli/git_hook_relay.py:72 src/omnibase_infra/cli/git_hook_relay.py:74 src/omnibase_infra/cli/infra_test/introspect.py:25 @@ -50,18 +50,13 @@ src/omnibase_infra/runtime/contract_registration_event_router.py:76 src/omnibase_infra/runtime/contract_registration_event_router.py:77 src/omnibase_infra/runtime/service_runtime_host_process.py:4299 src/omnibase_infra/runtime/service_runtime_host_process.py:4303 -src/omnibase_infra/services/observability/agent_actions/config.py:131 -src/omnibase_infra/services/observability/agent_actions/config.py:134 -src/omnibase_infra/services/observability/agent_actions/config.py:55 -src/omnibase_infra/services/observability/agent_actions/config.py:56 -src/omnibase_infra/services/observability/agent_actions/config.py:57 -src/omnibase_infra/services/observability/agent_actions/config.py:58 -src/omnibase_infra/services/observability/agent_actions/config.py:59 -src/omnibase_infra/services/observability/agent_actions/config.py:60 -src/omnibase_infra/services/observability/agent_actions/config.py:61 src/omnibase_infra/services/observability/agent_actions/config.py:62 src/omnibase_infra/services/observability/agent_actions/config.py:63 src/omnibase_infra/services/observability/agent_actions/config.py:64 +src/omnibase_infra/services/observability/agent_actions/config.py:65 +src/omnibase_infra/services/observability/agent_actions/config.py:66 +src/omnibase_infra/services/observability/agent_actions/config.py:67 +src/omnibase_infra/services/observability/agent_actions/config.py:68 src/omnibase_infra/services/observability/agent_actions/consumer.py:178 src/omnibase_infra/services/observability/agent_actions/consumer.py:179 src/omnibase_infra/services/observability/agent_actions/consumer.py:180 @@ -109,12 +104,8 @@ src/omnibase_infra/services/observability/injection_effectiveness/consumer.py:17 src/omnibase_infra/services/observability/injection_effectiveness/consumer.py:179 src/omnibase_infra/services/observability/llm_cost_aggregation/config.py:61 src/omnibase_infra/services/observability/llm_cost_aggregation/config.py:63 -src/omnibase_infra/services/observability/skill_lifecycle/config.py:124 -src/omnibase_infra/services/observability/skill_lifecycle/config.py:126 -src/omnibase_infra/services/observability/skill_lifecycle/config.py:53 -src/omnibase_infra/services/observability/skill_lifecycle/config.py:54 -src/omnibase_infra/services/observability/skill_lifecycle/config.py:55 -src/omnibase_infra/services/observability/skill_lifecycle/config.py:56 +src/omnibase_infra/services/observability/skill_lifecycle/config.py:57 +src/omnibase_infra/services/observability/skill_lifecycle/config.py:58 src/omnibase_infra/services/observability/skill_lifecycle/consumer.py:272 src/omnibase_infra/services/observability/skill_lifecycle/consumer.py:273 src/omnibase_infra/services/observability/skill_lifecycle/consumer.py:274 diff --git a/src/omnibase_infra/services/observability/agent_actions/config.py b/src/omnibase_infra/services/observability/agent_actions/config.py index 1b98e654c3..e9c99fedfc 100644 --- a/src/omnibase_infra/services/observability/agent_actions/config.py +++ b/src/omnibase_infra/services/observability/agent_actions/config.py @@ -16,6 +16,10 @@ from pydantic import Field, model_validator from pydantic_settings import BaseSettings, SettingsConfigDict +from omnibase_infra.topics.platform_topic_suffixes import ( + SUFFIX_OMNICLAUDE_AGENT_ACTIONS_DLQ, +) + logger = logging.getLogger(__name__) @@ -151,7 +155,7 @@ class ConfigAgentActionsConsumer(BaseSettings): # Dead Letter Queue (Phase 2 hardening - OMN-1768) dlq_topic: str = Field( - default="onex.evt.omniclaude.agent-actions-dlq.v1", + default=SUFFIX_OMNICLAUDE_AGENT_ACTIONS_DLQ, description=( "Dead letter topic for permanently failed messages. Messages that " "fail validation or exceed max retry count are forwarded here. " diff --git a/src/omnibase_infra/services/observability/skill_lifecycle/config.py b/src/omnibase_infra/services/observability/skill_lifecycle/config.py index bab543e11c..79d6da8716 100644 --- a/src/omnibase_infra/services/observability/skill_lifecycle/config.py +++ b/src/omnibase_infra/services/observability/skill_lifecycle/config.py @@ -13,6 +13,10 @@ from pydantic import Field, model_validator from pydantic_settings import BaseSettings, SettingsConfigDict +from omnibase_infra.topics.platform_topic_suffixes import ( + SUFFIX_OMNICLAUDE_SKILL_LIFECYCLE_DLQ, +) + logger = logging.getLogger(__name__) @@ -141,7 +145,7 @@ class ConfigSkillLifecycleConsumer(BaseSettings): # Dead Letter Queue dlq_topic: str = Field( - default="onex.evt.omniclaude.skill-lifecycle-dlq.v1", + default=SUFFIX_OMNICLAUDE_SKILL_LIFECYCLE_DLQ, description=( "Dead letter topic for permanently failed skill lifecycle messages. " "Configure via OMNIBASE_INFRA_SKILL_LIFECYCLE_DLQ_TOPIC env var." diff --git a/src/omnibase_infra/topics/platform_topic_suffixes.py b/src/omnibase_infra/topics/platform_topic_suffixes.py index e817eb7dad..ee8efec8c1 100644 --- a/src/omnibase_infra/topics/platform_topic_suffixes.py +++ b/src/omnibase_infra/topics/platform_topic_suffixes.py @@ -638,9 +638,24 @@ Consumer: observability alerting, incident recovery tooling """ +SUFFIX_OMNICLAUDE_SKILL_LIFECYCLE_DLQ: str = ( + "onex.evt.omniclaude.skill-lifecycle-dlq.v1" +) +"""Dead letter queue topic for the skill-lifecycle observability consumer. + +Messages that fail validation or exceed max retry count in +ConfigSkillLifecycleConsumer are forwarded to this topic. Matches the +hardcoded default in +``omnibase_infra.services.observability.skill_lifecycle.config.ConfigSkillLifecycleConsumer.dlq_topic``. + +Producer: skill-lifecycle consumer (ServiceSkillLifecycleConsumer) +Consumer: observability alerting, incident recovery tooling +""" + _OMNICLAUDE_OBSERVABILITY_DLQ_TOPIC_SUFFIXES: tuple[str, ...] = ( SUFFIX_OMNICLAUDE_AGENT_ACTIONS_DLQ, SUFFIX_OMNICLAUDE_AGENT_OBSERVABILITY_DLQ, + SUFFIX_OMNICLAUDE_SKILL_LIFECYCLE_DLQ, ) """DLQ topic suffixes for OmniClaude observability consumers. From f3db0bdab2edf7ea2760f67c0cc97676e58926e1 Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Thu, 19 Mar 2026 11:16:09 -0400 Subject: [PATCH 3/4] fix(topics): export SUFFIX_OMNICLAUDE_SKILL_LIFECYCLE_DLQ from topics __init__ [OMN-5445] --- src/omnibase_infra/topics/__init__.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/omnibase_infra/topics/__init__.py b/src/omnibase_infra/topics/__init__.py index 4094f3f1b2..233ae37261 100644 --- a/src/omnibase_infra/topics/__init__.py +++ b/src/omnibase_infra/topics/__init__.py @@ -63,6 +63,7 @@ SUFFIX_OMNICLAUDE_AUDIT_RETURN_BOUNDED, SUFFIX_OMNICLAUDE_AUDIT_SCOPE_VIOLATION, SUFFIX_OMNICLAUDE_CONTEXT_AUDIT_DLQ, + SUFFIX_OMNICLAUDE_SKILL_LIFECYCLE_DLQ, SUFFIX_OMNIINTELLIGENCE_ROUTING_DECISION_CMD, SUFFIX_OMNIMEMORY_ARCHIVE_MEMORY, SUFFIX_OMNIMEMORY_CRAWL_REQUESTED, @@ -194,6 +195,7 @@ "SUFFIX_OMNICLAUDE_AUDIT_RETURN_BOUNDED", "SUFFIX_OMNICLAUDE_AUDIT_COMPRESSION_TRIGGERED", "SUFFIX_OMNICLAUDE_CONTEXT_AUDIT_DLQ", + "SUFFIX_OMNICLAUDE_SKILL_LIFECYCLE_DLQ", # Validation domain topic suffix constants (OMN-5050) "SUFFIX_VALIDATION_CROSS_REPO_RUN_STARTED", "SUFFIX_VALIDATION_CROSS_REPO_VIOLATIONS_BATCH", From 49bc385cf73289a74a6c015f1f5781a37aa19861 Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Thu, 19 Mar 2026 11:58:57 -0400 Subject: [PATCH 4/4] fix(test): add SUFFIX_OMNICLAUDE_SKILL_LIFECYCLE_DLQ to 3-partition whitelist [OMN-5445] --- tests/unit/topics/test_platform_topic_suffixes.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/unit/topics/test_platform_topic_suffixes.py b/tests/unit/topics/test_platform_topic_suffixes.py index ec5b3104a3..dc5924d866 100644 --- a/tests/unit/topics/test_platform_topic_suffixes.py +++ b/tests/unit/topics/test_platform_topic_suffixes.py @@ -505,11 +505,13 @@ def test_omniclaude_skill_topics_use_1_partition(self) -> None: SUFFIX_OMNICLAUDE_AUDIT_RETURN_BOUNDED, SUFFIX_OMNICLAUDE_AUDIT_SCOPE_VIOLATION, SUFFIX_OMNICLAUDE_CONTEXT_AUDIT_DLQ, + SUFFIX_OMNICLAUDE_SKILL_LIFECYCLE_DLQ, ) three_partition_suffixes = { SUFFIX_OMNICLAUDE_AGENT_ACTIONS_DLQ, SUFFIX_OMNICLAUDE_AGENT_OBSERVABILITY_DLQ, + SUFFIX_OMNICLAUDE_SKILL_LIFECYCLE_DLQ, # OMN-5445 — skill-lifecycle consumer DLQ SUFFIX_OMNICLAUDE_AGENT_TRACE_FIX_TRANSITION, # Context audit topics (OMN-5240) — observability consumer throughput SUFFIX_OMNICLAUDE_AUDIT_COMPRESSION_TRIGGERED,