genlayerlabs · jmlago · Jun 24, 2026 · Jun 24, 2026 · Jun 24, 2026 · coderabbitai
diff --git a/codex_backend.py b/codex_backend.py
@@ -13,6 +13,8 @@
 import json
 from typing import Any, Iterable
 
+from llm_router_host import _cached_tokens
+
 CODEX_BASE_URL = "https://chatgpt.com/backend-api/codex"
 
 
@@ -258,6 +260,8 @@ def aggregate_codex_sse(lines: Iterable[str], latency_ms: int) -> dict:
             "tokens_in":     usage.get("input_tokens"),
             "tokens_out":    usage.get("output_tokens"),
             "tokens_total":  usage.get("total_tokens"),
+            "tokens_cached": _cached_tokens(usage),
+            "cost_reported": usage.get("cost"),
             "raw_model":     None,
         },
     }

diff --git a/llm_router_host.py b/llm_router_host.py
@@ -26,6 +26,21 @@
 import route_tool_capability as _route_tool_capability
 import route_cache as _route_cache
 
+
+def _cached_tokens(usage: dict) -> "int | None":
+    """Prompt-cache READ tokens the provider reports, across shapes — the metric
+    that proves cache_hot is paying off (same prompt_tokens, but this fraction is
+    billed at the cache-read rate, ~10x cheaper). OpenAI-compat:
+    prompt_tokens_details.cached_tokens; Codex Responses:
+    input_tokens_details.cached_tokens; Anthropic: cache_read_input_tokens."""
+    if not isinstance(usage, dict):
+        return None
+    for parent in ("prompt_tokens_details", "input_tokens_details"):
+        d = usage.get(parent)
+        if isinstance(d, dict) and d.get("cached_tokens") is not None:
+            return d.get("cached_tokens")
+    return usage.get("cache_read_input_tokens")
+
 import lupa
 from lupa import LuaRuntime
 
@@ -963,6 +978,8 @@ def _parse_openai_response(resp: "Any", latency: int, error_map: dict | None = N
                 "tokens_in":     usage.get("prompt_tokens"),
                 "tokens_out":    usage.get("completion_tokens"),
                 "tokens_total":  usage.get("total_tokens"),
+                "tokens_cached": _cached_tokens(usage),
+                "cost_reported": usage.get("cost"),
                 "raw_model":     data.get("model"),
             },
         }

diff --git a/route_session_meter.py b/route_session_meter.py
@@ -0,0 +1,59 @@
+"""Per-session usage meter: per-call is already on x_router; this accumulates the
+RUNNING TOTAL per session (the sid the caller sends), so a conversation/agent has
+both numbers — this call AND everything it has spent so far. Pairs with
+route_cache (same session key): route_cache says which peer is hot, this says what
+the session has cost and how much of it was served from cache.
+
+Tracks per session: calls, tokens_in, tokens_out, tokens_cached, cost_usd. The
+cache hit ratio (tokens_cached / tokens_in) and the realized cost are then exact
+per session, using the provider's reported cost — no model-price guessing.
+
+In-process (resets on restart), like route_cache / route_latency; fleet scale
+needs a shared store with TTL — the same debt the sibling forms carry.
+"""
+from __future__ import annotations
+
+import threading
+
+_lock = threading.Lock()
+_acc: dict[str, dict] = {}   # session -> running totals
+
+
+def observe(session: "str | None", *, tokens_in=0, tokens_out=0,
+            tokens_cached=0, cost_usd=0.0) -> "dict | None":
+    """Fold one call's usage into the session's running total; return the new
+    accumulated totals (so the caller can put per-call AND acc on the response).
+    No-op (returns None) when the caller named no session."""
+    if not session:
+        return None
+    with _lock:
+        a = _acc.get(session)
+        if a is None:
+            a = {"calls": 0, "tokens_in": 0, "tokens_out": 0,
+                 "tokens_cached": 0, "cost_usd": 0.0}
+            _acc[session] = a
+        a["calls"] += 1
+        a["tokens_in"] += int(tokens_in or 0)
+        a["tokens_out"] += int(tokens_out or 0)
+        a["tokens_cached"] += int(tokens_cached or 0)
+        a["cost_usd"] = round(a["cost_usd"] + float(cost_usd or 0.0), 6)
+        return dict(a)
+
+
+def get(session: "str | None") -> "dict | None":
+    if not session:
+        return None
+    with _lock:
+        a = _acc.get(session)
+        return dict(a) if a else None
+
+
+def snapshot() -> dict[str, dict]:
+    with _lock:
+        return {s: dict(a) for s, a in _acc.items()}
+
+
+def reset() -> None:
+    """Test hook."""
+    with _lock:
+        _acc.clear()
diff --git a/shim.py b/shim.py
@@ -37,6 +37,7 @@
 from pydantic import BaseModel, ConfigDict
 
 import route_cache
+import route_session_meter
 
 
 # Profile name used when nothing else can be inferred. Replaced via
@@ -235,6 +236,20 @@ def reload_config():
         import settings as _settings
         return {"ok": True, "overrides": _settings.reload()}
 
+    @app.get("/x/session/{sid}")
+    def session_meter(sid: str):
+        """Accumulated usage for a session: calls, tokens_in/out, tokens_cached,
+        cost_usd — the running total the per-call x_router.session_acc reflects.
+        Internal (/x/* hidden from consumers)."""
+        return route_session_meter.get(sid) or {
+            "calls": 0, "tokens_in": 0, "tokens_out": 0,
+            "tokens_cached": 0, "cost_usd": 0.0}
+
+    @app.get("/x/sessions")
+    def session_meters():
+        """All session meters (operator view of per-session spend/cache)."""
+        return {"sessions": route_session_meter.snapshot()}
+
     # ---- AntSeed buyer hot-wallet control (dashboard self-service) -----------
     # Proxy deposit/withdraw/refresh to the sidecar control server, then refresh
     # SOURCE_STATE so /x/market reflects the new escrow at once. Internal — /x/*
@@ -797,7 +812,8 @@ async def _handle_chat(req: ChatRequest, profile_name: str | None = None):
                 if not result.get("ok"):
                     return _openai_error_from_router(result)
                 return _router_response_to_openai(result, req.model,
-                                                  subscription_providers)
+                                                  subscription_providers,
+                                                  session=req.session)
             # Streaming flow: run as a task and wait briefly for a fast failure
             # (admission 400, or a node failing instantly) so it stays a clean
             # JSON error. Anything slower commits to SSE and HEARTBEATs while the
@@ -833,7 +849,8 @@ async def _handle_chat(req: ChatRequest, profile_name: str | None = None):
                 return _invalid_policy_response(admission)
             if result.get("ok"):
                 return _router_response_to_openai(result, req.model,
-                                                  subscription_providers)
+                                                  subscription_providers,
+                                                  session=req.session)
             return _openai_error_from_router(result)
         return await _handle_stream(contract, req)
 
@@ -902,6 +919,7 @@ def _final_chunk_parts(result: dict):
             "price_in": chosen.get("price_in"),
             "price_out": chosen.get("price_out"),
             "cost_usd": _executed_cost_usd(result, subscription_providers),
+            "tokens_cached": resp.get("tokens_cached"),
             "policy_fingerprint": (result.get("trace") or {}).get("policy_fingerprint"),
             "decision_trace": _trim_trace(result.get("trace")),
             "compact": _compact_suggested(resp),
@@ -1167,30 +1185,46 @@ def _trim_trace(trace):
     return out
 
 
+# Fraction of the input price billed for prompt-cache-READ tokens, used ONLY in
+# the computed fallback (providers that report no cost). Most caching providers
+# discount cache reads ~10x; this is the conservative typical factor.
+_CACHE_READ_FACTOR = 0.1
+
+
 def _executed_cost_usd(result: dict, subscription_providers=frozenset()) -> float | None:
-    """Cost of THIS request at the price the ranker actually used for the
-    chosen candidate. Subscription backends (codex) cost $0 regardless of
-    their ranking price (the scarcity ramp is a shadow price, not a bill).
-    None when the candidate had no price (then the dashboard's read-time
-    estimator is the fallback)."""
+    """Dollars actually spent on THIS request, accurate across providers.
+    Order: (1) subscription backends (codex) cost $0; (2) the provider's OWN
+    reported cost when present (e.g. OpenRouter `usage.cost`) — authoritative and
+    already net of prompt-cache discounts; (3) computed from the ranked price,
+    billing cache-read tokens at a fraction so a cache hit is not charged at full
+    input price. None when uncomputable (read-time estimator is the fallback)."""
     chosen = result.get("chosen") or {}
+    resp = result.get("response") or {}
     if chosen.get("provider_id") in subscription_providers:
         return 0.0
+    # (2) authoritative provider-reported cost — works for ANY provider that gives it
+    reported = resp.get("cost_reported")
+    if isinstance(reported, (int, float)) and not isinstance(reported, bool) and reported >= 0:
+        return round(float(reported), 6)
+    # (3) compute from the ranker price, discounting cache-read input tokens
     pin, pout = chosen.get("price_in"), chosen.get("price_out")
     if pin is None and pout is None:
         return None
-    resp = result.get("response") or {}
-    cost = round((resp.get("tokens_in") or 0) / 1e6 * (pin or 0)
-                 + (resp.get("tokens_out") or 0) / 1e6 * (pout or 0), 6)
-    # A negative price (an "unpriced"/sentinel value, or a shadow scarcity price)
-    # must never bill a negative cost — a call's cost is >= 0 by definition. Clamp
-    # at the source so no negative spend is recorded (we once saw a large
-    # negative-spend row that was exactly this: tokens × a negative chosen price).
-    return max(0.0, cost)
+    tin = resp.get("tokens_in") or 0
+    cached = resp.get("tokens_cached") or 0
+    uncached = max(0, tin - cached)
+    cost = (uncached / 1e6 * (pin or 0)
+            + cached / 1e6 * (pin or 0) * _CACHE_READ_FACTOR
+            + (resp.get("tokens_out") or 0) / 1e6 * (pout or 0))
+    # A negative price (unpriced sentinel / shadow scarcity price) must never bill
+    # negative — clamp at the source (we once saw a large negative-spend row that
+    # was exactly tokens × a negative chosen price).
+    return max(0.0, round(cost, 6))
 
 
 def _router_response_to_openai(result: dict, requested_model: str,
-                               subscription_providers=frozenset()) -> dict:
+                               subscription_providers=frozenset(),
+                               session: str | None = None) -> dict:
     response = result.get("response") or {}
     chosen = result.get("chosen") or {}
 
@@ -1233,10 +1267,21 @@ def _router_response_to_openai(result: dict, requested_model: str,
         "price_in": chosen.get("price_in"),
         "price_out": chosen.get("price_out"),
         "cost_usd": _executed_cost_usd(result, subscription_providers),
+        "tokens_cached": response.get("tokens_cached"),
         "policy_fingerprint": (result.get("trace") or {}).get("policy_fingerprint"),
         "decision_trace": _trim_trace(result.get("trace")),
         "compact": _compact_suggested(response),
     }
+    # Per-session meter: fold this call into the session's running total and put
+    # BOTH on the response — per-call (above) and accumulated (session_acc).
+    if session:
+        acc = route_session_meter.observe(
+            session,
+            tokens_in=response.get("tokens_in") or 0,
+            tokens_out=response.get("tokens_out") or 0,
+            tokens_cached=response.get("tokens_cached") or 0,
+            cost_usd=out["x_router"]["cost_usd"] or 0.0)
+        out["x_router"]["session_acc"] = acc
     return out
 
 

diff --git a/streaming.py b/streaming.py
@@ -13,6 +13,8 @@
 """
 from __future__ import annotations
 
+from llm_router_host import _cached_tokens
+
 import json
 import re
 import time
@@ -133,6 +135,8 @@ def _latency() -> int:
             "tokens_in": usage.get("prompt_tokens"),
             "tokens_out": usage.get("completion_tokens"),
             "tokens_total": usage.get("total_tokens"),
+            "tokens_cached": _cached_tokens(usage),
+            "cost_reported": usage.get("cost"),
             "raw_model": raw_model,
         },
     }

diff --git a/tests/test_metering.py b/tests/test_metering.py
@@ -0,0 +1,69 @@
+"""Essence for cross-provider cost metering + prompt-cache visibility.
+
+Proves the dollar number is accurate across providers: the provider's own
+reported cost is authoritative when present (already net of cache discounts);
+otherwise it is computed from the ranked price, billing cache-READ tokens at a
+fraction so a cache hit is not charged at full input price; subscription
+backends are $0. And `_cached_tokens` reads the cache-read metric across the
+OpenAI-compat / Codex-Responses / Anthropic usage shapes.
+"""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from shim import _executed_cost_usd, _CACHE_READ_FACTOR  # noqa: E402
+from llm_router_host import _cached_tokens                # noqa: E402
+
+
+def _result(resp, chosen):
+    return {"response": resp, "chosen": chosen}
+
+
+def test_prefers_provider_reported_cost():
+    # the provider's own cost wins over any computed estimate
+    out = _executed_cost_usd(_result(
+        {"tokens_in": 1000, "tokens_out": 10, "cost_reported": 0.00012},
+        {"provider_id": "openrouter", "price_in": 5.0, "price_out": 15.0}))
+    assert out == 0.00012
+
+
+def test_subscription_is_zero_even_if_priced():
+    out = _executed_cost_usd(_result(
+        {"tokens_in": 1000, "tokens_out": 10, "cost_reported": 0.5},
+        {"provider_id": "codex"}),
+        subscription_providers=frozenset(["codex"]))
+    assert out == 0.0
+
+
+def test_computed_fallback_discounts_cached_tokens():
+    # no reported cost -> compute; the 800 cached input tokens bill at the
+    # cache-read fraction, not full input price.
+    out = _executed_cost_usd(_result(
+        {"tokens_in": 1000, "tokens_out": 10, "tokens_cached": 800},
+        {"provider_id": "x", "price_in": 5.0, "price_out": 15.0}))
+    expected = round(200 / 1e6 * 5.0
+                     + 800 / 1e6 * 5.0 * _CACHE_READ_FACTOR
+                     + 10 / 1e6 * 15.0, 6)
+    assert out == expected
+    # and it is strictly cheaper than charging the cached tokens at full price
+    full = round(1000 / 1e6 * 5.0 + 10 / 1e6 * 15.0, 6)
+    assert out < full
+
+
+def test_negative_price_never_bills_negative():
+    out = _executed_cost_usd(_result(
+        {"tokens_in": 1000, "tokens_out": 10},
+        {"provider_id": "x", "price_in": -5.0, "price_out": -1.0}))
+    assert out == 0.0
+
+
+def test_cached_tokens_across_usage_shapes():
+    assert _cached_tokens({"prompt_tokens_details": {"cached_tokens": 1280}}) == 1280
+    assert _cached_tokens({"input_tokens_details": {"cached_tokens": 7}}) == 7
+    assert _cached_tokens({"cache_read_input_tokens": 42}) == 42
+    assert _cached_tokens({"prompt_tokens": 10}) is None
+    assert _cached_tokens(None) is None