fix(OMN-19968): the llm-cost writer speaks the shared usage-source vocabulary on SQLite and Postgres

contracts/OMN-19968.yaml

@@ -0,0 +1,33 @@
+---
+schema_version: "1.0.0"
+ticket_id: "OMN-19968"
+title: "Projection writers that run on a developer machine produce the same rows on SQLite and on Postgres"
+summary: "The four writers with a SQLite path (delegation_events, llm_call_metrics, usage_by_model_day, metering_summary) give equal normalized rows through the real SQLite adapter and the real Postgres adapter, and no Postgres-only SQL is reachable from the SQLite path. This change makes the llm-cost writer speak the shared usage-source vocabulary (measured/estimated/unknown) on both stores, which is what kept llm_call_metrics rows from matching on a database infra has migrated; AC1 binds to the test that pins it. AC2 was delivered earlier (omnimarket#3067) and is not changed here, so this contract does not claim it."
+is_seam_ticket: false
+interface_change: false
+interfaces_touched: []
+requirements:
+  - id: "req-store-neutral-writers"
+    statement: |-
+      Projection writers that run on a developer machine must produce the same rows on SQLite and on Postgres, with no Postgres-only SQL on the SQLite path.
+    acceptance:
+      - id: "AC1"
+        statement: |-
+          AC1: For each of the four writers with a SQLite path (delegation_events, llm_call_metrics, usage_by_model_day and metering_summary) the same recorded event fixtures give equal normalized rows through the real SQLite adapter and the real Postgres adapter -- falsifier: the omnimarket parity test run with uv run pytest shows a row difference
+evidence_requirements:
+  - kind: "tests"
+    description: "The llm-cost writer, its SQLite schema and the forward migration use the shared usage-source vocabulary, so its rows match on SQLite and on a Postgres infra has migrated."
+    command: "uv run pytest tests/unit/test_omn19968_usage_source_vocabulary.py -q"
+emergency_bypass:
+  enabled: false
+  justification: ""
+  follow_up_ticket_id: ""
+dod_evidence:
+  - id: "dod-omn19968-ac1-shared-usage-source"
+    description: "The llm-cost writer writes measured/estimated/unknown, the SQLite schema defaults to it, existing rows move onto it, and the forward migration renames the retired labels in place."
+    source: "manual"
+    checks:
+      - check_type: "test_passes"
+        check_value: "uv run pytest tests/unit/test_omn19968_usage_source_vocabulary.py -q"
+        cwd: "${OMNI_HOME}/omnimarket"
+    binds_ac: ["AC1"]

src/omnimarket/nodes/node_projection_llm_cost/handlers/row_llm_call_metrics.py

@@ -18,6 +18,7 @@
 import uuid
 from typing import Any
 
+from omnimarket.enums.enum_usage_source import EnumUsageSource
 from omnimarket.projection.runner import safe_parse_date
 
 # Ordered tuple of llm_call_metrics columns this node writes. Asserted equal to
@@ -43,18 +44,26 @@
     "created_at",
 )
 
-# DB enum usage_source_type accepts only these values. The upstream event uses
-# MEASURED (token usage measured via the provider's usage response) which maps to
-# the DB's API provenance; UNKNOWN/absent maps to MISSING.
-_USAGE_SOURCE_MAP: dict[str, str] = {
-    "MEASURED": "API",
-    "API": "API",
-    "ESTIMATED": "ESTIMATED",
-    "MISSING": "MISSING",
-    "UNKNOWN": "MISSING",
+# OMN-19968: rows carry the shared usage-source vocabulary (EnumUsageSource:
+# measured / estimated / unknown), the values omnibase_infra migration 077
+# (OMN-10382) gave the shared ``usage_source_type``. The retired provenance
+# labels an event may still send (API, MISSING, upper-case spellings) are
+# folded onto it here; anything unrecognised is ``unknown``.
+_USAGE_SOURCE_SYNONYMS: dict[str, EnumUsageSource] = {
+    "measured": EnumUsageSource.MEASURED,
+    "api": EnumUsageSource.MEASURED,
+    "estimated": EnumUsageSource.ESTIMATED,
+    "unknown": EnumUsageSource.UNKNOWN,
+    "missing": EnumUsageSource.UNKNOWN,
 }
 
 
+def _usage_source(raw: object) -> EnumUsageSource:
+    if not raw:
+        return EnumUsageSource.UNKNOWN
+    return _USAGE_SOURCE_SYNONYMS.get(str(raw).strip().lower(), EnumUsageSource.UNKNOWN)
+
+
 def _safe_int(value: Any, default: int = 0) -> int:
     if value is None:
         return default
@@ -102,8 +111,9 @@ def build_llm_call_metrics_row(data: dict[str, Any]) -> dict[str, Any]:
     """Map an inbound llm-call-completed event to an llm_call_metrics row.
 
     The returned dict has exactly the keys in ``LLM_CALL_METRICS_COLUMNS``.
-    ``usage_source`` is honest: MEASURED→API, anything else→ESTIMATED/MISSING,
-    with ``usage_is_estimated`` derived from it.
+    ``usage_source`` is honest and uses the shared vocabulary: ``measured``
+    when the provider reported the usage, else ``estimated`` or ``unknown``
+    (see ``_usage_source``), with ``usage_is_estimated`` derived from it.
     """
     model_id = str(data.get("model_id") or data.get("model_name") or "unknown")
     session_id = data.get("session_id") or data.get("sessionId")
@@ -149,15 +159,14 @@ def build_llm_call_metrics_row(data: dict[str, Any]) -> dict[str, Any]:
     # ContractLlmCallMetrics wire format nests it as usage_normalized.source
     # (values: "api" / "estimated" / "missing") — no top-level usage_source
     # field.  Fall back to that nested path when the flat field is absent so
-    # exp0/generation-consumer rows are not silently written as MISSING.
+    # exp0/generation-consumer rows are not silently written as unknown.
     _top_level = data.get("usage_source") or data.get("usageSource")
     if not _top_level:
         _normalized = data.get("usage_normalized")
         if isinstance(_normalized, dict):
             _top_level = _normalized.get("source")
-    usage_source_raw = str(_top_level or "MISSING").upper()
-    usage_source = _USAGE_SOURCE_MAP.get(usage_source_raw, "MISSING")
-    usage_is_estimated = usage_source != "API"
+    usage_source = _usage_source(_top_level).value
+    usage_is_estimated = usage_source != EnumUsageSource.MEASURED.value
 
     input_hash = _compute_input_hash(
         reporting_source=str(reporting_source) if reporting_source else None,

src/omnimarket/nodes/node_projection_llm_cost/migrations/0003_usage_source_shared_vocabulary.sql

@@ -0,0 +1,39 @@
+-- SPDX-FileCopyrightText: 2026 OmniNode.ai Inc.
+-- SPDX-License-Identifier: MIT
+--
+-- OMN-19968: llm_call_metrics.usage_source speaks the shared vocabulary.
+--
+-- 0001 created usage_source_type as ('API', 'ESTIMATED', 'MISSING'). omnibase_infra
+-- migration 077 (OMN-10382) moved the shared type to ('measured', 'estimated',
+-- 'unknown'), the EnumUsageSource values, and the writer now produces those. A
+-- database built from omnimarket's migrations alone still carries the old labels;
+-- this renames them in place. RENAME VALUE keeps every stored row and needs no
+-- table rewrite.
+--
+-- The type is resolved through search_path (to_regtype), never by name across
+-- every schema: 0001's typname guard matched a same-named type in any schema,
+-- which is how a database migrated by infra ended up rejecting this writer.
+--
+-- Idempotent: each label is renamed only while it is still present, so on a
+-- database that already carries the shared vocabulary (every infra-migrated one)
+-- the block changes nothing. Forward-only: applied migration bytes are immutable.
+
+DO $$
+DECLARE
+    usage_type regtype := to_regtype('usage_source_type');
+BEGIN
+    IF usage_type IS NULL THEN
+        RETURN;
+    END IF;
+    IF EXISTS (SELECT 1 FROM pg_enum WHERE enumtypid = usage_type AND enumlabel = 'API') THEN
+        ALTER TYPE usage_source_type RENAME VALUE 'API' TO 'measured';
+    END IF;
+    IF EXISTS (SELECT 1 FROM pg_enum WHERE enumtypid = usage_type AND enumlabel = 'ESTIMATED') THEN
+        ALTER TYPE usage_source_type RENAME VALUE 'ESTIMATED' TO 'estimated';
+    END IF;
+    IF EXISTS (SELECT 1 FROM pg_enum WHERE enumtypid = usage_type AND enumlabel = 'MISSING') THEN
+        ALTER TYPE usage_source_type RENAME VALUE 'MISSING' TO 'unknown';
+    END IF;
+END$$;
+
+ALTER TABLE llm_call_metrics ALTER COLUMN usage_source SET DEFAULT 'unknown';

src/omnimarket/nodes/node_projection_llm_cost/tests/test_row_and_backfill_llm_call_metrics.py

@@ -156,29 +156,29 @@ def test_builds_complete_row_shape(self) -> None:
         }
         assert expected_keys.issubset(row.keys())
 
-    def test_measured_maps_to_api(self) -> None:
+    def test_measured_maps_to_measured(self) -> None:
         row = _build_row(_minimal_event(usage_source="MEASURED"))
-        assert row["usage_source"] == "API"
+        assert row["usage_source"] == "measured"
         assert row["usage_is_estimated"] is False
 
     def test_api_usage_source_preserved(self) -> None:
         row = _build_row(_minimal_event(usage_source="API"))
-        assert row["usage_source"] == "API"
+        assert row["usage_source"] == "measured"
         assert row["usage_is_estimated"] is False
 
     def test_estimated_usage_source(self) -> None:
         row = _build_row(_minimal_event(usage_source="ESTIMATED"))
-        assert row["usage_source"] == "ESTIMATED"
+        assert row["usage_source"] == "estimated"
         assert row["usage_is_estimated"] is True
 
-    def test_unknown_usage_source_maps_to_missing(self) -> None:
+    def test_unknown_usage_source_maps_to_unknown(self) -> None:
         row = _build_row(_minimal_event(usage_source="UNKNOWN"))
-        assert row["usage_source"] == "MISSING"
+        assert row["usage_source"] == "unknown"
         assert row["usage_is_estimated"] is True
 
-    def test_garbage_usage_source_maps_to_missing(self) -> None:
+    def test_garbage_usage_source_maps_to_unknown(self) -> None:
         row = _build_row(_minimal_event(usage_source="GARBAGE"))
-        assert row["usage_source"] == "MISSING"
+        assert row["usage_source"] == "unknown"
 
     def test_source_from_reporting_source(self) -> None:
         row = _build_row(_minimal_event(reporting_source="ab-compare"))
@@ -309,7 +309,9 @@ def test_prompt_token_aliases_accepted(self) -> None:
 
     # --- OMN-12994: usage_normalized.source fallback ---
 
-    def test_usage_normalized_source_api_maps_to_api_when_no_top_level(self) -> None:
+    def test_usage_normalized_source_api_maps_to_measured_when_no_top_level(
+        self,
+    ) -> None:
         """ContractLlmCallMetrics wire shape: usage_normalized.source='api', no top-level usage_source."""
         data: dict[str, Any] = {
             "model_id": "Qwen3.6-35B-A3B",
@@ -319,8 +321,8 @@ def test_usage_normalized_source_api_maps_to_api_when_no_top_level(self) -> None
             "usage_normalized": {"source": "api"},
         }
         row = _build_row(data)
-        assert row["usage_source"] == "API", (
-            "usage_normalized.source='api' must resolve to API (MEASURED), not MISSING"
+        assert row["usage_source"] == "measured", (
+            "usage_normalized.source='api' must resolve to measured, not unknown"
         )
         assert row["usage_is_estimated"] is False
 
@@ -335,10 +337,10 @@ def test_usage_normalized_source_estimated_maps_to_estimated_when_no_top_level(
             "usage_normalized": {"source": "estimated"},
         }
         row = _build_row(data)
-        assert row["usage_source"] == "ESTIMATED"
+        assert row["usage_source"] == "estimated"
         assert row["usage_is_estimated"] is True
 
-    def test_usage_normalized_source_missing_maps_to_missing_when_no_top_level(
+    def test_usage_normalized_source_missing_maps_to_unknown_when_no_top_level(
         self,
     ) -> None:
         data: dict[str, Any] = {
@@ -347,7 +349,7 @@ def test_usage_normalized_source_missing_maps_to_missing_when_no_top_level(
             "usage_normalized": {"source": "missing"},
         }
         row = _build_row(data)
-        assert row["usage_source"] == "MISSING"
+        assert row["usage_source"] == "unknown"
 
     def test_top_level_usage_source_takes_priority_over_usage_normalized(self) -> None:
         """Top-level usage_source wins over usage_normalized.source."""
@@ -360,9 +362,9 @@ def test_top_level_usage_source_takes_priority_over_usage_normalized(self) -> No
             "usage_normalized": {"source": "missing"},
         }
         row = _build_row(data)
-        assert row["usage_source"] == "API"  # MEASURED → API from top-level
+        assert row["usage_source"] == "measured"  # MEASURED -> measured from top-level
 
-    def test_usage_normalized_absent_falls_back_to_missing(self) -> None:
+    def test_usage_normalized_absent_falls_back_to_unknown(self) -> None:
         """No top-level usage_source and no usage_normalized → MISSING (unchanged)."""
         data: dict[str, Any] = {
             "model_id": "some-model",
@@ -371,7 +373,7 @@ def test_usage_normalized_absent_falls_back_to_missing(self) -> None:
             "total_tokens": 150,
         }
         row = _build_row(data)
-        assert row["usage_source"] == "MISSING"
+        assert row["usage_source"] == "unknown"
 
 
 # ---------------------------------------------------------------------------
@@ -519,7 +521,7 @@ def test_canonical_event_maps_correctly(self) -> None:
         }
         row = _build_row(event)
 
-        assert row["usage_source"] == "API"
+        assert row["usage_source"] == "measured"
         assert row["usage_is_estimated"] is False
         assert row["latency_ms"] == pytest.approx(19820.0)
         assert row["source"] == "ab-compare"

src/omnimarket/projection/sqlite_database.py

@@ -211,7 +211,7 @@
     total_tokens       INTEGER,
     estimated_cost_usd REAL,
     latency_ms         REAL,
-    usage_source       TEXT NOT NULL DEFAULT 'MISSING',
+    usage_source       TEXT NOT NULL DEFAULT 'unknown',
     usage_is_estimated INTEGER NOT NULL DEFAULT 0,
     usage_raw          TEXT,
     input_hash         TEXT,
@@ -309,6 +309,16 @@ def _reconcile_legacy_llm_call_metrics(cls, conn: sqlite3.Connection) -> None:
             conn.execute(
                 f"ALTER TABLE llm_call_metrics RENAME TO {_LEGACY_LLM_CALL_METRICS_TABLE}"
             )
+        elif "usage_source" in columns:
+            # OMN-19968: rows written before the shared vocabulary move onto it
+            # (EnumUsageSource; omnibase_infra migration 077). Idempotent: it
+            # touches only rows still holding a retired label.
+            conn.execute(
+                "UPDATE llm_call_metrics SET usage_source = CASE usage_source "
+                "WHEN 'API' THEN 'measured' WHEN 'ESTIMATED' THEN 'estimated' "
+                "WHEN 'MISSING' THEN 'unknown' ELSE usage_source END "
+                "WHERE usage_source IN ('API', 'ESTIMATED', 'MISSING')"
+            )
 
     @staticmethod
     def _existing_columns(conn: sqlite3.Connection, table: str) -> set[str]:

tests/test_dispatch_path_projection_llm_cost.py

@@ -106,7 +106,7 @@ async def test_envelope_through_real_dispatch_writes_llm_call_metrics_row() -> N
     assert total_tokens == 28471
     assert estimated_cost_usd == pytest.approx(0.17897125)
     assert latency_ms == pytest.approx(19820.0)
-    assert usage_source == "API"  # MEASURED -> API
+    assert usage_source == "measured"  # MEASURED -> measured
     assert usage_is_estimated is False
     assert session_id == "b45a01ea-0000-0000-0000-000000000001"
     assert run_id is None

tests/test_golden_chain_projection_llm_cost.py

@@ -76,7 +76,7 @@ def test_project_batch(self) -> None:
         assert result.rows_upserted == 3
         assert len(db.query(TABLE)) == 3
 
-    def test_usage_source_measured_maps_to_api(self) -> None:
+    def test_usage_source_measured_maps_to_measured(self) -> None:
         db = InmemoryDatabaseAdapter()
         HANDLER.project(
             ModelLlmCallCompletedEvent(
@@ -85,7 +85,7 @@ def test_usage_source_measured_maps_to_api(self) -> None:
             db,
         )
         rows = db.query(TABLE)
-        assert rows[0]["usage_source"] == "API"
+        assert rows[0]["usage_source"] == "measured"
         assert rows[0]["usage_is_estimated"] is False
 
     def test_usage_source_estimated(self) -> None:
@@ -100,7 +100,7 @@ def test_usage_source_estimated(self) -> None:
             db,
         )
         rows = db.query(TABLE)
-        assert rows[0]["usage_source"] == "ESTIMATED"
+        assert rows[0]["usage_source"] == "estimated"
         assert rows[0]["usage_is_estimated"] is True
 
     def test_compute_cost_folds_into_estimated_cost(self, tmp_path: Path) -> None:

tests/test_omn19968_writer_store_parity.py

@@ -237,6 +237,33 @@ def _dsn(pg: _Postgres, schema: str) -> str:
     return pg.dsn(schema)
 
 
+async def _mirror_infra_usage_source_state(admin: Any, schema: str) -> None:
+    """Start the throwaway schema in the state a lab database is in.
+
+    omnibase_infra migration 031 creates ``usage_source_type`` together with
+    ``llm_call_metrics``, and migration 077 (OMN-10382) moves the type to the shared
+    vocabulary, all before omnimarket's node migrations run. 0001's typname guard
+    matches a type in any schema, so on such a server it skips creating its own
+    type. On a real lab database that is harmless because the table already exists;
+    in an empty throwaway schema it is not, so mirror the type and the table here.
+    On a server infra never migrated this is a no-op and 0001 creates everything.
+    """
+    shared = await admin.fetchval(
+        "SELECT EXISTS (SELECT 1 FROM pg_type t JOIN pg_enum e ON e.enumtypid = t.oid "
+        "WHERE t.typname = 'usage_source_type' AND e.enumlabel = 'measured')"
+    )
+    if not shared:
+        return
+    await admin.execute(
+        f"CREATE TYPE {schema}.usage_source_type AS ENUM ('measured', 'estimated', 'unknown')"
+    )
+    await admin.execute(
+        f"CREATE TABLE {schema}.llm_call_metrics (id BIGSERIAL PRIMARY KEY, "
+        f"model_id TEXT NOT NULL, usage_source {schema}.usage_source_type NOT NULL "
+        "DEFAULT 'unknown')"
+    )
+
+
 _WRITERS = {
     "delegation_events": _write_delegation,
     "llm_call_metrics": _write_llm,
@@ -255,6 +282,7 @@ async def test_writer_rows_equal_on_sqlite_and_postgres(
     sqlite_rows = _normalize(sqlite.query(table))
 
     async with _provisioned(pg) as (admin, schema):
+        await _mirror_infra_usage_source_state(admin, schema)
         for migration in _LLM_COST_MIGRATIONS:
             await admin.execute(migration.read_text(encoding="utf-8"))
         postgres = PostgresSyncProjectionAdapter(_dsn(pg, schema))

tests/test_projection_handlers.py

@@ -209,7 +209,7 @@ async def test_usage_source_normalization(self, mock_db: AsyncMock) -> None:
         # to the DB enum's MISSING value. call_args[0] is SQL; the 10th INSERT
         # column (usage_source) is the 10th positional bind -> index 10.
         call_args = mock_db.execute.call_args[0]
-        assert call_args[10] == "MISSING"
+        assert call_args[10] == "unknown"
 
 
 class TestDelegationHandler:

tests/unit/nodes/node_delegate_skill_orchestrator/test_local_dispatch_llm_call_metrics.py

@@ -127,7 +127,7 @@ def test_local_dispatch_materializes_llm_call_metrics_row(
     assert row["prompt_tokens"] == delegation[0]["tokens_input"] == 11
     assert row["completion_tokens"] == delegation[0]["tokens_output"] == 22
     assert row["total_tokens"] == 33
-    assert row["usage_source"] == "API"
+    assert row["usage_source"] == "measured"
 
 
 def test_two_local_runs_with_equal_tokens_are_two_rows(

tests/unit/test_cli_delegation_cost_demo.py

@@ -54,7 +54,7 @@ def test_cli_default_profile_prints_joined_projection_table(
     assert "savings_estimates:1" in result.output
     # OMN-13001: usage_source is now persisted in the DB enum's vocabulary
     # (MEASURED -> API) by the per-call llm_call_metrics writer.
-    assert "usage_source=API" in result.output
+    assert "usage_source=measured" in result.output
 
 
 @pytest.mark.unit
@@ -187,4 +187,4 @@ def test_cli_loads_explicit_profile_file(
     assert payload["profile"]["local_delegate"] == "tiny-delegate"
     assert payload["profile"]["local_model_id"] == "phi4-mini"
     # OMN-13001: persisted usage_source is the DB enum value (MEASURED -> API).
-    assert payload["joined"]["usage_source"] == "API"
+    assert payload["joined"]["usage_source"] == "measured"