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

contracts/OMN-13149.yaml

@@ -0,0 +1,66 @@
+# OMN-13149 contract file for omnibase_infra deploy-gate.
+# The canonical ticket contract and PASS receipts live in onex_change_control.
+# This file is the per-repo deploy-evidence carrier only.
+schema_version: "1.0.0"
+ticket_id: "OMN-13149"
+title: "fix(OMN-13149): savings_estimation consumer decodes typed ModelEventMessage instead of calling .get()"
+summary: >-
+  The savings-estimation consumer is wired in service_kernel via event_bus.subscribe;
+  both the Kafka and in-memory buses deliver a typed ModelEventMessage to the
+  on_message callback. The pre-fix _savings_on_message did json.loads only when msg
+  was str/bytes and otherwise passed the value through unchanged, so a ModelEventMessage
+  reached ServiceSavingsEstimator.ingest_event -> _session_id_for_topic, which calls
+  payload.get("session_id"). ModelEventMessage has no .get(), raising
+  AttributeError: 'ModelEventMessage' object has no attribute 'get' at runtime — the
+  consumer wrote zero rows. Fix adds module-level decode_event_message(message) that
+  reads the JSON payload off the typed .value field directly (strongly typed, fail-fast:
+  raises TypeError on a non-object payload — no getattr-default, no dict-style fallback).
+  The kernel callback now decodes then calls ingest_event(topic, payload). Same
+  dict/typed-model family as OMN-13141 / golden-chain BUG A.
+is_seam_ticket: false
+interface_change: false
+interfaces_touched: []
+evidence_requirements:
+  - kind: "tests"
+    description: "Typed-message decode correlates a session through the real bus; pre-fix path reproduces AttributeError"
+    command: "uv run pytest tests/unit/services/observability/savings_estimation/test_consumer.py -q"
+  - kind: "ci"
+    description: "CI gates pass on omnibase_infra PR"
+    command: "gh pr checks <PR> --repo OmniNode-ai/omnibase_infra"
+emergency_bypass:
+  enabled: false
+  justification: ""
+  follow_up_ticket_id: ""
+dod_evidence:
+  - id: "dod-contract-artifact"
+    description: "Repo-local deploy evidence contract is present for OMN-13149"
+    source: "manual"
+    checks:
+      - check_type: "command"
+        check_value: "test -f contracts/OMN-13149.yaml"
+  - id: "dod-typed-message-regression"
+    description: >-
+      Through the REAL consumer path (EventBusInmemory.subscribe/publish ->
+      on_message(ModelEventMessage) -> decode_event_message -> ingest_event) a typed
+      ModelEventMessage correlates a session; the pre-fix path (typed message straight
+      to ingest_event) reproduces AttributeError. This is the CI gate for the fix.
+    source: "manual"
+    checks:
+      - check_type: "command"
+        check_value: "uv run pytest tests/unit/services/observability/savings_estimation/test_consumer.py -q"
+  - id: "dod-runtime-suite"
+    description: "Runtime + event-bus + savings unit suites pass — no regression in kernel wiring or estimator."
+    source: "manual"
+    checks:
+      - check_type: "command"
+        check_value: "uv run pytest tests/unit/runtime/ tests/unit/services/observability/savings_estimation/ -q"
+  - id: "dod-deploy"
+    description: >-
+      Deploy validation: change is an internal correctness fix to the savings-estimation
+      consumer wiring in service_kernel (_savings_on_message) plus a module-level decode
+      helper in the consumer. No new Kafka topics, no schema changes, no container image
+      changes, no API surface changes. Runtime restart picks up the change on next deploy.
+    source: "manual"
+    checks:
+      - check_type: "command"
+        check_value: "echo 'deploy-gate: OMN-13149 is an internal savings-consumer wiring correctness fix; no new topics, schema, or container changes; runtime restart picks it up on next deploy'"

src/omnibase_infra/runtime/service_kernel.py

@@ -98,6 +98,7 @@
 from omnibase_infra.event_bus.event_bus_inmemory import EventBusInmemory
 from omnibase_infra.event_bus.event_bus_kafka import EventBusKafka
 from omnibase_infra.event_bus.models.config import ModelKafkaEventBusConfig
+from omnibase_infra.event_bus.models.model_event_message import ModelEventMessage
 from omnibase_infra.models import ModelNodeIdentity
 from omnibase_infra.models.health.model_llm_endpoint_health_config import (
     ModelLlmEndpointHealthConfig,
@@ -1484,6 +1485,7 @@ async def _baselines_loop() -> None:
                 )
                 from omnibase_infra.services.observability.savings_estimation.consumer import (
                     ServiceSavingsEstimator,
+                    decode_event_message,
                 )
                 from omnibase_infra.topics import topic_keys
                 from omnibase_infra.topics.service_topic_registry import (
@@ -1510,30 +1512,25 @@ async def _baselines_loop() -> None:
 
                 async def _savings_consumer_loop() -> None:
                     """Consume input events and produce savings estimates."""
-                    import json as _json
-
                     _poll_interval = 2.0
 
                     for _input_topic in _savings_input_topics:
                         try:
 
                             async def _savings_on_message(
-                                msg: object,
-                                *,
-                                _topic: str = _input_topic,
+                                message: ModelEventMessage,
                             ) -> None:
-                                _savings_estimator.ingest_event(
-                                    _topic,
-                                    # Why: Runtime wiring validates and narrows this payload shape before use.
-                                    _json.loads(msg)  # type: ignore[arg-type]
-                                    if isinstance(msg, (str, bytes))
-                                    else msg,
-                                )
+                                # The event bus delivers a typed ModelEventMessage,
+                                # not a raw dict/str. decode_event_message reads the
+                                # JSON payload off the typed `.value` field — it
+                                # must never call `.get()` on the message, which
+                                # has no such method (OMN-13149).
+                                _topic, _payload = decode_event_message(message)
+                                _savings_estimator.ingest_event(_topic, _payload)
 
                             await event_bus.subscribe(
                                 _input_topic,
-                                # Why: Runtime wiring validates and narrows this payload shape before use.
-                                on_message=_savings_on_message,  # type: ignore[arg-type]
+                                on_message=_savings_on_message,
                                 group_id=f"savings-estimator.{_input_topic}",
                             )
                         except Exception:  # noqa: BLE001
@@ -3231,10 +3228,6 @@ async def _wiring_health_loop() -> None:
                     version="v1",
                 )
 
-                from omnibase_infra.event_bus.models.model_event_message import (
-                    ModelEventMessage,
-                )
-
                 async def _triage_on_message(
                     message: ModelEventMessage,
                 ) -> None:

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

@@ -32,13 +32,15 @@
 
 from __future__ import annotations
 
+import json
 import logging
 import time
 from collections import OrderedDict
 from dataclasses import dataclass, field
 from enum import StrEnum
 from uuid import UUID
 
+from omnibase_infra.event_bus.models.model_event_message import ModelEventMessage
 from omnibase_infra.nodes.node_savings_estimation_compute.handlers.handler_savings_estimation import (
     HandlerSavingsEstimation,
 )
@@ -456,6 +458,41 @@ def _ingest_dispatch_eval(buf: SessionBuffer, payload: dict[str, object]) -> Non
     buf.outcome_received_at = time.monotonic()
 
 
+def decode_event_message(message: ModelEventMessage) -> tuple[str, dict[str, object]]:
+    """Decode a typed event-bus message into a (topic, payload) pair.
+
+    Both the Kafka and in-memory event buses deliver a
+    :class:`ModelEventMessage` to a consumer's ``on_message`` callback — never a
+    raw ``dict`` or ``str``. The message body is the JSON payload carried in the
+    typed ``value`` field (bytes). This decodes that field directly off the
+    typed model. It does NOT call ``.get()`` on the message, which has no such
+    method — the pre-fix wiring did, raising
+    ``AttributeError: 'ModelEventMessage' object has no attribute 'get'``
+    (OMN-13149).
+
+    Args:
+        message: The typed event-bus message delivered by the consumer callback.
+            ``message.topic`` is the correlation topic and ``message.value`` is
+            the JSON-encoded payload.
+
+    Returns:
+        A ``(topic, payload)`` pair ready for
+        :meth:`ServiceSavingsEstimator.ingest_event`.
+
+    Raises:
+        TypeError: If the decoded payload is not a JSON object. The savings
+            correlation logic requires a mapping keyed by ``session_id``; a
+            non-object payload is a contract violation and fails fast.
+    """
+    payload = json.loads(message.value)
+    if not isinstance(payload, dict):
+        raise TypeError(
+            "savings estimation payload must be a JSON object, "
+            f"got {type(payload).__name__} on topic {message.topic!r}"
+        )
+    return message.topic, payload
+
+
 class ServiceSavingsEstimator:
     """Session-event correlator producing savings-estimated.v1 events.
 
@@ -737,4 +774,4 @@ def _mark_finalized(self, session_id: str) -> None:
         self._finalized[session_id] = True
 
 
-__all__: list[str] = ["ServiceSavingsEstimator"]
+__all__: list[str] = ["ServiceSavingsEstimator", "decode_event_message"]

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

@@ -12,17 +12,42 @@
 
 from __future__ import annotations
 
+import json
+from datetime import UTC, datetime
 from unittest.mock import patch
 
 import pytest
 
+from omnibase_infra.event_bus.event_bus_inmemory import EventBusInmemory
+from omnibase_infra.event_bus.models.model_event_headers import ModelEventHeaders
+from omnibase_infra.event_bus.models.model_event_message import ModelEventMessage
 from omnibase_infra.services.observability.savings_estimation.config import (
     ConfigSavingsEstimation,
 )
 from omnibase_infra.services.observability.savings_estimation.consumer import (
     ServiceSavingsEstimator,
+    decode_event_message,
 )
 
+
+def _build_event_message(topic: str, payload: dict[str, object]) -> ModelEventMessage:
+    """Build a typed ModelEventMessage exactly as the event bus delivers it.
+
+    The Kafka and in-memory buses both encode the payload as JSON bytes in the
+    ``value`` field and deliver the typed model to the ``on_message`` callback.
+    """
+    return ModelEventMessage(
+        topic=topic,
+        key=None,
+        value=json.dumps(payload).encode("utf-8"),
+        headers=ModelEventHeaders(
+            source="test-producer",
+            event_type=topic,
+            timestamp=datetime.now(UTC),
+        ),
+    )
+
+
 # Base monotonic time for deterministic clock control in tests.
 _T0 = 1000.0
 
@@ -404,3 +429,124 @@ async def test_llm_only_session_produces_zero_savings_estimate(
     assert results[0]["session_id"] == "sess-llm-only"
     assert results[0]["direct_tokens_saved"] == 0
     assert results[0]["direct_savings_usd"] == 0.0
+
+
+# ---------------------------------------------------------------------------
+# OMN-13149: typed ModelEventMessage ingestion (regression)
+#
+# The Kafka and in-memory event buses deliver a typed ModelEventMessage to the
+# consumer's on_message callback — never a raw dict or str. The original
+# service_kernel wiring passed the typed message straight to ingest_event, whose
+# correlation path calls payload.get("session_id"). ModelEventMessage has no
+# .get(), so the live consumer raised:
+#     AttributeError: 'ModelEventMessage' object has no attribute 'get'
+# These tests pin the typed-message decode path through the REAL consumer.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_ingest_event_rejects_typed_message_documents_bug(
+    service: ServiceSavingsEstimator,
+) -> None:
+    """Passing a typed ModelEventMessage to ingest_event reproduces the bug.
+
+    This is the exact failure OMN-13149 fixes: the typed model has no ``.get()``,
+    so the dict-style correlation path raises AttributeError. decode_event_message
+    must be used to obtain the payload first (see tests below).
+    """
+    message = _build_event_message(
+        "onex.evt.omniintelligence.llm-call-completed.v1",
+        {
+            "session_id": "sess-typed-bug",
+            "model_id": "claude-opus-4-6",
+            "prompt_tokens": 100,
+            "completion_tokens": 50,
+        },
+    )
+    with pytest.raises(AttributeError, match="get"):
+        # Why: ModelEventMessage is not a Mapping; the correlation path calls
+        # payload.get(). This is the pre-fix crash being pinned.
+        service.ingest_event(message.topic, message)  # type: ignore[arg-type]
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_decode_event_message_correlates_typed_message(
+    service: ServiceSavingsEstimator,
+) -> None:
+    """decode_event_message reads the typed value field; ingest_event correlates."""
+    message = _build_event_message(
+        "onex.evt.omniintelligence.llm-call-completed.v1",
+        {
+            "session_id": "sess-typed-ok",
+            "model_id": "qwen3-coder-30b-a3b",
+            "prompt_tokens": 1000,
+            "completion_tokens": 500,
+        },
+    )
+
+    topic, payload = decode_event_message(message)
+    service.ingest_event(topic, payload)
+
+    assert service.active_session_count == 1
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_decode_event_message_rejects_non_object_payload() -> None:
+    """A non-object JSON payload fails fast — no silent drop, no dict coercion."""
+    message = ModelEventMessage(
+        topic="onex.evt.omniintelligence.llm-call-completed.v1",
+        key=None,
+        value=b"[1, 2, 3]",
+        headers=ModelEventHeaders(
+            source="test-producer",
+            event_type="onex.evt.omniintelligence.llm-call-completed.v1",
+            timestamp=datetime.now(UTC),
+        ),
+    )
+    with pytest.raises(TypeError, match="must be a JSON object"):
+        decode_event_message(message)
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_real_bus_delivers_typed_message_to_consumer(
+    service: ServiceSavingsEstimator,
+) -> None:
+    """End-to-end: the real in-memory bus delivers a typed ModelEventMessage.
+
+    This pins the REAL consumer path the kernel wires (event bus subscribe ->
+    on_message(ModelEventMessage) -> decode_event_message -> ingest_event), not
+    handler isolation. The callback mirrors service_kernel._savings_on_message.
+    Before the fix this raised AttributeError inside the subscriber callback;
+    after the fix the session is correlated.
+    """
+    bus = EventBusInmemory(environment="test", group="savings-test")
+    await bus.start()
+    try:
+        topic = "onex.evt.omniintelligence.llm-call-completed.v1"
+
+        async def _savings_on_message(message: ModelEventMessage) -> None:
+            _topic, _payload = decode_event_message(message)
+            service.ingest_event(_topic, _payload)
+
+        await bus.subscribe(
+            topic,
+            on_message=_savings_on_message,
+            group_id=f"savings-estimator.{topic}",
+        )
+
+        payload = {
+            "session_id": "sess-real-bus",
+            "model_id": "claude-opus-4-6",
+            "prompt_tokens": 1200,
+            "completion_tokens": 300,
+        }
+        await bus.publish(topic, None, json.dumps(payload).encode("utf-8"))
+
+        assert service.active_session_count == 1
+        assert service.is_finalized("sess-real-bus") is False
+    finally:
+        await bus.close()