scriptoriumadmin/GitNexus
1
0
Fork 0
mirror of https://github.com/abhigyanpatwari/GitNexus.git synced 2026-10-05 02:43:32 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 11

feat(eval): a scriptable stand-in for Anthropic and OpenAI

Browse source 
Every defect this harness shipped last round was invisible to its own tests for
one 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, not as a package. Failures went unrecorded because only the async hook was
overridden. CI or review caught all three; no unit test could, because each
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 - Anthropic Messages, streaming and not, and
OpenAI Responses - so a run can go through the real sandbox, the real CLI, the
real gateway and the real usage callback with only the model faked. The runner
already supports pointing at it: --base-url is the same path the free-model
proxy documentation uses.

Scripted rather than simulated. 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. One Reply served down both
protocols is also the cleanest demonstration that the same billed work is stated
as a sum on one side and as a whole on the other.

Tool blocks are the mechanism for artifact-producing cells. The CLI runs what it
is asked to run, so a scripted Write block makes it write that file inside the
sandbox for real - no model deciding anything.

The end-to-end test drives the real proxy against the mock and asserts the usage
log records the provider's own arithmetic through the Anthropic-shaped
translation. It SKIPS here, because litellm's console script is absent in this
environment, so it is unverified until CI runs it - the same footing the
bubblewrap canary started on, and that one found a real bug on its first CI run.

Not yet built: driving a whole sweep against this. That needs a scripted reply
sequence that carries a cell to a scored artifact, which is the next step and
the point of the exercise.

668 eval tests pass, 17 skipped; the two test_model_gateway.py failures are the
pre-existing environmental ones.
This commit is contained in:
Gergo Magyar 2026-09-08 17:33:59 +00:00
parent 18cbeb907c
commit 6fd443a941
2 changed files with 432 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

178
eval/tests/test_mock_provider.py Normal file
View file

@ -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

254
eval/workflow_bench/mock_provider.py Normal file
View file

@ -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)