Merge branch 'main' into fix/zig-value-ref-edges

eval/.gitignore

@@ -14,3 +14,4 @@ build/
 # Environment
 .env
 .venv/
+.venv

eval/tests/test_provider_usage.py

@@ -0,0 +1,132 @@
+"""The two providers' accounting equations, encoded literally.
+
+Adding OpenAI's cache fields to its input_tokens double-counts, because they are
+subsets of it. Subtracting Anthropic's under-counts, because they are additional
+categories. A single generic struct cannot be right for both, so these tests
+pin each equation rather than the field names.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from workflow_bench.provider_usage import (
+    ANTHROPIC,
+    OPENAI_RESPONSES,
+    UsageSemanticsError,
+    normalize_usage,
+)
+
+
+def _openai(input_tokens: int, cached: int | None = None, cache_write: int | None = None) -> dict:
+    details: dict[str, int] = {}
+    if cached is not None:
+        details["cached_tokens"] = cached
+    if cache_write is not None:
+        details["cache_write_tokens"] = cache_write
+    return {
+        "input_tokens": input_tokens,
+        "input_tokens_details": details,
+        "output_tokens": 300,
+        "output_tokens_details": {"reasoning_tokens": 250},
+    }
+
+
+def test_openai_uncached_request_is_all_ordinary_input() -> None:
+    usage = normalize_usage(OPENAI_RESPONSES, _openai(1000, cached=0, cache_write=0))
+    assert usage.ordinary_input_tokens == 1000
+    assert usage.total_input_tokens == 1000
+    assert (usage.cache_read_input_tokens, usage.cache_write_input_tokens) == (0, 0)
+
+
+def test_openai_cache_creation_keeps_the_parts_summing_to_input_tokens() -> None:
+    """The subsets must reconstruct the whole, never exceed it."""
+
+    usage = normalize_usage(OPENAI_RESPONSES, _openai(1000, cached=0, cache_write=400))
+    assert usage.ordinary_input_tokens == 600
+    assert (
+        usage.ordinary_input_tokens
+        + usage.cache_read_input_tokens
+        + usage.cache_write_input_tokens
+        == usage.total_input_tokens
+    )
+
+
+def test_openai_cache_hit_plus_new_write_uses_the_documented_subtraction() -> None:
+    usage = normalize_usage(OPENAI_RESPONSES, _openai(10_000, cached=7_000, cache_write=1_000))
+    assert usage.ordinary_input_tokens == 2_000
+    assert usage.total_input_tokens == 10_000, "input_tokens is the whole, not a component"
+
+
+def test_openai_reasoning_tokens_decompose_output_rather_than_adding_to_it() -> None:
+    usage = normalize_usage(OPENAI_RESPONSES, _openai(100, cached=0, cache_write=0))
+    assert usage.output_tokens == 300
+    assert usage.reasoning_output_tokens == 250
+    assert usage.reasoning_output_tokens <= usage.output_tokens
+
+
+def test_anthropic_uncached_total_is_just_input_tokens() -> None:
+    usage = normalize_usage(
+        ANTHROPIC,
+        {"input_tokens": 1000, "cache_creation_input_tokens": 0,
+         "cache_read_input_tokens": 0, "output_tokens": 200},
+    )
+    assert usage.total_input_tokens == 1000
+    assert usage.ordinary_input_tokens == 1000
+
+
+def test_anthropic_cached_total_adds_the_cache_categories() -> None:
+    """The opposite equation to OpenAI's, on deliberately identical numbers."""
+
+    usage = normalize_usage(
+        ANTHROPIC,
+        {"input_tokens": 2_000, "cache_creation_input_tokens": 1_000,
+         "cache_read_input_tokens": 7_000, "output_tokens": 200},
+    )
+    assert usage.total_input_tokens == 10_000
+    assert usage.ordinary_input_tokens == 2_000
+
+
+def test_the_same_numbers_mean_different_totals_on_the_two_providers() -> None:
+    """The whole reason a shared struct is unsafe, in one assertion."""
+
+    openai = normalize_usage(OPENAI_RESPONSES, _openai(10_000, cached=7_000, cache_write=1_000))
+    anthropic = normalize_usage(
+        ANTHROPIC,
+        {"input_tokens": 10_000, "cache_creation_input_tokens": 1_000,
+         "cache_read_input_tokens": 7_000, "output_tokens": 300},
+    )
+    assert openai.total_input_tokens == 10_000
+    assert anthropic.total_input_tokens == 18_000
+    assert openai.ordinary_input_tokens == 2_000
+    assert anthropic.ordinary_input_tokens == 10_000
+
+
+def test_missing_native_cache_fields_are_unknown_and_never_zero() -> None:
+    """A zero we invented is indistinguishable from a zero the provider reported."""
+
+    usage = normalize_usage(OPENAI_RESPONSES, {"input_tokens": 1000, "output_tokens": 10})
+    assert usage.cache_read_input_tokens is None
+    assert usage.cache_write_input_tokens is None
+    assert usage.ordinary_input_tokens is None, "cannot subtract what was never reported"
+    assert usage.total_input_tokens == 1000
+    assert not usage.complete
+    assert "cache_read_input_tokens" in usage.unknown_fields
+
+
+def test_an_absent_usage_object_is_entirely_unknown() -> None:
+    usage = normalize_usage(ANTHROPIC, None)
+    assert not usage.complete
+    assert usage.total_input_tokens is None
+
+
+def test_an_unknown_provider_is_refused_rather_than_guessed() -> None:
+    with pytest.raises(UsageSemanticsError, match="refusing to guess"):
+        normalize_usage("some-new-provider", {"input_tokens": 1})
+
+
+def test_cache_subsets_larger_than_the_whole_are_rejected() -> None:
+    """Nonsense arithmetic must surface, not silently produce a negative."""
+
+    with pytest.raises(UsageSemanticsError, match="exceed input_tokens"):
+        normalize_usage(OPENAI_RESPONSES, _openai(100, cached=90, cache_write=50))

eval/tests/test_provider_usage_capture.py

@@ -0,0 +1,313 @@
+"""What the proxy writes must outlive the translation that follows it.
+
+Claude Code receives an Anthropic-shaped response, which has nowhere to put
+OpenAI's cached_tokens, cache_write_tokens or reasoning_tokens. If those are not
+captured before the translation, the only remaining record of them is a bill.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import json
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+from workflow_bench import litellm_usage_callback, provider_usage
+from workflow_bench.litellm_usage_callback import USAGE_LOG_ENV_VAR, ProviderUsageLogger
+from workflow_bench.model_gateway import (
+    OpenAIGateway,
+    USAGE_CALLBACK_MODULE,
+    openai_litellm_config,
+    write_openai_litellm_config,
+)
+from workflow_bench.provider_usage import (
+    ANTHROPIC,
+    OPENAI_RESPONSES,
+    USAGE_ENV_VARS,
+    normalize_usage,
+)
+
+
+class _Usage:
+    """Stands in for the provider usage model LiteLLM hands the callback."""
+
+    def __init__(self, payload: dict) -> None:
+        self._payload = payload
+
+    def model_dump(self) -> dict:
+        return self._payload
+
+
+def _openai_response(usage: dict) -> SimpleNamespace:
+    return SimpleNamespace(
+        id="resp_68f2c1",
+        # The model that actually answered, which is not the role the caller asked for.
+        model="gpt-5.6-sol-2026-08-01",
+        usage=_Usage(usage),
+    )
+
+
+NATIVE = {
+    "input_tokens": 48_000,
+    "input_tokens_details": {"cached_tokens": 44_000, "cache_write_tokens": 1_000},
+    "output_tokens": 900,
+    "output_tokens_details": {"reasoning_tokens": 640},
+}
+
+
+@pytest.fixture
+def logged(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    log = tmp_path / "provider_usage.jsonl"
+    monkeypatch.setenv(USAGE_LOG_ENV_VAR, str(log))
+
+    def emit(usage: dict) -> dict:
+        ProviderUsageLogger()._append(
+            "success",
+            {"model": "claude-sonnet-4-5", "custom_llm_provider": "openai", "call_type": "responses"},
+            _openai_response(usage),
+            0.0,
+            1.0,
+        )
+        return json.loads(log.read_text().splitlines()[-1])
+
+    return emit
+
+
+def test_native_openai_usage_survives_the_anthropic_translation(logged) -> None:
+    event = logged(NATIVE)
+    native = event["native_usage"]
+    # Verbatim: the fields an Anthropic-shaped response cannot carry.
+    assert native["input_tokens_details"]["cached_tokens"] == 44_000
+    assert native["input_tokens_details"]["cache_write_tokens"] == 1_000
+    assert native["output_tokens_details"]["reasoning_tokens"] == 640
+    assert event["response_id"] == "resp_68f2c1"
+
+
+def test_the_actual_model_is_recorded_separately_from_the_requested_role(logged) -> None:
+    """Pricing must follow what answered, not what the caller named."""
+
+    event = logged(NATIVE)
+    assert event["requested_model"] == "claude-sonnet-4-5"
+    assert event["actual_model"] == "gpt-5.6-sol-2026-08-01"
+    assert "cell_id" not in event, "a proxy-wide variable cannot identify a cell"
+
+
+def test_the_captured_event_normalizes_with_openai_arithmetic(logged) -> None:
+    """Capture and normalization must agree end to end, not just in isolation."""
+
+    event = logged(NATIVE)
+    # The provider the LOG recorded, not one the test supplies - passing
+    # OPENAI_RESPONSES by hand here is what hid the adapter-key mismatch.
+    assert event["provider"] == OPENAI_RESPONSES
+    assert event["provider_label"] == "openai"
+    usage = normalize_usage(event["provider"], event["native_usage"])
+    assert usage.total_input_tokens == 48_000
+    assert usage.ordinary_input_tokens == 3_000
+    assert usage.cache_read_input_tokens == 44_000
+    assert usage.complete
+
+
+def test_usage_without_details_normalizes_to_unknown_rather_than_zero(logged) -> None:
+    """The mutation the accounting must not survive: dropped details, silent zeros."""
+
+    stripped = {k: v for k, v in NATIVE.items() if k != "input_tokens_details"}
+    event = logged(stripped)
+    usage = normalize_usage(event["provider"], event["native_usage"])
+    assert usage.cache_read_input_tokens is None
+    assert usage.ordinary_input_tokens is None
+    assert not usage.complete
+
+
+def test_a_failed_request_is_still_accounted_for(logged, tmp_path: Path) -> None:
+    """The money was spent whether or not the cell produced an artifact."""
+
+    import asyncio
+
+    logger = ProviderUsageLogger()
+    args = ({"model": "claude-sonnet-4-5"}, _openai_response(NATIVE), 0.0, 1.0)
+    # Every hook LiteLLM can call, not the private helper underneath them: the
+    # sync failure hook was missing entirely and _append could never show that.
+    logger.log_failure_event(*args)
+    asyncio.run(logger.async_log_failure_event(*args))
+    events = [json.loads(line) for line in (tmp_path / "provider_usage.jsonl").read_text().splitlines()]
+    assert len(events) == 2, "both failure hooks must record"
+    assert all(e["status"] == "failure" for e in events)
+
+
+def test_every_public_outcome_hook_records(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Overriding a subset silently drops whichever path LiteLLM actually uses."""
+
+    import asyncio
+
+    monkeypatch.setenv(USAGE_LOG_ENV_VAR, str(tmp_path / "usage.jsonl"))
+    logger = ProviderUsageLogger()
+    args = ({"model": "m"}, _openai_response(NATIVE), 0.0, 1.0)
+    logger.log_success_event(*args)
+    logger.log_failure_event(*args)
+    asyncio.run(logger.async_log_success_event(*args))
+    asyncio.run(logger.async_log_failure_event(*args))
+
+    events = [json.loads(line) for line in (tmp_path / "usage.jsonl").read_text().splitlines()]
+    assert [e["status"] for e in events] == ["success", "failure", "success", "failure"]
+
+
+def test_the_logger_never_raises_into_the_proxy(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Accounting is evidence, not control flow."""
+
+    monkeypatch.setenv(USAGE_LOG_ENV_VAR, str(tmp_path / "missing-dir" / "usage.jsonl"))
+    ProviderUsageLogger()._append("success", {}, object(), 0.0, 1.0)
+
+
+def test_no_log_is_written_when_the_destination_is_unset(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv(USAGE_LOG_ENV_VAR, raising=False)
+    ProviderUsageLogger()._append("success", {}, _openai_response(NATIVE), 0.0, 1.0)
+    assert not list(tmp_path.iterdir())
+
+
+def test_the_generated_config_loads_the_callback_from_beside_itself(tmp_path: Path) -> None:
+    """LiteLLM resolves the dotted path relative to the config directory."""
+
+    config = write_openai_litellm_config(tmp_path / "litellm.yaml", ["gpt-5.6-sol"])
+    assert openai_litellm_config(["gpt-5.6-sol"])["litellm_settings"]["callbacks"] == [
+        f"{USAGE_CALLBACK_MODULE}.handler"
+    ]
+    installed = config.parent / f"{USAGE_CALLBACK_MODULE}.py"
+    assert installed.is_file(), "the proxy cannot import a callback that was never placed"
+    # Importing it, not grepping it: a text search passes even when the module
+    # cannot load, which is exactly how a package-relative import survived
+    # review here. This is the deployment configuration, so load it the way the
+    # proxy does - by path, as a top-level module.
+    import importlib.util
+
+    spec = importlib.util.spec_from_file_location(USAGE_CALLBACK_MODULE, installed)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    assert isinstance(module.handler, module.ProviderUsageLogger)
+
+
+def test_the_gateway_forwards_the_usage_environment_into_the_proxy(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The proxy is a separate process with a constructed environment.
+
+    Popen(env=...) replaces the parent environment rather than extending it, so
+    a variable the callback reads is simply absent unless the gateway forwards
+    it by name. Without this the accounting looks configured and silently
+    records nothing on every request - the in-process tests above cannot see
+    that, because they never cross the subprocess boundary.
+    """
+
+    for name in USAGE_ENV_VARS:
+        monkeypatch.setenv(name, f"value-for-{name}")
+    captured: dict[str, dict[str, str]] = {}
+
+    class _Popen:
+        def __init__(self, *_a, **kwargs):
+            captured["env"] = kwargs["env"]
+            raise RuntimeError("stop before launching a real proxy")
+
+    # The console-script resolver runs before Popen and is absent in this
+    # environment (the same reason two gateway tests fail here); the argv it
+    # builds is not what this test is about.
+    monkeypatch.setattr(
+        "workflow_bench.model_gateway.litellm_proxy_argv",
+        lambda **_k: ["/bin/true"],
+    )
+    monkeypatch.setattr("workflow_bench.model_gateway.subprocess.Popen", _Popen)
+    gateway = OpenAIGateway(
+        openai_api_key="sk-test",
+        model_names=["gpt-5.6-sol"],
+        work_dir=tmp_path,
+    )
+    with contextlib.suppress(Exception):
+        gateway.__enter__()
+
+    env = captured.get("env")
+    assert env is not None, "the proxy was never constructed"
+    for name in USAGE_ENV_VARS:
+        assert env.get(name) == f"value-for-{name}", f"{name} never reached the proxy"
+    # The credential allowlist is still an allowlist, not the parent environment.
+    assert "PATH" in env and len(env) < 40
+
+
+def test_an_unresolvable_provider_is_refused_rather_than_guessed() -> None:
+    """LiteLLM says "openai" for Chat Completions too, and it counts differently."""
+
+    from workflow_bench.provider_usage import canonical_provider
+
+    assert canonical_provider("openai", "responses") == OPENAI_RESPONSES
+    assert canonical_provider("openai", "completion") is None
+    assert canonical_provider("openai", None) is None
+    assert canonical_provider("anthropic", "completion") == ANTHROPIC
+
+
+def test_request_identity_cannot_come_from_the_proxy_environment() -> None:
+    """One proxy serves the whole sweep, so its environment identifies the sweep.
+
+    attach_openai_gateway wraps all of _run_sweep, and cells run concurrently
+    under --workers, interleaving requests through that single process. Any
+    variable forwarded at launch is therefore constant for every event it ever
+    records. Pinned so a future change does not reintroduce a per-cell
+    environment variable that would silently stamp one value on every request.
+    """
+
+    assert USAGE_ENV_VARS == (
+        "GITNEXUS_BENCH_PROVIDER_USAGE",
+        "GITNEXUS_BENCH_SWEEP_ID",
+    ), "a per-cell variable here would be constant across concurrent cells"
+
+
+def test_a_request_records_its_session_so_attribution_stays_possible(logged) -> None:
+    """The per-request half of identity, recorded even when the provider omits it."""
+
+    event = logged(NATIVE)
+    assert "session_id" in event, "absent attribution is still a fact about the run"
+
+
+def test_the_callback_imports_the_way_litellm_actually_loads_it(tmp_path: Path) -> None:
+    """By path, as a top-level module, with no parent package and no sys.path entry.
+
+    LiteLLM resolves a dotted callback through spec_from_file_location against
+    the config directory, so the copied file is not part of workflow_bench when
+    it runs. A relative or sibling import therefore raises ImportError and the
+    proxy exits before becoming ready - which the in-package tests cannot see,
+    because they import it as workflow_bench.litellm_usage_callback.
+    """
+
+    import importlib.util
+    import shutil
+
+    source = Path(litellm_usage_callback.__file__)
+    installed = tmp_path / f"{USAGE_CALLBACK_MODULE}.py"
+    shutil.copy(source, installed)
+
+    spec = importlib.util.spec_from_file_location(USAGE_CALLBACK_MODULE, installed)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)  # ImportError here is the proxy refusing to start
+    assert hasattr(module, "handler")
+
+
+def test_the_callbacks_copied_constants_match_the_canonical_ones() -> None:
+    """The copies are deliberate; drifting apart silently is not.
+
+    The callback cannot import from the package (see the test above), so it
+    carries its own literals. These assertions are what keep the duplication
+    honest.
+    """
+
+    assert litellm_usage_callback.USAGE_LOG_ENV_VAR == provider_usage.USAGE_LOG_ENV_VAR
+    assert litellm_usage_callback.SWEEP_ID_ENV_VAR == provider_usage.SWEEP_ID_ENV_VAR
+    for label, call_type in (
+        ("openai", "responses"),
+        ("openai", "completion"),
+        ("openai", None),
+        ("anthropic", "completion"),
+        ("mystery", "responses"),
+    ):
+        assert litellm_usage_callback.canonical_provider(label, call_type) == provider_usage.canonical_provider(
+            label, call_type
+        ), f"resolver drifted for {label!r}/{call_type!r}"

eval/workflow_bench/litellm_usage_callback.py

@@ -0,0 +1,136 @@
+"""Append each upstream request's usage exactly as the provider reported it.
+
+This runs INSIDE the LiteLLM proxy, on the far side of the translation that
+turns an OpenAI response into the Anthropic shape Claude Code expects. That is
+the only point that still knows which provider served the request, what model
+actually answered, and what the native usage object said before its fields were
+renamed into someone else's semantics.
+
+Deliberately self-contained: the proxy loads this file by path from the config
+directory, so it cannot assume ``workflow_bench`` is importable. Normalization
+lives in workflow_bench.provider_usage and runs offline over what this writes -
+the native object is the evidence, and deriving from it here would mean the
+derivation could not be revisited without re-running a paid sweep.
+
+Never raises. A cell that fails still spent money upstream, and losing the
+accounting because the log write failed would be the worse outcome.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import threading
+from typing import Any
+
+from litellm.integrations.custom_logger import CustomLogger
+
+# Literals, not imports. LiteLLM loads this file BY PATH from the config
+# directory via spec_from_file_location, so it has no parent package and the
+# directory is not on sys.path - a relative or sibling import raises
+# ImportError and the proxy refuses to start. workflow_bench.provider_usage
+# holds the canonical copies and a test asserts these agree with them, which
+# catches drift without coupling at import time.
+USAGE_LOG_ENV_VAR = "GITNEXUS_BENCH_PROVIDER_USAGE"
+SWEEP_ID_ENV_VAR = "GITNEXUS_BENCH_SWEEP_ID"
+
+
+def canonical_provider(label, call_type):  # noqa: ANN001, ANN201
+    """Adapter key for the usage shape, or None when it cannot be resolved.
+
+    Mirrors workflow_bench.provider_usage.canonical_provider; see the note
+    above for why this is a copy rather than an import.
+    """
+
+    if label == "openai" and call_type and "responses" in call_type:
+        return "openai-responses"
+    if label == "anthropic":
+        return "anthropic"
+    return None
+
+SCHEMA_VERSION = 1
+_LOCK = threading.Lock()
+
+
+def _plain(value: Any) -> Any:
+    """Provider usage arrives as pydantic models; keep the shape, drop the class."""
+
+    for attr in ("model_dump", "dict"):
+        method = getattr(value, attr, None)
+        if callable(method):
+            try:
+                return method()
+            except Exception:
+                pass
+    if isinstance(value, dict):
+        return value
+    return None
+
+
+class ProviderUsageLogger(CustomLogger):
+    async def async_log_success_event(self, kwargs, response_obj, start_time, end_time) -> None:  # noqa: ANN001
+        self._append("success", kwargs, response_obj, start_time, end_time)
+
+    async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time) -> None:  # noqa: ANN001
+        # Failed requests are billed too, and a sweep that only accounts for
+        # successes understates what it spent.
+        self._append("failure", kwargs, response_obj, start_time, end_time)
+
+    def log_success_event(self, kwargs, response_obj, start_time, end_time) -> None:  # noqa: ANN001
+        self._append("success", kwargs, response_obj, start_time, end_time)
+
+    def log_failure_event(self, kwargs, response_obj, start_time, end_time) -> None:  # noqa: ANN001
+        # The synchronous counterpart. Overriding only the success hook here
+        # recorded successes and let failures fall through to the base class,
+        # which accounts for nothing - and a failed request is still billed, so
+        # a sweep missing them understates what it spent.
+        self._append("failure", kwargs, response_obj, start_time, end_time)
+
+    def _append(self, status, kwargs, response_obj, start_time, end_time) -> None:  # noqa: ANN001
+        path = os.environ.get(USAGE_LOG_ENV_VAR)
+        if not path:
+            return
+        try:
+            params = kwargs.get("litellm_params") or {}
+            call_type = kwargs.get("call_type")
+            provider_label = kwargs.get("custom_llm_provider") or params.get("custom_llm_provider")
+            metadata = params.get("metadata") or {}
+            event = {
+                "schema_version": SCHEMA_VERSION,
+                "status": status,
+                # Identity. The REQUESTED model is the caller's role name and the
+                # ACTUAL model is what answered; pricing must follow the second,
+                # because several roles map onto one upstream model here.
+                "requested_model": kwargs.get("model"),
+                "actual_model": getattr(response_obj, "model", None),
+                # Two fields, because they answer different questions. The raw
+                # label is what LiteLLM said; "provider" is the adapter key,
+                # which needs the call type too - LiteLLM reports "openai" for
+                # both Chat Completions and Responses and those report usage
+                # differently. Unresolvable stays None so normalize_usage
+                # refuses rather than guessing token semantics.
+                "provider_label": provider_label,
+                "provider": canonical_provider(provider_label, call_type),
+                "response_id": getattr(response_obj, "id", None),
+                "call_type": call_type,
+                "sweep_id": os.environ.get(SWEEP_ID_ENV_VAR),
+                # The per-request half of identity, and the only thing that can
+                # attribute a request to a cell: one proxy serves the whole
+                # sweep, so anything read from the environment is the same for
+                # every event. Recorded even when absent, because knowing the
+                # attribution is unavailable is itself a fact about the run.
+                "session_id": metadata.get("litellm_session_id") or metadata.get("session_id"),
+                "started_at": str(start_time),
+                "completed_at": str(end_time),
+                # Verbatim. Not flattened, not renamed, not summed.
+                "native_usage": _plain(getattr(response_obj, "usage", None)),
+            }
+            line = json.dumps(event, default=str) + "\n"
+            with _LOCK, open(path, "a", encoding="utf-8") as handle:
+                handle.write(line)
+        except Exception:
+            # Accounting is evidence, not control flow: never take the sweep down.
+            return
+
+
+handler = ProviderUsageLogger()

eval/workflow_bench/model_gateway.py

@@ -31,6 +31,8 @@ from typing import Any
 
 import yaml
 
+from .provider_usage import USAGE_ENV_VARS
+
 ANTHROPIC_API_KEY_ENV = "GITNEXUS_BENCH_ANTHROPIC_API_KEY"
 LEGACY_ANTHROPIC_API_KEY_ENV = "GITNEXUS_BENCH_AUTH_TOKEN"
 OPENAI_API_KEY_ENV = "GITNEXUS_BENCH_OPENAI_API_KEY"
@@ -173,6 +175,9 @@ def resolve_model_access(
     return ModelAccess(start_proxy=False)
 
 
+USAGE_CALLBACK_MODULE = "provider_usage_callback"
+
+
 def openai_litellm_config(model_names: Sequence[str]) -> dict[str, Any]:
     seen: list[str] = []
     for name in model_names:
@@ -196,7 +201,15 @@ def openai_litellm_config(model_names: Sequence[str]) -> dict[str, Any]:
             }
             for name in seen
         ],
-        "litellm_settings": {"request_timeout": GATEWAY_REQUEST_TIMEOUT_S},
+        "litellm_settings": {
+            "request_timeout": GATEWAY_REQUEST_TIMEOUT_S,
+            # Captures each upstream request's usage as the provider reported
+            # it, before translation renames OpenAI's fields into Anthropic's
+            # shape and loses which arithmetic applies. Resolved by LiteLLM
+            # relative to the config directory, which is why the module is
+            # copied next to the config rather than imported from the package.
+            "callbacks": [f"{USAGE_CALLBACK_MODULE}.handler"],
+        },
         "general_settings": {"master_key": "os.environ/LITELLM_MASTER_KEY"},
     }
 
@@ -204,9 +217,26 @@ def openai_litellm_config(model_names: Sequence[str]) -> dict[str, Any]:
 def write_openai_litellm_config(path: Path, model_names: Sequence[str]) -> Path:
     path.write_text(yaml.safe_dump(openai_litellm_config(model_names), sort_keys=False))
     path.chmod(0o600)
+    _install_usage_callback(path.parent)
     return path
 
 
+def _install_usage_callback(config_dir: Path) -> Path:
+    """Place the usage logger where LiteLLM resolves callbacks from.
+
+    LiteLLM loads a dotted callback path as a file relative to the config
+    directory before falling back to a package import, and the proxy runs as
+    its own process that need not have this package on sys.path. Copying the
+    one module is what makes the callback resolvable in both cases.
+    """
+
+    source = Path(__file__).with_name("litellm_usage_callback.py")
+    destination = config_dir / f"{USAGE_CALLBACK_MODULE}.py"
+    destination.write_text(source.read_text())
+    destination.chmod(0o600)
+    return destination
+
+
 def _free_loopback_port() -> int:
     with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
         sock.bind(("127.0.0.1", 0))
@@ -302,6 +332,17 @@ class OpenAIGateway(AbstractContextManager["OpenAIGateway"]):
             "OPENAI_API_KEY": self.openai_api_key,
             "LITELLM_MASTER_KEY": self.auth_token,
         }
+        # The proxy is a separate process and Popen(env=...) REPLACES the
+        # parent environment rather than extending it, so anything the usage
+        # callback reads has to be forwarded by name. Without this the callback
+        # loads, finds no destination, and returns silently on every request -
+        # the accounting looks configured and records nothing. Forwarded
+        # individually rather than by inheriting the environment, because the
+        # allowlist above is the gateway's credential boundary.
+        for name in USAGE_ENV_VARS:
+            value = os.environ.get(name)
+            if value:
+                env[name] = value
         if os.name == "nt":
             # Windows subprocess DLL/socket initialization needs SystemRoot.
             # Keep the rest of the gateway's credential boundary explicit.

eval/workflow_bench/provider_usage.py

@@ -0,0 +1,188 @@
+"""Per-request usage as the provider reported it, plus a derived cross-provider view.
+
+The benchmark has been reading token counts out of Claude Code's session output,
+which is Anthropic-shaped whatever actually served the request. That works until
+the upstream is OpenAI, because the two providers do not merely name their fields
+differently - they mean opposite things by them:
+
+    Anthropic:  total_input = input_tokens
+                            + cache_creation_input_tokens
+                            + cache_read_input_tokens
+                (input_tokens is only the UNCACHED remainder; cache fields ADD)
+
+    OpenAI:     total_input = input_tokens
+                ordinary    = input_tokens - cached_tokens - cache_write_tokens
+                (input_tokens is the WHOLE; cache fields are SUBSETS)
+
+Adding OpenAI's three together double-counts; subtracting Anthropic's
+under-counts. So the native object is authoritative and is stored verbatim, and
+the normalized view is derived from it per provider.
+
+The second rule is that a field nobody reported is UNKNOWN, not zero. A stored
+``cache_read = 0`` previously could mean either "the provider said zero" or "our
+adapter never looked", and those two must never be written identically again:
+the first says caching is not working, the second says we cannot tell.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any
+
+SCHEMA_VERSION = 1
+
+# Read by the in-proxy callback and forwarded by the gateway that launches it.
+# Defined here because this module is pure stdlib: model_gateway can import the
+# name without importing litellm, which only the callback needs.
+#
+# Both are SWEEP-scoped, and that is a constraint rather than an oversight.
+# attach_openai_gateway wraps the whole sweep (runner.py), so one proxy serves
+# every cell and its environment is fixed for that proxy's lifetime - while
+# cells run concurrently under --workers and interleave requests through it. An
+# environment variable therefore cannot carry a per-cell identity: it would
+# record one constant against every event. Attributing a request to a cell
+# needs an identifier that travels WITH the request; see the session fields the
+# callback records for the intended hook.
+USAGE_LOG_ENV_VAR = "GITNEXUS_BENCH_PROVIDER_USAGE"
+SWEEP_ID_ENV_VAR = "GITNEXUS_BENCH_SWEEP_ID"
+USAGE_ENV_VARS = (USAGE_LOG_ENV_VAR, SWEEP_ID_ENV_VAR)
+
+ANTHROPIC = "anthropic"
+OPENAI_RESPONSES = "openai-responses"
+
+
+class UsageSemanticsError(ValueError):
+    """The native usage object does not satisfy its own provider's arithmetic."""
+
+
+@dataclass(frozen=True)
+class NormalizedUsage:
+    """Cross-provider view. ``None`` means the provider did not report it.
+
+    Deliberately not defaulted to 0: see the module docstring. Every consumer
+    that sums these has to decide what to do about unknown, and making it None
+    forces that decision to be explicit instead of silently counting zero.
+    """
+
+    ordinary_input_tokens: int | None
+    cache_read_input_tokens: int | None
+    cache_write_input_tokens: int | None
+    total_input_tokens: int | None
+    output_tokens: int | None
+    reasoning_output_tokens: int | None
+
+    @property
+    def complete(self) -> bool:
+        return all(
+            value is not None
+            for value in (
+                self.ordinary_input_tokens,
+                self.cache_read_input_tokens,
+                self.cache_write_input_tokens,
+                self.total_input_tokens,
+                self.output_tokens,
+            )
+        )
+
+    @property
+    def unknown_fields(self) -> tuple[str, ...]:
+        return tuple(
+            name for name, value in sorted(vars(self).items()) if value is None
+        )
+
+
+def _int_or_none(source: Mapping[str, Any] | None, key: str) -> int | None:
+    """Absent, null, or non-numeric all read as unknown rather than zero."""
+
+    if not isinstance(source, Mapping):
+        return None
+    value = source.get(key)
+    if isinstance(value, bool) or not isinstance(value, int):
+        return None
+    return value
+
+
+def _normalize_openai_responses(usage: Mapping[str, Any]) -> NormalizedUsage:
+    """input_tokens is the WHOLE; cached and cache-write are subsets of it."""
+
+    total = _int_or_none(usage, "input_tokens")
+    details = usage.get("input_tokens_details")
+    cache_read = _int_or_none(details, "cached_tokens")
+    cache_write = _int_or_none(details, "cache_write_tokens")
+    output_details = usage.get("output_tokens_details")
+
+    ordinary: int | None = None
+    if total is not None and cache_read is not None and cache_write is not None:
+        ordinary = total - cache_read - cache_write
+        if ordinary < 0:
+            raise UsageSemanticsError(
+                f"OpenAI cached ({cache_read}) + cache_write ({cache_write}) "
+                f"exceed input_tokens ({total})"
+            )
+    return NormalizedUsage(
+        ordinary_input_tokens=ordinary,
+        cache_read_input_tokens=cache_read,
+        cache_write_input_tokens=cache_write,
+        total_input_tokens=total,
+        output_tokens=_int_or_none(usage, "output_tokens"),
+        # A decomposition of output_tokens, not an addition to it.
+        reasoning_output_tokens=_int_or_none(output_details, "reasoning_tokens"),
+    )
+
+
+def _normalize_anthropic(usage: Mapping[str, Any]) -> NormalizedUsage:
+    """input_tokens is the uncached REMAINDER; the cache fields add to it."""
+
+    ordinary = _int_or_none(usage, "input_tokens")
+    cache_read = _int_or_none(usage, "cache_read_input_tokens")
+    cache_write = _int_or_none(usage, "cache_creation_input_tokens")
+
+    total: int | None = None
+    if ordinary is not None and cache_read is not None and cache_write is not None:
+        total = ordinary + cache_read + cache_write
+    return NormalizedUsage(
+        ordinary_input_tokens=ordinary,
+        cache_read_input_tokens=cache_read,
+        cache_write_input_tokens=cache_write,
+        total_input_tokens=total,
+        output_tokens=_int_or_none(usage, "output_tokens"),
+        reasoning_output_tokens=None,
+    )
+
+
+def canonical_provider(label: str | None, call_type: str | None) -> str | None:
+    """Map LiteLLM's provider label onto an adapter key, or None if unsure.
+
+    LiteLLM reports ``custom_llm_provider`` as "openai" for both Chat
+    Completions and Responses, and those two report usage differently, so the
+    label alone cannot pick an adapter. The call type is what distinguishes
+    them. Returning None when it does not is deliberate: normalize_usage
+    refuses an unknown provider rather than guessing token semantics, which is
+    the whole point of keeping the native object authoritative.
+    """
+
+    if label == "openai" and call_type and "responses" in call_type:
+        return OPENAI_RESPONSES
+    if label in _ADAPTERS:
+        return label
+    return None
+
+
+_ADAPTERS = {
+    ANTHROPIC: _normalize_anthropic,
+    OPENAI_RESPONSES: _normalize_openai_responses,
+}
+
+
+def normalize_usage(provider: str, native_usage: Mapping[str, Any] | None) -> NormalizedUsage:
+    """Derive the cross-provider view. Never mutates or replaces the native object."""
+
+    adapter = _ADAPTERS.get(provider)
+    if adapter is None:
+        raise UsageSemanticsError(
+            f"no usage adapter for provider {provider!r}; refusing to guess its token semantics"
+        )
+    if not isinstance(native_usage, Mapping):
+        return NormalizedUsage(None, None, None, None, None, None)
+    return adapter(native_usage)

gitnexus/src/core/auto-sync/starter.ts

@@ -4,7 +4,7 @@ import path from 'node:path';
 import { execFileSync } from 'node:child_process';
 import { acquireFileLock, FileLockBusyError } from '../../storage/file-lock.js';
 import { getGlobalDir } from '../../storage/repo-manager.js';
-import { isProcessAlive, readProcessStartTime } from '../../utils/process-identity.js';
+import { isProcessAlive, readProcessStartTimeCached } from '../../utils/process-identity.js';
 import { loadAutoSyncConfig } from './config.js';
 import { runAutoSyncOnce } from './runner.js';
 import { getAutoSyncMutexPath, getAutoSyncWatchDir } from './state.js';
@@ -632,7 +632,7 @@ function resolveWatchDeps(deps: Partial<AutoSyncWatchControlDeps> = {}): AutoSyn
           return undefined;
         }
       }),
-    readProcessStartTime: deps.readProcessStartTime ?? readProcessStartTime,
+    readProcessStartTime: deps.readProcessStartTime ?? readProcessStartTimeCached,
     sleep:
       deps.sleep ??
       ((ms) =>

gitnexus/src/storage/file-lock.ts

@@ -3,7 +3,7 @@ import fs from 'node:fs/promises';
 import os from 'node:os';
 import path from 'node:path';
 import { setTimeout as sleep } from 'node:timers/promises';
-import { isProcessAlive, readProcessStartTime } from '../utils/process-identity.js';
+import { isProcessAlive, readProcessStartTimeCached } from '../utils/process-identity.js';
 
 const HOSTNAME = os.hostname();
 
@@ -46,7 +46,9 @@ export async function acquireFileLock(
     pid,
     ownerId: crypto.randomUUID(),
     processStartTime:
-      options.processStartTime ?? (options.readProcessStartTime ?? readProcessStartTime)(pid) ?? '',
+      options.processStartTime ??
+      (options.readProcessStartTime ?? readProcessStartTimeCached)(pid) ??
+      '',
     hostname: options.hostname ?? HOSTNAME,
   };
   if (!owner.processStartTime) {
@@ -69,7 +71,7 @@ export async function acquireFileLock(
             resolvedPath,
             owner,
             options.isProcessAlive ?? isProcessAlive,
-            options.readProcessStartTime ?? readProcessStartTime,
+            options.readProcessStartTime ?? readProcessStartTimeCached,
           )
         ) {
           continue;

gitnexus/src/utils/process-identity.ts

@@ -38,3 +38,22 @@ export function readProcessStartTime(pid: number): string | undefined {
     return undefined;
   }
 }
+
+let ownStartTime: string | undefined;
+
+/**
+ * `readProcessStartTime`, except this process's own start time is probed once.
+ * It cannot change while we are running, and every `acquireFileLock` — plus
+ * each retry attempt and each stale-lock reclaim guard — stamps the owner file
+ * with it. On Windows that probe is a `powershell.exe` spawn and a WMI query,
+ * so a process taking several locks pays it several times for one constant.
+ *
+ * A foreign pid is never cached: that process can exit and its pid can be
+ * reused, which is the very thing the stamp exists to detect. A failed probe
+ * is not cached either — one transient failure would otherwise leave the
+ * process unable to take a lock for its whole lifetime.
+ */
+export function readProcessStartTimeCached(pid: number): string | undefined {
+  if (pid !== process.pid) return readProcessStartTime(pid);
+  return (ownStartTime ??= readProcessStartTime(pid));
+}

gitnexus/test/unit/process-identity.test.ts

@@ -4,8 +4,24 @@ import { isProcessAlive, readProcessStartTime } from '../../src/utils/process-id
 
 afterEach(() => {
   vi.restoreAllMocks();
+  vi.doUnmock('node:child_process');
+  vi.resetModules();
 });
 
+/**
+ * Loads a fresh copy of the module (fresh memo) over a counted `execFileSync`,
+ * so "how many times did we actually shell out" is observable. `doMock` is not
+ * hoisted, so the statically imported functions used by the other tests keep
+ * the real implementation.
+ */
+async function withCountedProbe(probe: () => string) {
+  const execFileSync = vi.fn(probe);
+  vi.doMock('node:child_process', () => ({ execFileSync }));
+  vi.resetModules();
+  const identity = await import('../../src/utils/process-identity.js');
+  return { execFileSync, readProcessStartTimeCached: identity.readProcessStartTimeCached };
+}
+
 describe('process identity', () => {
   it('treats only ESRCH as a dead process', () => {
     const kill = vi.spyOn(process, 'kill');
@@ -39,4 +55,33 @@ describe('process identity', () => {
       }
     },
   );
+
+  it('probes this process once and re-probes a foreign pid every time', async () => {
+    const { execFileSync, readProcessStartTimeCached } = await withCountedProbe(() => 'STAMP\n');
+
+    expect(readProcessStartTimeCached(process.pid)).toBe('STAMP');
+    expect(readProcessStartTimeCached(process.pid)).toBe('STAMP');
+    // On Windows each probe is a powershell.exe spawn plus a WMI query.
+    expect(execFileSync).toHaveBeenCalledTimes(1);
+
+    // A foreign process can exit and its pid be reused — caching that stamp
+    // would blind the reuse check the stamp exists for.
+    expect(readProcessStartTimeCached(process.pid + 1)).toBe('STAMP');
+    expect(readProcessStartTimeCached(process.pid + 1)).toBe('STAMP');
+    expect(execFileSync).toHaveBeenCalledTimes(3);
+  });
+
+  it('retries after a failed self probe instead of caching the failure', async () => {
+    const { execFileSync, readProcessStartTimeCached } = await withCountedProbe(() => 'STAMP\n');
+    execFileSync.mockImplementationOnce(() => {
+      throw new Error('probe unavailable');
+    });
+
+    // A cached failure would leave acquireFileLock throwing "Unable to
+    // determine process start time" for the rest of the process's life.
+    expect(readProcessStartTimeCached(process.pid)).toBeUndefined();
+    expect(readProcessStartTimeCached(process.pid)).toBe('STAMP');
+    expect(readProcessStartTimeCached(process.pid)).toBe('STAMP');
+    expect(execFileSync).toHaveBeenCalledTimes(2);
+  });
 });