feat(OMN-20006): usage-by-model-day keeps measured cost apart from estimates

contracts/OMN-20006.yaml

@@ -0,0 +1,55 @@
+---
+schema_version: "1.0.0"
+ticket_id: "OMN-20006"
+title: "The Usage page renders tokens in, tokens out and cost per model from one served exposure"
+summary: "The data half of the Usage page. usage-by-model-day.v1 now keeps measured cost apart from cost that was estimated or not recorded: each call stores its usage_source, and each day-and-model row serves measured_cost_usd (NULL when no call was measured, never 0) and unmeasured_call_count beside cost_usd and call_count, which keep their meaning. The exposure is scoped to the reading tenant, the local onex delegate path writes the usage rows the page reads, a local store written before this change gains the new columns on connect, and the reasoned consumers opt-out is deleted because the omnidash Usage page is its reader. Tests: tests/test_omn19978_usage_by_model_day.py, tests/test_omn19978_usage_by_model_day_writer.py, tests/projection/test_omn20006_usage_by_model_day_served.py, tests/unit/nodes/node_delegate_skill_orchestrator/test_local_dispatch_usage_by_model_day.py."
+is_seam_ticket: false
+interface_change: false
+interfaces_touched: []
+requirements:
+  - id: "req-omn20006-usage-measured-cost-served"
+    statement: |-
+      usage-by-model-day.v1 serves, per tenant, UTC day and model, the tokens in and out of every call, the measured cost of the calls whose cost was measured, and the count of calls whose cost was not; the local onex delegate path writes those rows.
+    acceptance:
+      - id: "AC1"
+        statement: |-
+          AC1 (data half): one local onex delegate run writes one usage_by_model_day_calls row keyed by its correlation id with the terminal's tokens and its day-and-model row, two runs with equal tokens are two rows, a failing usage write never breaks the delegation, and a read through the projection read node returns the reading tenant's rows only -- falsifier: uv run pytest tests/unit/nodes/node_delegate_skill_orchestrator/test_local_dispatch_usage_by_model_day.py tests/projection/test_omn20006_usage_by_model_day_served.py -q
+      - id: "AC3"
+        statement: |-
+          AC3 (data half): one measured and one estimated call for the same day and model give measured_cost_usd equal to the measured call's cost and unmeasured_call_count 1, on SQLite and in the Postgres writer's recount, and the usage source is stored and replays without double counting -- falsifier: uv run pytest tests/test_omn19978_usage_by_model_day.py tests/test_omn19978_usage_by_model_day_writer.py -q
+      - id: "AC4"
+        statement: |-
+          AC4 (data half): a day and model with no measured call serves measured_cost_usd as null, never 0, a measured call costing nothing serves a measured 0, and a call that does not say how its cost was obtained is unmeasured -- falsifier: uv run pytest tests/test_omn19978_usage_by_model_day.py -q -k "typed_null or measured_zero or no_usage_source"
+evidence_requirements:
+  - kind: "tests"
+    description: "The usage tests pass at the head and fail at the merge base, where no usage source, measured cost, unmeasured count, tenant scope or local usage write exists."
+    command: "uv run pytest tests/test_omn19978_usage_by_model_day.py tests/test_omn19978_usage_by_model_day_writer.py tests/projection/test_omn20006_usage_by_model_day_served.py tests/unit/nodes/node_delegate_skill_orchestrator/test_local_dispatch_usage_by_model_day.py -q"
+emergency_bypass:
+  enabled: false
+  justification: ""
+  follow_up_ticket_id: ""
+dod_evidence:
+  - id: "dod-omn20006-ac1-local-usage-rows-per-tenant"
+    description: "One local delegate run writes its usage call and rollup rows with the terminal's tokens, equal-token runs stay apart, a failed usage write is swallowed, and the served read carries the reading tenant's rows only."
+    source: "manual"
+    checks:
+      - check_type: "test_passes"
+        check_value: "uv run pytest tests/unit/nodes/node_delegate_skill_orchestrator/test_local_dispatch_usage_by_model_day.py tests/projection/test_omn20006_usage_by_model_day_served.py -q"
+        cwd: "${OMNI_HOME}/omnimarket"
+    binds_ac: ["AC1"]
+  - id: "dod-omn20006-ac3-measured-cost-excludes-unmeasured"
+    description: "A measured and an estimated call give the measured call's cost as measured_cost_usd and one unmeasured call, on SQLite and in the Postgres recount statement; usage_source is stored and a replay changes nothing."
+    source: "manual"
+    checks:
+      - check_type: "test_passes"
+        check_value: "uv run pytest tests/test_omn19978_usage_by_model_day.py tests/test_omn19978_usage_by_model_day_writer.py -q"
+        cwd: "${OMNI_HOME}/omnimarket"
+    binds_ac: ["AC3"]
+  - id: "dod-omn20006-ac4-unmeasured-cost-is-null-not-zero"
+    description: "No measured call serves a null measured cost, a measured zero stays 0, and an unstated usage source is unmeasured."
+    source: "manual"
+    checks:
+      - check_type: "test_passes"
+        check_value: "uv run pytest tests/test_omn19978_usage_by_model_day.py -q -k \"typed_null or measured_zero or no_usage_source\""
+        cwd: "${OMNI_HOME}/omnimarket"
+    binds_ac: ["AC4"]

src/omnimarket/models/model_usage_call_event.py

@@ -0,0 +1,59 @@
+# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT
+"""Validated LLM call with an explicit event timestamp.
+
+Shared model (OMN-20006): the usage projection node and the local delegation
+dispatch port both build it, so it lives in ``omnimarket.models`` rather than in
+either node's private models package.
+"""
+
+from datetime import UTC, datetime
+
+from pydantic import AliasChoices, BaseModel, ConfigDict, Field, field_validator
+
+from omnimarket.enums.enum_usage_source import EnumUsageSource
+
+
+class ModelUsageCallEvent(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="ignore", populate_by_name=True)
+
+    call_id: str = Field(
+        default="",
+        validation_alias=AliasChoices("call_id", "correlation_id", "input_hash"),
+    )
+    model_name: str = Field(
+        min_length=1, validation_alias=AliasChoices("model_name", "model_id")
+    )
+    tenant_id: str | None = None
+    prompt_tokens: int = Field(
+        default=0, ge=0, validation_alias=AliasChoices("prompt_tokens", "input_tokens")
+    )
+    completion_tokens: int = Field(
+        default=0,
+        ge=0,
+        validation_alias=AliasChoices("completion_tokens", "output_tokens"),
+    )
+    estimated_cost_usd: float = Field(
+        default=0.0,
+        ge=0,
+        allow_inf_nan=False,
+        validation_alias=AliasChoices("estimated_cost_usd", "cost_usd"),
+    )
+    # How the call's cost was obtained. A call that does not say is UNKNOWN, never
+    # assumed measured: only a measured cost may reach measured_cost_usd.
+    usage_source: EnumUsageSource = EnumUsageSource.UNKNOWN
+    session_id: str | None = None
+    timestamp: datetime = Field(
+        validation_alias=AliasChoices("timestamp", "emitted_at", "created_at")
+    )
+
+    @field_validator("timestamp")
+    @classmethod
+    def aware_timestamp(cls, value: datetime) -> datetime:
+        """Interpret an explicit naive timestamp as UTC, never local time."""
+        if value.tzinfo is None:
+            return value.replace(tzinfo=UTC)
+        return value
+
+
+__all__ = ["ModelUsageCallEvent"]

src/omnimarket/nodes/node_delegate_skill_orchestrator/ports/port_local_delegation_dispatch.py

@@ -51,6 +51,7 @@
 import time
 import uuid
 from collections.abc import Callable, Mapping, Sequence
+from dataclasses import dataclass
 from datetime import datetime
 from decimal import Decimal
 from pathlib import Path
@@ -153,6 +154,7 @@
     ModelQualityGateResult,
     ModelQualityRuleEvaluation,
 )
+from omnimarket.models.model_usage_call_event import ModelUsageCallEvent
 from omnimarket.nodes.node_delegate_skill_orchestrator.dispatch_progress import (
     current_dispatch_progress,
     dispatch_stage,
@@ -221,6 +223,12 @@
     HandlerProjectionLlmCost,
     ModelLlmCallCompletedEvent,
 )
+from omnimarket.nodes.node_projection_usage_by_model_day.handlers.handler_projection_usage_by_model_day import (
+    HandlerProjectionUsageByModelDay,
+)
+from omnimarket.nodes.node_projection_usage_by_model_day.handlers.handler_usage_by_model_day_store import (
+    apply_usage_call,
+)
 from omnimarket.pricing import ModelBaselineSavings, compute_baseline_savings
 from omnimarket.projection.protocol_database import DatabaseAdapter
 from omnimarket.projection.snapshot_publisher import ModelSnapshotDeltaMessage
@@ -1227,6 +1235,10 @@ async def dispatch(
         # a rejected metered tier's real cost is never dropped (bus
         # ``_bank_attempt_spend`` parity). Projected as the row's cost_usd.
         cumulative_cost_usd = Decimal("0")
+        # OMN-20006: what each banked call consumed, on the model that served
+        # it, so the usage rows split the run's cost per model. Appended
+        # wherever cumulative_cost_usd is, so the two always agree.
+        attempt_usage: list[_AttemptUsage] = []
         attempts: list[dict[str, object]] = []
         escalation_count = 0
         # OMN-14220: best authored artifact seen across attempts (highest gate score,
@@ -1510,6 +1522,9 @@ async def dispatch(
                 # (typically zero) metered cost directly — mirrors the
                 # post-success banking below without requiring a gate verdict.
                 cumulative_cost_usd += transport_result.actual_cost_usd
+                attempt_usage.append(
+                    _AttemptUsage.of(transport_result, model_id=backend.model_id)
+                )
                 attempts.append(
                     {
                         "tier": current_tier,
@@ -1624,6 +1639,7 @@ async def dispatch(
                     baseline_savings=None,
                     escalation_count=escalation_count,
                     attempts=attempts,
+                    attempt_usage=attempt_usage,
                     # Transport failure: the gate never ran, so nothing was scored.
                     actual_score=None,
                     required_bar=None,
@@ -1688,6 +1704,11 @@ async def dispatch(
             # BEFORE deciding pass/fail so a rejected metered tier's spend is
             # counted even if we escalate away from it (OMN-13849).
             cumulative_cost_usd += result.actual_cost_usd
+            attempt_usage.append(
+                _AttemptUsage.of(
+                    result, model_id=result.served_model_id or backend.model_id
+                )
+            )
 
             attempt_tier = _routing_tier_name(backend)
 
@@ -1898,6 +1919,7 @@ async def dispatch(
                     baseline_savings=baseline_savings,
                     escalation_count=escalation_count,
                     attempts=attempts,
+                    attempt_usage=attempt_usage,
                     actual_score=gate_result.quality_score,
                     required_bar=_declared_required_bar(task_type),
                 )
@@ -2118,6 +2140,7 @@ async def dispatch(
                     baseline_savings=None,
                     escalation_count=escalation_count,
                     attempts=attempts,
+                    attempt_usage=attempt_usage,
                     actual_score=gate_result.quality_score,
                     required_bar=_declared_required_bar(task_type),
                 )
@@ -3205,6 +3228,7 @@ def _project_evidence(
         attempts: Sequence[Mapping[str, object]],
         actual_score: float | None,
         required_bar: float | None,
+        attempt_usage: Sequence[_AttemptUsage] = (),
     ) -> None:
         """Materialize a delegation_events row via the canonical projection.
 
@@ -3227,6 +3251,12 @@ def _project_evidence(
         routed every local terminal through the text fallback, where a run
         whose rungs answered and were refused on quality could be recorded as
         a provider quota failure.
+
+        OMN-20006: ``attempt_usage`` is what each banked call consumed, in
+        order. It becomes one usage-by-model-day call per call that reached a
+        provider, on that call's own model and with its own usage source, so an
+        escalated run's cost splits per model and still sums to ``cost_usd``.
+        Empty (the default, for callers that project no usage) writes none.
         """
         payload: dict[str, object] = {
             "status": "completed" if quality_passed else "failed",
@@ -3395,6 +3425,14 @@ def _project_evidence(
                     correlation_id,
                     exc_info=True,
                 )
+            # OMN-20006: the run's calls, each on its own model, folded into the
+            # usage-by-model-day tables the Usage page reads.
+            self._project_usage(
+                correlation_id=correlation_id,
+                tenant_id=tenant_id,
+                emitted_at=terminal.emitted_at,
+                attempt_usage=attempt_usage,
+            )
         except Exception:
             logger.warning(
                 "Failed to project local delegation evidence for correlation_id=%s",
@@ -3416,6 +3454,74 @@ def _project_evidence(
         # correlation_id.
         self._publish_terminal(payload, quality_passed=quality_passed)
 
+    def _project_usage(
+        self,
+        *,
+        correlation_id: UUID,
+        tenant_id: str,
+        emitted_at: datetime,
+        attempt_usage: Sequence[_AttemptUsage],
+    ) -> None:
+        """Fold each call into the usage-by-model-day tables (OMN-20006).
+
+        One call per attempt that reached a provider, on the model that served
+        it, with its own cost and usage source: an escalated run's spend is
+        split per model, and an attempt whose cost was not measured is never
+        booked as measured because a later one was. A call that carried no
+        usage at all (refused before a provider answered) is not a usage call;
+        its banked cost is zero, so the per-model sum still equals the run's
+        cost. A single call keeps the run's correlation id as its key; several
+        take ``<correlation id>:<n>`` in attempt order.
+
+        Tokens follow the one definition ``onex metering`` reads: the deciding
+        attempt's (the last one banked, whose tokens the delegation row
+        records). Every earlier attempt's call carries 0 tokens, so the window's
+        usage tokens equal metering's token totals (AC1). The columns are NOT
+        NULL, so 0 here means "not counted under that definition", not a
+        measured zero. Counting escalated-from attempts' tokens belongs in the
+        metering summary, so the CLI and the page move together.
+
+        Dated by the terminal's own emitted_at, so its UTC day is the
+        delegation row's. Each call has its own guard: a failed usage write
+        must neither break the response nor take the call-metrics row, or
+        another attempt's usage, down with it.
+        """
+        deciding = len(attempt_usage) - 1
+        calls = [
+            (index == deciding, usage)
+            for index, usage in enumerate(attempt_usage)
+            if usage.carries_usage
+        ]
+        for ordinal, (is_deciding, usage) in enumerate(calls, start=1):
+            call_id = (
+                str(correlation_id)
+                if len(calls) == 1
+                else f"{correlation_id}:{ordinal}"
+            )
+            try:
+                usage_event = ModelUsageCallEvent(
+                    call_id=call_id,
+                    model_name=usage.model_id,
+                    tenant_id=tenant_id,
+                    prompt_tokens=usage.tokens_in if is_deciding else 0,
+                    completion_tokens=usage.tokens_out if is_deciding else 0,
+                    estimated_cost_usd=float(usage.cost_usd),
+                    usage_source=usage.usage_source,
+                    timestamp=emitted_at,
+                )
+                apply_usage_call(
+                    HandlerProjectionUsageByModelDay().handle(usage_event),
+                    self._evidence_db,
+                )
+            except Exception:
+                logger.warning(
+                    "Failed to project local usage-by-model-day for "
+                    "correlation_id=%s call_id=%s",
+                    correlation_id,
+                    call_id,
+                    exc_info=True,
+                )
+
     def _refresh_metering_summary(
         self, *, correlation_id: UUID, tenant_id: str, terminal_at: datetime
     ) -> None:
@@ -3495,6 +3601,41 @@ def _local_terminal_topic(*, success: bool) -> str | None:
     return str(topic)
 
 
+@dataclass(frozen=True, slots=True)
+class _AttemptUsage:
+    """What one banked effect call consumed, on the model that served it."""
+
+    model_id: str
+    tokens_in: int
+    tokens_out: int
+    cost_usd: Decimal
+    usage_source: EnumUsageSource
+    answered: bool
+
+    @classmethod
+    def of(
+        cls, result: ModelLlmDelegationCallResult, *, model_id: str
+    ) -> _AttemptUsage:
+        return cls(
+            model_id=model_id,
+            tokens_in=result.tokens_in,
+            tokens_out=result.tokens_out,
+            cost_usd=result.actual_cost_usd,
+            usage_source=result.usage_source,
+            answered=result.success,
+        )
+
+    @property
+    def carries_usage(self) -> bool:
+        """A provider answered, or the call still recorded tokens or cost."""
+        return (
+            self.answered
+            or self.cost_usd != 0
+            or self.tokens_in != 0
+            or self.tokens_out != 0
+        )
+
+
 class _AttemptOutcome:
     """Outcome of one dispatch attempt (effect + gate) for the escalation loop.
 

src/omnimarket/nodes/node_projection_usage_by_model_day/contract.yaml

@@ -95,6 +95,11 @@ projection_api:
     - input_tokens
     - output_tokens
     - cost_usd
+    # OMN-20006: the cost of the calls whose cost was measured (NULL when none
+    # was), and how many calls were not measured. The Usage page shows these;
+    # cost_usd still sums every call.
+    - measured_cost_usd
+    - unmeasured_call_count
     - call_count
     - updated_at
     - projection_cursor
@@ -104,18 +109,11 @@ projection_api:
   limit: 500
   bus_backed: true
   key_columns: [tenant_id, usage_day, model_id]
-  # OMN-17199 -- the reader end. This exposure is `bus_backed: true` and no
-  # dashboard widget or typed backend reader renders it yet: OMN-19978 delivers
-  # the projection and its snapshot topic, not a rendering surface. The opt-out
-  # says so here instead of failing the exposure-reader gate on every
-  # omnibase_infra PR since the projection landed (omnimarket#3073). When a
-  # per-tenant usage surface reads it, delete both keys -- the gate fails on an
-  # opt-out left standing over a live reader.
-  consumers: none
-  consumers_reason: >-
-    Per-tenant model usage aggregates (tokens and cost per model per UTC day)
-    with no rendering surface built yet; the projection and its snapshot topic
-    ship first (OMN-19978) and a reader is a separate deliverable. Declared
-    here so the opt-out is visible rather than a silent gate failure.
+  # OMN-20006: a read is scoped to the reading tenant by the server, so the page
+  # never receives, or filters out, another tenant's rows.
+  tenant_column: tenant_id
+  # OMN-17199: the reader is the omnidash Usage page's usage-by-model-day
+  # component (OMN-20006), declared in omnidash's component registry, so the
+  # reasoned `consumers: none` opt-out is deleted rather than left standing.
 metadata:
   transport_type: kafka

src/omnimarket/nodes/node_projection_usage_by_model_day/handlers/handler_projection_usage_by_model_day.py

@@ -35,6 +35,7 @@ def handle(self, event: ModelUsageCallEvent) -> ModelUsageCallDelta:
             input_tokens=event.prompt_tokens,
             output_tokens=event.completion_tokens,
             cost_usd=Decimal(str(event.estimated_cost_usd)),
+            usage_source=event.usage_source,
             occurred_at=occurred_at,
         )