genlayerlabs · jmlago · Jun 24, 2026 · Jun 24, 2026 · Jun 24, 2026 · coderabbitai
diff --git a/auth_proxy.py b/auth_proxy.py
@@ -1945,6 +1945,35 @@ async def key_usage(request: Request) -> Response:
     return JSONResponse(content=_key_usage_snapshot(viewer=f"consumer:{caller}", key_sha256=str(auth.get("digest")), caller=caller, options=options))
 
 
+@app.get("/v1/session/{sid}")
+async def session_view(sid: str, request: Request) -> Response:
+    """Consumer-facing per-session economics: accumulated cost / tokens / cache,
+    plus `warm` — the routes (family / provider / served_by, i.e. the real peer
+    or backend behind a marketplace) currently holding the session's prompt-cache
+    prefix. Authed by the caller's consumer key; `sid` is the opaque id the client
+    itself sends as X-Unhardcoded-Session. Lets a harness (e.g. the opencode
+    plugin) show live router economics without operator access to /x/*."""
+    token = _extract_token(request)
+    auth = _caller_auth(token)
+    if not auth.get("ok"):
+        code = auth.get("error_code") or "caller_auth"
+        status_code = 403 if code in {"caller_inactive", "caller_key_revoked", "caller_key_expired"} else 401
+        return JSONResponse(status_code=status_code, content={"error": {"message": "caller not authorized", "type": "auth_error", "code": code}})
+    assert _client is not None
+    # Forward the authed consumer key so the upstream meter scopes the read to
+    # the session's OWNER (bound when the sid was first metered). A consumer must
+    # not read another consumer's session economics / warm peers; the upstream
+    # answers 404 (not 403) for a sid this caller does not own, so the endpoint
+    # never confirms that someone else's sid exists.
+    caller = str(auth.get("caller") or "")
+    try:
+        r = await _client.get(f"{UPSTREAM}/x/session/{sid}", timeout=5.0,
+                              headers={"x-llm-router-caller": caller})
+        return JSONResponse(status_code=r.status_code, content=r.json())
+    except Exception as exc:
+        return JSONResponse(status_code=502, content={"error": {"message": f"upstream: {exc}", "type": "api_error", "code": "upstream_error"}})
+
+
 @app.post("/dashboard/api/consumers/{consumer}")
 async def dashboard_update_consumer(consumer: str, request: Request) -> Response:
     ctx, error = _require_admin_dashboard_auth(request)

diff --git a/llm_router_host.py b/llm_router_host.py
@@ -25,6 +25,7 @@
 import route_latency as _route_latency
 import route_tool_capability as _route_tool_capability
 import route_cache as _route_cache
+import route_session_meter as _route_session_meter
 
 
 def _cached_tokens(usage: dict) -> "int | None":
@@ -293,6 +294,8 @@ async def run_node(nid, node, prompt):
                     "price_out": chosen.get("price_out"),
                     "tokens_in": resp.get("tokens_in"),
                     "tokens_out": resp.get("tokens_out"),
+                    "tokens_cached": resp.get("tokens_cached"),
+                    "cost_reported": resp.get("cost_reported"),
                     # this node's own latency, so the dashboard shows WHICH node is
                     # the slow one in a flow (e.g. a 12s antseed glm-5.2 node vs a
                     # 0.9s gpt-5.5 node).
@@ -305,6 +308,27 @@ async def run_node(nid, node, prompt):
         nodes = fr.get("trace") or []
         tok_in = sum((n.get("tokens_in") or 0) for n in nodes) or None
         tok_out = sum((n.get("tokens_out") or 0) for n in nodes) or None
+        tok_cached = sum((n.get("tokens_cached") or 0) for n in nodes) or None
+        # The flow's synthetic chosen ("flow") has no price, so the shim can't
+        # compute cost — aggregate it here, per node, the same way shim's
+        # _executed_cost_usd does: prefer the provider-reported cost, else compute
+        # from the node's ranked price discounting cache-read tokens (~10x). Sum
+        # is surfaced as the flow's cost_reported so the shim uses it verbatim.
+        def _node_cost(n):
+            rep = n.get("cost_reported")
+            if isinstance(rep, (int, float)) and not isinstance(rep, bool) and rep >= 0:
+                return float(rep)
+            pin, pout = n.get("price_in"), n.get("price_out")
+            if pin is None and pout is None:
+                return None
+            tin = n.get("tokens_in") or 0
+            cached = n.get("tokens_cached") or 0
+            uncached = max(0, tin - cached)
+            return (uncached / 1e6 * (pin or 0)
+                    + cached / 1e6 * (pin or 0) * 0.1
+                    + (n.get("tokens_out") or 0) / 1e6 * (pout or 0))
+        _costs = [c for c in (_node_cost(n) for n in nodes) if c is not None]
+        flow_cost = round(sum(_costs), 6) if _costs else None
         base_trace = {"policy_fingerprint": None, "flow_fingerprint": fp,
                       "flow_nodes": nodes}
         if not fr.get("ok"):
@@ -331,7 +355,8 @@ async def run_node(nid, node, prompt):
             "response": {"text": fr.get("text") or "",
                          "tool_calls": final_tool_calls or None,
                          "finish_reason": "tool_calls" if final_tool_calls else "stop",
-                         "tokens_in": tok_in, "tokens_out": tok_out},
+                         "tokens_in": tok_in, "tokens_out": tok_out,
+                         "tokens_cached": tok_cached, "cost_reported": flow_cost},
             "chosen": {"provider_id": "flow", "model_family": "flow:" + fp,
                        "served_model_id": "flow:" + fp},
             "trace": base_trace,
@@ -848,6 +873,11 @@ def _fold_route_outcome(request: dict, result: dict,
     # cache_hot field marks it and a cache-aware policy keeps it sticky. Same one
     # route identity as reliability/latency; no-op when the caller named no session.
     _route_cache.observe(session, rkey, ok)
+    # Record the warm route for the per-session display panel (which family is
+    # warm, on which provider, via which peer/backend). Success only, like the
+    # affinity fold. Display-only; the routing decision stays in route_cache.
+    if ok and session:
+        _route_session_meter.observe_route(session, pid, fam, peer_id or pid)
     # Learned tool capability is a marketplace concern only (static/partner routes
     # declare their capabilities in config), so keep it peer-scoped.
     if peer_id:

diff --git a/route_session_meter.py b/route_session_meter.py
@@ -17,16 +17,31 @@
 
 _lock = threading.Lock()
 _acc: dict[str, dict] = {}   # session -> running totals
+# session -> {model_family -> {family, provider, served_by}}: the routes that
+# successfully served the session, i.e. which models/peers are WARM (hold the
+# session's prompt-cache prefix). For DISPLAY (the warm panel); the affinity
+# decision stays in route_cache. Per family, so a flow's glm AND gpt both show.
+_warm: dict[str, dict[str, dict]] = {}
+# session -> owning consumer key (the authed caller that FIRST wrote this sid).
+# A session's economics (cost/tokens/cache + warm peers) belong to one consumer;
+# this binding lets the consumer-facing view refuse to disclose another
+# consumer's session (cross-consumer isolation). First-writer-wins: a different
+# consumer reusing someone else's opaque sid must NOT steal or overwrite it.
+_owner: dict[str, str] = {}
 
 
 def observe(session: "str | None", *, tokens_in=0, tokens_out=0,
-            tokens_cached=0, cost_usd=0.0) -> "dict | None":
+            tokens_cached=0, cost_usd=0.0, owner: "str | None" = None) -> "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."""
+    No-op (returns None) when the caller named no session. When `owner` (the
+    authed consumer key) is given, bind sid->owner first-writer-wins so the
+    consumer-facing view can scope reads to the owning consumer."""
     if not session:
         return None
     with _lock:
+        if owner:
+            _owner.setdefault(session, owner)
         a = _acc.get(session)
         if a is None:
             a = {"calls": 0, "tokens_in": 0, "tokens_out": 0,
@@ -40,6 +55,40 @@ def observe(session: "str | None", *, tokens_in=0, tokens_out=0,
         return dict(a)
 
 
+def observe_route(session: "str | None", provider: "str | None",
+                  family: "str | None", served_by: "str | None") -> None:
+    """Record (for DISPLAY) that `family` was served warm for this session by
+    `provider` via `served_by` (the peer / real backend behind a marketplace, or
+    the provider itself for direct routes). Keyed per family so a multi-family
+    flow shows all warm models. No-op without a session/family."""
+    if not session or not family:
+        return
+    with _lock:
+        w = _warm.get(session)
+        if w is None:
+            w = {}
+            _warm[session] = w
+        w[family] = {"family": family, "provider": provider,
+                     "served_by": served_by or provider}
+
+
+def warm(session: "str | None") -> list[dict]:
+    """The session's warm routes: [{family, provider, served_by}], one per family."""
+    if not session:
+        return []
+    with _lock:
+        return list((_warm.get(session) or {}).values())
+
+
+def owner(session: "str | None") -> "str | None":
+    """The consumer key that owns this session (first writer), or None for an
+    unknown / unnamed session. Used to scope the consumer-facing session view."""
+    if not session:
+        return None
+    with _lock:
+        return _owner.get(session)
+
+
 def get(session: "str | None") -> "dict | None":
     if not session:
         return None
@@ -57,3 +106,5 @@ def reset() -> None:
     """Test hook."""
     with _lock:
         _acc.clear()
+        _warm.clear()
+        _owner.clear()