Skip to content

fix(OMN-15700): derive savings consumer groups canonically - #2681

Merged
jonahgabriel merged 6 commits into
devfrom
daniyal/omn-15700-savings-consumer-group
Aug 10, 2026
Merged

jonahgabriel merged 6 commits into
devfrom
daniyal/omn-15700-savings-consumer-group

Conversation

@daniyalabbas96

@daniyalabbas96 daniyalabbas96 commented Aug 8, 2026 •

Copy link
Copy Markdown
Contributor

Evidence-Ticket: OMN-15700

Summary

Replace the savings estimator's explicit savings-estimator.<topic> Kafka consumer-group override with the runtime's typed ModelNodeIdentity derivation path. This keeps the group environment-qualified and inside the existing onex-dev.* MSK IAM boundary; no IAM widening is included.

Changes

  • define a stable savings-estimator node identity from the live runtime environment and service contract
  • pass that identity to every configured input subscription and let the event bus apply its canonical per-topic discriminator
  • replace the previous literal-group AST assertion with a default-deny guard for this call site
  • exercise the identity-derived path through the real in-memory event bus

Test plan

  • Focused savings tests: 18 passed
  • Effective-group regression: every configured topic resolves to a distinct .__t.<topic> broker-facing group
  • Kafka consumer-group/event-bus suite: 230 passed
  • Ruff check and format: clean
  • mypy on all touched files: clean
  • AWS readback: session valid and managed Kafka cluster ACTIVE
  • Post-deploy proof: consume all three live inputs without GroupAuthorizationFailedError and show the savings projection advancing

Runtime startup gate

This touches service_kernel.py only at an optional Kafka subscription call site; it does not change contract discovery, handler construction, wire_from_manifest, or readiness enforcement.

  • Test loads the real contract manifest from disk (not changed by this PR)
  • Test calls wire_from_manifest with production arguments (not changed by this PR)
  • Test asserts zero wiring failures (not changed by this PR)
  • CI boots omninode-runtime in its governed runtime checks

Type safety checklist

  • No new untyped metadata access
  • No new dict[str, Any] fields
  • No new exception boundaries
  • No metadata keys added
  • No compose service or environment changes

Strict-gate ordering

  • No strict or blocking gate introduced

Scope and residual proof

This PR is code-only evidence. After Jonah reviews and merges it, the staging runtime must be rebuilt/deployed. OMN-15700 remains In Progress until the three live inputs are exercised and the downstream cost/savings projection advances. No AWS, IAM, MSK topic, or Kubernetes mutation is part of this PR.

Refs OMN-15700

Summary by CodeRabbit

Bug Fixes

  • Improved reliability of savings estimation event processing through consistent service and runtime identification.
  • Updated subscription handling to prevent topic-specific settings from causing unintended processing conflicts.
  • Ensured events from different savings-related topics continue to be processed independently.
  • Improved consistency when processing savings estimation events across runtime environments.

Tests

  • Expanded automated coverage for subscription behavior, identity consistency, and independent topic processing.

Consumer-group migration behavior

The former savings-estimator.<topic> groups never passed MSK authorization, so they have no usable committed offsets to transfer. The runtime bus defaults a newly derived group to latest; the deployment proof will therefore use fresh, uniquely identified post-rollout events on each of the three inputs rather than replaying history or resetting offsets manually. The estimator's source event ID is deterministic from session ID and schema version, and its finalized-session cache suppresses duplicate finalization during the bounded proof.

Evidence-Source: OCC#6261

@coderabbitai

coderabbitai Bot commented Aug 8, 2026 •

Copy link
Copy Markdown

Review Change Stack

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info
⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: 48c1c8ad-74ea-43ec-aa6e-9bf3800681c9

📥 Commits

Reviewing files that changed from the base of the PR and between ae8c7dd and 3ece9a8.

📒 Files selected for processing (1)
  • tests/integration/runtime/test_savings_estimator_subscription_group.py
🚧 Files skipped from review as they are similar to previous changes (1)
  • tests/integration/runtime/test_savings_estimator_subscription_group.py

📝 Walkthrough

Walkthrough

The savings estimator subscription now uses a typed ModelNodeIdentity containing runtime environment, service, node, and version metadata. The subscription no longer passes a topic-specific group_id. Unit and integration tests verify the updated contract and distinct topic consumer groups.

Changes

Savings estimator subscription identity

Layer / File(s) Summary
Create and pass node identity
src/omnibase_infra/runtime/service_kernel.py
The runtime creates ModelNodeIdentity for the savings estimator and passes it to the subscription instead of group_id.
Validate subscription metadata and consumer groups
tests/integration/runtime/test_savings_estimator_subscription_group.py, tests/unit/services/observability/savings_estimation/test_consumer.py
Tests verify the identity fields, confirm that subscriptions use node_identity without group_id, and verify distinct topic-qualified consumer groups.

Estimated code review effort: 2 (Simple) | ~10 minutes

Suggested reviewers: jonahgabriel

🚥 Pre-merge checks | ✅ 5
✅ Passed checks (5 passed)
Check name Status Explanation
Description Check ✅ Passed Check skipped - CodeRabbit’s high-level summary is enabled.
Title check ✅ Passed The title clearly and concisely describes the canonical derivation of savings consumer groups, which is the main change.
Docstring Coverage ✅ Passed No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.
Linked Issues check ✅ Passed Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check ✅ Passed Check skipped because no linked issues were found for this pull request.
✨ Finishing Touches
📝 Generate docstrings
  • Create stacked PR
  • Commit on current branch
🧪 Generate unit tests (beta)
  • Create PR with unit tests
  • Commit unit tests in branch daniyal/omn-15700-savings-consumer-group

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

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

@coderabbitai coderabbitai Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

🧹 Nitpick comments (2)
tests/integration/runtime/test_savings_estimator_subscription_group.py (1)

45-70: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick win

Strengthen the AST guard for the complete subscription contract.

The test only checks that service and version keywords exist. A wrong value still passes and changes the canonical group identity. The call search is module-wide and checks only calls that already contain _savings_node_identity; a second savings subscription could retain group_id without being detected. Scope the search to the _input_topic loop and assert the intended service and version values.

🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/integration/runtime/test_savings_estimator_subscription_group.py`
around lines 45 - 70, Strengthen the AST assertions around the _input_topic
loop: scope savings_subscribe_calls to subscribe calls within that loop rather
than searching module-wide by _savings_node_identity alone, assert exactly one
matching subscription, and validate its service and version keyword values
against the intended canonical values instead of only checking their presence.
Preserve the existing node_identity and group_id assertions.
src/omnibase_infra/runtime/service_kernel.py (1)

1710-1711: 🗄️ Data Integrity & Integration | 🔵 Trivial

Plan the consumer-group migration before rollout.

This change removes the explicit savings-estimator.<topic> groups. Unless the resolver aliases the old values, the canonical identity creates new Kafka group IDs, so existing offsets do not transfer automatically. Confirm offset reset, replay, and idempotency behavior for all three input topics.

🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/omnibase_infra/runtime/service_kernel.py` around lines 1710 - 1711,
Before rolling out the consumer-group identity change around node_identity and
on_message, verify how the resolver maps the former savings-estimator.<topic>
groups to the canonical identities. Define and validate the migration plan for
all three input topics, including offset reset, replay, and idempotent
processing behavior; preserve existing group IDs or explicitly handle offset
transfer where required.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Nitpick comments:
In `@src/omnibase_infra/runtime/service_kernel.py`:
- Around line 1710-1711: Before rolling out the consumer-group identity change
around node_identity and on_message, verify how the resolver maps the former
savings-estimator.<topic> groups to the canonical identities. Define and
validate the migration plan for all three input topics, including offset reset,
replay, and idempotent processing behavior; preserve existing group IDs or
explicitly handle offset transfer where required.

In `@tests/integration/runtime/test_savings_estimator_subscription_group.py`:
- Around line 45-70: Strengthen the AST assertions around the _input_topic loop:
scope savings_subscribe_calls to subscribe calls within that loop rather than
searching module-wide by _savings_node_identity alone, assert exactly one
matching subscription, and validate its service and version keyword values
against the intended canonical values instead of only checking their presence.
Preserve the existing node_identity and group_id assertions.

ℹ️ Review info
⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: 8c39dee0-667a-4129-bb99-9f961023b138

📥 Commits

Reviewing files that changed from the base of the PR and between 4bf964a and 12b2f25.

📒 Files selected for processing (3)
  • src/omnibase_infra/runtime/service_kernel.py
  • tests/integration/runtime/test_savings_estimator_subscription_group.py
  • tests/unit/services/observability/savings_estimation/test_consumer.py

@daniyalabbas96

Copy link
Copy Markdown
Contributor Author

Addressed both review notes in head bc14b30 and the PR handoff: the AST guard is now scoped to the actual _input_topic loop and pins the full env/service/node/version identity contract. The rollout section now records that the denied legacy groups have no transferable committed offsets, the new groups start at the runtime's latest policy, and acceptance uses fresh unique sessions with deterministic source IDs and finalized-session deduplication. Focused guard tests, Ruff, and mypy remain green.

@jonahgabriel

Copy link
Copy Markdown
Collaborator

Temporarily closed to relieve CI runner-fleet congestion (merge-sweep 2026-08-08, operator-approved — see ROLLING_WORK_LEDGER [mergesweep-0808-close]). Nothing is lost: branch, commits, and review state remain; this PR will be reopened in batches once the queue drains (reopen order recorded in the ledger).

auto-merge was automatically disabled August 8, 2026 10:12

Pull request was closed

@daniyalabbas96 daniyalabbas96 reopened this Aug 8, 2026

@coderabbitai coderabbitai Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Actionable comments posted: 1

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)
tests/integration/runtime/test_savings_estimator_subscription_group.py (1)

80-95: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick win

Check every subscribe call in the loop.

Lines [80-89] exclude calls that do not already contain node_identity. A legacy call with only group_id is therefore invisible to Lines [92-95]. Collect all subscribe calls first, then assert that every call uses _savings_node_identity and omits group_id.

Suggested guard change
     savings_subscribe_calls = [
         node
         for node in ast.walk(savings_input_loops[0])
         if isinstance(node, ast.Call) and _call_name(node.func) == "subscribe"
-        if any(
-            keyword.arg == "node_identity"
-            and isinstance(keyword.value, ast.Name)
-            and keyword.value.id == "_savings_node_identity"
-            for keyword in node.keywords
-        )
     ]

     assert len(savings_subscribe_calls) == 1
     assert all(
-        keyword.arg != "group_id" for keyword in savings_subscribe_calls[0].keywords
+        any(
+            keyword.arg == "node_identity"
+            and isinstance(keyword.value, ast.Name)
+            and keyword.value.id == "_savings_node_identity"
+            for keyword in node.keywords
+        )
+        and all(keyword.arg != "group_id" for keyword in node.keywords)
+        for node in savings_subscribe_calls
     )
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/integration/runtime/test_savings_estimator_subscription_group.py`
around lines 80 - 95, Update the savings_subscribe_calls collection to include
every subscribe ast.Call within savings_input_loops[0], without filtering on
node_identity. Then assert all collected calls use the _savings_node_identity
Name for node_identity and omit group_id, while preserving the existing
single-call assertion if required by the test.
🧹 Nitpick comments (1)
tests/integration/runtime/test_savings_estimator_subscription_group.py (1)

21-21: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick win

Anchor SERVICE_KERNEL_PATH to __file__.

Line [21] depends on the current working directory. The test can fail when pytest starts outside the repository root. Build the repository path from __file__.

Suggested path fix
-SERVICE_KERNEL_PATH = Path("src/omnibase_infra/runtime/service_kernel.py")
+REPO_ROOT = Path(__file__).parent.parent.parent.parent
+SERVICE_KERNEL_PATH = REPO_ROOT / "src/omnibase_infra/runtime/service_kernel.py"

Based on learnings, integration tests in this repository use lexical Path(__file__).parent... traversal without .resolve().

🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/integration/runtime/test_savings_estimator_subscription_group.py` at
line 21, Update the SERVICE_KERNEL_PATH definition to derive the
repository-relative path from __file__ using the existing lexical
Path(__file__).parent traversal style, rather than relying on the current
working directory; preserve the target service_kernel.py location.

Source: Learnings

🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@tests/integration/runtime/test_savings_estimator_subscription_group.py`:
- Around line 111-125: Update the test around effective_group_ids to exercise
EventBusKafka.subscribe rather than calling the private
_resolve_effective_group_id directly. Provide a ModelNodeIdentity and a spy or
test double for the subscription registration, invoke subscribe for each
savings_config.consumed_topics topic, and assert that each registration uses the
expected effective subscription group without requiring a real broker.

---

Outside diff comments:
In `@tests/integration/runtime/test_savings_estimator_subscription_group.py`:
- Around line 80-95: Update the savings_subscribe_calls collection to include
every subscribe ast.Call within savings_input_loops[0], without filtering on
node_identity. Then assert all collected calls use the _savings_node_identity
Name for node_identity and omit group_id, while preserving the existing
single-call assertion if required by the test.

---

Nitpick comments:
In `@tests/integration/runtime/test_savings_estimator_subscription_group.py`:
- Line 21: Update the SERVICE_KERNEL_PATH definition to derive the
repository-relative path from __file__ using the existing lexical
Path(__file__).parent traversal style, rather than relying on the current
working directory; preserve the target service_kernel.py location.
🪄 Autofix

Fix all unresolved CodeRabbit comments on this PR:

  • Push a commit to this branch (recommended)
  • Create a new PR with the fixes

ℹ️ Review info
⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: d53f584f-1203-4def-9c68-ccfa033659db

📥 Commits

Reviewing files that changed from the base of the PR and between 12b2f25 and 13cce04.

📒 Files selected for processing (1)
  • tests/integration/runtime/test_savings_estimator_subscription_group.py

Comment thread tests/integration/runtime/test_savings_estimator_subscription_group.py Outdated
@daniyalabbas96

Copy link
Copy Markdown
Contributor Author

Correction applied after independent review: Evidence-Source now points to OCC#6192, the deploy assessment is a falsifiable exact-head content probe, and both Receipt Gate and the rerun Deploy Gate pass. I also added a behavioral regression at head 13cce04 proving every configured savings topic receives a distinct broker-facing .__t. group. OMN-15700 remains In Progress pending trusted OCC review, merge/deploy, and the live MSK/projection proof.

@github-actions

github-actions Bot commented Aug 8, 2026 •

Copy link
Copy Markdown
Contributor

✅ Hostile Reviewer — PASSED

Blocking findings (critical): 0
Total findings: 0
Models succeeded: qwen3-review,qwen3-review-b


Gate semantics (pilot phase)

Verdict Meaning Blocks merge?
passed No critical findings No
blocked CRITICAL findings found Yes
degraded All models unavailable (infra) No (pilot)

Powered by omniintelligence.review_pairing.cli_review — node-based adversarial review via HandlerLlmCliSubprocess (OMN-8468/OMN-8524)

@daniyalabbas96

Copy link
Copy Markdown
Contributor Author

Follow-up: addressed CodeRabbit's valid test-quality finding. Head 0e21e0c now exercises the public subscribe(node_identity=...) path for every configured savings topic and captures the effective group at the consumer-start boundary. Focused integration tests pass 3/3; Ruff and mypy pass. OCC#6192 is refreshed through fed650ea for this exact product head.

@jonahgabriel

Copy link
Copy Markdown
Collaborator

Heads up: this branch is now behind dev past the OMN-15732 migration-vendor deadlock fix (211e81e), which superseded the vendored node_canary_score_reducer/0003 migration + the application-migrations TSV/manifest count (now 96). Recommend rebasing onto origin/dev and taking dev's bytes for the vendored migration file/TSV/manifest-count on conflict, keeping your own PR content — this is not something we'll force-push on your behalf.

Advances the merge base to origin/dev @ 303e008 to unblock PR #2681
(OMN-15700) on the required CI Summary external-context check.

The PR previously pinned merge SHA a90460f replayed over the Aug-6
base 4bf964a, which contained neither:
  - PR #2700 (dev squash 6b6cdc1): monitor_logs test timeout bump 10s -> 30s
  - PR #2714 (303e008): measured-comment fix for the same test

Merging origin/dev in (not rebasing) so 0e21e0c
stays in ancestry, keeping the prior hostile-review approval at that
exact head citable against an unrewritten commit.

No conflicts: git merge --no-commit --no-ff reported "Automatic merge
went well" with a single auto-merged file
(src/omnibase_infra/runtime/service_kernel.py, the savings-estimator
seam file) and zero UU entries / conflict markers.

Refs OMN-15700
…set_override seam

PR #2681 (OMN-15700) landed on top of dev's #2705 (OMN-15789), which
made EventBusKafka.subscribe() unconditionally pass
auto_offset_reset_override= into _start_consumer_for_topic_unlocked().
The contractor's fake capture_consumer_start(topic, group_id) in
test_savings_estimator_topics_keep_distinct_effective_consumer_groups
predates that seam and only accepted the original two positional
params, so the keyword-only argument raised TypeError deterministically
in CI (Tests Split 3/15) — a seam collision between two individually
correct, individually green PRs (OMN-14208 class), not a defect in
either.

Widen the fake's signature to field-match the real
_start_consumer_for_topic_unlocked seam exactly (name + keyword-only +
default None) and record the captured value alongside topic/group_id
so the double mirrors the seam instead of swallowing it. No production
code touched; the test's consumer-group-distinctness assertions are
unchanged.
@jonahgabriel
jonahgabriel merged commit d098bc0 into dev Aug 10, 2026
141 of 179 checks passed
@jonahgabriel
jonahgabriel deleted the daniyal/omn-15700-savings-consumer-group branch August 10, 2026 17:33
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants