feat: display token usage in USD

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.

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,

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:

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,

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