[jonahgabriel]

Evidence-Source: OCC#3743
Evidence-Ticket: OMN-14155

Summary

• Daniyal's AWS side for the MSK-IAM cloud dev cutover is DONE (SG on :9098, runtime EC2 instance-profile role, wildcard DescribeTopic). His own framing: "Our runtime doesn't speak AWS_MSK_IAM yet ... a blind flip would break local emission. That's code on my side."
• Recon confirmed the MSK-IAM OAUTHBEARER foundation is already fully built and unit-tested under OMN-2793 (kafka_auth.py, ModelKafkaEventBusConfig, 8 call sites already correctly spread build_aiokafka_auth_kwargs_from_env()), and the CORE runtime bus (EventBusKafka via backends/auto_configure.py) is already fully wired via env-var overrides — zero code change needed there.
• The actual gap: ~14 OTHER direct AIOKafkaConsumer(/AIOKafkaProducer( construction sites bypassed kafka_auth.py entirely, passing only bootstrap_servers= (implicit PLAINTEXT). Several are docker-catalog-DEPLOYED cloud runtime containers (context-audit, llm-cost-aggregation, skill-lifecycle, agent-actions, infra-routing-decisions, consumer-health-projection, injection-effectiveness, waitlist-signup-notifier, plus the registration-orchestrator snapshot producer) that would fail the SASL handshake outright once the broker flips to KAFKA_SASL_MECHANISM=AWS_MSK_IAM.
• This PR spreads the existing build_aiokafka_auth_kwargs_from_env() helper into every remaining site. It is a no-op today (returns {} under PLAINTEXT) and becomes load-bearing only once the Infisical cloud-k8s secrets are cut over (separate, config-only step — see design doc).
• Adds an AST-based CI regression guard (tests/ci/test_no_raw_aiokafka_construction_without_auth.py) so a future direct construction site can't silently skip auth kwargs, with a documented allowlist for cli_kafka.py (pre-existing, self-documented V2 SASL/TLS deferral).

Design doc: docs/plans/2026-07-08-runtime-msk-iam-oauthbearer-design.md
Closes OMN-14155 (parent epic OMN-12986, Done)

Risk flagged (not a code blocker — verify before cutover)

MSKTokenProvider resolves AWS credentials via the default botocore chain (EC2 IMDS for the instance-profile role). This same k3s cluster previously had pod-level IMDS reachability broken by an IMDSv2 hop-limit issue (root cause of the recurring ECR-token-refresh outage, OMN-13749/13818/13819) that may not survive node replacement. Verify aws sts get-caller-identity succeeds from a live onex-dev pod before flipping the Infisical secrets — documented in the design doc §6.

Test plan

• tests/unit/event_bus/test_msk_auth_direct_clients.py — existing MSK auth coverage, unaffected, passes
• New AST-based regression guard passes (tests/ci/test_no_raw_aiokafka_construction_without_auth.py) — both the main invariant test and the anti-permanence allowlist-staleness test
• Existing unit tests for every touched service with coverage (context_audit, infra_routing_decisions, waitlist_signup_notifier, consumer_status/TUI, skill_lifecycle, llm_cost_aggregation, injection_effectiveness, node_registration_orchestrator plugin, snapshot_publisher_registration, node_kafka_replay_compute, session consumer, runner_health CLI, post_merge) — 1058+57 passed, 0 failed, 4 pre-existing unrelated skips
• ruff format --check / ruff check clean on all touched files
• mypy clean on all touched files
• pre-commit run --files <touched> — full hook suite green
• CI (gh pr checks) — verifying after push
• Cutover verification (separate step, after this merges): flip onex-dev Infisical secrets to MSK-IAM, confirm the 9 deployed catalog consumers reach a Stable, non-Empty consumer group against the real MSK cluster (see design doc §8)

[coderabbitai]

Warning

Review limit reached

You’ve reached a temporary PR review limit under our Fair Usage Limits Policy.

Your recent review volume is higher than typical usage, so adaptive limits are currently applied.

Next review available in: 3 minutes

Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available.
You're only billed for reviews past your plan's rate limits ($0.25/file).

How can I continue?

After more reviews become available, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews.

How do review limits work?

CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability.

For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window.

Please refer docs for additional details.

Review details

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: f946594a-3c71-441f-95b9-2e48955e5a50

📥 Commits

Reviewing files that changed from the base of the PR and between 9f66d39 and aba5507.

📒 Files selected for processing (19)

• docker/migrations/forward/nodes/node_projection_delegation/0022_delegation_events_tenant_id.sql
• docker/migrations/forward/nodes/node_projection_savings/080_savings_estimates_tenant_id.sql
• src/omnibase_infra/nodes/node_kafka_replay_compute/contract.yaml
• src/omnibase_infra/nodes/node_kafka_replay_compute/handlers/handler_replay.py
• src/omnibase_infra/nodes/node_registration_orchestrator/plugin.py
• src/omnibase_infra/observability/runner_health/cli_runner_health.py
• src/omnibase_infra/projectors/snapshot_publisher_registration.py
• src/omnibase_infra/services/observability/agent_actions/consumer.py
• src/omnibase_infra/services/observability/consumer_health/consumer.py
• src/omnibase_infra/services/observability/context_audit/consumer.py
• src/omnibase_infra/services/observability/infra_routing_decisions/consumer.py
• src/omnibase_infra/services/observability/injection_effectiveness/consumer.py
• src/omnibase_infra/services/observability/llm_cost_aggregation/consumer.py
• src/omnibase_infra/services/observability/skill_lifecycle/consumer.py
• src/omnibase_infra/services/post_merge/consumer.py
• src/omnibase_infra/services/session/consumer.py
• src/omnibase_infra/services/waitlist_signup_notifier/consumer.py
• src/omnibase_infra/tui/consumers/consumer_status.py
• tests/ci/test_no_raw_aiokafka_construction_without_auth.py

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch jonah/omn-14155-runtime-msk-iam-oauthbearer-wiring

---

_{Comment @coderabbitai help to get the list of available commands.}