feat(kafka): wire session timeout to all consumers [OMN-5445]

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",

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: <repo-relative-path>:<lineno>
 # 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

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(),
         )
 

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

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:

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

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__)
 
 
@@ -76,6 +80,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=(
@@ -131,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. "

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()

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=(

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()

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=(

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()

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(
         ...,

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()

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__)
 
 
@@ -66,6 +70,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=(
@@ -121,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."