#!/usr/bin/env python3 """Shared efficiency + cost collection for Loki Mode. This module is the single source of truth for reading per-iteration efficiency records out of .loki/metrics/efficiency/ and turning them into a cost dict. It was extracted from autonomy/lib/proof-generator.py (R1) so that both the proof generator and the R2 benchmark adapters compute cost the same way ("bench cost == proof cost"). proof-generator.py has a hyphen in its name and is therefore not importable as a module, which is why the logic lives here instead. Behavior contract (preserved from the R1 proof generator): - cost.usd is None when NO valid efficiency record was read (cost was never collected for this run). A skeptic seeing "$0.00" assumes the artifact is fake; "cost not recorded" is the honest signal. - Token counts are also None and available is False when no record exists. - A genuine 0.0 (records existed but summed to zero) is preserved as 0.0. - usd is rounded to 4 decimals only when records were collected. The token->USD pricing helper (price_from_tokens) lives here too so that adapters and the report compute uniform cost from a single dated price table (benchmarks/bench/prices.json) when a tool reports tokens but no native cost_usd. This is additive: the proof generator does not use it. """ import json import os __all__ = [ "collect_efficiency", "record_is_measured", "read_child_ledger", "load_prices", "price_from_tokens", "inference_geo_multiplier", "DEFAULT_PRICES_PATH", ] # benchmarks/bench/prices.json relative to the repo root. Resolved lazily so # importing this module never depends on cwd or on the file existing. _HERE = os.path.dirname(os.path.abspath(__file__)) _REPO_ROOT = os.path.dirname(os.path.dirname(_HERE)) # autonomy/lib -> repo DEFAULT_PRICES_PATH = os.path.join( _REPO_ROOT, "benchmarks", "bench", "prices.json" ) # --------------------------------------------------------------------------- # small helpers (self-contained copies; proof-generator.py keeps its own) # --------------------------------------------------------------------------- # Price multiplier for LOKI_INFERENCE_GEO, read from the same file the TS gate # reads (inference_geo_multipliers in loki-ts/data/model-pricing.json). Unset, # global and unknown values are 1.0; an unreadable file falls back to 1.1 for # "us" so a budget never silently under-counts. MODEL_PRICING_PATH = os.path.join(_REPO_ROOT, "loki-ts", "data", "model-pricing.json") def inference_geo_multiplier(env=None, pricing_path=None): geo = ((env if env is not None else os.environ).get("LOKI_INFERENCE_GEO") or "").strip().lower() if not geo: return 1.0 table = None try: with open(pricing_path or MODEL_PRICING_PATH, "r") as f: table = json.load(f).get("inference_geo_multipliers") except Exception: table = None if not isinstance(table, dict): table = {"us": 1.1} m = table.get(geo) return float(m) if isinstance(m, (int, float)) and m > 0 else 1.0 def _read_json(path, default=None): try: with open(path, "r") as f: return json.load(f) except Exception: return default def _to_int(v, default=0): try: return int(v) except Exception: return default def _to_float(v, default=0.0): try: return float(v) except Exception: return default _MEASURED_FIELDS = ( "cost_usd", "input_tokens", "output_tokens", "cache_read_tokens", "cache_creation_tokens", ) def record_is_measured(rec): """True when ONE efficiency record actually carries an observed value. THE SINGLE DEFINITION of "measured" for a per-iteration record. collect_ efficiency() applies the same rule to the SUM; anything wanting the rule per iteration (cost-summary.py) must import this rather than restate it. A second copy of this predicate is how the honesty rule drifts: the four surfaces that once rendered an unmeasured run as "$0.00" each had their own idea of what counted as measured. A present file is not a measurement. A run that did work necessarily consumed tokens, so all-zeros means we FAILED TO MEASURE, and unmeasured must read as unknown, never as free. """ if not isinstance(rec, dict): return False for key in _MEASURED_FIELDS: v = rec.get(key) if isinstance(v, bool): continue if isinstance(v, (int, float)) and v: return True return False # --------------------------------------------------------------------------- # efficiency collection (extracted verbatim from proof-generator.py) # --------------------------------------------------------------------------- def collect_efficiency(loki_dir): """Sum cost + tokens across .loki/metrics/efficiency/iteration-*.json. Returns (cost dict, best-effort model name (last non-empty seen)). Credibility: unavailable cost and token values are None when NO valid efficiency record was read. A skeptic seeing "$0.00" assumes the artifact is fake; "cost not recorded" is the honest signal. A genuine 0.0 from a present record is preserved as 0.0. """ cost = { "usd": 0.0, "input_tokens": 0, "output_tokens": 0, "cache_read_tokens": 0, "cache_creation_tokens": 0, "available": False, } model = "" collected = False _records = [] eff_dir = os.path.join(loki_dir, "metrics", "efficiency") try: names = sorted(os.listdir(eff_dir)) except Exception: names = [] for name in names: if not (name.startswith("iteration-") and name.endswith(".json")): continue rec = _read_json(os.path.join(eff_dir, name), default=None) if not isinstance(rec, dict): continue collected = True _records.append(rec) cost["usd"] += _to_float(rec.get("cost_usd")) cost["input_tokens"] += _to_int(rec.get("input_tokens")) cost["output_tokens"] += _to_int(rec.get("output_tokens")) cost["cache_read_tokens"] += _to_int(rec.get("cache_read_tokens")) cost["cache_creation_tokens"] += _to_int(rec.get("cache_creation_tokens")) if rec.get("model"): model = str(rec.get("model")) # A record EXISTING is not the same as a record carrying data. `collected` # was true for any parseable file, so a run whose records were all zeros # reported usd=0.0 with available=True -- the receipt asserting the run cost # NOTHING. That is a fabricated fact, and the receipt exists to prevent # exactly that. # # Measured on the real FireLater run: four efficiency records, every token # field 0 (codex wrote no usage before v8.51.0), and collect_efficiency # returned {'usd': 0.0, ..., 'available': True}. # # A run that did work necessarily consumed tokens. So "available" now means # at least one record carried a non-zero token count or cost -- an OBSERVED # value, not a present file. Zero everywhere means we failed to measure, and # unmeasured must read as unknown. # # Applied through record_is_measured() so the aggregate rule and the # per-iteration rule are literally the same code (see that docstring). # The cost dict keys "usd" where a record says "cost_usd", so map across. _observed = record_is_measured({ "cost_usd": cost["usd"], "input_tokens": cost["input_tokens"], "output_tokens": cost["output_tokens"], "cache_read_tokens": cost["cache_read_tokens"], "cache_creation_tokens": cost["cache_creation_tokens"], }) # COST AND TOKENS ARE MEASURED SEPARATELY, so they must be reported # separately. `_observed` is true when EITHER a cost or a token count was # seen, which is right for "did we measure anything" -- but it let a run with # real token counts and NO priced record report usd=0.0 available=True. The # receipt then asserts the run cost nothing while its own token counts prove # work happened, which is a fabricated fact of exactly the kind this module # exists to prevent. Measured, before this fix: # [{"input_tokens":100,"output_tokens":200}] -> usd=0.0 available=True # A partially-priced run has the same shape: an iteration missing cost_usd # contributes 0.0 to the sum and silently understates total spend. # # So usd is now null unless at least one record actually carried a cost, and # cost_partial marks the case where some did and some did not. Tokens are # unaffected: they keep reporting whatever was observed. # PRESENCE of the key, not its value. An explicit {"cost_usd": 0} is a # GENUINE measured zero and must stay 0.0 -- a free cache-hit iteration is a # real observation, and nulling it would be its own dishonesty (existing # contract: test_genuine_zero_cost_stays_zero_not_null). What is unknown is a # record that never carried the field at all. # a record flagged tokens_measured:false (an ambiguous resumed session) carries no usage keys. One of them makes every # token total NOT RECORDED (None); the flag keys are added only then, so a complete run's dict is unchanged. _unmeasured_tokens = sum(1 for r in _records if r.get("tokens_measured") is False) _any_cost = any("cost_usd" in r for r in _records) _missing_cost = sum(1 for r in _records if "cost_usd" not in r) if collected and _observed: # Round usd to a sane precision but keep it precise (anti-pattern: # round suspiciously-clean numbers). 4 decimals preserves odd values. cost["usd"] = round(cost["usd"], 4) cost["available"] = True if not _any_cost: # Tokens were observed but nothing was priced: unknown, not zero. cost["usd"] = None cost["cost_available"] = False else: cost["cost_available"] = True # Some records priced, others not: the total is a LOWER BOUND. cost["cost_partial"] = _missing_cost > 0 cost["cost_unpriced_records"] = _missing_cost if _unmeasured_tokens: for key in ("input_tokens", "output_tokens", "cache_read_tokens", "cache_creation_tokens"): cost[key] = None cost["tokens_partial"] = True cost["tokens_unmeasured_records"] = _unmeasured_tokens else: # No record means unavailable, never an observed zero. for key in ( "usd", "input_tokens", "output_tokens", "cache_read_tokens", "cache_creation_tokens", ): cost[key] = None _merge_child_ledger(loki_dir, cost) return cost, model # --------------------------------------------------------------------------- # child-session ledger (S1 of the cost program) # --------------------------------------------------------------------------- LEDGER_RELPATH = os.path.join("metrics", "child-sessions.jsonl") _LEDGER_TOKEN_KEYS = ( "input_tokens", "output_tokens", "cache_read_tokens", "cache_creation_tokens", ) def cost_block_is_partial(cost): """The one partial-cost check (FC-COST-PARTIAL-b): True when a cost block's dollar total is a lower bound or unknown, so consumers label it and never print it as complete.""" if not isinstance(cost, dict): return False if cost.get("cost_partial") is True or cost.get("cost_available") is False: return True return any( isinstance(cost.get(k), int) and not isinstance(cost.get(k), bool) and cost.get(k) > 0 for k in ("ledger_invalid_lines", "cost_unpriced_records")) def read_child_ledger(loki_dir): """Read .loki/metrics/child-sessions.jsonl, one JSON object per child session. Record format: phase, model, input_tokens, output_tokens, cache_creation_tokens, cache_read_tokens, usd, source. Returns (records, invalid_lines) or (None, 0) when the file is absent or unreadable. None is the honest "no ledger" signal: callers must not treat it as zero child spend. Blank lines are ignored; non-JSON lines and non-object lines are skipped and counted. """ path = os.path.join(loki_dir, LEDGER_RELPATH) try: with open(path, "r", encoding="utf-8") as fh: lines = fh.read().splitlines() except Exception: return None, 0 records, invalid = [], 0 for line in lines: line = line.strip() if not line: continue try: rec = json.loads(line) except Exception: invalid += 1 continue if isinstance(rec, dict): records.append(rec) else: invalid += 1 return records, invalid def _merge_child_ledger(loki_dir, cost): """Fold the child-session ledger into the cost dict, in place. The main RARV session's efficiency record is only one of 11-20 child `claude -p` sessions per run (grill, reviewers, council, docgen); summing it alone undercounts true spend. Always sets cost["ledger"] to "present" or "absent" so a missing ledger is visible, never a silent zero. A ledger line without a usd value is unpriced (counted, never treated as 0.0), and the total then becomes a lower bound via cost_partial. """ records, invalid = read_child_ledger(loki_dir) main_usd = cost.get("usd") by_phase = {} if main_usd is not None: by_phase["implement"] = main_usd if records is None: cost["ledger"] = "absent" if by_phase: cost["cost_by_phase"] = by_phase return cost["ledger"] = "present" cost["ledger_records"] = len(records) cost["ledger_invalid_lines"] = invalid if invalid > 0: # A line that could not be parsed is spend we cannot see: lower bound. cost["cost_partial"] = True if not records: if by_phase: cost["cost_by_phase"] = by_phase return ledger_usd, priced, unpriced = 0.0, 0, 0 main_known = bool(cost.get("available")) # Main records exist but none carried a dollar figure (cost_available is # False, set only by the token-only branch): the ledger-only total is a # lower bound, never complete. main_unpriced = main_known and cost.get("cost_available") is False tokens = {k: cost.get(k) for k in _LEDGER_TOKEN_KEYS} if not main_known: tokens = {k: 0 for k in _LEDGER_TOKEN_KEYS} for rec in records: usd = rec.get("usd") if isinstance(usd, (int, float)) and not isinstance(usd, bool): ledger_usd += float(usd) priced += 1 phase = str(rec.get("phase") or "unknown") by_phase[phase] = by_phase.get(phase, 0.0) + float(usd) else: unpriced += 1 for k in _LEDGER_TOKEN_KEYS: if tokens[k] is not None: tokens[k] += _to_int(rec.get(k)) cost.update(tokens) cost["available"] = True if priced: cost["usd"] = round((main_usd or 0.0) + ledger_usd, 4) cost["cost_available"] = True partial = bool(cost.get("cost_partial")) or unpriced > 0 or not main_known or main_unpriced cost["cost_partial"] = partial cost["cost_unpriced_records"] = int(cost.get("cost_unpriced_records") or 0) + unpriced if by_phase: cost["cost_by_phase"] = {k: round(v, 4) for k, v in by_phase.items()} # --------------------------------------------------------------------------- # uniform token -> USD pricing (R2 benchmark; not used by the proof generator) # --------------------------------------------------------------------------- def load_prices(path=None): """Load the shared dated price table. Returns {} if missing/unreadable. Never raises: a missing price table means we cannot price tokens, which is an honest null, not a fabricated zero. """ p = path or DEFAULT_PRICES_PATH data = _read_json(p, default=None) if not isinstance(data, dict): return {} return data def price_from_tokens(model, input_tokens, output_tokens, cache_read_tokens=0, prices=None, prices_path=None): """Compute a USD cost from raw token counts and the shared price table. Returns a float (rounded to 6 decimals, times inference_geo_multiplier()) when the model is present in the price table, else None. None means "cannot price" (honest), never 0.0. Pricing is per million tokens (mtok). cache_read is priced separately from fresh input tokens. Used for tools that expose token counts but not a native cost_usd; keeps cross-tool cost comparison on one pricing basis. """ table = prices if prices is not None else load_prices(prices_path) models = table.get("models") if isinstance(table, dict) else None if not isinstance(models, dict): return None entry = models.get(model) if not isinstance(entry, dict): return None in_rate = _to_float(entry.get("input_per_mtok"), None) out_rate = _to_float(entry.get("output_per_mtok"), None) cache_rate = _to_float(entry.get("cache_read_per_mtok"), 0.0) if in_rate is None or out_rate is None: return None it = _to_int(input_tokens) ot = _to_int(output_tokens) ct = _to_int(cache_read_tokens) usd = ( (it / 1_000_000.0) * in_rate + (ot / 1_000_000.0) * out_rate + (ct / 1_000_000.0) * cache_rate ) # LOKI_INFERENCE_GEO: same multiplier as the budget gate and dashboard. Only # token-estimated cost reaches here; a provider-reported cost_usd never does. return round(usd * inference_geo_multiplier(), 6)