fix(OMN-13149): savings consumer decodes typed ModelEventMessage instead of calling .get()

[jonahgabriel]

OMN-13149 — savings_estimation consumer AttributeError (ModelEventMessage has no .get())

Problem

The savings-estimation consumer is wired in service_kernel.py via event_bus.subscribe. Both the Kafka bus (EventBusKafka) and the in-memory bus (EventBusInmemory) deliver a typed ModelEventMessage to the on_message callback — never a raw dict or str (callback signature is Callable[[ModelEventMessage], Awaitable[None]]).

The pre-fix _savings_on_message did:

_json.loads(msg) if isinstance(msg, (str, bytes)) else msg

A ModelEventMessage is neither str nor bytes, so it fell into the else msg branch and was passed straight to ServiceSavingsEstimator.ingest_event(topic, payload). The correlation path _session_id_for_topic then calls payload.get("session_id"). ModelEventMessage is a frozen Pydantic model with no .get(), so this raised at runtime:

AttributeError: 'ModelEventMessage' object has no attribute 'get'

The subscriber callback crashed on every consumed event → the consumer correlated nothing and wrote zero rows. Same dict/typed-model family as OMN-13141 / golden-chain BUG A.

The repo's own runtime-error-triage consumer in the same kernel already showed the correct pattern: it reads message.value and json.loads it. The savings callback never decoded .value.

Fix

• New module-level decode_event_message(message: ModelEventMessage) -> tuple[str, dict[str, object]] in consumer.py reads the JSON payload off the typed .value field directly — strongly typed, fail-fast: raises TypeError if the decoded payload is not a JSON object. No getattr-default, no dict-style fallback, no silent guard.
• service_kernel._savings_on_message now takes message: ModelEventMessage, calls decode_event_message(message), then ingest_event(topic, payload).
• Removed the now-redundant local ModelEventMessage import in the runtime-error-triage consumer (module-level import added).

decode_event_message is a module-level function (not a class method) so ServiceSavingsEstimator stays at 10 methods — keeping the ONEX Pattern Validation gate green rather than tripping the per-class method-count threshold.

Regression tests (REAL consumer path, not handler isolation)

tests/unit/services/observability/savings_estimation/test_consumer.py:

1. test_ingest_event_rejects_typed_message_documents_bug — pins the pre-fix crash: a typed ModelEventMessage straight into ingest_event raises AttributeError: ...has no attribute 'get'.
2. test_decode_event_message_correlates_typed_message — decode + ingest_event correlates a session.
3. test_decode_event_message_rejects_non_object_payload — non-object JSON fails fast (TypeError).
4. test_real_bus_delivers_typed_message_to_consumer — end-to-end through the real EventBusInmemory (subscribe/publish → on_message(ModelEventMessage) → decode_event_message → ingest_event), mirroring the kernel callback. Reproduced failing-first: with the pre-fix decode the callback raises the exact AttributeError inside EventBusInmemory.publish; with the fix the session is correlated.

These run under @pytest.mark.unit (no infra markers) → wired as a pre-merge CI gate.

Verification (local, in worktree)

• uv run ruff format src/ tests/ && uv run ruff check --fix src/ tests/ — clean
• uv run mypy src/omnibase_infra/services/observability/savings_estimation/consumer.py src/omnibase_infra/runtime/service_kernel.py --strict — Success, no issues
• uv run pytest tests/unit/services/observability/savings_estimation/test_consumer.py -q — 16 passed
• uv run pytest tests/unit/runtime/ tests/unit/services/observability/savings_estimation/ -q — 4902 passed, 5 skipped (pre-existing)
• uv run pytest tests/unit/ -m "not slow and not kafka and not postgres and not consul" — 20472 passed, 1 failure (tests/unit/verification/test_cli.py::test_registration_only, pre-existing — fails identically on clean origin/dev, unrelated to this change)
• pre-commit run on changed files — all hooks pass (incl. ONEX Pattern Validation)

dod_evidence

Per contracts/OMN-13149.yaml (this PR): dod-contract-artifact, dod-typed-message-regression, dod-runtime-suite, dod-deploy.

OCC pairing

Paired OCC receipt PR: OmniNode-ai/onex_change_control#2629 (contract contracts/OMN-13149.yaml + PASS receipts under drift/dod_receipts/OMN-13149/; verifier receipt-gate-local ≠ runner; honesty gate 0 violations).

Receipt Gate / DoD Gate evidence for OMN-13149:

Evidence-Source: OCC#2629
Evidence-Ticket: OMN-13149

Note on repo

The OMN-13149 ticket text guessed omnimarket for the consumer. The actual savings_estimation consumer lives in omnibase_infra (src/omnibase_infra/services/observability/savings_estimation/consumer.py), packaged into omnimarket's venv as a dependency. The fix and the runtime wiring (service_kernel.py) are both in omnibase_infra.

Closes OMN-13149. Parent: OMN-13119.

[coderabbitai]

Warning

Review limit reached

@jonahgabriel, we couldn't start this review because you've reached your PR review rate limit.

More reviews will be available in 2 hours, 5 minutes, and 32 seconds. Learn how PR review limits work.

Your organization has used up its prepaid credits, and credit purchases are no longer available. Enable the review add-on in the billing tab to keep reviews running — you're only billed for reviews past your plan's rate limits ($0.25/file).

⌛ How to resolve this issue?

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.

We recommend that you space out your commits to avoid hitting the rate limit.

🚦 How do rate limits work?

CodeRabbit enforces hourly rate limits for each developer per organization.

Our paid plans include higher PR review limits than trial, open-source, and free plans. In all cases, reviews become available again over time. During sustained high-volume PR review activity, CodeRabbit may temporarily slow when the next review becomes available.

Please see our Fair Usage Limits Policy for further information.

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: a9b9f89a-af3d-4e9e-a99b-8e3596c14f37

📥 Commits

Reviewing files that changed from the base of the PR and between b56a3ff and 3ab6b0e.

📒 Files selected for processing (4)

• contracts/OMN-13149.yaml
• src/omnibase_infra/runtime/service_kernel.py
• src/omnibase_infra/services/observability/savings_estimation/consumer.py
• tests/unit/services/observability/savings_estimation/test_consumer.py

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch jonah/omn-13149-savings-consumer-typed-message

---

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