fix(OMN-14155): wire MSK-IAM auth kwargs into remaining direct aiokafka construction sites

docker/migrations/forward/nodes/node_projection_delegation/0022_delegation_events_tenant_id.sql

@@ -0,0 +1,25 @@
+-- OMN-14058: add tenant_id to delegation_events.
+--
+-- HandlerProjectionDelegation stamps row["tenant_id"] on delegation_events
+-- writes (interim per-tenant identity threading, OMN-14058) whenever the
+-- source event carries a non-blank tenant identity -- the key is omitted,
+-- never written as NULL, when no tenant is resolved. The column did not
+-- exist on this table: PostgresSyncProjectionAdapter.upsert() builds its
+-- INSERT column list directly from row.keys(), so the first write with
+-- ONEX_TENANT_ID set raised `column "tenant_id" does not exist` against real
+-- Postgres. The in-memory dict adapter used by unit tests has no schema to
+-- violate, so it masked the gap entirely.
+--
+-- DEFAULT 'omninode' mirrors the interim single-tenant convention already
+-- used elsewhere on this projection surface (0019_delegation_budget_state.sql's
+-- DEFAULT_TENANT, and the cloud-only omnidash/db/migrations/0001_tenant_rls.sql,
+-- which applies the same default to a different database via a manual,
+-- operator-run apply plan and never reaches this table). Existing rows and
+-- unstamped writes land under the default; a caller-supplied tenant_id
+-- overrides it on INSERT.
+
+ALTER TABLE delegation_events
+    ADD COLUMN IF NOT EXISTS tenant_id TEXT NOT NULL DEFAULT 'omninode';
+
+CREATE INDEX IF NOT EXISTS idx_delegation_events_tenant_id
+    ON delegation_events (tenant_id);

docker/migrations/forward/nodes/node_projection_savings/080_savings_estimates_tenant_id.sql

@@ -0,0 +1,25 @@
+-- OMN-14058: add tenant_id to savings_estimates.
+--
+-- HandlerProjectionSavings.project_delegate_skill_savings() stamps
+-- row["tenant_id"] whenever the source delegate-skill terminal event carries
+-- a non-blank tenant identity -- the key is omitted, never written as NULL,
+-- when no tenant is resolved. The column did not exist on this table:
+-- PostgresSyncProjectionAdapter.upsert() builds its INSERT column list
+-- directly from row.keys(), so the first write with ONEX_TENANT_ID set
+-- raised `column "tenant_id" does not exist` against real Postgres. The
+-- in-memory dict adapter used by unit tests has no schema to violate, so it
+-- masked the gap entirely.
+--
+-- DEFAULT 'omninode' mirrors the interim single-tenant convention already
+-- used elsewhere on this projection surface (0019_delegation_budget_state.sql's
+-- DEFAULT_TENANT, and the cloud-only omnidash/db/migrations/0001_tenant_rls.sql,
+-- which applies the same default to a different database via a manual,
+-- operator-run apply plan and never reaches this table). Existing rows and
+-- unstamped writes land under the default; a caller-supplied tenant_id
+-- overrides it on INSERT.
+
+ALTER TABLE savings_estimates
+    ADD COLUMN IF NOT EXISTS tenant_id TEXT NOT NULL DEFAULT 'omninode';
+
+CREATE INDEX IF NOT EXISTS idx_savings_estimates_tenant_id
+    ON savings_estimates (tenant_id);

src/omnibase_infra/nodes/node_kafka_replay_compute/contract.yaml

@@ -8,10 +8,10 @@ node_type: "COMPUTE_GENERIC"
 contract_version:
   major: 1
   minor: 0
-  patch: 1
-node_version: "1.0.1"
+  patch: 2
+node_version: "1.0.2"
 description: >
-  Compute node that replays Kafka event-bus envelopes from an explicitly injected target Kafka/Redpanda bootstrap for deterministic projection-equality proofs. The node never reads production bootstrap configuration or mutates runtime .201. Contract patch 1 records handler-internal dataclass annotations only; replay command, output, and transport semantics are unchanged.
+  Compute node that replays Kafka event-bus envelopes from an explicitly injected target Kafka/Redpanda bootstrap for deterministic projection-equality proofs. The node never reads production bootstrap configuration or mutates runtime .201. Contract patch 1 records handler-internal dataclass annotations only; replay command, output, and transport semantics are unchanged. Contract patch 2 records that the default replay consumer spreads the shared aiokafka auth kwargs builder into direct AIOKafkaConsumer construction so MSK-IAM env settings are honored during cloud cutover while remaining a no-op under plaintext local defaults.
 
 input_model:
   name: "ModelKafkaReplayInput"

src/omnibase_infra/nodes/node_kafka_replay_compute/handlers/handler_replay.py

@@ -17,6 +17,7 @@
 
 from omnibase_core.models.events.model_event_envelope import ModelEventEnvelope
 from omnibase_infra.enums import EnumHandlerType, EnumHandlerTypeCategory
+from omnibase_infra.event_bus.kafka_auth import build_aiokafka_auth_kwargs_from_env
 from omnibase_infra.models.projection import ModelSequenceInfo
 from omnibase_infra.nodes.node_kafka_replay_compute.models import (
     ModelKafkaReplayInput,
@@ -49,6 +50,7 @@ def _default_consumer_factory(
         group_id=command.target_consumer_group,
         enable_auto_commit=False,
         auto_offset_reset="earliest",
+        **build_aiokafka_auth_kwargs_from_env(),
     )
     return cast("ProtocolKafkaReplayConsumer", consumer)
 

src/omnibase_infra/nodes/node_registration_orchestrator/plugin.py

@@ -828,6 +828,9 @@ async def _initialize_snapshot_publisher(
         try:
             from aiokafka import AIOKafkaProducer
 
+            from omnibase_infra.event_bus.kafka_auth import (
+                build_aiokafka_auth_kwargs_from_env,
+            )
             from omnibase_infra.models.projection import ModelSnapshotTopicConfig
             from omnibase_infra.projectors.snapshot_publisher_registration import (
                 SnapshotPublisherRegistration,
@@ -836,6 +839,7 @@ async def _initialize_snapshot_publisher(
             snapshot_config = ModelSnapshotTopicConfig.default()
             snapshot_producer = AIOKafkaProducer(
                 bootstrap_servers=kafka_bootstrap_servers,
+                **build_aiokafka_auth_kwargs_from_env(),
             )
             self._snapshot_publisher = SnapshotPublisherRegistration(
                 snapshot_producer,

src/omnibase_infra/observability/runner_health/cli_runner_health.py

@@ -250,6 +250,9 @@ async def _emit_to_kafka(
     try:
         from aiokafka import AIOKafkaProducer
 
+        from omnibase_infra.event_bus.kafka_auth import (
+            build_aiokafka_auth_kwargs_from_env,
+        )
         from omnibase_infra.topics.platform_topic_suffixes import (
             SUFFIX_RUNNER_HEALTH_SNAPSHOT,
         )
@@ -262,6 +265,7 @@ async def _emit_to_kafka(
         producer = AIOKafkaProducer(
             bootstrap_servers=bootstrap,
             value_serializer=lambda v: json.dumps(v).encode(),
+            **build_aiokafka_auth_kwargs_from_env(),
         )
         await producer.start()
         try:

src/omnibase_infra/projectors/snapshot_publisher_registration.py

@@ -906,6 +906,10 @@ async def _load_cache_from_topic(self, correlation_id: UUID) -> None:
             from aiokafka import AIOKafkaConsumer
             from pydantic import ValidationError
 
+            from omnibase_infra.event_bus.kafka_auth import (
+                build_aiokafka_auth_kwargs_from_env,
+            )
+
             # Create consumer with unique group ID for this publisher instance
             # Using a unique group ensures we get our own offset tracking
             consumer_group = f"snapshot-reader-{self._config.topic}-{uuid4()!s}"
@@ -918,6 +922,7 @@ async def _load_cache_from_topic(self, correlation_id: UUID) -> None:
                 session_timeout_ms=_KAFKA_SESSION_TIMEOUT_MS,
                 heartbeat_interval_ms=_KAFKA_HEARTBEAT_INTERVAL_MS,
                 max_poll_interval_ms=_KAFKA_MAX_POLL_INTERVAL_MS,
+                **build_aiokafka_auth_kwargs_from_env(),
             )
 
             try:

src/omnibase_infra/services/observability/agent_actions/consumer.py

@@ -74,6 +74,7 @@
 from omnibase_core.errors import OnexError
 from omnibase_core.types import JsonType
 from omnibase_infra.event_bus.consumer_health_emitter import ConsumerHealthEmitter
+from omnibase_infra.event_bus.kafka_auth import build_aiokafka_auth_kwargs_from_env
 from omnibase_infra.event_bus.mixin_consumer_health import MixinConsumerHealth
 from omnibase_infra.services.observability.agent_actions.config import (
     ConfigAgentActionsConsumer,
@@ -712,6 +713,7 @@ async def start(self) -> None:
                 heartbeat_interval_ms=self._config.heartbeat_interval_ms,
                 max_poll_interval_ms=self._config.max_poll_interval_ms,
                 max_poll_records=self._config.batch_size,
+                **build_aiokafka_auth_kwargs_from_env(),
             )
 
             await self._consumer.start()
@@ -729,6 +731,7 @@ async def start(self) -> None:
             if self._config.dlq_enabled:
                 self._dlq_producer = AIOKafkaProducer(
                     bootstrap_servers=self._config.kafka_bootstrap_servers,
+                    **build_aiokafka_auth_kwargs_from_env(),
                 )
                 await self._dlq_producer.start()
                 logger.info(
@@ -753,6 +756,7 @@ async def start(self) -> None:
                 if health_prod is None:
                     self._health_producer = AIOKafkaProducer(
                         bootstrap_servers=self._config.kafka_bootstrap_servers,
+                        **build_aiokafka_auth_kwargs_from_env(),
                     )
                     await self._health_producer.start()
                     health_prod = self._health_producer

src/omnibase_infra/services/observability/consumer_health/consumer.py

@@ -47,6 +47,7 @@
 from aiokafka import AIOKafkaConsumer
 from aiokafka.errors import KafkaError
 
+from omnibase_infra.event_bus.kafka_auth import build_aiokafka_auth_kwargs_from_env
 from omnibase_infra.services.observability.consumer_health.config import (
     ConfigConsumerHealthProjection,
 )
@@ -151,6 +152,7 @@ async def start(self) -> None:
             heartbeat_interval_ms=self._config.heartbeat_interval_ms,
             max_poll_interval_ms=self._config.max_poll_interval_ms,
             value_deserializer=lambda v: json.loads(v.decode("utf-8")) if v else None,
+            **build_aiokafka_auth_kwargs_from_env(),
         )
         await self._consumer.start()
         logger.info(

src/omnibase_infra/services/observability/context_audit/consumer.py

@@ -65,6 +65,7 @@
 from aiokafka import AIOKafkaConsumer, AIOKafkaProducer, TopicPartition
 from aiokafka.errors import KafkaError
 
+from omnibase_infra.event_bus.kafka_auth import build_aiokafka_auth_kwargs_from_env
 from omnibase_infra.services.observability.context_audit.config import (
     ConfigContextAuditConsumer,
 )
@@ -327,6 +328,7 @@ async def start(self) -> None:
             heartbeat_interval_ms=self.config.heartbeat_interval_ms,
             max_poll_interval_ms=self.config.max_poll_interval_ms,
             value_deserializer=lambda v: v,
+            **build_aiokafka_auth_kwargs_from_env(),
         )
         await self._consumer.start()
 
@@ -335,6 +337,7 @@ async def start(self) -> None:
             self._producer = AIOKafkaProducer(
                 bootstrap_servers=self.config.kafka_bootstrap_servers,
                 value_serializer=lambda v: v,
+                **build_aiokafka_auth_kwargs_from_env(),
             )
             await self._producer.start()
 

src/omnibase_infra/services/observability/infra_routing_decisions/consumer.py

@@ -43,6 +43,7 @@
 from aiokafka import AIOKafkaConsumer, AIOKafkaProducer, TopicPartition
 from aiokafka.errors import KafkaError
 
+from omnibase_infra.event_bus.kafka_auth import build_aiokafka_auth_kwargs_from_env
 from omnibase_infra.services.observability.infra_routing_decisions.config import (
     ConfigInfraRoutingDecisionsConsumer,
 )
@@ -286,13 +287,15 @@ async def start(self) -> None:
             heartbeat_interval_ms=self.config.heartbeat_interval_ms,
             max_poll_interval_ms=self.config.max_poll_interval_ms,
             value_deserializer=lambda v: v,
+            **build_aiokafka_auth_kwargs_from_env(),
         )
         await self._consumer.start()
 
         if self.config.dlq_enabled:
             self._producer = AIOKafkaProducer(
                 bootstrap_servers=self.config.kafka_bootstrap_servers,
                 value_serializer=lambda v: v,
+                **build_aiokafka_auth_kwargs_from_env(),
             )
             await self._producer.start()
 

src/omnibase_infra/services/observability/injection_effectiveness/consumer.py

@@ -73,6 +73,7 @@
 from omnibase_core.errors import OnexError
 from omnibase_core.types import JsonType
 from omnibase_infra.event_bus.consumer_health_emitter import ConsumerHealthEmitter
+from omnibase_infra.event_bus.kafka_auth import build_aiokafka_auth_kwargs_from_env
 from omnibase_infra.event_bus.mixin_consumer_health import MixinConsumerHealth
 from omnibase_infra.services.observability.injection_effectiveness.config import (
     ConfigInjectionEffectivenessConsumer,
@@ -548,6 +549,7 @@ async def start(self) -> None:
                 heartbeat_interval_ms=self._config.heartbeat_interval_ms,
                 max_poll_interval_ms=self._config.max_poll_interval_ms,
                 max_poll_records=self._config.batch_size,
+                **build_aiokafka_auth_kwargs_from_env(),
             )
 
             await self._consumer.start()
@@ -571,6 +573,7 @@ async def start(self) -> None:
             if ConsumerHealthEmitter.is_enabled():
                 self._health_producer = AIOKafkaProducer(
                     bootstrap_servers=self._config.kafka_bootstrap_servers,
+                    **build_aiokafka_auth_kwargs_from_env(),
                 )
                 await self._health_producer.start()
                 self._init_health_emitter(

src/omnibase_infra/services/observability/llm_cost_aggregation/consumer.py

@@ -74,6 +74,7 @@
 from omnibase_infra.enums import EnumInfraTransportType
 from omnibase_infra.errors import InfraConnectionError, ModelInfraErrorContext
 from omnibase_infra.event_bus.consumer_health_emitter import ConsumerHealthEmitter
+from omnibase_infra.event_bus.kafka_auth import build_aiokafka_auth_kwargs_from_env
 from omnibase_infra.event_bus.mixin_consumer_health import MixinConsumerHealth
 from omnibase_infra.services.observability.llm_cost_aggregation.config import (
     ConfigLlmCostAggregation,
@@ -480,6 +481,7 @@ async def start(self) -> None:
                 heartbeat_interval_ms=self._config.heartbeat_interval_ms,
                 max_poll_interval_ms=self._config.max_poll_interval_ms,
                 max_poll_records=self._config.batch_size,
+                **build_aiokafka_auth_kwargs_from_env(),
             )
 
             await self._consumer.start()
@@ -494,6 +496,7 @@ async def start(self) -> None:
             if ConsumerHealthEmitter.is_enabled():
                 self._health_producer = AIOKafkaProducer(
                     bootstrap_servers=self._config.kafka_bootstrap_servers,
+                    **build_aiokafka_auth_kwargs_from_env(),
                 )
                 await self._health_producer.start()
                 self._init_health_emitter(

src/omnibase_infra/services/observability/skill_lifecycle/consumer.py

@@ -61,6 +61,7 @@
 from aiokafka import AIOKafkaConsumer, AIOKafkaProducer, TopicPartition
 from aiokafka.errors import KafkaError
 
+from omnibase_infra.event_bus.kafka_auth import build_aiokafka_auth_kwargs_from_env
 from omnibase_infra.services.observability.skill_lifecycle.config import (
     ConfigSkillLifecycleConsumer,
 )
@@ -355,6 +356,7 @@ async def start(self) -> None:
             heartbeat_interval_ms=self.config.heartbeat_interval_ms,
             max_poll_interval_ms=self.config.max_poll_interval_ms,
             value_deserializer=lambda v: v,
+            **build_aiokafka_auth_kwargs_from_env(),
         )
         await self._consumer.start()
 
@@ -363,6 +365,7 @@ async def start(self) -> None:
             self._producer = AIOKafkaProducer(
                 bootstrap_servers=self.config.kafka_bootstrap_servers,
                 value_serializer=lambda v: v,
+                **build_aiokafka_auth_kwargs_from_env(),
             )
             await self._producer.start()
 

src/omnibase_infra/services/post_merge/consumer.py

@@ -68,6 +68,7 @@
 from aiokafka.errors import KafkaError
 from pydantic import ValidationError
 
+from omnibase_infra.event_bus.kafka_auth import build_aiokafka_auth_kwargs_from_env
 from omnibase_infra.models.github.model_pr_merged_event import ModelPRMergedEvent
 from omnibase_infra.services.post_merge.checks import (
     run_contract_sweep,
@@ -160,12 +161,14 @@ async def start(self) -> None:
             auto_offset_reset=self._config.auto_offset_reset,
             enable_auto_commit=False,
             value_deserializer=lambda v: json.loads(v.decode("utf-8")),
+            **build_aiokafka_auth_kwargs_from_env(),
         )
         await self._consumer.start()
 
         self._producer = AIOKafkaProducer(
             bootstrap_servers=self._config.kafka_bootstrap_servers,
             value_serializer=lambda v: json.dumps(v, default=str).encode("utf-8"),
+            **build_aiokafka_auth_kwargs_from_env(),
         )
         await self._producer.start()
 

src/omnibase_infra/services/session/consumer.py

@@ -65,6 +65,7 @@
 from pydantic import ValidationError
 
 from omnibase_infra.event_bus.consumer_health_emitter import ConsumerHealthEmitter
+from omnibase_infra.event_bus.kafka_auth import build_aiokafka_auth_kwargs_from_env
 from omnibase_infra.event_bus.mixin_consumer_health import MixinConsumerHealth
 from omnibase_infra.services.session.config_consumer import ConfigSessionConsumer
 from omnibase_infra.services.session.protocol_session_aggregator import (
@@ -374,6 +375,7 @@ async def start(self) -> None:
                 heartbeat_interval_ms=self._config.heartbeat_interval_ms,
                 max_poll_interval_ms=self._config.max_poll_interval_ms,
                 max_poll_records=self._config.max_poll_records,
+                **build_aiokafka_auth_kwargs_from_env(),
             )
 
             await self._consumer.start()
@@ -384,6 +386,7 @@ async def start(self) -> None:
             if ConsumerHealthEmitter.is_enabled():
                 self._health_producer = AIOKafkaProducer(
                     bootstrap_servers=self._config.bootstrap_servers,
+                    **build_aiokafka_auth_kwargs_from_env(),
                 )
                 await self._health_producer.start()
                 self._init_health_emitter(

src/omnibase_infra/services/waitlist_signup_notifier/consumer.py

@@ -32,6 +32,7 @@
 from aiohttp import web
 from aiokafka import AIOKafkaConsumer
 
+from omnibase_infra.event_bus.kafka_auth import build_aiokafka_auth_kwargs_from_env
 from omnibase_infra.handlers.handler_slack_webhook import HandlerSlackWebhook
 from omnibase_infra.handlers.models.model_slack_alert import (
     EnumAlertSeverity,
@@ -69,6 +70,7 @@ async def start(self) -> None:
             auto_offset_reset="latest",
             enable_auto_commit=True,
             value_deserializer=lambda v: json.loads(v.decode("utf-8")),
+            **build_aiokafka_auth_kwargs_from_env(),
         )
         await self._consumer.start()
         self._healthy = True

src/omnibase_infra/tui/consumers/consumer_status.py

@@ -32,6 +32,7 @@
 if TYPE_CHECKING:
     from omnibase_core.types import JsonType
 
+from omnibase_infra.event_bus.kafka_auth import build_aiokafka_auth_kwargs_from_env
 from omnibase_infra.topics.platform_topic_suffixes import (
     SUFFIX_GIT_HOOK,
     SUFFIX_GITHUB_PR_STATUS,
@@ -127,6 +128,7 @@ 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,
+            **build_aiokafka_auth_kwargs_from_env(),
         )
         await consumer.start()
         logger.info("TUI consumer started. Topics: %s.", _ALL_TOPICS)