diff --git a/eval/tests/test_mock_provider.py b/eval/tests/test_mock_provider.py
new file mode 100644
index 000000000..f300649c8
--- /dev/null
+++ b/eval/tests/test_mock_provider.py
@@ -0,0 +1,178 @@
+"""The mock has to be right about the wire, or every test built on it lies."""
+
+from __future__ import annotations
+
+import json
+import urllib.request
+
+from workflow_bench.mock_provider import MockProvider, Reply
+from workflow_bench.provider_usage import ANTHROPIC, OPENAI_RESPONSES, normalize_usage
+
+
+def _post(url: str, payload: dict) -> tuple[int, bytes]:
+    request = urllib.request.Request(
+        url, data=json.dumps(payload).encode(), headers={"Content-Type": "application/json"}
+    )
+    with urllib.request.urlopen(request, timeout=10) as response:
+        return response.status, response.read()
+
+
+def test_anthropic_messages_returns_a_usable_message() -> None:
+    with MockProvider([Reply(text="reviewed")]) as provider:
+        _status, raw = _post(provider.base_url + "/v1/messages", {"model": "m", "messages": []})
+    body = json.loads(raw)
+    assert body["role"] == "assistant"
+    assert body["content"][0]["text"] == "reviewed"
+    assert body["stop_reason"] == "end_turn"
+
+
+def test_a_scripted_tool_call_is_carried_as_a_tool_use_block() -> None:
+    """Tool blocks are how a mocked run produces real artifacts.
+
+    The CLI executes what it is asked to run, so a Write block makes it write
+    that file for real inside the sandbox - which is how an artifact-producing
+    cell can be exercised with no model involved.
+    """
+
+    write = {"name": "Write", "input": {"file_path": "/review-output/review-output.json", "content": "{}"}}
+    with MockProvider([Reply(text="writing", tools=[write])]) as provider:
+        _status, raw = _post(provider.base_url + "/v1/messages", {"model": "m", "messages": []})
+    body = json.loads(raw)
+    block = body["content"][1]
+    assert block["type"] == "tool_use" and block["name"] == "Write"
+    assert block["input"]["file_path"] == "/review-output/review-output.json"
+    assert body["stop_reason"] == "tool_use", "a turn ending in a tool call must say so"
+
+
+def test_streaming_emits_the_event_sequence_a_consumer_expects() -> None:
+    with MockProvider([Reply(text="hi")]) as provider:
+        request = urllib.request.Request(
+            provider.base_url + "/v1/messages",
+            data=json.dumps({"model": "m", "messages": [], "stream": True}).encode(),
+            headers={"Content-Type": "application/json"},
+        )
+        with urllib.request.urlopen(request, timeout=10) as response:
+            assert response.headers["Content-Type"] == "text/event-stream"
+            body = response.read().decode()
+
+    events = [line[len("event: ") :] for line in body.splitlines() if line.startswith("event: ")]
+    assert events[0] == "message_start"
+    assert events[-1] == "message_stop"
+    assert "content_block_delta" in events
+    # message_delta carries the final usage, which is where output tokens land.
+    assert events[-2] == "message_delta"
+
+
+def test_each_protocol_reports_usage_in_its_own_arithmetic() -> None:
+    """The whole point: the two providers count the same numbers differently.
+
+    Anthropic's cache fields ADD to input_tokens; OpenAI's are SUBSETS of it.
+    Scripting one Reply and serving it both ways is what makes that asymmetry
+    testable without a paid request.
+    """
+
+    reply = Reply(input_tokens=2_000, output_tokens=300, cache_read_input_tokens=7_000, cache_creation_input_tokens=1_000)
+
+    with MockProvider([reply, reply]) as provider:
+        _s, anthropic_raw = _post(provider.base_url + "/v1/messages", {"model": "m", "messages": []})
+        _s, openai_raw = _post(provider.base_url + "/v1/responses", {"model": "m", "input": []})
+
+    anthropic = normalize_usage(ANTHROPIC, json.loads(anthropic_raw)["usage"])
+    openai = normalize_usage(OPENAI_RESPONSES, json.loads(openai_raw)["usage"])
+
+    assert anthropic.total_input_tokens == 10_000
+    assert openai.total_input_tokens == 10_000, "same billed work, stated as the whole"
+    assert anthropic.ordinary_input_tokens == 2_000
+    assert openai.ordinary_input_tokens == 2_000, "recovered by subtraction, not addition"
+    assert openai.cache_read_input_tokens == 7_000
+
+
+def test_a_scripted_failure_is_returned_as_one() -> None:
+    """Billed failures are part of what the accounting must survive."""
+
+    with MockProvider([Reply(status_code=529, error_body={"error": {"type": "overloaded_error"}})]) as provider:
+        try:
+            _post(provider.base_url + "/v1/messages", {"model": "m", "messages": []})
+            raise AssertionError("the scripted failure was not returned")
+        except urllib.error.HTTPError as exc:
+            assert exc.code == 529
+
+
+def test_requests_are_recorded_for_assertions() -> None:
+    with MockProvider() as provider:
+        _post(provider.base_url + "/v1/messages", {"model": "claude-sonnet-4-5", "messages": [{"role": "user"}]})
+    assert len(provider.requests) == 1
+    assert provider.requests[0].body["model"] == "claude-sonnet-4-5"
+    assert provider.requests[0].path.endswith("/v1/messages")
+
+
+def test_an_unscripted_turn_gets_the_default_rather_than_stalling() -> None:
+    """A real run makes more calls than a test wants to enumerate."""
+
+    with MockProvider([Reply(text="first")], default=Reply(text="fallback")) as provider:
+        _s, one = _post(provider.base_url + "/v1/messages", {"model": "m", "messages": []})
+        _s, two = _post(provider.base_url + "/v1/messages", {"model": "m", "messages": []})
+    assert json.loads(one)["content"][0]["text"] == "first"
+    assert json.loads(two)["content"][0]["text"] == "fallback"
+
+
+def test_a_request_through_the_real_gateway_records_native_usage(tmp_path, monkeypatch) -> None:
+    """The whole stack minus the model: proxy, translation, callback, log.
+
+    This is the path that shipped three separate defects invisible to unit
+    tests - the usage variable never reaching the proxy subprocess, the
+    callback failing to import when loaded by path, and failures never
+    recorded. All three live between the gateway and the provider, which is
+    exactly the span this exercises.
+    """
+
+    import shutil
+
+    import yaml
+
+    from workflow_bench import model_gateway
+    from workflow_bench.model_gateway import OpenAIGateway
+    from workflow_bench.provider_usage import USAGE_LOG_ENV_VAR
+
+    if shutil.which("litellm") is None:
+        import pytest
+
+        pytest.skip("litellm console script absent; the proxy cannot start here")
+
+    usage_log = tmp_path / "provider_usage.jsonl"
+    monkeypatch.setenv(USAGE_LOG_ENV_VAR, str(usage_log))
+
+    reply = Reply(input_tokens=2_000, output_tokens=300, cache_read_input_tokens=7_000, cache_creation_input_tokens=1_000)
+    with MockProvider(default=reply) as provider:
+        original = model_gateway.write_openai_litellm_config
+
+        def config(path, names):
+            original(path, names)
+            document = yaml.safe_load(path.read_text())
+            for entry in document["model_list"]:
+                entry["litellm_params"]["api_base"] = f"{provider.base_url}/v1"
+            path.write_text(yaml.safe_dump(document))
+            return path
+
+        monkeypatch.setattr(model_gateway, "write_openai_litellm_config", config)
+        with OpenAIGateway(
+            openai_api_key="mock-key", model_names=["gpt-4.1"], work_dir=tmp_path / "gw", ready_timeout_s=60
+        ) as gateway:
+            request = urllib.request.Request(
+                gateway.base_url + "/v1/messages",
+                data=json.dumps({"model": "gpt-4.1", "max_tokens": 32, "messages": [{"role": "user", "content": "ping"}]}).encode(),
+                headers={"Content-Type": "application/json", "x-api-key": gateway.auth_token, "anthropic-version": "2023-06-01"},
+            )
+            with urllib.request.urlopen(request, timeout=60):
+                pass
+
+    assert usage_log.exists(), "the callback never wrote - the env did not reach the proxy"
+    events = [json.loads(line) for line in usage_log.read_text().splitlines()]
+    assert events, "the proxy started but recorded nothing"
+    native = events[-1]["native_usage"]
+    # The provider's own arithmetic survived the Anthropic-shaped translation.
+    assert native["input_tokens_details"]["cached_tokens"] == 7_000
+    assert native["input_tokens_details"]["cache_write_tokens"] == 1_000
+    usage = normalize_usage(events[-1]["provider"], native)
+    assert usage.total_input_tokens == 10_000
+    assert usage.ordinary_input_tokens == 2_000
diff --git a/eval/workflow_bench/mock_provider.py b/eval/workflow_bench/mock_provider.py
new file mode 100644
index 000000000..4b045a114
--- /dev/null
+++ b/eval/workflow_bench/mock_provider.py
@@ -0,0 +1,254 @@
+"""A scriptable stand-in for Anthropic and OpenAI, for running the harness offline.
+
+Every defect this benchmark shipped in the last round was invisible to its own
+tests for the same reason: the tests exercised a layer BELOW where the code
+runs. The usage log was never written because the proxy is a subprocess with a
+constructed environment. The callback could not be imported because LiteLLM
+loads it by path. Failures went unrecorded because only the async hook was
+overridden. Each was caught by CI or review, never by a unit test, because the
+unit test called the function directly instead of driving the path that calls
+it.
+
+This closes that gap without spending money. It speaks the two wire protocols
+the harness actually depends on, so a run can go through the real sandbox, the
+real Claude Code CLI, the real gateway and the real usage callback, and only
+the model is fake:
+
+    POST /v1/messages   Anthropic Messages, streaming and non-streaming
+    POST /v1/responses  OpenAI Responses, which the gateway translates into
+
+Point the runner at it with ``--base-url http://127.0.0.1:<port>``, which is
+the same supported path the free-model proxy documentation already uses, or
+give it to LiteLLM as ``api_base`` to exercise the gateway.
+
+Scripted, not simulated: replies are supplied by the caller, so a test decides
+what the model "says", which tools it asks for, and exactly what usage it
+reports. That last part is what makes provider-native accounting testable at
+all - real cache hits are not reproducible on demand, but a declared
+``cache_read`` of 44_000 is.
+"""
+
+from __future__ import annotations
+
+import json
+import threading
+import time
+from collections import deque
+from dataclasses import dataclass, field
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from typing import Any
+
+
+@dataclass
+class Reply:
+    """One scripted model turn.
+
+    ``tools`` drives real tool execution: Claude Code runs what it is asked to
+    run, so a reply carrying a Write block makes the CLI write that file inside
+    the sandbox for real. That is how an artifact-producing cell can be
+    exercised without a model deciding anything.
+    """
+
+    text: str = "ok"
+    tools: list[dict[str, Any]] = field(default_factory=list)
+    stop_reason: str = "end_turn"
+    # Anthropic accounting: input_tokens is the UNCACHED remainder and the
+    # cache fields add to it. Defaults are deliberately non-zero so a test that
+    # forgets to script usage still cannot mistake silence for a measurement.
+    input_tokens: int = 11
+    output_tokens: int = 7
+    cache_read_input_tokens: int = 0
+    cache_creation_input_tokens: int = 0
+    status_code: int = 200
+    error_body: dict[str, Any] | None = None
+
+
+@dataclass
+class Request:
+    """What the harness actually sent, kept so a test can assert on it."""
+
+    path: str
+    headers: dict[str, str]
+    body: dict[str, Any]
+
+
+class _Handler(BaseHTTPRequestHandler):
+    provider: MockProvider
+
+    def log_message(self, *_args: Any) -> None:  # noqa: A003 - silence the default stderr spam
+        return
+
+    def do_POST(self) -> None:  # noqa: N802 - BaseHTTPRequestHandler's interface
+        length = int(self.headers.get("Content-Length") or 0)
+        raw = self.rfile.read(length) if length else b"{}"
+        try:
+            body = json.loads(raw or b"{}")
+        except json.JSONDecodeError:
+            body = {"_unparsed": raw.decode("utf-8", "replace")}
+        self.provider.record(Request(self.path, dict(self.headers), body))
+        reply = self.provider.next_reply()
+
+        if reply.status_code != 200:
+            self._send_json(reply.status_code, reply.error_body or {"error": {"message": "scripted failure"}})
+            return
+        if self.path.rstrip("/").endswith("/responses"):
+            self._send_json(200, _openai_response(reply))
+            return
+        if body.get("stream"):
+            self._send_anthropic_stream(reply)
+            return
+        self._send_json(200, _anthropic_message(reply))
+
+    def _send_json(self, status: int, payload: dict[str, Any]) -> None:
+        encoded = json.dumps(payload).encode()
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(encoded)))
+        self.end_headers()
+        self.wfile.write(encoded)
+
+    def _send_anthropic_stream(self, reply: Reply) -> None:
+        self.send_response(200)
+        self.send_header("Content-Type", "text/event-stream")
+        self.send_header("Cache-Control", "no-cache")
+        self.end_headers()
+        for event, data in _anthropic_stream_events(reply):
+            self.wfile.write(f"event: {event}\ndata: {json.dumps(data)}\n\n".encode())
+            self.wfile.flush()
+
+
+def _content_blocks(reply: Reply) -> list[dict[str, Any]]:
+    blocks: list[dict[str, Any]] = [{"type": "text", "text": reply.text}]
+    for index, tool in enumerate(reply.tools):
+        blocks.append(
+            {
+                "type": "tool_use",
+                "id": f"toolu_mock_{index}",
+                "name": tool["name"],
+                "input": tool.get("input", {}),
+            }
+        )
+    return blocks
+
+
+def _anthropic_usage(reply: Reply) -> dict[str, int]:
+    return {
+        "input_tokens": reply.input_tokens,
+        "output_tokens": reply.output_tokens,
+        "cache_read_input_tokens": reply.cache_read_input_tokens,
+        "cache_creation_input_tokens": reply.cache_creation_input_tokens,
+    }
+
+
+def _anthropic_message(reply: Reply) -> dict[str, Any]:
+    return {
+        "id": "msg_mock",
+        "type": "message",
+        "role": "assistant",
+        "model": "mock-model",
+        "content": _content_blocks(reply),
+        "stop_reason": "tool_use" if reply.tools else reply.stop_reason,
+        "stop_sequence": None,
+        "usage": _anthropic_usage(reply),
+    }
+
+
+def _anthropic_stream_events(reply: Reply) -> list[tuple[str, dict[str, Any]]]:
+    """The SSE sequence a Messages consumer expects, in order."""
+
+    message = _anthropic_message(reply)
+    events: list[tuple[str, dict[str, Any]]] = [
+        ("message_start", {"type": "message_start", "message": {**message, "content": [], "usage": _anthropic_usage(reply)}})
+    ]
+    for index, block in enumerate(message["content"]):
+        if block["type"] == "text":
+            events.append(("content_block_start", {"type": "content_block_start", "index": index, "content_block": {"type": "text", "text": ""}}))
+            events.append(("content_block_delta", {"type": "content_block_delta", "index": index, "delta": {"type": "text_delta", "text": block["text"]}}))
+        else:
+            events.append(("content_block_start", {"type": "content_block_start", "index": index, "content_block": {"type": "tool_use", "id": block["id"], "name": block["name"], "input": {}}}))
+            events.append(("content_block_delta", {"type": "content_block_delta", "index": index, "delta": {"type": "input_json_delta", "partial_json": json.dumps(block["input"])}}))
+        events.append(("content_block_stop", {"type": "content_block_stop", "index": index}))
+    events.append(("message_delta", {"type": "message_delta", "delta": {"stop_reason": message["stop_reason"], "stop_sequence": None}, "usage": {"output_tokens": reply.output_tokens}}))
+    events.append(("message_stop", {"type": "message_stop"}))
+    return events
+
+
+def _openai_response(reply: Reply) -> dict[str, Any]:
+    """OpenAI Responses shape: input_tokens is the WHOLE, cache fields subsets."""
+
+    total_input = reply.input_tokens + reply.cache_read_input_tokens + reply.cache_creation_input_tokens
+    return {
+        "id": "resp_mock",
+        "object": "response",
+        "created_at": int(time.time()),
+        "status": "completed",
+        "model": "mock-model",
+        "error": None,
+        "output": [
+            {
+                "id": "msg_mock",
+                "type": "message",
+                "role": "assistant",
+                "status": "completed",
+                "content": [{"type": "output_text", "text": reply.text, "annotations": []}],
+            }
+        ],
+        "usage": {
+            "input_tokens": total_input,
+            "output_tokens": reply.output_tokens,
+            "total_tokens": total_input + reply.output_tokens,
+            "input_tokens_details": {
+                "cached_tokens": reply.cache_read_input_tokens,
+                "cache_write_tokens": reply.cache_creation_input_tokens,
+            },
+            "output_tokens_details": {"reasoning_tokens": 0},
+        },
+    }
+
+
+class MockProvider:
+    """Loopback-only provider stand-in. Use as a context manager."""
+
+    def __init__(self, replies: list[Reply] | None = None, *, default: Reply | None = None) -> None:
+        self._replies: deque[Reply] = deque(replies or [])
+        # A run makes more requests than a test wants to script; the default
+        # keeps it going rather than failing on the first unscripted turn.
+        self._default = default or Reply()
+        self._requests: list[Request] = []
+        self._lock = threading.Lock()
+        self._server: ThreadingHTTPServer | None = None
+
+    def __enter__(self) -> MockProvider:
+        handler = type("_BoundHandler", (_Handler,), {"provider": self})
+        # Loopback only: this answers with no authentication at all.
+        self._server = ThreadingHTTPServer(("127.0.0.1", 0), handler)
+        threading.Thread(target=self._server.serve_forever, daemon=True).start()
+        return self
+
+    def __exit__(self, *_exc: object) -> bool:
+        if self._server is not None:
+            self._server.shutdown()
+            self._server.server_close()
+        return False
+
+    @property
+    def port(self) -> int:
+        assert self._server is not None, "provider is not running"
+        return self._server.server_port
+
+    @property
+    def base_url(self) -> str:
+        return f"http://127.0.0.1:{self.port}"
+
+    def record(self, request: Request) -> None:
+        with self._lock:
+            self._requests.append(request)
+
+    def next_reply(self) -> Reply:
+        with self._lock:
+            return self._replies.popleft() if self._replies else self._default
+
+    @property
+    def requests(self) -> list[Request]:
+        with self._lock:
+            return list(self._requests)