diff --git a/.github/workflows/ci-tests.yml b/.github/workflows/ci-tests.yml index 3d98783d6..01cfa24b6 100644 --- a/.github/workflows/ci-tests.yml +++ b/.github/workflows/ci-tests.yml @@ -858,6 +858,11 @@ jobs: timeout-minutes: 20 env: GITNEXUS_REQUIRE_BWRAP_CANARY: '1' + # This job installs bubblewrap, the pinned runtime and a built GitNexus, + # so the offline sweep runs here with nothing provisioning-stubbed: real + # containment, real mounts, real graph. A missing piece fails the job + # rather than silently falling back to the stubbed path. + GITNEXUS_REQUIRE_FULL_SWEEP: '1' GITNEXUS_REQUIRE_CLAUDE_CANARY: '1' steps: - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 @@ -916,7 +921,9 @@ jobs: tests/test_process_control.py tests/test_proposer_sandbox.py tests/test_workflow_bench_sessions.py - tests/test_ce_plugin_runtime.py -q + tests/test_ce_plugin_runtime.py + tests/test_offline_sweep_integration.py + tests/test_mock_provider.py -q working-directory: eval # Native Windows Job Object canary. POSIX-only tests skip by platform, while diff --git a/eval/tests/fixtures/fake_claude.py b/eval/tests/fixtures/fake_claude.py new file mode 100755 index 000000000..fbcedcdbe --- /dev/null +++ b/eval/tests/fixtures/fake_claude.py @@ -0,0 +1,175 @@ +#!/usr/bin/env python3 +"""A stand-in for the Claude Code CLI: real HTTP, real tool execution, real stream-json. + +Not a mock of the harness's own code. It does what the CLI does at the two +boundaries the harness depends on - it calls ANTHROPIC_BASE_URL for a turn, it +EXECUTES the tool blocks that come back, and it prints the stream-json event +sequence the parent parses. Only Write really executes - it is what produces the +review artifact, so the artifact path has to be genuine end to end. Skill is +MODELLED: it validates the request and returns a synthetic result, because the +parent's evidence gate keys on the request/result pair rather than on a skill +having loaded, and a fixture cannot load a real one. Bash is stubbed outright: +arbitrary shell from a scripted reply buys no fidelity for the paths this +exercises and plenty of ways to damage the host. Everything between +those boundaries (the sandbox, +the artifact capture, the scoring, the row) stays real, which is the whole +point: those are the layers that shipped bugs no unit test could see. + +Reads the prompt from stdin, as the real CLI does under "-p --input-format text". +""" + +from __future__ import annotations + +import json +import os +import pathlib +import sys +import urllib.request + + +def _turn(base_url: str, prompt: str) -> dict: + request = urllib.request.Request( + base_url.rstrip("/") + "/v1/messages", + data=json.dumps({"model": os.environ.get("ANTHROPIC_MODEL", "mock"), "max_tokens": 1024, + "messages": [{"role": "user", "content": prompt}]}).encode(), + headers={"Content-Type": "application/json", + "x-api-key": os.environ.get("ANTHROPIC_API_KEY", ""), + "anthropic-version": "2023-06-01"}, + ) + with urllib.request.urlopen(request, timeout=30) as response: + return json.load(response) + + +def _run_tool(name: str, params: dict) -> str: + """Write executes for real - it is what produces the review artifact. + + Skill and Bash do not: see the module docstring for which is modelled and + which is stubbed, and why neither can be genuine here. + """ + + if name == "Write": + target = pathlib.Path(params["file_path"]) + target.parent.mkdir(parents=True, exist_ok=True) + # Atomic, exactly as the real Write tool does it: temp file beside the + # target, then rename. This is the operation the read-only workspace + # boundary has to permit for the artifact directory and refuse for the + # workspace, so a stand-in that wrote in place would prove nothing. + staging = target.with_name(target.name + ".tmp.fake") + staging.write_text(params.get("content", "")) + os.replace(staging, target) + return f"wrote {target}" + if name == "Skill": + # Modelled explicitly rather than falling through to a generic success. + # The parent's evidence gate keys on a Skill request with a non-error + # result, so leaving this unimplemented let an unexecuted skill satisfy + # the gate - the gate would have been measuring the fixture, not a skill. + skill = params.get("skill") or params.get("command") or params.get("name") + if not skill: + raise NotImplementedError("Skill request carried no skill name") + return f"loaded skill {skill}" + if name == "Bash": + return "(bash suppressed in the stand-in)" + # An unsupported tool is a FAILED tool run, not a quiet success. Returning a + # plain string here made the parent's evidence gate read an unexecuted Skill + # request as a successful invocation. + raise NotImplementedError(f"unsupported tool {name}") + + +def main() -> int: + # stdin, because that is where the real CLI takes it under + # "-p --input-format text": the parent pipes prompt bytes in. Scanning argv + # for a non-flag token picks up a flag's VALUE instead ("text"), which is + # exactly what the prompt-fidelity test caught. + prompt = sys.stdin.read() + base_url = os.environ.get("ANTHROPIC_BASE_URL") + if not base_url: + print(json.dumps({"type": "result", "subtype": "error", "is_error": True, + "session_id": "fake-session", "num_turns": 0}), flush=True) + return 1 + + emit = lambda event: print(json.dumps(event), flush=True) # noqa: E731 + emit({"type": "system", "subtype": "init", "session_id": "fake-session"}) + + try: + message = _turn(base_url, prompt) + except (OSError, ValueError) as exc: + # A provider failure is a failed SESSION, not a crashed process: dying + # here leaves no terminal result event, so the parent reports a generic + # stream error instead of the upstream failure it actually saw. + emit({"type": "result", "subtype": "error", "is_error": True, + "session_id": "fake-session", "num_turns": 0, + "error": f"provider request failed: {type(exc).__name__}: {exc}"}) + return 1 + blocks = message.get("content", []) + emit({"type": "assistant", "message": {"role": "assistant", "content": blocks}}) + + tool_results = [] + for block in blocks: + if block.get("type") == "tool_use": + # A refused write is a tool ERROR the session reports and carries + # on from, not a crash. Letting it kill the process would lose the + # result event and misreport a working boundary as a broken run. + failed = False + try: + output = _run_tool(block["name"], block.get("input", {})) + except (OSError, NotImplementedError) as exc: + output, failed = f"error: {type(exc).__name__}: {exc}", True + # is_error is load-bearing: the parent treats an ABSENT is_error as + # success, so a refused or unsupported tool would otherwise be + # scored as a completed one. + tool_results.append({ + "type": "tool_result", "tool_use_id": block["id"], + "content": output, "is_error": failed, + }) + if tool_results: + emit({"type": "user", "message": {"role": "user", "content": tool_results}}) + + # Unknown is not zero. A reply carrying no usage used to become four + # zero-valued fields plus a fabricated cost, which the harness then treats + # as a real measurement - the exact confusion the accounting this fixture + # feeds exists to prevent. + usage = message.get("usage") + # Every field that gets forwarded is validated, not just the required two. + # The parent's well_formed check tests only that the four keys are PRESENT, + # so an unvalidated cache value rides into a success result and is recorded + # as a real measurement. A field good enough to report is good enough to + # check. + countable = lambda v: isinstance(v, int) and not isinstance(v, bool) and v >= 0 # noqa: E731 + if not isinstance(usage, dict) or not all( + countable(usage.get(f)) for f in ("input_tokens", "output_tokens") + ) or not all( + countable(usage[f]) + for f in ("cache_read_input_tokens", "cache_creation_input_tokens") + if f in usage + ): + emit({"type": "result", "subtype": "error", "is_error": True, + "session_id": "fake-session", "num_turns": 1, + "error": "provider reply carried no usable usage; refusing to report a measured run"}) + return 1 + emit({ + "type": "result", + "subtype": "success", + "is_error": False, + "session_id": "fake-session", + "num_turns": 1, + "duration_ms": 1200, + # A measured zero is not the same as unmeasured; the parent rejects a + # collapsed cost, so report a real one. + "total_cost_usd": 0.42, + # Forward exactly the fields the provider reported. Defaulting the + # absent ones to 0 fabricated a complete measurement out of an + # incomplete reply - and worse, it made the parent's own completeness + # check (runner_sessions.USAGE_FIELDS / well_formed) unfirable from any + # offline test, because the stand-in always satisfied it. + "usage": { + field: usage[field] + for field in ("input_tokens", "output_tokens", + "cache_read_input_tokens", "cache_creation_input_tokens") + if field in usage + }, + }) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/eval/tests/test_mock_provider.py b/eval/tests/test_mock_provider.py new file mode 100644 index 000000000..201845fbf --- /dev/null +++ b/eval/tests/test_mock_provider.py @@ -0,0 +1,280 @@ +"""The mock has to be right about the wire, or every test built on it lies.""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +import urllib.request +from pathlib import Path + +import pytest + +from workflow_bench.mock_provider import MockProvider, Reply +from workflow_bench.provider_usage import ( + ANTHROPIC, + LITELLM_NORMALIZED, + 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 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" + event = events[-1] + native = event["native_usage"] + # LiteLLM hands a callback its OWN normalised object, not the upstream body: + # an OpenAI Responses reply arrives as prompt_tokens / prompt_tokens_details. + # Asserting the wire shape here is what proved the shipped adapter read keys + # that are never present. + assert native["prompt_tokens_details"]["cached_tokens"] == 7_000 + assert native["prompt_tokens_details"]["cache_write_tokens"] == 1_000 + assert event["provider"] == LITELLM_NORMALIZED + assert event["call_type"] == "anthropic_messages", "the observed call type, not a Responses one" + + usage = normalize_usage(event["provider"], native) + assert usage.total_input_tokens == 10_000 + assert usage.cache_read_input_tokens == 7_000 + assert usage.cache_write_input_tokens == 1_000 + assert usage.ordinary_input_tokens == 2_000 + assert usage.complete, "a run that cannot interpret its own usage measured nothing" + + +def test_probe_what_identity_the_real_cli_actually_sends(tmp_path: Path) -> None: + """An experiment, not an assertion: which fields could correlate a request to a cell? + + Per-cell usage attribution is unbuilt because one proxy serves the whole + sweep, so anything read from the proxy environment is identical for every + request. Attribution needs something that travels WITH the request, and + what the Claude Code CLI actually sends is not documented anywhere I can + check - guessing it is how the last three accounting bugs happened. + + So this drives the REAL pinned CLI against the mock and prints the + identity-bearing fields that arrive. It asserts only that a request was + made; the value is the recorded evidence, which the job log preserves. + """ + + claude = os.environ.get("CLAUDE_CANARY_BIN") + if not claude or not Path(claude).exists(): + pytest.skip("no pinned Claude CLI here; the containment job supplies CLAUDE_CANARY_BIN") + + with MockProvider(default=Reply(text="ok")) as provider: + subprocess.run( + [claude, "-p", "--input-format", "text", "--output-format", "stream-json", "--verbose"], + input=b"say ok", + capture_output=True, + timeout=120, + env={ + **os.environ, + "ANTHROPIC_BASE_URL": provider.base_url, + "ANTHROPIC_API_KEY": "offline-probe", + "HOME": str(tmp_path), + }, + ) + + assert provider.requests, "the real CLI never reached the mock provider" + request = provider.requests[0] + interesting = { + "header:" + name: value + for name, value in request.headers.items() + if any(k in name.lower() for k in ("session", "user", "trace", "request-id", "conversation", "metadata")) + } + interesting.update( + {f"body:{key}": request.body[key] for key in ("metadata", "user", "session_id") if key in request.body} + ) + print("\nIDENTITY FIELDS THE REAL CLI SENDS:") + print(" body keys:", sorted(request.body)) + print(" candidate correlators:", interesting or "NONE — per-cell attribution needs another mechanism") + + +def test_scripted_tools_survive_the_responses_protocol_too() -> None: + """The gateway uses Responses BECAUSE it carries tool use. + + Emitting only output_text there meant a scripted Write or Skill crossed the + gateway with the tool dropped, so a mock claiming to serve both protocols + was wrong about the one the gateway actually runs. + """ + + 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/responses", {"model": "m", "input": []}) + + output = json.loads(raw)["output"] + calls = [item for item in output if item["type"] == "function_call"] + assert len(calls) == 1, "the scripted tool must cross the Responses path" + assert calls[0]["name"] == "Write" + assert json.loads(calls[0]["arguments"])["file_path"] == "/review-output/review-output.json" + + +def test_an_omitted_cache_field_stays_omitted_on_the_responses_wire_too() -> None: + """Absence must survive both protocols, not just the Anthropic one. + + `_int_or_none` reads an absent detail key as unknown and a present 0 as a + measured zero, so serializing 0 for a scripted None would claim a + measurement the reply never made. + """ + + with MockProvider([Reply(input_tokens=2_000, cache_read_input_tokens=None)]) as provider: + _status, raw = _post(provider.base_url + "/v1/responses", {"model": "m", "input": []}) + + details = json.loads(raw)["usage"]["input_tokens_details"] + assert "cached_tokens" not in details, "an omitted field must not serialize as a measured zero" + assert details["cache_write_tokens"] == 0, "a scripted 0 is still a real measurement" diff --git a/eval/tests/test_offline_session_integration.py b/eval/tests/test_offline_session_integration.py new file mode 100644 index 000000000..cb5d1fe8f --- /dev/null +++ b/eval/tests/test_offline_session_integration.py @@ -0,0 +1,195 @@ +"""A session end to end with only the model faked. + +The layers between the CLI and the row are where this harness has actually +shipped bugs - the artifact that could not be written, the usage that was never +recorded, the evidence that was scored from the wrong directory. Every one of +them sat below the level its tests exercised. These run the real session path +against a scripted provider, so the only thing not real is what the model says. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import pytest + +from workflow_bench.mock_provider import MockProvider, Reply +from workflow_bench.proposer_sandbox import ( + host_workspace_write_boundary, + prepare_review_workspace, + prepare_sandbox, +) +from workflow_bench.review_scoring import REVIEW_OUTPUT, parse_review_output +from workflow_bench.runner_sessions import run_claude + +FAKE_CLI = Path(__file__).parent / "fixtures" / "fake_claude.py" +REVIEW_JSON = '{"schema_version": 1, "verdict": "approve", "findings": []}' + + +def _session(clone: Path, provider: MockProvider, **overrides): + return run_claude( + "review the change", + clone, + claude_bin=str(FAKE_CLI), + timeout=60, + env={ + "ANTHROPIC_BASE_URL": provider.base_url, + "ANTHROPIC_API_KEY": "offline", + "PATH": "/usr/bin:/bin", + }, + **overrides, + ) + + +@pytest.fixture +def clone(tmp_path: Path) -> Path: + workspace = tmp_path / "clone" + workspace.mkdir() + (workspace / "source.ts").write_text("export const answer = 42;\n") + return workspace + + +def test_a_session_records_the_usage_the_provider_reported(clone: Path) -> None: + """Token counts must survive the CLI boundary, not be invented after it.""" + + 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: + record = _session(clone, provider) + + assert record["ok"] is True, record.get("error_detail") + assert record["input_tokens"] == 2_000 + assert record["cache_read_input_tokens"] == 7_000 + assert record["cache_creation_input_tokens"] == 1_000 + assert record["output_tokens"] == 300 + # A measured zero would be indistinguishable from an unmeasured one. + assert record["cost_usd"] == 0.42 + assert record["num_turns"] == 1 + + +def test_a_scripted_write_produces_a_review_artifact_the_scorer_accepts(clone: Path) -> None: + """The full artifact path: model asks, CLI writes atomically, scorer reads. + + This is the operation that shipped empty for a whole run. Nothing here + fakes the write, the directory, or the parse - only the decision to write. + """ + + with prepare_sandbox( + clone=clone, claude_bin=Path(sys.executable), backend="host-unsafe", preflight=False + ) as sandbox: + artifact = prepare_review_workspace(sandbox, REVIEW_OUTPUT) + write = {"name": "Write", "input": {"file_path": str(artifact), "content": REVIEW_JSON}} + with MockProvider(default=Reply(text="reviewing", tools=[write])) as provider: + # Take the command configuration from the sandbox the way run_arm + # does, rather than calling run_claude bare. On host-unsafe the + # prefix is [] by construction, so this pins the WIRING, not the + # isolation - a bwrap run would carry a real prefix through here. + record = _session( + clone, + provider, + command_prefix=sandbox.command_prefix_for(), + require_pid_namespace=sandbox.require_pid_namespace, + ) + assert record["ok"] is True, record.get("error_detail") + # Read inside the scope: prepare_sandbox removes the private root on exit. + verdict, findings = parse_review_output(artifact) + + assert verdict == "approve" + assert findings == () + + +def test_the_provider_saw_the_prompt_the_harness_meant_to_send(clone: Path) -> None: + """A run that measures the wrong prompt measures nothing.""" + + with MockProvider() as provider: + _session(clone, provider) + + assert provider.requests, "the session never reached the provider" + sent = provider.requests[0].body["messages"][0]["content"] + assert "review the change" in sent + + +def test_a_provider_failure_surfaces_as_a_failed_session_not_a_silent_pass(clone: Path) -> None: + """An upstream 529 must not be recorded as a usable measurement.""" + + failing = Reply(status_code=529, error_body={"error": {"type": "overloaded_error"}}) + with MockProvider(default=failing) as provider: + record = _session(clone, provider) + + assert record["ok"] is False + assert record["error_kind"] is not None + + +def test_the_write_boundary_refuses_the_workspace_and_permits_the_artifact(clone: Path, tmp_path: Path) -> None: + """The contract the empty-artifact run violated, on the backend available here. + + A review must not change the workspace, and must still be able to write its + artifact ATOMICALLY - temp file beside the target, then rename - which is + what needs a writable parent DIRECTORY rather than a writable file. Both + halves are asserted through the real session, with the real boundary + applied, and the model scripted to attempt each one. + + Scope: this is the host-unsafe boundary, which its own docstring calls + best-effort because a session that can chmod can undo it. The kernel-enforced + version is bubblewrap's --ro-bind, which needs namespaces this machine cannot + create; that half stays with the real-sandbox canary in CI. + """ + + artifacts = tmp_path / "artifacts" + artifacts.mkdir() + target = artifacts / REVIEW_OUTPUT + protected = clone / "source.ts" + before = protected.read_text() + + write_artifact = {"name": "Write", "input": {"file_path": str(target), "content": REVIEW_JSON}} + tamper = {"name": "Write", "input": {"file_path": str(protected), "content": "tampered"}} + + # No writable= entry: the boundary only governs paths INSIDE the workspace + # (it refuses one that escapes), and the artifact directory deliberately + # lives outside it - that relocation is the fix for the empty-artifact run. + with host_workspace_write_boundary(clone): + with MockProvider(default=Reply(text="writing", tools=[write_artifact, tamper])) as provider: + record = _session(clone, provider) + + assert record["ok"] is True, record.get("error_detail") + # The artifact landed, written the way the agent's Write tool does it. + verdict, _findings = parse_review_output(target) + assert verdict == "approve" + assert not list(artifacts.glob("*.tmp.*")), "the rename landed rather than a copy" + # The workspace did not move. + assert protected.read_text() == before, "the read-only workspace was modified" + + +def test_a_reply_missing_cache_usage_is_refused_not_zero_filled(clone: Path) -> None: + """An omitted cache field must not arrive as a measured zero. + + The parent already demands all four USAGE_FIELDS before it calls a session + measured (runner_sessions.well_formed). The stand-in used to default the + absent ones to 0, which both fabricated a complete measurement AND made + that parent guard unfirable from any offline test - it was always + satisfied. Scripting the absence is what proves the guard still fires. + """ + + partial = Reply(input_tokens=2_000, output_tokens=300, cache_read_input_tokens=None) + with MockProvider(default=partial) as provider: + record = _session(clone, provider) + + assert record["ok"] is False, "an incomplete usage report is not a usable measurement" + assert record["error_kind"] == "session-error" + + +@pytest.mark.parametrize("bad", [-5, True, "1200"], ids=["negative", "boolean", "string"]) +def test_a_nonsense_cache_value_is_refused_rather_than_forwarded(clone: Path, bad: object) -> None: + """A field good enough to report is good enough to validate. + + The parent's well_formed check tests only that the four keys are PRESENT, + so an unvalidated cache value would ride into a success result and be + recorded as a real measurement. + """ + + reply = Reply(input_tokens=2_000, output_tokens=300) + object.__setattr__(reply, "cache_read_input_tokens", bad) + with MockProvider(default=reply) as provider: + record = _session(clone, provider) + + assert record["ok"] is False, f"{bad!r} must not be recorded as a measured cache value" diff --git a/eval/tests/test_offline_sweep_integration.py b/eval/tests/test_offline_sweep_integration.py new file mode 100644 index 000000000..7ffb6db25 --- /dev/null +++ b/eval/tests/test_offline_sweep_integration.py @@ -0,0 +1,348 @@ +"""A whole sweep, offline: real runner, real sessions, scripted model. + +The layers between a model turn and a promotion decision had never been +exercised together. Unit tests covered each in isolation and the paid runs that +would have covered the composition kept dying, so the contracts BETWEEN them +went unverified - and that is where this harness has repeatedly shipped bugs. + +This drives runner.main() the way the workflow does. Everything is real: task +selection, hidden-oracle capture, the sandbox, the CLI subprocess, artifact +capture, review scoring against the oracle, aggregation, the health guard, and +the promotion gate. Only the model is scripted, through MockProvider. + +Two provisioning steps are stubbed because this environment cannot supply them, +and neither is harness logic: the pinned gitnexus runtime mounts (no +node_modules in a worktree) and the sanitized graph build (needs the gitnexus +CLI at a mounted path). Containment is host-unsafe here; bubblewrap stays with +the real-sandbox canary in the containment job. +""" + +from __future__ import annotations + +import json +import re +import subprocess +import sys +from pathlib import Path +from types import SimpleNamespace + +import os +import shutil + +import pytest + +from workflow_bench import oracle_assets, runner +from workflow_bench.mock_provider import MockProvider, Reply + +FAKE_CLI = Path(__file__).parent / "fixtures" / "fake_claude.py" +ARMS = ("ce_review", "review", "candidate_review") + +# When set, the sweep runs with NOTHING provisioning-stubbed: real bubblewrap +# containment, the real pinned runtime mounts, and the real sanitized graph +# build. The named CI job installs all three, so a missing one there is a +# regression rather than an unsupported machine - it FAILS instead of quietly +# degrading to the stubbed path, which is the whole point of the gate. +FULL_SWEEP_ENV = "GITNEXUS_REQUIRE_FULL_SWEEP" +FULL_SWEEP = os.environ.get(FULL_SWEEP_ENV) == "1" + +# The runner refuses --unsafe-no-bwrap whenever CI is set, because that mode runs +# sessions with bypassPermissions behind a boundary its own docstring calls "not +# a security boundary". Deleting CI to get past that refusal would run an +# uncontained agent sweep on the runner holding the checkout and credentials, so +# the stubbed path is skipped under CI instead. The containment job sets +# GITNEXUS_REQUIRE_FULL_SWEEP=1 and takes the real bubblewrap path, so CI keeps +# its coverage; only the uncontained convenience run is given up. +pytestmark = pytest.mark.skipif( + not FULL_SWEEP and bool(os.environ.get("CI")), + reason="an uncontained sweep must not run in CI; the containment job runs it with GITNEXUS_REQUIRE_FULL_SWEEP=1", +) + +# The review output and the hidden labels are DELIBERATELY different shapes - +# the labels carry line_start/line_end and no recommendation. Only a real run +# surfaces that; it is why these are written out rather than shared. +FINDING = { + "id": "f1", "severity": "high", "category": "correctness", "path": "src/sum.js", + "line": 1, "end_line": 1, "blocking": True, "scenario": "review-defect", + "evidence": "export const total = (a, b) => a - b;", "recommendation": "use a + b", +} +LABEL = {"id": "f1", "severity": "high", "category": "correctness", + "path": "src/sum.js", "line_start": 1, "line_end": 1} +SECOND_LABEL = {"id": "f2", "severity": "high", "category": "correctness", + "path": "src/scale.js", "line_start": 1, "line_end": 1} +SECOND_FINDING = { + "id": "f2", "severity": "high", "category": "correctness", "path": "src/scale.js", + "line": 1, "end_line": 1, "blocking": True, "scenario": "review-defect", + "evidence": "export const twice = (n) => n + 2;", "recommendation": "use n * 2", +} + + +def _git(repo: Path, *args: str) -> str: + return subprocess.run(["git", "-C", str(repo), *args], check=True, + capture_output=True, text=True).stdout.strip() + + +@pytest.fixture +def bench(tmp_path: Path): + """A self-contained corpus: one repo, one task, one hidden label.""" + + repo = tmp_path / "repo" + (repo / "src").mkdir(parents=True) + (repo / "src" / "sum.js").write_text("export const total = (a, b) => a - b;\n") + (repo / "src" / "scale.js").write_text("export const twice = (n) => n + 2;\n") + _git(repo, "init", "-q", ".") + _git(repo, "config", "user.email", "t@t") + _git(repo, "config", "user.name", "t") + _git(repo, "add", "-A") + _git(repo, "commit", "-q", "-m", "fixture") + sha = _git(repo, "rev-parse", "HEAD") + + oracles = tmp_path / "oracles" + oracles.mkdir() + (oracles / "review-fixture-defect.labels.json").write_text( + json.dumps({"schema_version": 1, "findings": [LABEL]}) + ) + (oracles / "review-fixture-second.labels.json").write_text( + json.dumps({"schema_version": 1, "findings": [SECOND_LABEL]}) + ) + + tasks = tmp_path / "tasks.yaml" + tasks.write_text( + "tasks:\n" + " - id: review-fixture-defect\n" + " class: review-defect\n" + f" repo: {repo}\n" + f" ref: {sha}\n" + " prompt: Review this change and report actionable defects.\n" + ' verify: test -s "$GITNEXUS_BENCH_REVIEW_OUTPUT"\n' + " oracle:\n" + ' command: test -s "$GITNEXUS_BENCH_REVIEW_OUTPUT"\n' + " files: [{ source: review-fixture-defect.labels.json, target: review-labels.json }]\n" + # A SECOND task, because the thing a cross-task scheduler changes is + # invisible with one: waves are per-task, so a single task cannot show + # ordering, packing, or a breaker that spans a task boundary. + " - id: review-fixture-second\n" + " class: review-defect\n" + f" repo: {repo}\n" + f" ref: {sha}\n" + " prompt: Review the scaling helper and report actionable defects.\n" + ' verify: test -s "$GITNEXUS_BENCH_REVIEW_OUTPUT"\n' + " oracle:\n" + ' command: test -s "$GITNEXUS_BENCH_REVIEW_OUTPUT"\n' + " files: [{ source: review-fixture-second.labels.json, target: review-labels.json }]\n" + ) + + plugin = tmp_path / "ce-plugin" + (plugin / ".claude-plugin").mkdir(parents=True) + (plugin / ".claude-plugin" / "plugin.json").write_text( + json.dumps({"name": "compound-engineering", "version": "0.0.0-fixture"}) + ) + for skill in ("ce-plan", "ce-work", "ce-code-review"): + directory = plugin / "skills" / skill + directory.mkdir(parents=True) + (directory / "SKILL.md").write_text(f"---\nname: {skill}\ndescription: fixture\n---\nFixture.\n") + + overlay = tmp_path / "overlay" / ".claude" / "skills" / "gitnexus-review" + overlay.mkdir(parents=True) + (overlay / "SKILL.md").write_text("---\nname: gitnexus-review\ndescription: fixture\n---\nCandidate.\n") + + return SimpleNamespace(tasks=tasks, oracles=oracles, plugin=plugin, + overlay=tmp_path / "overlay", out=tmp_path / "out") + + +def _stub_provisioning(monkeypatch: pytest.MonkeyPatch) -> None: + """Replace what this machine cannot supply - and nothing else. + + Under FULL_SWEEP nothing is replaced: the runtime mounts and the graph are + built for real, so the sweep exercises containment and provisioning too. + """ + + if FULL_SWEEP: + if shutil.which("bwrap") is None: + pytest.fail(f"{FULL_SWEEP_ENV}=1 but bubblewrap is absent") + return + + monkeypatch.setattr(runner, "trusted_gitnexus_runtime_mounts", lambda: ()) + + def materialize(worktree, *, sanitized_head=None, **_kwargs): + # The one-clone registry guard reads this before any session runs. + meta = Path(worktree) / ".gitnexus" + meta.mkdir(parents=True, exist_ok=True) + (meta / "meta.json").write_text( + json.dumps({"indexedAt": "2026-09-08T00:00:00Z", "lastCommit": sanitized_head or "0" * 40}) + ) + + def fake_graph(**kwargs): + kwargs["env"].graph_snapshots[kwargs["graph_key"]] = SimpleNamespace( + digest="fixture-graph", manifest_digest="fixture-graph-manifest", + dependency_content_digest=None, dependency_manifest_digest=None, + materialize=materialize, + ) + + monkeypatch.setattr(runner, "ensure_task_graph", fake_graph) + + +def _sweep(bench, monkeypatch: pytest.MonkeyPatch, findings: list[dict], verdict: str, *, invoke_skill: bool = True): + """Run the real CLI against a model scripted to return `findings`.""" + + _stub_provisioning(monkeypatch) + monkeypatch.setattr( + oracle_assets, "ORACLE_ROOT", bench.oracles, raising=False + ) + monkeypatch.setattr( + runner, "capture_task_oracles", + lambda tasks, root=bench.oracles: oracle_assets.capture_task_oracles(tasks, root=root), + ) + def review_for(body: str) -> str: + # Per task: the second task's defect is in another file, so replying + # with the first task's finding would score it wrong. A cross-task + # scheduler makes which task a request belongs to load-bearing. + chosen = findings + if findings and "scaling helper" in body: + chosen = [SECOND_FINDING if f is FINDING else f for f in findings] + return json.dumps({"schema_version": 1, "verdict": verdict, "findings": chosen}) + + class Scripted(MockProvider): + def next_reply(self) -> Reply: + body = json.dumps(self.requests[-1].body if self.requests else {}) + target = re.search(r"(/[^\s\"']*review-output\.json)", body) + skill = re.search(r"\b(gitnexus-review|ce-code-review)\b", body) + return Reply( + text="reviewing", + tools=[ + # The evidence gate needs a Skill request with a non-error + # result: a review that never invoked its skill measured the + # model, not the skill. + *([{"name": "Skill", "input": {"skill": skill.group(1) if skill else "gitnexus-review"}}] + if invoke_skill else []), + {"name": "Write", "input": { + "file_path": target.group(1) if target else str(bench.out / "unmatched-review-output.json"), + "content": review_for(body)}}, + ], + input_tokens=2_000, output_tokens=300, + cache_read_input_tokens=7_000, cache_creation_input_tokens=1_000, + ) + + with Scripted() as provider: + monkeypatch.setattr(sys, "argv", [ + "runner", "--tasks", str(bench.tasks), "--arms", *ARMS, + "--runs", "1", "--workers", "1", "--out", str(bench.out), + "--base-url", provider.base_url, "--anthropic-api-key", "offline", + "--claude-bin", str(FAKE_CLI), + *([] if FULL_SWEEP else ["--unsafe-no-bwrap"]), + "--model", "mock-model", + "--ce-plugin-dir", str(bench.plugin), "--ce-plugin-version", "0.0.0-fixture", + "--candidate-overlay", str(bench.overlay), + ]) + + try: + code = runner.main() + except SystemExit as exc: + code = exc.code + + rows = [json.loads(line) for line in (bench.out / "results.jsonl").read_text().splitlines()] + return code, rows, provider + + +def _row(rows: list[dict], arm: str, task: str = "review-fixture-defect") -> dict: + return next(r for r in rows if r["arm"] == arm and r["task"] == task) + + +def test_a_correct_review_scores_and_the_sweep_exits_clean(bench, monkeypatch) -> None: + """The whole path, green: every arm measured, scored, and accounted for.""" + + code, rows, provider = _sweep(bench, monkeypatch, [FINDING], "request_changes") + + assert code in (None, 0), f"sweep did not succeed: {code}" + tasks = {"review-fixture-defect", "review-fixture-second"} + assert len(rows) == len(ARMS) * len(tasks) + assert len(provider.requests) == len(ARMS) * len(tasks), "each cell must reach the provider once" + assert {r["task"] for r in rows} == tasks, "both tasks must have run" + + row = _row(rows, "review") + assert row["ok"] is True and row["resolved"] is True + assert row["skill_invoked"] is True + assert (row["review_true_positives"], row["review_false_positives"], row["review_false_negatives"]) == (1, 0, 0) + assert row["review_f1"] == 1.0 + # The provider's own numbers survived the CLI, the parser and the row. + assert row["cache_read_input_tokens"] == 7_000 + assert row["input_tokens"] == 2_000 + + # Each task scored against ITS OWN oracle. This is what a cross-task + # scheduler puts at risk: interleaving cells from different tasks means a + # mis-routed context or artifact scores one task against another's labels, + # and both would still look "green" per row. + second = _row(rows, "review", task="review-fixture-second") + assert second["resolved"] is True and second["review_f1"] == 1.0 + assert second["review_artifact"] == "review-fixture-second-review-run0.review.json" + + for name in ("results.jsonl", "report.md", "promotion.json"): + assert (bench.out / name).is_file(), f"{name} was not written" + assert (bench.out / "review-fixture-defect-review-run0.review.json").is_file() + + +def test_one_run_cannot_promote_a_candidate(bench, monkeypatch) -> None: + """The gate refuses on insufficient paired runs, and says so.""" + + _sweep(bench, monkeypatch, [FINDING], "request_changes") + promotion = json.loads((bench.out / "promotion.json").read_text()) + + assert promotion["run_status"] == "complete" + decision = next(d for d in promotion["decisions"] if d["candidate_arm"] == "candidate_review") + assert decision["decision"] == "insufficient_evidence" + assert any("valid paired runs" in reason for reason in decision["reasons"]) + + +def test_a_finding_in_the_wrong_place_scores_zero_but_stays_valid_evidence(bench, monkeypatch) -> None: + """Being wrong is a quality result, not a broken measurement. + + The negative control that makes the passing case mean something: same + harness, same well-formed artifact, only the answer changed. + """ + + wrong = {**FINDING, "path": "src/WRONG.js", "line": 99, "end_line": 99} + _code, rows, _provider = _sweep(bench, monkeypatch, [wrong], "request_changes") + + row = _row(rows, "review") + assert (row["review_true_positives"], row["review_false_positives"], row["review_false_negatives"]) == (0, 1, 1) + assert row["review_f1"] == 0.0 + assert row["resolved"] is False + assert row["error_kind"] == "oracle-failed", "a wrong answer is not a session or evidence failure" + assert row["review_evidence_valid"] is True, "the artifact was well formed; only the answer was wrong" + + +def test_approving_defective_code_is_a_miss_with_no_false_positive(bench, monkeypatch) -> None: + """The other half of the control: silence scores differently from a wrong guess.""" + + _code, rows, _provider = _sweep(bench, monkeypatch, [], "approve") + + row = _row(rows, "review") + assert (row["review_true_positives"], row["review_false_positives"], row["review_false_negatives"]) == (0, 0, 1) + assert row["review_precision"] is None, "precision is undefined with no predictions, not zero" + assert row["review_verdict_correct"] is False, "approving defective code is the wrong verdict" + assert row["review_evidence_valid"] is True + + +def test_a_review_that_never_invoked_its_skill_is_not_a_measurement(bench, monkeypatch) -> None: + """The gate that separates measuring a SKILL from measuring a model. + + Added because a mutation exposed it: forcing skill_was_invoked_events to + return True left every other test here passing, so nothing pinned the gate. + The artifact is written and correct in this run - only the skill request is + missing - so a pass would mean the arm scored a review it never performed. + """ + + code, rows, _provider = _sweep(bench, monkeypatch, [FINDING], "request_changes", invoke_skill=False) + + row = _row(rows, "review") + assert row["skill_invoked"] is False + assert row["error_kind"] == "skill-not-invoked" + assert code not in (None, 0), "the sweep must not report success on unusable evidence" + + # The row still carries its own score - the artifact was well formed - and + # aggregate() DOES count it in the arm's quality median (the KNOWN GAP noted + # above aggregate(); test_workflow_bench pins the resulting 0.5). Filtering + # it out of the median alone inverted a promotion, because valid_runs and + # excluded_runs kept counting it. It counts for cost either way: the session + # ran and was billed. + assert row["review_weighted_f1"] == 1.0 + assert row["review_evidence_valid"] is True diff --git a/eval/tests/test_provider_usage_capture.py b/eval/tests/test_provider_usage_capture.py index 99e542e4d..91e4956e5 100644 --- a/eval/tests/test_provider_usage_capture.py +++ b/eval/tests/test_provider_usage_capture.py @@ -24,7 +24,7 @@ from workflow_bench.model_gateway import ( ) from workflow_bench.provider_usage import ( ANTHROPIC, - OPENAI_RESPONSES, + LITELLM_NORMALIZED, USAGE_ENV_VARS, normalize_usage, ) @@ -49,11 +49,16 @@ def _openai_response(usage: dict) -> SimpleNamespace: ) +# The shape a callback actually receives: LiteLLM normalises usage into its own +# Chat-Completions-style object before any logger sees it, so an OpenAI reply +# arrives as prompt_tokens / prompt_tokens_details. Confirmed against a real +# proxy in tests/test_mock_provider.py; a fixture in the wire shape would test +# an object this code path never gets. 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}, + "prompt_tokens": 48_000, + "prompt_tokens_details": {"cached_tokens": 44_000, "cache_write_tokens": 1_000}, + "completion_tokens": 900, + "completion_tokens_details": {"reasoning_tokens": 640}, } @@ -79,9 +84,9 @@ 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 native["prompt_tokens_details"]["cached_tokens"] == 44_000 + assert native["prompt_tokens_details"]["cache_write_tokens"] == 1_000 + assert native["completion_tokens_details"]["reasoning_tokens"] == 640 assert event["response_id"] == "resp_68f2c1" @@ -100,7 +105,10 @@ def test_the_captured_event_normalizes_with_openai_arithmetic(logged) -> None: 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 + # LITELLM_NORMALIZED, not OPENAI_RESPONSES: a proxy callback never sees the + # upstream body. Measured against a real gateway - the Responses adapter + # found none of its keys there and reported every field unknown. + assert event["provider"] == LITELLM_NORMALIZED assert event["provider_label"] == "openai" usage = normalize_usage(event["provider"], event["native_usage"]) assert usage.total_input_tokens == 48_000 @@ -112,7 +120,7 @@ def test_the_captured_event_normalizes_with_openai_arithmetic(logged) -> None: 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"} + stripped = {k: v for k, v in NATIVE.items() if k != "prompt_tokens_details"} event = logged(stripped) usage = normalize_usage(event["provider"], event["native_usage"]) assert usage.cache_read_input_tokens is None @@ -238,10 +246,16 @@ def test_an_unresolvable_provider_is_refused_rather_than_guessed() -> None: 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 + # Every openai call reaching this callback has already been normalised by + # LiteLLM, whatever endpoint it used - the observed call_type for a Claude + # Code request through the gateway is "anthropic_messages". The adapter has + # to match the object in hand, not the protocol on the wire. + assert canonical_provider("openai", "responses") == LITELLM_NORMALIZED + assert canonical_provider("openai", "anthropic_messages") == LITELLM_NORMALIZED assert canonical_provider("anthropic", "completion") == ANTHROPIC + # An unrecognised provider is still refused rather than guessed. + assert canonical_provider("some-new-provider", "responses") is None + assert canonical_provider(None, None) is None def test_request_identity_cannot_come_from_the_proxy_environment() -> None: diff --git a/eval/tests/test_workflow_bench.py b/eval/tests/test_workflow_bench.py index a53b2b8cd..51212ceda 100644 --- a/eval/tests/test_workflow_bench.py +++ b/eval/tests/test_workflow_bench.py @@ -195,6 +195,11 @@ def test_eval_ci_uses_locked_uv_and_blocking_native_containment_jobs(): assert containment["env"] == { "GITNEXUS_REQUIRE_BWRAP_CANARY": "1", "GITNEXUS_REQUIRE_CLAUDE_CANARY": "1", + # This job is the only place with bubblewrap, the pinned runtime and a + # built GitNexus together, so it is where the offline sweep runs with + # nothing provisioning-stubbed. Pinned here so the gate cannot be + # dropped and leave the sweep silently running the stubbed path. + "GITNEXUS_REQUIRE_FULL_SWEEP": "1", } assert containment["timeout-minutes"] == 20 assert containment_node_setup["with"] == { @@ -239,6 +244,12 @@ def test_eval_ci_uses_locked_uv_and_blocking_native_containment_jobs(): "tests/test_proposer_sandbox.py", "tests/test_workflow_bench_sessions.py", "tests/test_ce_plugin_runtime.py", + # The offline sweep, run here with nothing stubbed: this job is the only + # one carrying bubblewrap, the pinned runtime and a built GitNexus. + "tests/test_offline_sweep_integration.py", + # Carries the real-CLI identity probe, which needs CLAUDE_CANARY_BIN - + # set only on this job. Omitted from this list it skipped everywhere. + "tests/test_mock_provider.py", "-q", ] bwrap_canary_marker = re.compile( @@ -1052,3 +1063,33 @@ def test_a_raising_packed_cell_still_persists_its_settled_siblings(): ) assert (1, "review") in folded, "the sibling that completed was never recorded" + + +def test_an_uninvoked_skill_still_counts_toward_the_arm_median(): + """Pins a KNOWN GAP, not a desired behaviour. + + A cell whose skill never ran still moves the arm's quality median, even + though an arm exists to measure a SKILL. The narrow fix - filtering those + rows out of the quality metrics - is worse than the gap: valid_runs and + excluded_runs keep counting them, so the promotion gate sees N clean runs + while the median came from fewer. Since the dropped rows are systematically + an arm's worst, that biases toward promoting, and it was measured flipping + keep_incumbent to promote. + + Closing it honestly needs a scored-run count and a paired-equality check in + the promotion gate. Pinned here so the half-fix cannot be reapplied without + someone reading why it was reverted. + """ + + good = record(review_weighted_f1=1.0, cost_usd=2.0) + uninvoked = record( + review_weighted_f1=0.0, cost_usd=4.0, error_kind="skill-not-invoked", skill_invoked=False + ) + agg = aggregate([good, uninvoked]) + + assert agg["review_weighted_f1"] == 0.5, "the uninvoked row is counted - the known gap" + assert agg["cost_usd"] == 3.0 + # The invariant that makes the half-fix unsafe: the median and the run count + # the gate reads must cover the same rows. + assert agg["valid_runs"] == 2 + diff --git a/eval/workflow_bench/litellm_usage_callback.py b/eval/workflow_bench/litellm_usage_callback.py index 843413765..8277d0102 100644 --- a/eval/workflow_bench/litellm_usage_callback.py +++ b/eval/workflow_bench/litellm_usage_callback.py @@ -42,8 +42,8 @@ def canonical_provider(label, call_type): # noqa: ANN001, ANN201 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 == "openai": + return "litellm-normalized" if label == "anthropic": return "anthropic" return None @@ -104,11 +104,10 @@ class ProviderUsageLogger(CustomLogger): "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. + # label is what LiteLLM said; "provider" is the adapter key for + # the object actually in hand, which is always LiteLLM's own + # normalised shape here. An unrecognised label 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), diff --git a/eval/workflow_bench/mock_provider.py b/eval/workflow_bench/mock_provider.py new file mode 100644 index 000000000..dcc328251 --- /dev/null +++ b/eval/workflow_bench/mock_provider.py @@ -0,0 +1,284 @@ +"""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:``, 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 + # None means the field is OMITTED from the reply, which is not the same as + # reporting 0. A consumer that cannot tell those apart is the bug this + # harness exists to catch, so the mock has to be able to script absence. + cache_read_input_tokens: int | None = 0 + cache_creation_input_tokens: int | None = 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]: + usage = { + "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, + } + return {field: value for field, value in usage.items() if value is not None} + + +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.""" + + # An omitted cache field contributes nothing to the Responses total; that + # is arithmetic, not a claim the value was measured as zero. + cache_read = reply.cache_read_input_tokens or 0 + cache_write = reply.cache_creation_input_tokens or 0 + total_input = reply.input_tokens + cache_read + cache_write + 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": []}], + }, + # Tool calls belong here too. Responses is the protocol the gateway + # is configured for BECAUSE it carries tool use, so emitting only + # output_text meant a reply scripted with a Write or Skill crossed + # the gateway with the tool silently dropped - the mock would have + # been wrong about the wire on the one path that matters most. + *( + { + "id": f"fc_mock_{index}", + "type": "function_call", + "status": "completed", + "call_id": f"call_mock_{index}", + "name": tool["name"], + "arguments": json.dumps(tool.get("input", {})), + } + for index, tool in enumerate(reply.tools) + ), + ], + "usage": { + "input_tokens": total_input, + "output_tokens": reply.output_tokens, + "total_tokens": total_input + reply.output_tokens, + # Omitted stays omitted here too. Collapsing None to 0 is right for + # the total above (an unreported field adds nothing) but wrong on + # the wire: _int_or_none reads an absent key as unknown and a + # present 0 as a measured zero, so serializing 0 would claim a + # measurement the reply never made - the same confusion the + # Anthropic path already refuses. + "input_tokens_details": { + **({"cached_tokens": cache_read} if reply.cache_read_input_tokens is not None else {}), + **({"cache_write_tokens": cache_write} if reply.cache_creation_input_tokens is not None else {}), + }, + "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) diff --git a/eval/workflow_bench/provider_usage.py b/eval/workflow_bench/provider_usage.py index 782eb18fa..a56582be1 100644 --- a/eval/workflow_bench/provider_usage.py +++ b/eval/workflow_bench/provider_usage.py @@ -50,6 +50,14 @@ USAGE_ENV_VARS = (USAGE_LOG_ENV_VAR, SWEEP_ID_ENV_VAR) ANTHROPIC = "anthropic" OPENAI_RESPONSES = "openai-responses" +# What the gateway's callback actually receives. LiteLLM does not hand a logger +# the upstream body: it normalises usage into its own Chat-Completions-shaped +# object first, so an OpenAI Responses reply arrives as prompt_tokens / +# prompt_tokens_details even though the wire carried input_tokens / +# input_tokens_details. Measured against a real proxy, not assumed - the +# Responses adapter below found none of its keys and reported every field +# unknown. The arithmetic is still OpenAI's (the whole, with subsets). +LITELLM_NORMALIZED = "litellm-normalized" class UsageSemanticsError(ValueError): @@ -131,6 +139,35 @@ def _normalize_openai_responses(usage: Mapping[str, Any]) -> NormalizedUsage: ) +def _normalize_litellm(usage: Mapping[str, Any]) -> NormalizedUsage: + """prompt_tokens is the WHOLE; the details are subsets of it.""" + + total = _int_or_none(usage, "prompt_tokens") + details = usage.get("prompt_tokens_details") + cache_read = _int_or_none(details, "cached_tokens") + cache_write = _int_or_none(details, "cache_write_tokens") + if cache_write is None: + cache_write = _int_or_none(details, "cache_creation_tokens") + output_details = usage.get("completion_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"LiteLLM cached ({cache_read}) + cache_write ({cache_write}) " + f"exceed prompt_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, "completion_tokens"), + 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.""" @@ -154,16 +191,24 @@ def _normalize_anthropic(usage: Mapping[str, Any]) -> NormalizedUsage: 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. + Every "openai" label maps to LITELLM_NORMALIZED regardless of call type, + because anything reaching a proxy callback has already been normalised by + LiteLLM into its own object - measured against a real gateway, where the + observed call type is "anthropic_messages" and the upstream Responses shape + never arrives. OPENAI_RESPONSES stays in the adapter table for a RAW + upstream body, which only direct callers and the wire-shape tests pass. + + An unrecognised label still returns None, so normalize_usage refuses rather + than guessing token semantics. """ - if label == "openai" and call_type and "responses" in call_type: - return OPENAI_RESPONSES + if label == "openai": + # Anything reaching a proxy callback has already been normalised by + # LiteLLM, whichever endpoint the caller used - the observed call_type + # for a Claude Code request through this gateway is "anthropic_messages", + # not a Responses one. The adapter has to match the object in hand, not + # the protocol on the wire. + return LITELLM_NORMALIZED if label in _ADAPTERS: return label return None @@ -172,6 +217,7 @@ def canonical_provider(label: str | None, call_type: str | None) -> str | None: _ADAPTERS = { ANTHROPIC: _normalize_anthropic, OPENAI_RESPONSES: _normalize_openai_responses, + LITELLM_NORMALIZED: _normalize_litellm, } diff --git a/eval/workflow_bench/runner.py b/eval/workflow_bench/runner.py index a2223a20a..42cbfc9cb 100644 --- a/eval/workflow_bench/runner.py +++ b/eval/workflow_bench/runner.py @@ -1591,6 +1591,21 @@ def aggregate(records: list[dict[str, Any]]) -> dict[str, Any]: "review_category_accuracy", "review_grounded_evidence", ) + # NOTE: a skill-not-invoked row still contributes to these medians. That is + # a real measurement gap - an arm exists to measure a SKILL, and a cell + # where the skill never ran did not measure it - but the narrow fix is + # WORSE than the gap, so it is deliberately not applied here. + # + # Filtering those rows out of the quality metrics alone leaves valid_runs + # and excluded_runs counting them, so the promotion gate sees N clean runs + # while the median was taken over fewer. Because the dropped rows are + # systematically an arm's worst, that biases toward PROMOTING: measured on + # one real run at 0.9 plus two uninvoked rows at 0.0, the gate flipped from + # keep_incumbent to promote. The three verdict fields below compound it - + # they are all() reducers, so one uninvoked cell flips a whole arm. + # Closing this honestly needs a scored-run count and a paired-equality + # check in the gate itself: a promotion-semantics change, not an + # aggregation fix. if any("review_weighted_f1" in record for record in valid): for metric in review_metrics: values = [record[metric] for record in valid if record.get(metric) is not None]