Capture per-trial token/latency metrics; enable trusted attribution on officeqa

Tokens (not dollars) are the reported unit: with a fixed target model, dollars
are a downstream linear function of the (input, cached, output) triple, so they
are not stored. Turn on the gateway's per-trial thread attribution for officeqa
(request_log_attribution: true) so per_trial_tokens.py attributes gateway tokens
to trials via stamped thread_ids instead of content-matching. Extend the
aggregator with a multi-session grid rollup (flat CSV keyed by run/eval/task),
per-trial wall time, and a gateway-vs-attributed-vs-agent-reported reconciliation.
Document the token surfaces in CONFIGURATION.md and the run-benchmark skill.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: varunursekar

harness-engineering-bench/CONFIGURATION.md

@@ -28,6 +28,20 @@ benchmark can be checked against the others at a glance.
   budget-blind ablation: enforcement is unchanged but all budget signal is
   hidden from the agent, including a redaction of `inference_*` metrics from
   agent-facing receipts and context (latency metrics stay visible).
+- **Per-trial token/latency accounting**: the trusted gateway meters each
+  evaluation's usage as an input/cached/output/total split
+  (`inference_input_tokens`, `inference_cached_input_tokens`,
+  `inference_output_tokens`, `inference_total_tokens`, `inference_requests`) in
+  `reward_metrics`; each trial's `result.json` also carries agent-self-reported
+  tokens (`agent_reported_*`) and its wall window. With
+  `inference_gateway.request_log_attribution: true` the gateway stamps every
+  request-log record with a `thread_id`, so `scripts/per_trial_tokens.py`
+  attributes gateway tokens to individual trials (trusted, versus a content-match
+  fallback when off) and rolls a run — or a grid of runs — up to a flat CSV.
+  Enabled on officeqa; extended across the suite as the grid rolls out. Because
+  each scope's target model is fixed, **dollar cost is a linear function of the
+  (input, cached, output) token triple** with a per-model rate vector — computed
+  downstream, deliberately not stored anywhere in the run.
 - **Target model**: `fireworks_ai/deepseek-v4-flash` by default (see the
   per-benchmark table — a benchmark may pin a different evaluated model), fixed
   per run by the gateway's evaluation scope (the `evaluation.allowed_models`

harness-engineering-bench/officeqa/baseline/build.yaml

@@ -108,6 +108,9 @@ wandb:
 inference_gateway:
   upstream_api_key_env: OPENAI_API_KEY
   upstream_base_url_env: OPENAI_BASE_URL
+  # Stamp request-log records with a thread_id so per_trial_tokens.py can attribute
+  # gateway token usage to individual trials (trusted, vs. content-matching).
+  request_log_attribution: true
   producer:
     allowed_models: ["${optimizer_model:-openai/gpt-5.4}"]
     max_concurrency: 8

harness-engineering-bench/scripts/per_trial_tokens.py

@@ -24,13 +24,20 @@
 with empty bodies are unrecoverable and land in the residual. Enable
 attribution at build time for complete coverage.
 
-Usage: per_trial_tokens.py SESSION_DIR [--requests-dir DIR] [--tasks-dir DIR]
-       [--json]
+Pass one session dir for a single run, or several to roll up a grid; ``--csv``
+writes a flat per-trial table across all of them (one row per
+run/evaluation/task) for spreadsheet analysis. Tokens are the reported unit;
+dollars are a downstream linear function of the (input, cached, output) triple
+with a per-model rate vector, so they are deliberately not computed here.
+
+Usage: per_trial_tokens.py SESSION_DIR [SESSION_DIR ...] [--requests-dir DIR]
+       [--tasks-dir DIR] [--json] [--csv OUT.csv]
 """
 
 from __future__ import annotations
 
 import argparse
+import csv
 import json
 import re
 import sys
@@ -209,30 +216,27 @@ def assign_thread(
     return None
 
 
-def main() -> int:
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("session", type=Path)
-    parser.add_argument("--requests-dir", type=Path, default=None)
-    parser.add_argument("--tasks-dir", type=Path, default=None)
-    parser.add_argument("--json", action="store_true", dest="as_json")
-    arguments = parser.parse_args()
+def trial_wall_seconds(trial: dict) -> float | None:
+    started, finished = trial.get("started"), trial.get("finished")
+    if started is None or finished is None:
+        return None
+    return round((finished - started).total_seconds(), 1)
 
-    session = arguments.session
-    requests_dir = (
-        arguments.requests_dir or session / "artifacts" / "inference" / "requests"
-    )
-    if not requests_dir.is_dir():
-        print(f"no request log at {requests_dir}", file=sys.stderr)
-        return 1
 
+def analyze_session(
+    session: Path, requests_dir: Path, task_texts: list[tuple[str, str]]
+) -> dict:
+    """Per evaluation id: per-trial token triples + latency + wall, with the
+    trusted gateway envelope, the attributed sum, the independent agent-reported
+    sum (for reconciliation), and the unattributed residual."""
     trials_by_evaluation = load_trials(session)
     threads_by_evaluation = load_threads(requests_dir)
-    task_texts = load_task_texts(arguments.tasks_dir)
 
-    report = {}
+    report: dict[str, dict] = {}
     for evaluation_id, threads in sorted(threads_by_evaluation.items()):
         trials = trials_by_evaluation.get(evaluation_id, [])
         per_trial: dict[str, dict] = {}
+        seen: dict[str, set] = defaultdict(set)
         residual = {"requests": 0, **{key: 0 for key in TOKEN_KEYS}}
         for thread in threads.values():
             trial = assign_thread(thread, trials, task_texts)
@@ -241,11 +245,13 @@ def main() -> int:
                 for key in TOKEN_KEYS:
                     residual[key] += thread[key]
                 continue
+            task_name = trial["task_name"]
             entry = per_trial.setdefault(
-                trial["task_name"],
+                task_name,
                 {
                     "requests": 0,
                     "latency_ms": 0.0,
+                    "wall_s": 0.0,
                     "agent_reported": trial["agent_reported"],
                     **{key: 0 for key in TOKEN_KEYS},
                 },
@@ -254,53 +260,182 @@ def main() -> int:
             entry["latency_ms"] += thread["latency_ms"]
             for key in TOKEN_KEYS:
                 entry[key] += thread[key]
-        total = {
+            # Wall is a per-trial property, so add each contributing trial once:
+            # repeated passes of one task sum, but a trial's many threads don't
+            # double-count its wall.
+            trial_name = trial.get("trial_name")
+            if trial_name not in seen[task_name]:
+                seen[task_name].add(trial_name)
+                wall = trial_wall_seconds(trial)
+                if wall is not None:
+                    entry["wall_s"] = round(entry["wall_s"] + wall, 1)
+        gateway_total = {
             key: sum(thread[key] for thread in threads.values()) for key in TOKEN_KEYS
         }
         attributed = {
             key: sum(entry[key] for entry in per_trial.values()) for key in TOKEN_KEYS
         }
+        agent_reported_total = {
+            key: sum(
+                entry["agent_reported"].get(key, 0) for entry in per_trial.values()
+            )
+            for key in ("n_input_tokens", "n_cache_tokens", "n_output_tokens")
+        }
         report[evaluation_id] = {
             "trials": per_trial,
-            "gateway_total": total,
+            "gateway_total": gateway_total,
             "attributed": attributed,
+            "agent_reported_total": agent_reported_total,
             "residual": residual,
             "coverage_pct": (
-                round(100.0 * attributed["total_tokens"] / total["total_tokens"], 1)
-                if total["total_tokens"]
+                round(
+                    100.0 * attributed["total_tokens"] / gateway_total["total_tokens"],
+                    1,
+                )
+                if gateway_total["total_tokens"]
                 else None
             ),
         }
+    return report
 
-    if arguments.as_json:
-        print(json.dumps(report, indent=1, default=str))
-        return 0
 
+_CSV_FIELDS = [
+    "run",
+    "evaluation",
+    "task",
+    "requests",
+    "input_tokens",
+    "cached_input_tokens",
+    "output_tokens",
+    "total_tokens",
+    "latency_ms",
+    "wall_s",
+    "agent_reported_input_tokens",
+    "agent_reported_cache_tokens",
+    "agent_reported_output_tokens",
+]
+
+
+def csv_rows(run_label: str, report: dict):
+    """Flat per-trial rows (plus one residual row per evaluation) for the grid."""
     for evaluation_id, data in report.items():
-        print(f"\n=== evaluation {evaluation_id} ===")
-        header = f"{'task':<44} {'req':>5} {'input':>10} {'cached':>10} {'output':>8} {'agent-reported in/cache/out':>28}"
-        print(header)
+        for task_name, entry in sorted(data["trials"].items()):
+            reported = entry["agent_reported"]
+            yield {
+                "run": run_label,
+                "evaluation": evaluation_id,
+                "task": task_name,
+                "requests": entry["requests"],
+                "input_tokens": entry["input_tokens"],
+                "cached_input_tokens": entry["cached_input_tokens"],
+                "output_tokens": entry["output_tokens"],
+                "total_tokens": entry["total_tokens"],
+                "latency_ms": round(entry["latency_ms"], 1),
+                "wall_s": entry["wall_s"],
+                "agent_reported_input_tokens": reported.get("n_input_tokens"),
+                "agent_reported_cache_tokens": reported.get("n_cache_tokens"),
+                "agent_reported_output_tokens": reported.get("n_output_tokens"),
+            }
+        residual = data["residual"]
+        if residual["requests"]:
+            yield {
+                "run": run_label,
+                "evaluation": evaluation_id,
+                "task": "(unattributed)",
+                "requests": residual["requests"],
+                "input_tokens": residual["input_tokens"],
+                "cached_input_tokens": residual["cached_input_tokens"],
+                "output_tokens": residual["output_tokens"],
+                "total_tokens": residual["total_tokens"],
+                "latency_ms": "",
+                "wall_s": "",
+                "agent_reported_input_tokens": "",
+                "agent_reported_cache_tokens": "",
+                "agent_reported_output_tokens": "",
+            }
+
+
+def print_report(run_label: str, report: dict) -> None:
+    for evaluation_id, data in report.items():
+        print(f"\n=== {run_label} / evaluation {evaluation_id} ===")
+        print(
+            f"{'task':<40} {'req':>5} {'input':>10} {'cached':>10} "
+            f"{'output':>8} {'wall_s':>8} {'agent in/cache/out':>24}"
+        )
         for task_name, entry in sorted(data["trials"].items()):
             reported = entry["agent_reported"]
             reported_text = "/".join(
                 str(reported.get(key, "-"))
                 for key in ("n_input_tokens", "n_cache_tokens", "n_output_tokens")
             )
             print(
-                f"{str(task_name)[:44]:<44} {entry['requests']:>5} "
+                f"{str(task_name)[:40]:<40} {entry['requests']:>5} "
                 f"{entry['input_tokens']:>10} {entry['cached_input_tokens']:>10} "
-                f"{entry['output_tokens']:>8} {reported_text:>28}"
+                f"{entry['output_tokens']:>8} {entry['wall_s']:>8} {reported_text:>24}"
             )
         residual = data["residual"]
         print(
-            f"{'(unattributed)':<44} {residual['requests']:>5} "
+            f"{'(unattributed)':<40} {residual['requests']:>5} "
             f"{residual['input_tokens']:>10} {residual['cached_input_tokens']:>10} "
             f"{residual['output_tokens']:>8}"
         )
+        gateway, attributed = data["gateway_total"], data["attributed"]
+        agent = data["agent_reported_total"]
         print(
-            f"coverage: {data['coverage_pct']}% of "
-            f"{data['gateway_total']['total_tokens']} gateway-metered tokens attributed"
+            f"coverage: {data['coverage_pct']}% of {gateway['total_tokens']} "
+            f"gateway-metered tokens attributed to trials"
         )
+        # Reconcile the trusted envelope against the two independent token sources.
+        print(
+            f"  input tokens  gateway {gateway['input_tokens']} | "
+            f"attributed {attributed['input_tokens']} | "
+            f"agent-reported {agent['n_input_tokens']}"
+        )
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("sessions", type=Path, nargs="+", help="one or more session dirs")
+    parser.add_argument(
+        "--requests-dir", type=Path, default=None, help="single session only"
+    )
+    parser.add_argument("--tasks-dir", type=Path, default=None)
+    parser.add_argument("--json", action="store_true", dest="as_json")
+    parser.add_argument("--csv", type=Path, default=None, help="write a flat per-trial CSV")
+    arguments = parser.parse_args()
+
+    task_texts = load_task_texts(arguments.tasks_dir)
+    grid: dict[str, dict] = {}
+    for session in arguments.sessions:
+        if arguments.requests_dir is not None and len(arguments.sessions) == 1:
+            requests_dir = arguments.requests_dir
+        else:
+            requests_dir = session / "artifacts" / "inference" / "requests"
+        if not requests_dir.is_dir():
+            print(f"no request log at {requests_dir}", file=sys.stderr)
+            continue
+        grid[session.name] = analyze_session(session, requests_dir, task_texts)
+
+    if not grid:
+        return 1
+
+    if arguments.csv is not None:
+        with open(arguments.csv, "w", newline="", encoding="utf-8") as handle:
+            writer = csv.DictWriter(handle, fieldnames=_CSV_FIELDS)
+            writer.writeheader()
+            for run_label, report in grid.items():
+                writer.writerows(csv_rows(run_label, report))
+        print(f"wrote {arguments.csv}", file=sys.stderr)
+
+    if arguments.as_json:
+        # One session keeps the original flat {evaluation: ...} schema; several
+        # nest under their run label.
+        payload = grid[arguments.sessions[0].name] if len(grid) == 1 else grid
+        print(json.dumps(payload, indent=1, default=str))
+        return 0
+
+    for run_label, report in grid.items():
+        print_report(run_label, report)
     return 0