diff --git a/CHANGELOG.md b/CHANGELOG.md index 7376a60c..d74b5825 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,10 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/) and this project adheres to [Semantic Versioning](https://semver.org/). +## [Unreleased] +### Added +- Added display of live token cost estimation during the run. + ## [0.4.1] - 2026-05-23 ### Fixed - Fixed several mismatches in task definitions and the provided extra information. diff --git a/src/clawbench/runner/run.py b/src/clawbench/runner/run.py index 04dca5c5..e6b3b235 100644 --- a/src/clawbench/runner/run.py +++ b/src/clawbench/runner/run.py @@ -191,7 +191,9 @@ def main(): time_limit_s = int(float(task["time_limit"]) * 60) except (OSError, json.JSONDecodeError, ValueError) as e: duration = time.time() - start_time - classification = classify_run(output_dir, False, "task_data") + classification = classify_run( + output_dir, False, "task_data", model_cfg=model_cfg + ) meta = make_run_meta( task=task, task_json_sha256=task_json_sha256, @@ -222,7 +224,9 @@ def main(): docker_build(args.harness) except SystemExit as e: duration = time.time() - start_time - classification = classify_run(output_dir, False, "infra_failure") + classification = classify_run( + output_dir, False, "infra_failure", model_cfg=model_cfg + ) meta = make_run_meta( task=task, task_json_sha256=task_json_sha256, @@ -336,7 +340,7 @@ def handle_sigint(sig, frame): step(f"Agent running (max {task['time_limit']}min)") phase = "waiting_for_container" - docker_wait(container) + docker_wait(container, model_cfg=None if args.human else model_cfg) phase = "container_logs" step("Container logs") @@ -352,7 +356,10 @@ def handle_sigint(sig, frame): phase = "printing_results" step("Results") - intercepted = print_results(output_dir) + intercepted = print_results( + output_dir, + model_cfg=None if args.human else model_cfg, + ) # Stage 2 — LLM judge (default on, --no-judge to skip). # Only invoked when stage 1 (intercepted) succeeded; otherwise the @@ -408,7 +415,7 @@ def handle_sigint(sig, frame): # Write run metadata phase = "writing_run_meta" duration = time.time() - start_time - classification = classify_run(output_dir, intercepted) + classification = classify_run(output_dir, intercepted, model_cfg=model_cfg) meta = make_run_meta( task=task, task_json_sha256=task_json_sha256, @@ -461,7 +468,7 @@ def handle_sigint(sig, frame): except Exception: pass duration = time.time() - start_time - classification = classify_run(output_dir, False, category) + classification = classify_run(output_dir, False, category, model_cfg=model_cfg) meta = make_run_meta( task=task, task_json_sha256=task_json_sha256, diff --git a/src/clawbench/runner/run_support/docker.py b/src/clawbench/runner/run_support/docker.py index daa3af5e..75a80f6d 100644 --- a/src/clawbench/runner/run_support/docker.py +++ b/src/clawbench/runner/run_support/docker.py @@ -21,6 +21,11 @@ IMAGE, harness_image, ) +from clawbench.runner.run_support.usage import ( + fetch_openrouter_pricing, + format_usage_status, + summarize_usage_text, +) from clawbench.utils.paths import DOCKER_CONTEXT_ROOT, HARNESS_ROOT console = Console() @@ -402,13 +407,71 @@ def docker_run( run([*env_flags, harness_image(harness)]) -def docker_wait(name: str) -> None: +def _container_usage_summary( + name: str, + model_cfg: dict | None, + pricing_models: dict[str, dict] | None, +) -> dict | None: + try: + r = subprocess.run( + [ + ENGINE, + "exec", + name, + "sh", + "-c", + ( + "if [ -s /data/agent-messages.jsonl ]; then " + "cat /data/agent-messages.jsonl; " + "elif [ -s /root/.openclaw/agents/main/sessions/clawbench.jsonl ]; then " + "cat /root/.openclaw/agents/main/sessions/clawbench.jsonl; " + "elif [ -s /tmp/hermes-live-agent-messages.jsonl ]; then " + "cat /tmp/hermes-live-agent-messages.jsonl; " + "elif [ -s /data/agent-messages.raw.jsonl ]; then " + "cat /data/agent-messages.raw.jsonl; " + "elif [ -s /tmp/codex-stdout.jsonl ]; then " + "cat /tmp/codex-stdout.jsonl; " + "else " + "p=$(find /root/.codex/sessions -name 'rollout-*.jsonl' " + "-type f -printf '%T@ %p\\n' 2>/dev/null | sort -rn | " + "head -1 | cut -d' ' -f2-); " + 'if [ -n "$p" ] && [ -s "$p" ]; then cat "$p"; ' + "else " + "p=$(ls -t /root/workspace/.claw/sessions/*/*.jsonl " + "2>/dev/null | head -1); " + 'if [ -n "$p" ] && [ -s "$p" ]; then cat "$p"; fi; ' + "fi; " + "fi" + ), + ], + capture_output=True, + text=True, + timeout=10, + ) + except (subprocess.TimeoutExpired, subprocess.SubprocessError): + return None + if r.returncode != 0: + return None + return summarize_usage_text( + r.stdout, + model_cfg=model_cfg, + pricing_models=pricing_models, + ) + + +def docker_wait(name: str, model_cfg: dict | None = None) -> None: """Block until the container exits, showing a live status line.""" start = time.time() proc = subprocess.Popen( [ENGINE, "wait", name], stdout=subprocess.PIPE, stderr=subprocess.PIPE ) last_actions = 0 + usage_summary: dict | None = None + pricing_models: dict[str, dict] | None = None + if model_cfg and "openrouter.ai" in str(model_cfg.get("base_url", "")): + pricing_models = fetch_openrouter_pricing( + base_url=str(model_cfg.get("base_url") or "") + ) with Status("[dim]starting...[/]", console=console) as status: while proc.poll() is None: elapsed = int(time.time() - start) @@ -427,14 +490,30 @@ def docker_wait(name: str) -> None: pass except (subprocess.TimeoutExpired, subprocess.SubprocessError): pass - status.update(f"[dim]{mins:02d}:{secs:02d} • {last_actions} actions[/]") + usage_summary = _container_usage_summary(name, model_cfg, pricing_models) + usage_part = ( + format_usage_status(usage_summary) + if usage_summary is not None + else "tokens pending" + ) + status.update( + f"[dim]{mins:02d}:{secs:02d} • {last_actions} actions • " + f"{usage_part}[/]" + ) try: proc.wait(timeout=5) except subprocess.TimeoutExpired: pass elapsed = int(time.time() - start) mins, secs = divmod(elapsed, 60) - console.print(f" Container exited ({mins}m{secs:02d}s, {last_actions} actions)") + usage_part = ( + f", {format_usage_status(usage_summary)}" + if usage_summary is not None and usage_summary.get("total_tokens") + else "" + ) + console.print( + f" Container exited ({mins}m{secs:02d}s, {last_actions} actions{usage_part})" + ) def docker_copy(name: str, output_dir: Path) -> None: diff --git a/src/clawbench/runner/run_support/metadata.py b/src/clawbench/runner/run_support/metadata.py index 0c6da649..b695d42e 100644 --- a/src/clawbench/runner/run_support/metadata.py +++ b/src/clawbench/runner/run_support/metadata.py @@ -245,6 +245,7 @@ def make_run_meta( "adjusted_eligible": classification["adjusted_eligible"], "infra_flags": classification["infra_flags"], "run_metrics": classification["metrics"], + "usage": classification["metrics"].get("usage"), "runtime": _runtime_meta(harness), "task": _task_meta( task=task, diff --git a/src/clawbench/runner/run_support/results.py b/src/clawbench/runner/run_support/results.py index 72e4f63b..f5c9e0ae 100644 --- a/src/clawbench/runner/run_support/results.py +++ b/src/clawbench/runner/run_support/results.py @@ -4,6 +4,11 @@ from pathlib import Path from typing import Any +from clawbench.runner.run_support.usage import ( + format_usage_summary, + summarize_usage_file, +) + INFRA_STOP_REASONS = { "chrome_cdp_timeout", "gateway_failed", @@ -81,7 +86,10 @@ def _line_has_api_or_credit_evidence(line: str) -> bool: return any(pattern in lowered for pattern in API_OR_CREDIT_PATTERNS) -def collect_run_metrics(output_dir: Path) -> dict[str, Any]: +def collect_run_metrics( + output_dir: Path, + model_cfg: dict[str, Any] | None = None, +) -> dict[str, Any]: data_dir = output_dir / "data" actions_file = data_dir / "actions.jsonl" requests_file = data_dir / "requests.jsonl" @@ -196,6 +204,11 @@ def collect_run_metrics(output_dir: Path) -> dict[str, Any]: elif browser_use_model_outputs: metrics["api_calls"] = browser_use_model_outputs + usage = summarize_usage_file(messages_file, model_cfg=model_cfg) + if usage["api_calls"] > metrics["api_calls"]: + metrics["api_calls"] = usage["api_calls"] + metrics["usage"] = usage + if metrics["api_or_credit_evidence"] is None and data_dir.exists(): for log_file in data_dir.glob("*.log"): try: @@ -218,8 +231,9 @@ def classify_run( output_dir: Path, intercepted: bool, default_failure_category: str | None = None, + model_cfg: dict[str, Any] | None = None, ) -> dict[str, Any]: - metrics = collect_run_metrics(output_dir) + metrics = collect_run_metrics(output_dir, model_cfg=model_cfg) infra_flags: list[str] = [] if metrics["api_calls"] == 0: infra_flags.append("zero_api_calls") @@ -311,7 +325,10 @@ def ensure_interception(output_dir: Path): interception_file.write_text(json.dumps(result, indent=2)) -def print_results(output_dir: Path) -> bool: +def print_results( + output_dir: Path, + model_cfg: dict[str, Any] | None = None, +) -> bool: data_dir = output_dir / "data" actions_file = data_dir / "actions.jsonl" @@ -345,4 +362,6 @@ def print_results(output_dir: Path) -> bool: print(f"Request method: {result['request']['method']}") if result["request"].get("body"): print(f"Body: {json.dumps(result['request']['body'])[:300]}") + usage = summarize_usage_file(data_dir / "agent-messages.jsonl", model_cfg=model_cfg) + print(format_usage_summary(usage)) return intercepted diff --git a/src/clawbench/runner/run_support/usage.py b/src/clawbench/runner/run_support/usage.py new file mode 100644 index 00000000..d513dcf9 --- /dev/null +++ b/src/clawbench/runner/run_support/usage.py @@ -0,0 +1,469 @@ +"""Token and estimated-cost accounting for agent transcripts.""" + +from __future__ import annotations + +import json +import time +import urllib.error +import urllib.request +from decimal import Decimal, InvalidOperation +from pathlib import Path +from typing import Any, Iterable + +OPENROUTER_MODELS_URL = "https://openrouter.ai/api/v1/models" +_CACHE_TTL_S = 300 +_MODELS_CACHE: tuple[float, dict[str, dict[str, Any]]] | None = None + + +def _to_int(value: Any) -> int: + if isinstance(value, bool) or value is None: + return 0 + if isinstance(value, int): + return max(value, 0) + if isinstance(value, float): + return max(int(value), 0) + if isinstance(value, str): + try: + return max(int(float(value)), 0) + except ValueError: + return 0 + return 0 + + +def _to_decimal(value: Any) -> Decimal | None: + if value is None: + return None + try: + return Decimal(str(value)) + except (InvalidOperation, ValueError): + return None + + +def _first_int(data: dict[str, Any], keys: Iterable[str]) -> int: + for key in keys: + value = _to_int(data.get(key)) + if value: + return value + return 0 + + +def _normalize_usage(raw: dict[str, Any]) -> dict[str, int]: + input_details = raw.get("input_tokens_details") + output_details = raw.get("output_tokens_details") + if not isinstance(input_details, dict): + input_details = {} + if not isinstance(output_details, dict): + output_details = {} + + explicit_cache_read = _first_int( + raw, + ( + "cacheRead", + "cache_read", + "cache_read_tokens", + "cache_read_input_tokens", + ), + ) + nested_cache_read = _first_int( + input_details, ("cached_tokens", "cache_read_tokens") + ) + + usage = { + "input_tokens": _first_int( + raw, + ( + "input", + "prompt", + "prompt_tokens", + "input_tokens", + ), + ), + "output_tokens": _first_int( + raw, + ( + "output", + "completion", + "completion_tokens", + "output_tokens", + ), + ), + "cache_read_tokens": explicit_cache_read or nested_cache_read, + "cache_write_tokens": _first_int( + raw, + ( + "cacheWrite", + "cache_write", + "cache_write_tokens", + "cache_creation_input_tokens", + ), + ), + "reasoning_tokens": _first_int( + raw, + ( + "reasoning", + "reasoning_tokens", + "internal_reasoning", + "internal_reasoning_tokens", + ), + ) + or _first_int(output_details, ("reasoning_tokens",)), + "reported_total_tokens": _first_int( + raw, + ( + "totalTokens", + "total_tokens", + "total", + ), + ), + } + if nested_cache_read and not explicit_cache_read: + usage["input_tokens"] = max(usage["input_tokens"] - nested_cache_read, 0) + usage["total_tokens"] = ( + usage["input_tokens"] + + usage["output_tokens"] + + usage["cache_read_tokens"] + + usage["cache_write_tokens"] + + usage["reasoning_tokens"] + ) or usage["reported_total_tokens"] + return usage + + +def _merge_usage(total: dict[str, int], usage: dict[str, int]) -> None: + for key in ( + "input_tokens", + "output_tokens", + "cache_read_tokens", + "cache_write_tokens", + "reasoning_tokens", + "reported_total_tokens", + "total_tokens", + ): + total[key] += usage.get(key, 0) + + +def fetch_openrouter_pricing( + *, + base_url: str | None = None, + timeout: float = 3.0, +) -> dict[str, dict[str, Any]]: + """Fetch OpenRouter model pricing, returning an id -> model map.""" + global _MODELS_CACHE + now = time.time() + if _MODELS_CACHE and now - _MODELS_CACHE[0] < _CACHE_TTL_S: + return _MODELS_CACHE[1] + + url = OPENROUTER_MODELS_URL + if base_url and "openrouter.ai" in base_url: + url = base_url.rstrip("/") + "/models" + + try: + req = urllib.request.Request(url, headers={"User-Agent": "clawbench/usage"}) + with urllib.request.urlopen(req, timeout=timeout) as resp: + payload = json.loads(resp.read().decode("utf-8")) + except ( + OSError, + TimeoutError, + urllib.error.URLError, + json.JSONDecodeError, + ): + return {} + + models: dict[str, dict[str, Any]] = {} + for row in payload.get("data", []): + if isinstance(row, dict) and isinstance(row.get("id"), str): + models[row["id"]] = row + _MODELS_CACHE = (now, models) + return models + + +def resolve_openrouter_model( + candidates: Iterable[str], + models: dict[str, dict[str, Any]], +) -> dict[str, Any] | None: + """Resolve model aliases against OpenRouter model ids.""" + normalized = [c.strip() for c in candidates if c and c.strip()] + for candidate in normalized: + if candidate in models: + return models[candidate] + for candidate in normalized: + suffix = f"/{candidate}" + matches = [row for model_id, row in models.items() if model_id.endswith(suffix)] + if len(matches) == 1: + return matches[0] + return None + + +def _pricing_rates(model_row: dict[str, Any] | None) -> dict[str, Decimal | None]: + pricing = model_row.get("pricing") if isinstance(model_row, dict) else None + if not isinstance(pricing, dict): + pricing = {} + return { + "input_tokens": _to_decimal(pricing.get("prompt")), + "output_tokens": _to_decimal(pricing.get("completion")), + "cache_read_tokens": _to_decimal(pricing.get("input_cache_read")), + "cache_write_tokens": _to_decimal(pricing.get("input_cache_write")), + "reasoning_tokens": _to_decimal(pricing.get("internal_reasoning")), + } + + +def _estimate_cost_usd( + totals: dict[str, int], + model_row: dict[str, Any] | None, +) -> tuple[float | None, list[str]]: + rates = _pricing_rates(model_row) + if rates["input_tokens"] is None and rates["output_tokens"] is None: + return None, [] + + missing: list[str] = [] + cost = Decimal("0") + for key, rate in rates.items(): + tokens = totals.get(key, 0) + if not tokens: + continue + if rate is None: + missing.append(key) + continue + cost += Decimal(tokens) * rate + return float(cost), missing + + +def _event_usage_key(event: dict[str, Any], usage_owner: dict[str, Any]) -> str | None: + message_id = usage_owner.get("id") + if message_id: + return f"message:{message_id}" + response_id = usage_owner.get("responseId") or usage_owner.get("response_id") + if response_id: + return f"response:{response_id}" + event_id = event.get("id") + if event_id: + return f"event:{event_id}" + uuid = event.get("uuid") + if uuid: + return f"uuid:{uuid}" + timestamp = usage_owner.get("timestamp") or event.get("timestamp") + model = usage_owner.get("model") + if timestamp and model: + return f"{model}:{timestamp}" + return None + + +def _extract_usage_events( + lines: Iterable[str], +) -> tuple[dict[str, int], int, set[str], int]: + totals = { + "input_tokens": 0, + "output_tokens": 0, + "cache_read_tokens": 0, + "cache_write_tokens": 0, + "reasoning_tokens": 0, + "reported_total_tokens": 0, + "total_tokens": 0, + } + seen_keys: set[str] = set() + observed_models: set[str] = set() + api_calls = 0 + session_aggregate: dict[str, int] | None = None + session_api_calls: int | None = None + + for line in lines: + if not line.strip(): + continue + try: + event = json.loads(line) + except json.JSONDecodeError: + continue + if not isinstance(event, dict): + continue + + for key in ("model", "model_id", "modelId"): + value = event.get(key) + if isinstance(value, str) and value: + observed_models.add(value.removeprefix("openrouter/")) + + if event.get("type") == "session_meta": + for key in ("model", "model_id"): + value = event.get(key) + if isinstance(value, str) and value: + observed_models.add(value.removeprefix("openrouter/")) + normalized = _normalize_usage(event) + if normalized["total_tokens"]: + session_aggregate = normalized + if isinstance(event.get("api_call_count"), int): + session_api_calls = event["api_call_count"] + continue + + usage_owner: dict[str, Any] | None = None + raw_usage = event.get("usage") + if isinstance(raw_usage, dict): + usage_owner = event + else: + message = event.get("message") + if isinstance(message, dict): + for key in ("model", "model_id", "modelId"): + value = message.get(key) + if isinstance(value, str) and value: + observed_models.add(value.removeprefix("openrouter/")) + raw_usage = message.get("usage") + if isinstance(raw_usage, dict): + usage_owner = message + + if not isinstance(raw_usage, dict) or usage_owner is None: + continue + + usage_key = _event_usage_key(event, usage_owner) + if usage_key and usage_key in seen_keys: + continue + if usage_key: + seen_keys.add(usage_key) + + normalized = _normalize_usage(raw_usage) + if normalized["total_tokens"]: + api_calls += 1 + _merge_usage(totals, normalized) + + if session_aggregate is not None: + totals = session_aggregate + if session_api_calls is not None: + api_calls = session_api_calls + elif api_calls == 0: + api_calls = 1 + elif session_api_calls is not None and api_calls == 0: + api_calls = session_api_calls + + if totals["total_tokens"] == 0: + totals["total_tokens"] = ( + totals["input_tokens"] + + totals["output_tokens"] + + totals["cache_read_tokens"] + + totals["cache_write_tokens"] + + totals["reasoning_tokens"] + ) or totals["reported_total_tokens"] + + return totals, api_calls, observed_models, len(seen_keys) + + +def summarize_usage_lines( + lines: Iterable[str], + *, + model_cfg: dict[str, Any] | None = None, + pricing_models: dict[str, dict[str, Any]] | None = None, +) -> dict[str, Any]: + totals, api_calls, observed_models, usage_events = _extract_usage_events(lines) + configured_model = str(model_cfg.get("model", "")) if model_cfg else "" + base_url = str(model_cfg.get("base_url", "")) if model_cfg else "" + candidates = [configured_model, *sorted(observed_models)] + + models = pricing_models + if models is None and "openrouter.ai" in base_url: + models = fetch_openrouter_pricing(base_url=base_url or None) + models = models or {} + model_row = resolve_openrouter_model(candidates, models) + cost, missing_rates = _estimate_cost_usd(totals, model_row) + if totals["total_tokens"] == 0: + cost = None + + if totals["total_tokens"] == 0 and api_calls == 0: + status = "usage_unavailable" + elif cost is None: + status = "price_unavailable" + else: + status = "estimated" + + return { + "status": status, + "api_calls": api_calls, + "usage_events": usage_events, + "input_tokens": totals["input_tokens"], + "output_tokens": totals["output_tokens"], + "cache_read_tokens": totals["cache_read_tokens"], + "cache_write_tokens": totals["cache_write_tokens"], + "reasoning_tokens": totals["reasoning_tokens"], + "total_tokens": totals["total_tokens"], + "estimated_cost_usd": round(cost, 6) if cost is not None else None, + "pricing_source_url": OPENROUTER_MODELS_URL, + "matched_openrouter_model_id": ( + model_row.get("id") if isinstance(model_row, dict) else None + ), + "pricing_missing_components": missing_rates, + "observed_models": sorted(observed_models), + } + + +def summarize_usage_file( + path: Path, + *, + model_cfg: dict[str, Any] | None = None, + pricing_models: dict[str, dict[str, Any]] | None = None, +) -> dict[str, Any]: + try: + with path.open(encoding="utf-8", errors="replace") as f: + return summarize_usage_lines( + f, + model_cfg=model_cfg, + pricing_models=pricing_models, + ) + except OSError: + return summarize_usage_lines( + (), + model_cfg=model_cfg, + pricing_models=pricing_models, + ) + + +def summarize_usage_text( + text: str, + *, + model_cfg: dict[str, Any] | None = None, + pricing_models: dict[str, dict[str, Any]] | None = None, +) -> dict[str, Any]: + return summarize_usage_lines( + text.splitlines(), + model_cfg=model_cfg, + pricing_models=pricing_models, + ) + + +def format_usage_status(summary: dict[str, Any]) -> str: + total = _to_int(summary.get("total_tokens")) + if total <= 0: + return "tokens pending" + calls = _to_int(summary.get("api_calls")) + call_part = f"{calls} calls" if calls else "calls pending" + cost = summary.get("estimated_cost_usd") + if isinstance(cost, (int, float)): + cost_part = f"${cost:.4f}" + elif summary.get("status") == "price_unavailable": + cost_part = "price unavailable" + else: + cost_part = "cost pending" + return f"{call_part} • {total:,} tok • {cost_part}" + + +def format_usage_summary(summary: dict[str, Any]) -> str: + total = _to_int(summary.get("total_tokens")) + if total <= 0: + return "Usage: unavailable" + parts = [ + f"{total:,} total", + f"{_to_int(summary.get('input_tokens')):,} input", + f"{_to_int(summary.get('output_tokens')):,} output", + ] + cache_read = _to_int(summary.get("cache_read_tokens")) + cache_write = _to_int(summary.get("cache_write_tokens")) + reasoning = _to_int(summary.get("reasoning_tokens")) + if cache_read: + parts.append(f"{cache_read:,} cache read") + if cache_write: + parts.append(f"{cache_write:,} cache write") + if reasoning: + parts.append(f"{reasoning:,} reasoning") + + cost = summary.get("estimated_cost_usd") + if isinstance(cost, (int, float)): + cost_part = f"estimated cost ${cost:.6f}" + elif summary.get("status") == "price_unavailable": + cost_part = "price unavailable" + else: + cost_part = "cost unavailable" + model = summary.get("matched_openrouter_model_id") or "unmatched model" + return f"Usage: {', '.join(parts)}; {cost_part} ({summary.get('status')}, {model})" diff --git a/tests/test_results_and_metadata.py b/tests/test_results_and_metadata.py index 06050a66..cabcea25 100644 --- a/tests/test_results_and_metadata.py +++ b/tests/test_results_and_metadata.py @@ -69,6 +69,44 @@ def test_classify_run_detects_api_or_credit_evidence(tmp_path: Path) -> None: assert "429" in result["metrics"]["api_or_credit_evidence"] +def test_print_results_includes_usage_summary( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + from clawbench.runner.run_support import results + + data = tmp_path / "data" + data.mkdir() + _write_jsonl(data / "actions.jsonl", [{"type": "click", "url": "https://e.test"}]) + _write_jsonl(data / "requests.jsonl", [{"url": "https://e.test"}]) + _write_jsonl(data / "agent-messages.jsonl", []) + (data / "interception.json").write_text( + json.dumps({"intercepted": True, "stop_reason": "eval_matched"}) + ) + monkeypatch.setattr( + results, + "summarize_usage_file", + lambda _path, model_cfg=None: { + "status": "estimated", + "total_tokens": 123, + "input_tokens": 100, + "output_tokens": 23, + "cache_read_tokens": 0, + "cache_write_tokens": 0, + "reasoning_tokens": 0, + "estimated_cost_usd": 0.0042, + "matched_openrouter_model_id": "provider/model", + }, + ) + + assert results.print_results(tmp_path) is True + + out = capsys.readouterr().out + assert "Usage: 123 total" in out + assert "estimated cost $0.004200" in out + + def test_run_metadata_redacts_model_and_judge_secrets( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: @@ -100,7 +138,14 @@ def test_run_metadata_redacts_model_and_judge_secrets( "infra_failure": False, "adjusted_eligible": True, "infra_flags": [], - "metrics": {}, + "metrics": { + "usage": { + "status": "estimated", + "total_tokens": 123, + "estimated_cost_usd": 0.0042, + "matched_openrouter_model_id": "provider/model", + } + }, } model_cfg = { "model": "provider/model", @@ -141,3 +186,5 @@ def test_run_metadata_redacts_model_and_judge_secrets( assert "hidden" not in dumped assert meta["model_config"]["api_key_count"] == 2 assert meta["judge_config"]["api_key_count"] == 2 + assert meta["usage"]["estimated_cost_usd"] == 0.0042 + assert meta["run_metrics"]["usage"]["total_tokens"] == 123 diff --git a/tests/test_usage.py b/tests/test_usage.py new file mode 100644 index 00000000..1b042aaa --- /dev/null +++ b/tests/test_usage.py @@ -0,0 +1,198 @@ +"""Token and estimated-cost accounting tests.""" + +from __future__ import annotations + +import json +from pathlib import Path + +from clawbench.runner.run_support.usage import ( + format_usage_status, + resolve_openrouter_model, + summarize_usage_file, + summarize_usage_lines, +) + + +PRICING = { + "provider/test-model": { + "id": "provider/test-model", + "pricing": { + "prompt": "0.001", + "completion": "0.002", + "input_cache_read": "0.0001", + "input_cache_write": "0.0005", + "internal_reasoning": "0.003", + }, + }, + "provider/other-model": { + "id": "provider/other-model", + "pricing": {"prompt": "0.01", "completion": "0.02"}, + }, +} + + +def _line(row: dict) -> str: + return json.dumps(row) + + +def test_resolve_openrouter_model_exact_then_suffix() -> None: + exact = resolve_openrouter_model(["provider/test-model"], PRICING) + assert exact is not None + assert exact["id"] == "provider/test-model" + + suffix = resolve_openrouter_model(["test-model"], PRICING) + assert suffix is not None + assert suffix["id"] == "provider/test-model" + + assert resolve_openrouter_model(["missing"], PRICING) is None + + +def test_openclaw_usage_and_cost_math() -> None: + lines = [ + _line( + { + "type": "message", + "id": "a1", + "message": { + "role": "assistant", + "model": "test-model", + "usage": { + "input": 100, + "output": 20, + "cacheRead": 50, + "cacheWrite": 10, + "totalTokens": 180, + }, + }, + } + ), + _line( + { + "type": "message", + "id": "a2", + "message": { + "role": "assistant", + "model": "test-model", + "usage": {"input": 10, "output": 5, "totalTokens": 15}, + }, + } + ), + ] + + summary = summarize_usage_lines( + lines, + model_cfg={"model": "test-model", "base_url": "https://openrouter.ai/api/v1"}, + pricing_models=PRICING, + ) + + assert summary["status"] == "estimated" + assert summary["api_calls"] == 2 + assert summary["total_tokens"] == 195 + assert summary["input_tokens"] == 110 + assert summary["output_tokens"] == 25 + assert summary["cache_read_tokens"] == 50 + assert summary["cache_write_tokens"] == 10 + assert summary["estimated_cost_usd"] == 0.17 + assert "195 tok" in format_usage_status(summary) + + +def test_claude_code_duplicate_stream_rows_count_once() -> None: + row = { + "type": "assistant", + "message": { + "id": "gen-1", + "role": "assistant", + "model": "test-model", + "usage": { + "input_tokens": 100, + "output_tokens": 30, + "cache_read_input_tokens": 70, + "total_tokens": 130, + }, + }, + } + lines = [_line(row), _line(row), _line({**row, "uuid": "different-fragment"})] + + summary = summarize_usage_lines( + lines, + model_cfg={"model": "test-model", "base_url": "https://openrouter.ai/api/v1"}, + pricing_models=PRICING, + ) + + assert summary["api_calls"] == 1 + assert summary["input_tokens"] == 100 + assert summary["output_tokens"] == 30 + assert summary["cache_read_tokens"] == 70 + assert summary["total_tokens"] == 200 + + +def test_openai_nested_cached_tokens_are_not_double_counted() -> None: + lines = [ + _line( + { + "usage": { + "input_tokens": 100, + "output_tokens": 10, + "input_tokens_details": {"cached_tokens": 40}, + } + } + ) + ] + + summary = summarize_usage_lines( + lines, + model_cfg={"model": "test-model", "base_url": "https://openrouter.ai/api/v1"}, + pricing_models=PRICING, + ) + + assert summary["input_tokens"] == 60 + assert summary["cache_read_tokens"] == 40 + assert summary["total_tokens"] == 110 + + +def test_hermes_session_meta_usage_wins_over_message_rows() -> None: + lines = [ + _line( + { + "type": "session_meta", + "model": "provider/test-model", + "input_tokens": 300, + "output_tokens": 40, + "cache_read_tokens": 500, + "api_call_count": 7, + } + ), + _line( + { + "type": "message", + "message": { + "id": "m1", + "role": "assistant", + "model": "provider/test-model", + "usage": {"input_tokens": 1, "output_tokens": 1}, + }, + } + ), + ] + + summary = summarize_usage_lines( + lines, + model_cfg={ + "model": "provider/test-model", + "base_url": "https://openrouter.ai/api/v1", + }, + pricing_models=PRICING, + ) + + assert summary["api_calls"] == 7 + assert summary["input_tokens"] == 300 + assert summary["output_tokens"] == 40 + assert summary["cache_read_tokens"] == 500 + assert summary["total_tokens"] == 840 + + +def test_summarize_usage_file_handles_missing_file(tmp_path: Path) -> None: + summary = summarize_usage_file(tmp_path / "missing.jsonl") + + assert summary["status"] == "usage_unavailable" + assert summary["total_tokens"] == 0