fix(#000057): real token usage + cost per input/output separately

fox 2026-05-21: (1) use REAL API token usage, not len//4; (2) the
substrate prefills a large retrieved CONTEXT as INPUT while solo feeds
~nothing, so per-completion-token over-charges the substrate — and per-
TOTAL-token UNDER-charges it (its mix is ~98% cheap prefill tokens).
Measured n=30 qwen-nothink/4090: substrate prefills ~6.6k input tok/query
(claim_lattice) vs solo ~52 — ~127x. Neither single per-token denominator
is honest; prefill (parallel, cheap/tok) and decode (autoregressive,
dear/tok) must be costed separately.

- OpenAICompatibleClient stashes data['usage'] as .last_usage (non-
  invasive; return type unchanged).
- watt_bench captures real prompt_tokens + completion_tokens per call
  (both arms), aggregates per cell, and energy_cogs reports gross +
  marginal per BOTH 1k-total-tok and 1k-completion-tok plus the context
  size. Prints the prompt/completion split.
- 12 tests incl. the prompt-context artifact (per-total cheap, per-
  completion dear). Full suite 2540 passed.

The clean per-input-tok / per-output-tok split rides bench/watt_calibrate
(slope calibration; separate commit once validated live).

arborist/qa/client.py

@@ -99,6 +99,11 @@ class OpenAICompatibleClient:
         # connection pool serializes pool access. Closed via close().
         import httpx
         self._http = httpx.Client(timeout=self.timeout)
+        # Token usage from the most recent chat_completion (prompt /
+        # completion / total tokens), or None. Lets energy benches cost
+        # per ACTUAL tokens — incl. the prompt/context the substrate
+        # prefills — instead of a char//4 estimate. See bench/watt_bench.
+        self.last_usage: dict | None = None
 
     def close(self) -> None:
         """Release the underlying connection pool."""
@@ -188,6 +193,7 @@ class OpenAICompatibleClient:
                     continue
                 resp.raise_for_status()
                 data = resp.json()
+                self.last_usage = data.get("usage")
                 return data["choices"][0]["message"]["content"]
             except httpx.HTTPStatusError as e:
                 # Retry only the configured transient codes; raise others.

bench/watt_bench.py

@@ -351,64 +351,82 @@ def _make_client(endpoint: str):
         api_key=os.environ.get("ARBORIST_LLM_API_KEY"))
 
 
-def _solo_call(client, mkey: str, question: str) -> tuple[str, int]:
+def _usage_tokens(client, answer: str) -> tuple[int, int]:
-    """One solo completion. Returns (answer, completion_tokens). The
+    """(prompt_tokens, completion_tokens) from the client's last call.
-    bench client returns only the string; for token counts we re-issue
+
-    via a thin httpx call when usage is wanted — but to keep the energy
+    Real API usage when the endpoint reports it; falls back to a char//4
-    path simple we estimate tokens as len//4 when usage isn't surfaced."""
+    estimate for completion (prompt 0) only when usage is absent."""
+    u = getattr(client, "last_usage", None) or {}
+    p = u.get("prompt_tokens")
+    c = u.get("completion_tokens")
+    if c is None:
+        c = max(1, len(answer) // 4)
+    return int(p or 0), int(c)
+
+
+def _solo_call(client, mkey: str, question: str) -> tuple[str, int, int]:
+    """One solo completion. Returns (answer, prompt_tokens, completion_tokens)
+    from the endpoint's real usage. Solo's prompt is just system+question —
+    near-zero context — which is exactly the input asymmetry vs the
+    substrate arm (which prefills a large retrieved context)."""
     cfg = MODELS[mkey]
     ans = client.chat_completion(
         [{"role": "system", "content": _SOLO_SYS},
          {"role": "user", "content": question}],
         model=cfg["model"], max_tokens=cfg["max_tokens"],
         extra_body=cfg["extra"])
-    return ans, max(1, len(ans) // 4)
+    p, c = _usage_tokens(client, ans)
+    return ans, p, c
 
 
 def energy_cogs(gpu_joules: float | None, window_s: float | None,
-                tot_tok: int, idle_mean_w: float | None,
+                completion_tok: int, idle_mean_w: float | None,
                 price_per_kwh: float,
                 serving_floor_w: float | None = None,
                 gen_draw_w: float | None = None,
-                gen_duty: float | None = None) -> dict:
+                gen_duty: float | None = None,
-    """Energy cost-of-goods-sold from MEASURED power states — no hardcodes.
+                prompt_tok: int = 0) -> dict:
+    """Energy cost-of-goods-sold from MEASURED power + REAL token usage.
 
-    fox 2026-05-21: a card occupies DISTINCT power states (idle, middle-
+    Two corrections fox forced (2026-05-21):
-    idle = model resident between requests, generation), and they differ
-    for every card × model × inference-server. So every power number is
-    measured at runtime and the states are kept DISTINCT — we do NOT
-    collapse them into ``joules/window`` and call that "generation"
-    (that blend is a duty-cycle artifact, not a state the card sits at).
 
-      * idle_mean_w     — deep idle (no-request baseline window).
+    1. Power STATES, not a blend. A card occupies distinct states (idle,
-      * serving_floor_w — middle-idle: model resident, between requests
+       middle-idle = model resident between requests, generation), which
-                          (low band of the driven window). The right floor
+       differ per card×model×server. The marginal (energy above the
-                          for marginal cost — the standing cost of being
+       serving floor) is the query's GPU cost; ``gross`` (window integral)
-                          ready to answer.
+       is NOT query-attributable — it charges the always-on floor for the
-      * gen_draw_w      — the actual generation draw (high band), with
+       seconds spent in CPU-side retrieval (GPU free for other traffic).
-                          ``gen_duty`` = fraction of the window generating.
-      * marginal COGS   — THE GPU COST OF A QUERY: joules ABOVE the
-                          serving floor / tokens — the generation burst.
-                          Retrieval/verify are CPU/SQLite (GPU untouched),
-                          so this generation-only number is what the query
-                          actually causes on the GPU. Falls back to idle
-                          floor if no band split.
-      * gross           — wall-clock window joules / tokens. NOT this
-                          query's GPU cost: it charges the always-on
-                          model-resident floor for the seconds we spend in
-                          CPU-side retrieval (the card is free for other
-                          traffic then). Kept for reference, not as COGS.
 
-    kWh = J / 3.6e6; ``$/1k-tok`` is the unit that compares to API pricing.
+    2. Cost per TOTAL tokens processed, not output only. The substrate
-    Only ``price_per_kwh`` is an operator input. See docs/stock-v1-config.md.
+       prefills a large retrieved CONTEXT (``prompt_tok``); solo prefills
+       almost nothing. An LLM request = prefill(all input) + decode(output),
+       so most of the substrate's GPU work is prefilling that context.
+       Dividing energy by completion tokens alone wrongly charges the
+       cost-of-reading-the-evidence to a few output tokens. We report BOTH
+       bases — per total (prompt+completion) is the apples-to-apples
+       compute cost; per completion is the answer-delivery view — and the
+       prompt-token count so the input asymmetry is explicit.
+
+    kWh = J / 3.6e6. ``price_per_kwh`` is the only operator input.
     """
-    if not gpu_joules or not tot_tok or not window_s:
+    completion_tok = completion_tok or 0
+    prompt_tok = prompt_tok or 0
+    total_tok = completion_tok + prompt_tok
+    if not gpu_joules or not window_s or total_tok <= 0:
         return {"available": False, "price_per_kwh": price_per_kwh}
     J_PER_KWH = 3.6e6
+
+    def per_1k(usd, tok):
+        return round(usd / tok * 1000, 6) if tok else None
+
     gross_usd = gpu_joules / J_PER_KWH * price_per_kwh
     out = {
         "available": True,
         "price_per_kwh": price_per_kwh,
+        # real token usage (the substrate's prompt_tok carries the context)
+        "prompt_tokens": prompt_tok,
+        "completion_tokens": completion_tok,
+        "total_tokens": total_tok,
         # measured power states (distinct — not a blend)
         "idle_w_measured": (round(idle_mean_w, 1)
                             if idle_mean_w is not None else None),
@@ -421,7 +439,8 @@ def energy_cogs(gpu_joules: float | None, window_s: float | None,
         "window_s": round(window_s, 1),
         "gross_joules": round(gpu_joules, 1),
         "gross_usd": round(gross_usd, 6),
-        "gross_usd_per_1k_tok": round(gross_usd / tot_tok * 1000, 6),
+        "gross_usd_per_1k_total_tok": per_1k(gross_usd, total_tok),
+        "gross_usd_per_1k_completion_tok": per_1k(gross_usd, completion_tok),
     }
     # Marginal against the serving floor (preferred) or idle (fallback).
     floor_w = serving_floor_w if serving_floor_w is not None else idle_mean_w
@@ -435,7 +454,9 @@ def energy_cogs(gpu_joules: float | None, window_s: float | None,
             "marginal_floor_w": round(floor_w, 1),
             "marginal_joules": round(marginal_joules, 1),
             "marginal_usd": round(marginal_usd, 6),
-            "marginal_usd_per_1k_tok": round(marginal_usd / tot_tok * 1000, 6),
+            "marginal_usd_per_1k_total_tok": per_1k(marginal_usd, total_tok),
+            "marginal_usd_per_1k_completion_tok": per_1k(marginal_usd,
+                                                         completion_tok),
         })
     return out
 
@@ -552,7 +573,7 @@ def main() -> int:
                     tq = time.time()
                     cache_status = "solo_no_cache"
                     if arm == "solo":
-                        ans, ctoks = _solo_call(client, mkey, q)
+                        ans, ptoks, ctoks = _solo_call(client, mkey, q)
                     else:  # arborist
                         from arborist.qa.query import query
                         from bench.stock_v1 import (
@@ -583,7 +604,10 @@ def main() -> int:
                                   extra_body=MODELS[mkey]["extra"],
                                   policy=pol, burn_existing=True)
                         ans = r.get("raw_answer") or r.get("answer_text") or ""
-                        ctoks = max(1, len(ans) // 4)
+                        # Real usage from query()'s internal synthesis call
+                        # (one call — STOCK V.1 repair is off). prompt_tokens
+                        # carries the retrieved CONTEXT the substrate prefills.
+                        ptoks, ctoks = _usage_tokens(client, ans)
                         cache_status = r.get("status", "?")
                         if (cache_status == "cache_hit"
                                 or "cache_hit" in str(r.get("lookup_path", ""))):
@@ -593,7 +617,8 @@ def main() -> int:
                                  "variant": variant,
                                  "latency_s": round(dt, 2),
                                  "answer_chars": len(ans),
-                                 "est_completion_tokens": ctoks,
+                                 "prompt_tokens": ptoks,
+                                 "completion_tokens": ctoks,
                                  "cache_status": cache_status})
         finally:
             client.close()
@@ -603,7 +628,11 @@ def main() -> int:
         bands = probe.band_stats()
         elapsed = time.time() - t0
         nq = len(rows)
-        tot_tok = sum(r["est_completion_tokens"] for r in rows)
+        tot_completion_tok = sum(r.get("completion_tokens",
+                                       r.get("est_completion_tokens", 0))
+                                 for r in rows)
+        tot_prompt_tok = sum(r.get("prompt_tokens", 0) for r in rows)
+        tot_tok = tot_completion_tok  # back-compat alias for joules/token
         gpu_j = st.get("joules")
         cpu_j = cpu_st.get("joules")
         total_j = (gpu_j or 0) + (cpu_j or 0) if (gpu_j or cpu_j) else None
@@ -620,9 +649,11 @@ def main() -> int:
                                         if cpu_j and nq else None),
             "total_joules_per_question": (round(total_j / nq, 1)
                                           if total_j and nq else None),
-            "est_total_completion_tokens": tot_tok,
+            "total_completion_tokens": tot_completion_tok,
-            "gpu_joules_per_token": (round(gpu_j / tot_tok, 3)
+            "total_prompt_tokens": tot_prompt_tok,
-                                     if gpu_j and tot_tok else None),
+            "total_tokens_processed": tot_completion_tok + tot_prompt_tok,
+            "gpu_joules_per_completion_token": (round(gpu_j / tot_completion_tok, 3)
+                                                if gpu_j and tot_completion_tok else None),
             "mean_latency_s": (round(sum(r["latency_s"] for r in rows) / nq, 2)
                                if nq else None),
             # Measured power-state decomposition (idle / serving-floor /
@@ -639,12 +670,13 @@ def main() -> int:
             # input. Window timestamps let a post-hoc load_monitor cross-ref
             # flag organic-traffic contamination under non-isolation.
             "energy_cogs": energy_cogs(
-                gpu_j, st.get("window_s"), tot_tok,
+                gpu_j, st.get("window_s"), tot_completion_tok,
                 (idle or {}).get("mean_w"), a.price_per_kwh,
                 serving_floor_w=(bands.get("low_band_w")
                                  if bands.get("bimodal") else None),
                 gen_draw_w=bands.get("high_band_w"),
-                gen_duty=bands.get("duty_cycle")),
+                gen_duty=bands.get("duty_cycle"),
+                prompt_tok=tot_prompt_tok),
             "window_start_unix": round(t0, 3),
             "window_end_unix": round(t0 + elapsed, 3),
             "watt_samples": (probe.watt_samples()
@@ -673,17 +705,25 @@ def main() -> int:
                   + ("" if b.get("bimodal") else " [unimodal — no gen state]"))
         cg = cell["energy_cogs"]
         if cg.get("available"):
-            # GPU COGS = generation energy ABOVE the always-on model-
+            # Cost per TOTAL tokens processed (prompt+completion) is the
-            # resident floor (the marginal). Retrieval/verify are CPU/
+            # apples-to-apples compute cost: the substrate prefills a large
-            # SQLite — the GPU is untouched then, so the gross window
+            # retrieved CONTEXT (prompt_tokens) that solo lacks, so per
-            # integral OVER-counts (it charges the shared floor during our
+            # completion-token alone over-charges the substrate. GPU COGS =
-            # retrieval gap) and is NOT this query's GPU cost (fox).
+            # generation energy above the model-resident floor (retrieval/
-            print(f"           GPU COGS @${cg['price_per_kwh']}/kWh: "
+            # verify are CPU; GPU untouched). At our scale there is no
-                  + (f"${cg['marginal_usd_per_1k_tok']}/1k-tok generation "
+            # prefill/KV-cache reuse (no policy yet for what's worth keeping
-                     f"(above {cg['marginal_floor']}-floor {cg['marginal_floor_w']}W)"
+            # hot), so that context is prefilled fresh every query.
-                     if "marginal_usd_per_1k_tok" in cg else "n/a")
+            print(f"           ctx: prompt={cg['prompt_tokens']}tok "
-                  + f"  ·  [window-total ${cg['gross_usd_per_1k_tok']}/1k-tok "
+                  f"completion={cg['completion_tokens']}tok "
-                  f"incl. shared idle during retrieval — NOT query-attributable]")
+                  f"(total {cg['total_tokens']})")
+            if "marginal_usd_per_1k_total_tok" in cg:
+                print(f"           GPU COGS @${cg['price_per_kwh']}/kWh: "
+                      f"${cg['marginal_usd_per_1k_total_tok']}/1k-total-tok "
+                      f"· ${cg['marginal_usd_per_1k_completion_tok']}/1k-compl-tok "
+                      f"(gen above {cg['marginal_floor']}-floor "
+                      f"{cg['marginal_floor_w']}W)")
+            print(f"           [window-total (ref, not attributable): "
+                  f"${cg['gross_usd_per_1k_total_tok']}/1k-total-tok]")
 
     for mkey in models:
         run_cell("solo", mkey)

tests/test_watt_cogs.py

@@ -14,17 +14,31 @@ from bench.watt_probe import classify_power_bands
 # ---- energy_cogs -------------------------------------------------------
 
 def test_gross_cogs_one_kwh():
-    c = energy_cogs(gpu_joules=3.6e6, window_s=100.0, tot_tok=1000,
+    c = energy_cogs(gpu_joules=3.6e6, window_s=100.0, completion_tok=1000,
                     idle_mean_w=30.0, price_per_kwh=0.33)
     assert c["available"] is True
     assert c["gross_usd"] == 0.33                  # 1 kWh * $0.33
-    assert c["gross_usd_per_1k_tok"] == 0.33
+    assert c["gross_usd_per_1k_total_tok"] == 0.33
+    assert c["gross_usd_per_1k_completion_tok"] == 0.33
     assert c["window_mean_w"] == 36000.0           # blend, honestly labelled
 
 
+def test_prompt_context_makes_per_total_cheap_per_completion_dear():
+    # The fox correction: substrate prefills a big CONTEXT (prompt_tok).
+    # Per TOTAL token processed it is cheap; charging it all to the few
+    # completion tokens makes it look expensive — that's the artifact.
+    c = energy_cogs(gpu_joules=3.6e6, window_s=100.0, completion_tok=1000,
+                    idle_mean_w=30.0, price_per_kwh=0.33, prompt_tok=9000)
+    assert c["prompt_tokens"] == 9000
+    assert c["total_tokens"] == 10000
+    assert c["gross_usd_per_1k_total_tok"] == 0.033       # 0.33 / 10000 *1k
+    assert c["gross_usd_per_1k_completion_tok"] == 0.33   # 10x dearer
+    assert (c["gross_usd_per_1k_completion_tok"]
+            > c["gross_usd_per_1k_total_tok"])
+
+
 def test_marginal_uses_serving_floor_when_present():
-    # Serving floor (middle-idle 60W) is the right marginal floor, not idle.
+    c = energy_cogs(gpu_joules=3.6e6, window_s=100.0, completion_tok=1000,
-    c = energy_cogs(gpu_joules=3.6e6, window_s=100.0, tot_tok=1000,
                     idle_mean_w=30.0, price_per_kwh=0.33,
                     serving_floor_w=60.0, gen_draw_w=400.0, gen_duty=0.5)
     assert c["marginal_floor"] == "serving"
@@ -35,15 +49,16 @@ def test_marginal_uses_serving_floor_when_present():
 
 
 def test_marginal_falls_back_to_idle_without_serving_floor():
-    c = energy_cogs(gpu_joules=3.6e6, window_s=100.0, tot_tok=1000,
+    c = energy_cogs(gpu_joules=3.6e6, window_s=100.0, completion_tok=1000,
                     idle_mean_w=30.0, price_per_kwh=0.33)
     assert c["marginal_floor"] == "idle"
     assert c["marginal_joules"] == 3.6e6 - 30.0 * 100.0
-    assert c["marginal_usd_per_1k_tok"] < c["gross_usd_per_1k_tok"]
+    assert (c["marginal_usd_per_1k_total_tok"]
+            < c["gross_usd_per_1k_total_tok"])
 
 
 def test_marginal_clamps_to_zero_when_floor_exceeds_load():
-    c = energy_cogs(gpu_joules=5000.0, window_s=100.0, tot_tok=10,
+    c = energy_cogs(gpu_joules=5000.0, window_s=100.0, completion_tok=10,
                     idle_mean_w=30.0, price_per_kwh=0.33,
                     serving_floor_w=100.0)
     assert c["marginal_joules"] == 0.0
@@ -51,16 +66,18 @@ def test_marginal_clamps_to_zero_when_floor_exceeds_load():
 
 
 def test_no_floor_gives_gross_only():
-    c = energy_cogs(gpu_joules=3.6e6, window_s=100.0, tot_tok=1000,
+    c = energy_cogs(gpu_joules=3.6e6, window_s=100.0, completion_tok=1000,
                     idle_mean_w=None, price_per_kwh=0.33)
     assert c["available"] is True
-    assert "marginal_usd_per_1k_tok" not in c   # no floor -> can't decompose
+    assert "marginal_usd_per_1k_total_tok" not in c
 
 
 def test_unavailable_when_inputs_missing():
     assert energy_cogs(None, 100.0, 1000, 30.0, 0.33)["available"] is False
     assert energy_cogs(3.6e6, None, 1000, 30.0, 0.33)["available"] is False
-    assert energy_cogs(3.6e6, 100.0, 0, 30.0, 0.33)["available"] is False
+    # zero completion AND zero prompt -> no tokens at all
+    assert energy_cogs(3.6e6, 100.0, 0, 30.0, 0.33,
+                       prompt_tok=0)["available"] is False
 
 
 def test_price_is_the_only_lever():