Record whether each prediction row's responses were live calls or cache replays

docs/runbook.md

@@ -437,3 +437,51 @@ for model_dir in sorted((run / "chunks").glob("*")):
     )
 PY
 ```
+
+## 8. Call provenance
+
+The harness caches provider responses on disk (`policybench/cache.py`). A
+request identical to an earlier one returns the cached response, and the
+timer measures the lookup, a few hundredths of a second, as if it were the
+call. Each prediction row therefore records where its responses came from:
+
+| Column | Meaning |
+| --- | --- |
+| `live_calls` | responses behind the row that the provider answered |
+| `cache_hit_calls` | responses behind the row that the disk cache replayed |
+| `live_elapsed_seconds` | the live calls' share of `elapsed_seconds` |
+
+The counts cover the request group whose time the row's `elapsed_seconds`
+apportions: the whole household for `run_no_tools_eval` (every chunk, repair
+round, budget escalation and explanation repair), the one output for
+single-output runs and row repairs, and the one request for batch rows.
+Every row of a group repeats them, so never sum them across rows.
+`live_elapsed_seconds` is divided among the rows like `elapsed_seconds`, so a
+household's live time is the sum over its rows. A replay adds its lookup time
+to `elapsed_seconds` and nothing to `live_elapsed_seconds`. Failed attempts
+are not counted here; the spend ledger records them.
+
+Batch rows have `cache_hit_calls` 0 (batch requests never read the disk
+cache), `live_calls` 1 when a response answered them, and no live time, like
+their empty `elapsed_seconds`. Rows written before these columns existed,
+including the frozen board's, leave all three empty, and a supervised run
+resumed across the change accepts those scenario files. A row-repair merge
+that applies an older repair, written without the columns, empties them on
+the row it replaces rather than keep the replaced request's values.
+
+The values hold these invariants, which `tests/test_call_provenance.py`
+checks with property-based tests that drive the real request path, failed
+requests included, against the calls actually made and the spend ledger's
+records of them:
+
+- the counts are non-negative integers;
+- `0 <= live_elapsed_seconds <= elapsed_seconds`, and both are empty together;
+- a row whose responses were all replays has `live_calls` 0 and
+  `live_elapsed_seconds` exactly 0;
+- a row with no replays has `live_elapsed_seconds` equal to `elapsed_seconds`.
+
+`call_provenance.household_call_provenance` classifies each household as `live`,
+`partial_replay`, `full_replay`, `no_response` or `unknown` (rows that
+predate the columns), with its summed live time. No published figure uses it:
+`latencySeconds` still takes the median of each household's
+`elapsed_seconds`, replays included.

docs/runstore.md

@@ -94,6 +94,7 @@ The CSV has **22 columns** (one row per `(model, scenario_id, variable)`):
 | `prompt_tokens`, `completion_tokens`, `total_tokens`, `reasoning_tokens`, `cached_prompt_tokens`, `cache_write_prompt_tokens` | `responses.usage_json` |
 | `provider_reported_cost_usd`, `reconstructed_cost_usd`, `total_cost_usd`, `cost_is_estimated`, `estimated_cost_usd` | `responses.cost_usd` (+ the exact per-row values for byte-identical export) |
 | `provider_response_id`, `provider_system_fingerprint`, `provider_resolved_model` | dedicated `responses` columns |
+| `live_calls`, `cache_hit_calls`, `live_elapsed_seconds` (call provenance, see the runbook; absent from CSVs written before October 2026) | `responses.usage_json` (+ the exact per-row values) |
 | any future/unknown column | `predictions.extra_json` |
 
 The `run_id` column is optional in the CSV (the eval loop only writes it for

policybench/batch_eval.py

@@ -13,6 +13,9 @@
   is not model latency. Models evaluated in batch mode show no latency on the
   leaderboard rather than a misleading one. Batch turnaround is recorded in
   the run's ``batches/`` state files.
+- ``live_calls`` is 1 on rows a batch response answered and 0 otherwise, and
+  ``cache_hit_calls`` is always 0: batch requests bypass the LiteLLM disk
+  cache. ``live_elapsed_seconds`` is empty, like ``elapsed_seconds``.
 - Cost columns are reconstructed from token usage at standard synchronous
   rates (litellm's price map plus ``PRICE_OVERRIDES_PER_1M``), matching the
   leaderboard's comparison basis. Actual spend is ~half of the reported
@@ -687,6 +690,12 @@ def split(value):
                 "provider_resolved_model": (
                     result.provider_resolved_model if result else None
                 ),
+                # A batch request goes to the provider's batch endpoint, never
+                # through the LiteLLM disk cache, so its one response is live.
+                # It is untimed, like elapsed_seconds.
+                "live_calls": 1 if result is not None else 0,
+                "cache_hit_calls": 0,
+                "live_elapsed_seconds": None,
             }
         )
     return rows

policybench/call_provenance.py

@@ -0,0 +1,84 @@
+"""Where each household's responses came from: provider calls or cache replays.
+
+Prediction rows record ``live_calls``, ``cache_hit_calls`` and
+``live_elapsed_seconds`` (``eval_no_tools.NO_TOOLS_CALL_PROVENANCE_COLUMNS``;
+docs/runbook.md, "Call provenance"). This module reads them back per
+household. It sits outside ``analysis.py`` because nothing published reads it:
+``model_cost_latency`` still counts a replay's lookup time as a call (#228).
+"""
+
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+
+from policybench.eval_no_tools import NO_TOOLS_CALL_PROVENANCE_COLUMNS
+
+HOUSEHOLD_CALL_STATUSES = (
+    "live",
+    "partial_replay",
+    "full_replay",
+    "no_response",
+    "unknown",
+)
+
+
+def household_call_provenance(predictions: pd.DataFrame) -> pd.DataFrame:
+    """Where each household's responses came from, by (model, scenario_id).
+
+    ``status`` is ``live`` when the provider answered every response behind the
+    household's rows, ``full_replay`` when the LiteLLM disk cache replayed
+    every one, ``partial_replay`` for a mix, and ``no_response`` when no
+    response reached any row. It is ``unknown`` when any row lacks the
+    call-provenance columns, as every row written before they existed does.
+    The test is per row, so it holds whether a household's rows share one
+    request group (household runs) or each have their own (single-output runs,
+    row repairs, batch chunks).
+
+    ``live_elapsed_seconds`` sums the rows' live time, as the rows'
+    ``elapsed_seconds`` sum to the household's latency: zero for a full
+    replay, NaN when unknown or untimed (batch rows carry no time).
+    """
+    keys = ["model", "scenario_id"]
+    if predictions.empty or not set(keys) <= set(predictions.columns):
+        return pd.DataFrame(
+            {
+                "status": pd.Series(dtype=object),
+                "live_elapsed_seconds": pd.Series(dtype=float),
+            },
+            index=pd.MultiIndex.from_tuples([], names=keys),
+        )
+    frame = predictions[keys].copy()
+    for column in NO_TOOLS_CALL_PROVENANCE_COLUMNS:
+        frame[column] = (
+            pd.to_numeric(predictions[column], errors="coerce")
+            if column in predictions.columns
+            else np.nan
+        )
+    frame["known"] = frame["live_calls"].notna() & frame["cache_hit_calls"].notna()
+    frame["any_live"] = frame["live_calls"].fillna(0) > 0
+    frame["any_hit"] = frame["cache_hit_calls"].fillna(0) > 0
+    grouped = frame.groupby(keys)
+    flags = grouped.agg(
+        known=("known", "all"),
+        any_live=("any_live", "any"),
+        any_hit=("any_hit", "any"),
+    )
+    status = np.select(
+        [
+            ~flags["known"],
+            flags["any_live"] & flags["any_hit"],
+            flags["any_hit"],
+            flags["any_live"],
+        ],
+        ["unknown", "partial_replay", "full_replay", "live"],
+        default="no_response",
+    )
+    live_elapsed = grouped["live_elapsed_seconds"].sum(min_count=1)
+    return pd.DataFrame(
+        {
+            "status": pd.Series(status, index=flags.index, dtype=object),
+            "live_elapsed_seconds": live_elapsed.where(flags["known"]),
+        },
+        index=flags.index,
+    )

policybench/eval_no_tools.py

@@ -54,6 +54,21 @@
 
 logger = logging.getLogger(__name__)
 
+# Where the responses behind a row came from. ``live_calls`` counts the
+# responses the provider answered and ``cache_hit_calls`` the ones the LiteLLM
+# disk cache (policybench/cache.py) replayed, over the request group whose time
+# the row's ``elapsed_seconds`` apportions. Every row of one group repeats the
+# counts, so they are never summed across rows. ``live_elapsed_seconds`` is the
+# live calls' share of ``elapsed_seconds``, apportioned the same way; a replay
+# contributes its cache lookup to ``elapsed_seconds`` and nothing here. Failed
+# attempts appear only in the spend ledger. Rows written before these columns
+# existed leave all three empty (docs/runbook.md, "Call provenance").
+NO_TOOLS_CALL_PROVENANCE_COLUMNS = (
+    "live_calls",
+    "cache_hit_calls",
+    "live_elapsed_seconds",
+)
+
 # Canonical columns emitted by both no-tools result writers. ``run_id`` is
 # added only by repeated runs. Keep the worker-schema parity test in sync.
 NO_TOOLS_RESULT_COLUMNS = (
@@ -83,6 +98,7 @@
     "provider_response_id",
     "provider_system_fingerprint",
     "provider_resolved_model",
+    *NO_TOOLS_CALL_PROVENANCE_COLUMNS,
 )
 
 # litellm resolves an unprefixed model's provider (and prices it) through its
@@ -515,6 +531,37 @@ def _response_cache_hit(response) -> bool:
     return isinstance(hidden, dict) and hidden.get("cache_hit") is True
 
 
+def _request_call_provenance(elapsed_seconds: float, cache_hit: bool) -> dict:
+    """Provenance of one answered request: a provider call or a cache replay."""
+    return {
+        "live_calls": 0 if cache_hit else 1,
+        "cache_hit_calls": 1 if cache_hit else 0,
+        "live_elapsed_seconds": 0.0 if cache_hit else elapsed_seconds,
+    }
+
+
+def _aggregate_call_provenance(results: list[dict]) -> dict:
+    """Sum request provenance over the responses a row's time is summed from.
+
+    No responses means zero calls of either kind and no live time, matching
+    ``elapsed_seconds``, which is then empty too. A result without provenance
+    (one built before the fields existed) makes the whole aggregate unknown
+    rather than undercounting it.
+    """
+    if any(
+        result.get("live_calls") is None or result.get("cache_hit_calls") is None
+        for result in results
+    ):
+        return dict.fromkeys(NO_TOOLS_CALL_PROVENANCE_COLUMNS)
+    return {
+        "live_calls": sum(int(result["live_calls"]) for result in results),
+        "cache_hit_calls": sum(int(result["cache_hit_calls"]) for result in results),
+        "live_elapsed_seconds": _sum_optional_numbers(
+            result.get("live_elapsed_seconds") for result in results
+        ),
+    }
+
+
 def _extract_usage_metadata(
     response, model_id: str, messages: list[dict], content: str
 ) -> dict:
@@ -1382,6 +1429,7 @@ def _aggregate_request_results(results: list[dict]) -> dict:
             "provider_response_id": None,
             "provider_system_fingerprint": None,
             "provider_resolved_model": None,
+            **_aggregate_call_provenance([]),
         }
 
     return {
@@ -1437,6 +1485,7 @@ def _aggregate_request_results(results: list[dict]) -> dict:
         "provider_resolved_model": _first_non_null(
             result.get("provider_resolved_model") for result in results
         ),
+        **_aggregate_call_provenance(results),
     }
 
 
@@ -2140,6 +2189,7 @@ def _request_explanations_once(
             if not _numeric_values_match(explanation_value, float(answer)):
                 explanations[variable] = None
         usage = _extract_usage_metadata(response, model_id, messages, content or "")
+        cache_hit = _response_cache_hit(response)
         spend_record = _spend_ledger_record(
             call_key=call_key,
             phase="explanation_repair",
@@ -2156,7 +2206,7 @@ def _request_explanations_once(
             request_completed_at=request_completed_at,
             elapsed_seconds=elapsed_seconds,
             usage=usage,
-            cache_hit=_response_cache_hit(response),
+            cache_hit=cache_hit,
             completion_budget_tokens=request_completion_budget,
             escalated_from_budget_tokens=escalated_from_budget_tokens,
         )
@@ -2170,6 +2220,7 @@ def _request_explanations_once(
             "finish_reason": finish_reason,
             "completion_budget": request_completion_budget,
             **usage,
+            **_request_call_provenance(elapsed_seconds, cache_hit),
             "spend_ledger": [spend_record],
         }
     except SensitivityKnobError:
@@ -2386,6 +2437,7 @@ def _request_predictions_once(
                 messages,
                 raw_response or content,
             )
+            cache_hit = _response_cache_hit(response)
             has_parse_error = any(
                 predictions.get(variable) is None for variable in variables
             ) or (
@@ -2408,7 +2460,7 @@ def _request_predictions_once(
                     request_completed_at=request_completed_at,
                     elapsed_seconds=elapsed_seconds,
                     usage=usage,
-                    cache_hit=_response_cache_hit(response),
+                    cache_hit=cache_hit,
                     completion_budget_tokens=request_completion_budget,
                     escalated_from_budget_tokens=(
                         escalated_from_budget_tokens if attempt == 0 else None
@@ -2429,6 +2481,7 @@ def _request_predictions_once(
                     escalated_from_budget_tokens is not None
                 ),
                 **usage,
+                **_request_call_provenance(elapsed_seconds, cache_hit),
                 "spend_ledger": spend_records,
             }
         except SensitivityKnobError:
@@ -2704,6 +2757,7 @@ def run_single_no_tools(
             "provider_resolved_model": _first_non_null(
                 result.get("provider_resolved_model") for result in chunk_results
             ),
+            **_aggregate_call_provenance(chunk_results),
             "repair_disagreements": [
                 disagreement
                 for result in chunk_results
@@ -3471,6 +3525,7 @@ def run_no_tools_eval(
                     "total_cost_usd": None,
                     "cost_is_estimated": None,
                     "estimated_cost_usd": None,
+                    **_aggregate_call_provenance([]),
                     "failure_sources": {},
                     "budget_escalation_count": 0,
                 }
@@ -3491,6 +3546,7 @@ def run_no_tools_eval(
             total_cost_usd = result.get("total_cost_usd")
             cost_is_estimated = result.get("cost_is_estimated")
             estimated_cost_usd = result.get("estimated_cost_usd")
+            live_elapsed_seconds = result.get("live_elapsed_seconds")
             for variable in scenario_programs:
                 prediction = result["predictions"].get(variable)
                 explanation = result.get("explanations", {}).get(variable)
@@ -3576,6 +3632,16 @@ def run_no_tools_eval(
                         "provider_resolved_model": result.get(
                             "provider_resolved_model"
                         ),
+                        # Counts cover the household's whole request group,
+                        # so every row repeats them; live time splits like
+                        # elapsed_seconds.
+                        "live_calls": result.get("live_calls"),
+                        "cache_hit_calls": result.get("cache_hit_calls"),
+                        "live_elapsed_seconds": (
+                            live_elapsed_seconds / batch_size
+                            if live_elapsed_seconds is not None
+                            else None
+                        ),
                     }
                 )
 
@@ -3723,6 +3789,7 @@ def run_no_tools_single_output_eval(
                         "provider_response_id": None,
                         "provider_system_fingerprint": None,
                         "provider_resolved_model": None,
+                        **_aggregate_call_provenance([]),
                         "failure_sources": {},
                         "budget_escalation_count": 0,
                     }
@@ -3773,6 +3840,9 @@ def run_no_tools_single_output_eval(
                         "provider_resolved_model": result.get(
                             "provider_resolved_model"
                         ),
+                        "live_calls": result.get("live_calls"),
+                        "cache_hit_calls": result.get("cache_hit_calls"),
+                        "live_elapsed_seconds": result.get("live_elapsed_seconds"),
                     }
                 )