diff --git a/.github/workflows/release-evaluation.yml b/.github/workflows/release-evaluation.yml index 996c18aad..f5001a377 100644 --- a/.github/workflows/release-evaluation.yml +++ b/.github/workflows/release-evaluation.yml @@ -414,6 +414,8 @@ jobs: if [[ -n "$INPUT_PR" && "$runtime" == stable ]]; then continue; fi remaining=$(( ${{ needs.start-runner.outputs.stop_deadline_epoch }} - $(date +%s) - 5400 )) (( remaining > 0 )) || { echo '::error::No time remains before the stop and evidence reserve.'; exit 1; } + GITNEXUS_BENCH_PROVIDER_USAGE="$RUNNER_TEMP/release-evaluation/$runtime/provider-usage.jsonl" \ + GITNEXUS_BENCH_SWEEP_ID="release-${{ github.run_id }}-${{ github.run_attempt }}-$runtime" \ timeout --signal=TERM --kill-after=60 "${remaining}s" \ uv run --locked --extra dev python -m workflow_bench.runner \ --tasks "$RUNNER_TEMP/release-evaluation/$runtime/tasks.yaml" \ @@ -437,6 +439,8 @@ jobs: set -euo pipefail remaining=$(( ${{ needs.start-runner.outputs.stop_deadline_epoch }} - $(date +%s) - 5400 )) (( remaining > 0 )) || { echo '::warning::No time remains for the CLI-only arm.'; exit 1; } + GITNEXUS_BENCH_PROVIDER_USAGE="$RUNNER_TEMP/release-evaluation/candidate/cli-provider-usage.jsonl" \ + GITNEXUS_BENCH_SWEEP_ID="release-${{ github.run_id }}-${{ github.run_attempt }}-cli" \ timeout --signal=TERM --kill-after=60 "${remaining}s" \ uv run --locked --extra dev python -m workflow_bench.runner \ --tasks "$RUNNER_TEMP/release-evaluation/candidate/tasks.yaml" \ @@ -456,6 +460,8 @@ jobs: set -euo pipefail remaining=$(( ${{ needs.start-runner.outputs.stop_deadline_epoch }} - $(date +%s) - 5400 )) (( remaining > 0 )) || { echo '::error::No time remains for deferred sessions before the evidence reserve.'; exit 1; } + GITNEXUS_BENCH_PROVIDER_USAGE="$RUNNER_TEMP/release-evaluation/candidate/deferred-provider-usage.jsonl" \ + GITNEXUS_BENCH_SWEEP_ID="release-${{ github.run_id }}-${{ github.run_attempt }}-deferred" \ timeout --signal=TERM --kill-after=60 "${remaining}s" \ uv run --locked --extra dev python -m workflow_bench.runner \ --tasks "$RUNNER_TEMP/release-evaluation/candidate/tasks.yaml" \ @@ -539,6 +545,49 @@ jobs: --paired "$candidate/raw/results.jsonl" --cli "$candidate/cli-raw/results.jsonl" \ --out "$RUNNER_TEMP/release-evaluation/public" cat "$RUNNER_TEMP/release-evaluation/public/cli-arm-comparison.md" >> "$GITHUB_STEP_SUMMARY" + - name: Summarize time, cost, and tool measurements + if: ${{ always() && steps.resolve.outputs.sha != '' }} + working-directory: eval + env: + DEFERRED_ARM: ${{ inputs.mcp_tool_loading }} + run: | + set -euo pipefail + root="$RUNNER_TEMP/release-evaluation" + public="$root/public" + mkdir -p "$public" + failed=0 + for variant in stable candidate cli deferred; do + case "$variant" in + stable) + if [[ -n "$INPUT_PR" ]]; then continue; fi + results="$root/stable/raw/results.jsonl" + ledger="$root/stable/provider-usage.jsonl" ;; + candidate) + results="$root/candidate/raw/results.jsonl" + ledger="$root/candidate/provider-usage.jsonl" ;; + cli) + if [[ "$CLI_ARM" != true ]]; then continue; fi + results="$root/candidate/cli-raw/results.jsonl" + ledger="$root/candidate/cli-provider-usage.jsonl" ;; + deferred) + if [[ "$DEFERRED_ARM" != true || -n "$INPUT_PR" ]]; then continue; fi + results="$root/candidate/deferred/results.jsonl" + ledger="$root/candidate/deferred-provider-usage.jsonl" ;; + esac + if ! uv run --locked --extra dev python -m workflow_bench.cost_diagnostics \ + --results "$results" --provider-usage "$ledger" --allow-incomplete \ + --out "$public/$variant-diagnostics.json" --markdown-out "$public/$variant-diagnostics.md"; then + printf '\nDetailed measurements for %s could not be validated; inspect the step log.\n' "$variant" >> "$GITHUB_STEP_SUMMARY" + failed=1 + continue + fi + { + printf '\n
Detailed measurements: %s\n\n' "$variant" + cat "$public/$variant-diagnostics.md" + printf '\n
\n' + } >> "$GITHUB_STEP_SUMMARY" + done + exit "$failed" - name: Upload sanitized release evidence if: ${{ always() && steps.resolve.outputs.sha != '' && inputs.candidate_pr == '' }} uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 @@ -553,6 +602,8 @@ jobs: ${{ runner.temp }}/release-evaluation/public/release-quality-gate.md ${{ runner.temp }}/release-evaluation/public/mcp-tool-loading.json ${{ runner.temp }}/release-evaluation/public/mcp-tool-loading.md + ${{ runner.temp }}/release-evaluation/public/*-diagnostics.json + ${{ runner.temp }}/release-evaluation/public/*-diagnostics.md if-no-files-found: error retention-days: 30 - name: Upload the opt-in CLI-only arm comparison @@ -563,6 +614,8 @@ jobs: path: | ${{ runner.temp }}/release-evaluation/public/cli-arm-comparison.json ${{ runner.temp }}/release-evaluation/public/cli-arm-comparison.md + ${{ runner.temp }}/release-evaluation/public/cli-diagnostics.json + ${{ runner.temp }}/release-evaluation/public/cli-diagnostics.md if-no-files-found: warn retention-days: 30 - name: Upload sanitized PR cost comparison @@ -573,6 +626,8 @@ jobs: path: | ${{ runner.temp }}/release-evaluation/public/pr-cost-comparison.json ${{ runner.temp }}/release-evaluation/public/pr-cost-comparison.md + ${{ runner.temp }}/release-evaluation/public/*-diagnostics.json + ${{ runner.temp }}/release-evaluation/public/*-diagnostics.md if-no-files-found: error retention-days: 30 diff --git a/eval/tests/test_cli_shim.py b/eval/tests/test_cli_shim.py index 8686d65b2..4b23b3a68 100644 --- a/eval/tests/test_cli_shim.py +++ b/eval/tests/test_cli_shim.py @@ -129,13 +129,17 @@ class _FakeMcpServer: self.listener = socket.socket(socket.AF_UNIX) self.listener.bind(str(path)) self.listener.listen(1) + self.listener.settimeout(0.1) + self.stopped = threading.Event() self.thread = threading.Thread(target=self._serve, daemon=True) self.thread.start() def _serve(self) -> None: - while True: + while not self.stopped.is_set(): try: connection, _ = self.listener.accept() + except TimeoutError: + continue except OSError: return with connection, connection.makefile("rwb") as stream: @@ -157,7 +161,20 @@ class _FakeMcpServer: stream.flush() def close(self) -> None: - self.listener.close() + self.stopped.set() + try: + # Wait for accept to exit before releasing its fd for another server. + self.thread.join(timeout=5) + finally: + self.listener.close() + assert not self.thread.is_alive(), "fake MCP server did not stop" + + +@pytest.mark.skipif(not hasattr(socket, "AF_UNIX"), reason="the shim speaks MCP over a Unix socket") +def test_fake_mcp_server_close_stops_idle_accept_thread(tmp_path): + server = _FakeMcpServer(tmp_path / "mcp.sock", []) + server.close() + assert not server.thread.is_alive() @pytest.mark.skipif(not hasattr(socket, "AF_UNIX"), reason="the shim speaks MCP over a Unix socket") diff --git a/eval/tests/test_cost_diagnostics.py b/eval/tests/test_cost_diagnostics.py new file mode 100644 index 000000000..38bbe582c --- /dev/null +++ b/eval/tests/test_cost_diagnostics.py @@ -0,0 +1,526 @@ +"""Offline measurements must not invent attribution, completeness or savings.""" + +import copy +import json +import os +import subprocess +import sys +from collections import Counter +from pathlib import Path + +import pytest +import yaml + +from tests.test_session_diagnostics import result, use +from workflow_bench.cost_diagnostics import build_cost_report, provider_summary +from workflow_bench.diagnostic_experiments import VARIANTS, prepare_plan +from workflow_bench.session_diagnostics import diagnose_session, public_diagnostics +from workflow_bench.workload_diagnostics import classify_bash + + +def trace(actions): + events = [] + for index, (tool, args, body) in enumerate(actions): + events += [use(tool, tool_id=str(index), args=args), result(body, tool_id=str(index))] + return events + + +def test_graph_reference_is_not_confused_with_supplied_source_and_paths_stay_private(): + events = trace([ + ("mcp__gitnexus__query", {}, {"definitions": [{"filePath": "src/private.ts"}]}), + ("mcp__gitnexus__context", {}, {"symbol": {"filePath": "src/private.ts", "content": "secret source"}}), + ("Read", {"file_path": "/workspace/src/private.ts", "offset": 10, "limit": 5}, "source"), + ("Read", {"file_path": "/workspace/src/private.ts", "offset": 10, "limit": 5}, "source"), + ("Read", {"file_path": "/workspace/other/private.ts"}, "unrelated"), + ]) + report = diagnose_session(events, None, 10) + work = report["workload"] + assert work["source_reads"] == 3 + assert work["repeated_source_reads"] == 1 + assert work["reads_after_graph_reference"] == work["reads_after_graph_content_reference"] == 2 + assert work["query_context_read_chains"] == 2 + assert "private.ts" not in json.dumps(public_diagnostics([report])) + assert "secret source" not in json.dumps(report) + + +def test_read_before_graph_result_is_not_credited_as_a_followup(): + events = [use(), use("Read", tool_id="read", args={"file_path": "src/a.ts"}), + result({"filePath": "src/a.ts"}), result("source", tool_id="read")] + work = diagnose_session(events, None, 10)["workload"] + assert work["reads_after_graph_reference"] == 0 + + +def test_server_reported_time_is_separate_from_host_wait_and_missing_stays_unknown(): + events = trace([("mcp__gitnexus__query", {}, {"timing": {"wall": 125}, "epistemic": "lower-bound"}), + ("mcp__gitnexus__impact", {}, {"risk": "LOW"})]) + report = diagnose_session(events, [0, 2, 3, 5], 6) + safe = public_diagnostics([report])[0] + assert safe["tools"]["mcp__gitnexus__query"]["server_wall_s"] == 0.125 + assert safe["tools"]["mcp__gitnexus__query"]["elapsed_s"] == 2 + assert safe["tools"]["mcp__gitnexus__query"]["lower_bound"] == 1 + assert safe["tools"]["mcp__gitnexus__impact"]["server_wall_s"] is None + + +@pytest.mark.parametrize("command", ["cd gitnexus && npm test", "uv run --locked --extra dev python -m pytest"]) +def test_verification_repetition_respects_edits_and_unknown_shell_mutations(command): + check = ("Bash", {"command": command}, "ok") + events = trace([check, check, ("Edit", {"file_path": "secret"}, "ok"), check, + ("Bash", {"command": "python custom_script.py"}, "ok"), check]) + work = diagnose_session(events, None, 10)["workload"] + assert work["verification_calls"] == 4 + assert work["verification_repeated_without_observed_change"] == 1 + assert work["verification_repeated_after_possible_change"] == 2 + assert work["edits_after_verification"] == 1 + + +@pytest.mark.parametrize(("command", "category"), [ + ("uv run python -m pytest -q", "test"), ("npm run test:unit", "test"), + ("uv run --locked --extra dev python -m pytest -q", "test"), + ("uv run --frozen --no-sync --offline --extra=dev pytest", "test"), + ("uv run --project ./eval --group dev --python 3.12 -- pytest", "test"), + ("uv run --locked --extra dev ruff check .", "lint"), + ("uv run --locked --extra dev ruff format .", "mutation"), + ("uv run --unknown pytest", "other"), ("uv run --extra", "other"), + ("uv run --extra= pytest", "other"), ("uv run --extra --locked pytest", "other"), + ("uv run --locked=true pytest", "other"), ("uv run --locked", "other"), + ("uv run -- --locked pytest", "other"), + ("npx tsc --noEmit", "typecheck"), ("npx eslint .", "lint"), + ("npx eslint . --fix", "mutation"), ("npm run build", "build"), + ("cat source.ts", "source"), ("rg 'x' src", "source"), + ("echo pytest", "other"), ("npm test && npm run build", "mixed"), + ("cd gitnexus && npm test && touch x", "mixed"), + ("cat x > y", "mixed"), ("sed -i 's/x/y/' a", "mutation"), + ("sed --in-place 's/x/y/' a", "mutation"), ("npx prettier -w .", "mutation"), + ("uv run ruff format .", "mutation"), ("uv run ruff format --check .", "lint"), + ("echo $(npm test)", "mixed"), ("unterminated '", "other"), +]) +def test_shell_classification_is_conservative(command, category): + assert classify_bash(command) == category + + +def test_usage_missing_fields_are_unknown_and_cache_counts_are_not_double_counted(): + first = {"type": "assistant", "message": {"id": "a", "usage": { + "input_tokens": 10, "cache_read_input_tokens": 100, "cache_creation_input_tokens": 20, "output_tokens": 4, + }}} + second = {"type": "assistant", "message": {"id": "b", "usage": {"input_tokens": 30}}} + report = diagnose_session([first, second, {"type": "system", "subtype": "compact_boundary"}], None, 5) + work = report["workload"] + assert work["usage_snapshots"] == 2 and work["complete_input_snapshots"] == 1 + assert work["max_input_tokens"] == work["first_input_tokens"] == 130 + assert work["sum_input_tokens"] is work["last_input_tokens"] is None + assert work["sum_cache_read_tokens"] is work["sum_output_tokens"] is None + assert work["compactions"] == 1 + work["prompt"] = "secret" + assert "secret" not in json.dumps(public_diagnostics([report])) + + +def provider_event(): + return {"status": "success", "response_id": "one", "sweep_id": "sweep", "provider": "litellm-normalized", + "native_usage": {"prompt_tokens": 100, "completion_tokens": 20, + "prompt_tokens_details": {"cached_tokens": 60, "cache_write_tokens": 10}}, + "request_metrics": {"tools_bytes": 300, "duration_s": 2}} + + +def test_provider_deduplicates_identical_hooks_and_refuses_conflicting_totals(): + event = provider_event() + report = provider_summary([event, copy.deepcopy(event)]) + assert report["observed_requests"] == 1 and report["duplicate_callbacks"] == 1 + assert report["token_totals"]["total_input_tokens"] == 100 + assert report["token_totals"]["ordinary_input_tokens"] == 30 + assert report["token_totals"]["reasoning_output_tokens"] is None + assert report["attribution"] == "ledger_only_no_cell_join" + conflicting = copy.deepcopy(event) + conflicting["native_usage"]["prompt_tokens"] = 120 + report = provider_summary([event, conflicting]) + assert report["conflicting_response_ids"] == 1 + assert all(value is None for value in report["token_totals"].values()) + + +def test_missing_identity_failed_usage_and_unknown_providers_remain_visible(): + event = provider_event() + del event["response_id"] + report = provider_summary([event, event, {"status": "failure", "provider": "unknown"}]) + assert report["requests_without_response_id"] == 3 and not report["identity_complete"] + assert report["failures"] == report["invalid_usage_records"] == 1 + assert report["token_totals"]["total_input_tokens"] is None + assert report["request_metrics"]["tools_bytes"]["missing"] == 1 + + +def test_cost_report_keeps_failures_in_spend_and_missing_telemetry_is_not_zero(): + rows = [{"arm": "baseline", "task": "test", "resolved": True, "ok": True, "cost_usd": 2}, + {"arm": "baseline", "task": "test", "resolved": False, "ok": False, "cost_usd": 3}] + report = build_cost_report(rows) + group = report["groups"][0] + assert group["cost_per_solved"] == 5 + assert group["failed_cells"] == 1 and group["diagnostic_cells"] == 0 + assert group["metrics"]["duration_s"]["mean"] is None + rows[1]["cost_usd"] = float("nan") + assert build_cost_report(rows)["groups"][0]["cost_per_solved"] is None + + +def test_different_model_or_prompt_cohorts_and_duplicate_exports_cannot_blend(): + row = {"arm": "baseline", "task": "test", "run": 0, "model": "model-a", "cost_usd": 2} + other = row | {"model": "model-b", "cost_usd": 20} + assert len(build_cost_report([row, other])["groups"]) == 2 + with pytest.raises(ValueError, match="duplicate cell"): + build_cost_report([row, row]) + + +def test_tool_summaries_preserve_cohort_boundaries_and_budget_treatment(): + diagnostic = diagnose_session(trace([("Read", {"file_path": "source.ts"}, "source")]), None, 5) + row = {"arm": "baseline", "task": "test", "run": 0, "max_budget_usd": None, + "session_diagnostics": [diagnostic]} + report = build_cost_report([row, row | {"max_budget_usd": 5}]) + assert len(report["groups"]) == 2 + assert report["tool_markdown"].count("Diagnostics available for 1/1 cells") == 2 + assert "Diagnostics available for 2/2 cells" not in report["tool_markdown"] + + +def test_proxy_envelope_metrics_preserve_accounting_without_recording_content(tmp_path, monkeypatch): + from types import SimpleNamespace + + from workflow_bench.litellm_usage_callback import USAGE_LOG_ENV_VAR, ProviderUsageLogger + + log = tmp_path / "provider.jsonl" + monkeypatch.setenv(USAGE_LOG_ENV_VAR, str(log)) + kwargs = {"messages": [{"role": "user", "content": "SECRET PROMPT"}], + "optional_params": {"tools": [{"name": "PRIVATE TOOL"}]}, "response_cost": 0.25} + ProviderUsageLogger()._append("success", kwargs, SimpleNamespace(usage={"prompt_tokens": 20}), 10, 12) + payload = log.read_text() + row = json.loads(payload) + assert "SECRET" not in payload and "PRIVATE" not in payload + assert row["request_metrics"]["tools_count"] == 1 + assert row["request_metrics"]["messages_count"] == 1 + assert row["request_metrics"]["duration_s"] == 2 + assert row["request_metrics"]["proxy_estimated_cost_usd"] == 0.25 + ProviderUsageLogger()._append("failure", {"optional_params": [], "response_cost": float("nan")}, + SimpleNamespace(usage={"prompt_tokens": 20}), None, None) + row = json.loads(log.read_text().splitlines()[-1]) + assert row["native_usage"]["prompt_tokens"] == 20 + assert row["request_metrics"]["tools_bytes"] is row["request_metrics"]["duration_s"] is None + + +@pytest.mark.parametrize("fail", [False, True]) +def test_cell_phases_include_cleanup_and_partial_failure_without_changing_credit(monkeypatch, tmp_path, fail): + from tests.test_runner_hardening import _cell_context, _stub_cell_dependencies + from workflow_bench import runner + from workflow_bench.session_diagnostics import CELL_PHASES + + _stub_cell_dependencies(monkeypatch, tmp_path) + if fail: + def explode(*args, **kwargs): + raise RuntimeError("synthetic failure") + monkeypatch.setattr(runner, "run_arm", explode) + row = runner.run_cell(_cell_context(tmp_path), 2, "workflow") + phases = row["cell_phase_seconds"] + assert set(phases) <= set(CELL_PHASES) + assert "cleanup" in phases and "agent_and_verification" in phases + assert ("artifacts" not in phases) == fail + assert sum(phases.values()) == pytest.approx(row["cell_duration_s"]) + assert row["cell_finished_at"] >= row["cell_started_at"] + assert row["resolved"] is (not fail) + + +def test_release_and_cli_artifacts_publish_safe_diagnostics_and_keep_phase_validation(tmp_path): + from tests.test_cli_shim import _report_row, _write_results + from tests.test_release_report import DIGEST, NOW, SHA, metadata, rows + from workflow_bench import cli_arm_report, release_report + + diagnostic = diagnose_session(trace([("Bash", {"command": "gitnexus query secret"}, "ok")]), None, 10) + extra = {"session_diagnostics": [diagnostic], "cell_duration_s": 12, + "cell_phase_seconds": {"clone": 1.0, "secret content": 99, "mcp_startup": float("nan")}} + measurements = rows() + measurements[1].update(extra) + report = release_report.build_report(measurements, metadata(), now=NOW) + release_report.validate_report(report, runtime_sha=SHA, task_set_digest=DIGEST, now=NOW) + assert report["per_run"][1]["cell_phase_seconds"] == {"clone": 1.0, "mcp_startup": None} + assert "secret" not in json.dumps(report) + paired, cli = _write_results(tmp_path, [_report_row("baseline", 0, 1), _report_row("baseline_nomcp", 0, 1)], + [_report_row("baseline_cli", 0, 1, **extra)]) + report = cli_arm_report.build_report(paired, cli) + assert "secret" not in json.dumps(report) + assert "Tool diagnostics: baseline_cli" in cli_arm_report.render_markdown(report) + + +def plan_args(): + return dict(repository="/repo", runtime="/runtime", runtime_sha="a" * 40, harness_sha="b" * 40, + model="gpt-5.6-sol", effort="xhigh", repeats=5, seed=3762, variants=list(VARIANTS)) + + +def task_document(): + return {"tasks": [ + {"id": name, "ref": "c" * 40, "prompt": "Fix this", "class": task_class, "verify": "true", + "oracle": {"command": "true", "files": [{"source": "fixture.txt", "target": "fixture.txt"}]}} + for name, task_class in (("one", "trivial"), ("two", "cross-module")) + ]} + + +def test_plan_is_deterministic_balanced_isolated_and_never_runs_a_model(): + original = task_document() + plan = prepare_plan(original, **plan_args()) + assert plan == prepare_plan(original, **plan_args()) + assert len(plan["cells"]) == 50 and plan["position_balanced"] + assert plan["execution"] == "not_started" and plan["release_gate_eligible"] is False + assert original == task_document() + for task in ("one", "two"): + for variant in VARIANTS: + positions = Counter(c["position"] for c in plan["cells"] if c["task"] == task and c["variant"] == variant) + assert set(positions.values()) == {1} and len(positions) == 5 + ledgers = {c["env"]["GITNEXUS_BENCH_PROVIDER_USAGE"] for c in plan["cells"]} + assert len(ledgers) == 50 + for cell in plan["cells"]: + assert (cell["prompt_sha256"] != cell["base_prompt_sha256"]) == (cell["variant"] == "mcp_optional_prompt") + assert "--unsafe-no-bwrap" not in cell["argv"] + + +@pytest.mark.parametrize("override", [{"runtime_sha": "main"}, {"model": "latest"}, {"repeats": 0}, + {"variants": ["cli", "cli"]}, {"task_ids": ["missing"]}]) +def test_plan_rejects_unpinned_or_ambiguous_inputs(override): + with pytest.raises(ValueError): + prepare_plan(task_document(), **(plan_args() | override)) + + +@pytest.mark.parametrize("document", [None, [], "tasks", 42]) +def test_plan_rejects_non_mapping_yaml_documents(document): + with pytest.raises(ValueError, match="task document must contain tasks"): + prepare_plan(document, **plan_args()) + + +@pytest.mark.parametrize("model", [None, 1, "", " ", " auto ", "claude-sonnet-LATEST", + "provider/latest", "provider:auto", "provider@AUTO"]) +def test_plan_uses_runner_model_validation(model): + with pytest.raises(ValueError): + prepare_plan(task_document(), **(plan_args() | {"model": model})) + + +def test_plan_normalizes_model_in_metadata_and_commands(): + plan = prepare_plan(task_document(), **(plan_args() | {"model": " gpt-5.6-sol "})) + assert plan["model"] == "gpt-5.6-sol" + for cell in plan["cells"]: + assert cell["argv"][cell["argv"].index("--model") + 1] == "gpt-5.6-sol" + + +@pytest.mark.parametrize("task_id", ["_task", "-task", "a" * 129]) +def test_plan_rejects_task_ids_the_runner_cannot_use(task_id): + document = task_document() + document["tasks"][0]["id"] = task_id + with pytest.raises(ValueError, match="artifact-safe slug"): + prepare_plan(document, **plan_args()) + + +def test_every_prepared_task_satisfies_the_runner_metadata_contract(): + from workflow_bench.runner_tasks import select_tasks + + document = task_document() + document["tasks"][0]["id"] = "a.task-1" + plan = prepare_plan(document, **plan_args()) + for cell in plan["cells"]: + selected, skipped = select_tasks(cell["task_document"]["tasks"], include_expensive=True) + assert selected == cell["task_document"]["tasks"] and skipped == [] + document["tasks"][0]["verify"] = "" + with pytest.raises(ValueError, match="verify"): + prepare_plan(document, **plan_args()) + + +@pytest.mark.parametrize("intervening_read", [False, True]) +def test_post_graph_input_totals_require_every_relevant_snapshot(intervening_read): + complete = {"type": "assistant", "message": {"id": "complete", "usage": { + "input_tokens": 10, "cache_read_input_tokens": 100, "cache_creation_input_tokens": 20, + }}} + incomplete = {"type": "assistant", "message": {"id": "incomplete", "usage": {"input_tokens": 30}}} + events = [use(), result({}), complete] + if intervening_read: + events += [use("Read", tool_id="read", args={"file_path": "src/a.ts"}), result("source", tool_id="read")] + events.append(incomplete) + work = diagnose_session(events, None, 5)["workload"] + assert work["sum_input_tokens"] is None + assert work["usage_after_graph_snapshots"] == (1 if intervening_read else 2) + assert work["input_tokens_after_graph"] == (130 if intervening_read else None) + # Preserve the actual first/last positions, even when a snapshot is incomplete. + reversed_usage = diagnose_session([incomplete, complete], None, 5)["workload"] + assert reversed_usage["first_input_tokens"] is None and reversed_usage["last_input_tokens"] == 130 + complete_usage = diagnose_session([complete], None, 5)["workload"] + assert complete_usage["sum_input_tokens"] == 130 + + +def test_trial_plan_passes_each_cell_budget_without_starting_a_model(): + args = plan_args() | {"repeats": 1, "variants": ["no_graph", "mcp_eager"], "task_ids": ["one"], "max_budget_usd": 5.0} + plan = prepare_plan(task_document(), **args) + assert len(plan["cells"]) == 2 + assert plan["planned_cli_budget_usd"] == 10.0 + assert plan["execution"] == "not_started" + for cell in plan["cells"]: + assert cell["argv"][cell["argv"].index("--max-budget-usd") + 1] == "5.0" + + +@pytest.mark.parametrize("budget", [0, -1, float("nan"), float("inf"), True, 10 ** 1000]) +def test_invalid_budget_fails_before_any_process_starts(monkeypatch, tmp_path, budget): + from workflow_bench import runner_sessions + + def must_not_start(*args, **kwargs): + pytest.fail("invalid budget launched a process") + + monkeypatch.setattr(runner_sessions, "run_managed", must_not_start) + with pytest.raises(ValueError, match="max-budget-usd"): + runner_sessions.run_claude("task", tmp_path, claude_bin="claude", timeout=1, max_budget_usd=budget) + + +def test_cli_budget_stop_remains_a_failed_billed_session(monkeypatch, tmp_path): + from tests.test_workflow_bench_sessions import fake_cli_result, report_variant + from workflow_bench import runner_sessions + + captured = [] + + def run(command, **kwargs): + captured.extend(command) + return fake_cli_result(report_variant(subtype="error_max_budget_usd", total_cost_usd=5.1)) + + monkeypatch.setattr(runner_sessions, "run_managed", run) + record = runner_sessions.run_claude("task", tmp_path, claude_bin="claude", timeout=1, max_budget_usd=5) + assert captured[captured.index("--max-budget-usd") + 1] == "5.0" + assert record["ok"] is False and record["cost_usd"] == 5.1 + assert record["max_budget_usd"] == 5 + + +def test_budget_rejects_multisession_arms_before_reading_task_inputs(monkeypatch, capsys): + from workflow_bench import runner + + monkeypatch.setattr("sys.argv", ["runner", "--tasks", "nonexistent.yaml", "--model", "gpt-5.6-sol", + "--arms", "workflow", "--max-budget-usd", "5"]) + with pytest.raises(SystemExit) as failure: + runner.main() + assert failure.value.code == 2 + assert "--max-budget-usd is limited to single-session baseline arms" in capsys.readouterr().err + + +def test_summary_cli_gathers_safe_measurements_and_provider_coverage(tmp_path, monkeypatch): + from workflow_bench import cost_diagnostics + + diagnostic = diagnose_session(trace([ + ("mcp__gitnexus__query", {"query": "SECRET QUERY"}, {"staleness": {"status": "behind"}}), + ("Bash", {"command": "uv run --locked --extra dev pytest"}, "SECRET RESULT"), + ]), [0, 2, 3, 5], 8) + rows = [{"task": "one", "arm": "baseline", "run": 0, "cost_usd": 2, "duration_s": 8, + "ok": False, "resolved": False, "session_diagnostics": [diagnostic], "private": "SECRET PATH", + "cell_phase_seconds": {"clone": 1, "agent_and_verification": 9, "PRIVATE PHASE": 123}}] + results, ledger = tmp_path / "results.jsonl", tmp_path / "provider.jsonl" + results.write_text(json.dumps(rows[0]) + "\n") + event = provider_event() | {"messages": "SECRET PROMPT", "response_id": "SECRET ID"} + ledger.write_text(json.dumps(event) + "\n") + out, markdown = tmp_path / "diagnostics.json", tmp_path / "diagnostics.md" + monkeypatch.setattr("sys.argv", ["cost_diagnostics", "--results", str(results), "--provider-usage", str(ledger), + "--out", str(out), "--markdown-out", str(markdown)]) + cost_diagnostics.main() + body = markdown.read_text() + assert "SECRET" not in out.read_text() + body and "PRIVATE PHASE" not in body + assert "Solved: 0/1; failed sessions: 1" in body + assert "| clone | 1 / 0 | 1 |" in body + assert "| verification_calls | 1 | 1/1 |" in body + assert "| tools_bytes | 1 / 0 | 300 |" in body + assert "| total_input_tokens | 100 |" in body + assert "Largest, slowest, and problematic tool calls" in body + assert "requests without identity: 0" in body + assert json.loads(out.read_text())["provider"]["ledger_complete"] is True + + +def test_summary_cli_reports_missing_legacy_telemetry_and_torn_provider_rows(tmp_path, monkeypatch): + from workflow_bench import cost_diagnostics + + results, ledger = tmp_path / "results.jsonl", tmp_path / "provider.jsonl" + results.write_text(json.dumps({"task": "old", "arm": "baseline", "cost_usd": 1}) + "\n{torn") + ledger.write_text(json.dumps(provider_event()) + "\n{torn") + out, markdown = tmp_path / "diagnostics.json", tmp_path / "diagnostics.md" + monkeypatch.setattr("sys.argv", ["cost_diagnostics", "--results", str(results), "--provider-usage", str(ledger), + "--allow-incomplete", "--out", str(out), "--markdown-out", str(markdown)]) + cost_diagnostics.main() + report = json.loads(out.read_text()) + assert len(report["input_problems"]) == 2 + assert report["provider"]["observed_requests"] == 1 + assert report["provider"]["ledger_complete"] is False + assert all(value is None for value in report["provider"]["token_totals"].values()) + body = markdown.read_text() + assert "Partial input" in body and "diagnostics: 0/1" in body + assert "| clone | 0 / 1 | n/a |" in body + assert "| total_input_tokens | n/a |" in body + + +def test_summary_cli_reports_unavailable_inputs_after_early_failure(tmp_path, monkeypatch): + from workflow_bench import cost_diagnostics + + out, markdown = tmp_path / "diagnostics.json", tmp_path / "diagnostics.md" + monkeypatch.setattr("sys.argv", ["cost_diagnostics", "--results", str(tmp_path / "missing"), + "--provider-usage", str(tmp_path / "missing-ledger"), "--allow-incomplete", + "--out", str(out), "--markdown-out", str(markdown)]) + cost_diagnostics.main() + assert "Observed cells: 0" in markdown.read_text() + assert "Provider ledger unavailable" in markdown.read_text() + assert json.loads(out.read_text())["provider"] is None + + +def test_summary_escapes_markdown_and_html_labels(): + from workflow_bench.cost_diagnostics import render_markdown + + report = build_cost_report([{"arm": "baseline", "task": "|bad![x](https://example.test)"}]) + markdown = render_markdown(report) + assert " 0 and metrics["tools_bytes"] > 0 + assert metrics["duration_s"] >= 0 + assert "PRIVATE" not in usage_log.read_text(), "numeric envelope evidence must not retain request content" usage = normalize_usage(event["provider"], native) assert usage.total_input_tokens == 10_000 diff --git a/eval/tests/test_pr_evaluation.py b/eval/tests/test_pr_evaluation.py index c480b934a..2fbe74400 100644 --- a/eval/tests/test_pr_evaluation.py +++ b/eval/tests/test_pr_evaluation.py @@ -73,6 +73,23 @@ def test_pr_comparison_rejects_readable_runtime_cells(evidence): assert not report["complete"] and not report["improved"] +def test_pr_summary_preserves_safe_diagnostics_in_complete_and_partial_reports(evidence): + from tests.test_session_diagnostics import result, use + from workflow_bench.session_diagnostics import diagnose_session + + diagnostic = diagnose_session([use(args={"query": "SECRET"}), result("SECRET")], [1, 2], 3) + evidence[1][0]["session_diagnostics"] = [diagnostic] + evidence[1][0]["cell_phase_seconds"] = {"clone": 2, "SECRET": 9} + report = compare(evidence) + assert report["per_run"][0]["cell_phase_seconds"] == {"clone": 2} + assert "SECRET" not in json.dumps(report) + assert "Tool diagnostics: baseline" in pr_evaluation.render_markdown(report) + evidence[1].pop() + partial = compare(evidence) + assert not partial["complete"] + assert "Tool diagnostics: baseline" in pr_evaluation.render_markdown(partial) + + def test_comparison_measures_only_new_mcp_cells_and_preserves_historical_cost(evidence): before = deepcopy(evidence) report = compare(evidence) diff --git a/eval/tests/test_retrieval_diagnostics.py b/eval/tests/test_retrieval_diagnostics.py new file mode 100644 index 000000000..c80139161 --- /dev/null +++ b/eval/tests/test_retrieval_diagnostics.py @@ -0,0 +1,65 @@ +"""Partial retrieval judgments must not turn unjudged code into false positives.""" + +import copy +import json +from pathlib import Path + +import pytest + +from workflow_bench.retrieval_diagnostics import score_retrieval + +CASES = {"source_sha": "a" * 40, "query_options": {"limit": 5, "max_symbols": 10, "include_content": False}, "cases": [ + {"id": "case", "query": "find handler", "anchors": [{"file": "src/a.ts", "name": "handle"}]}, + {"id": "missing", "query": "another query", "anchors": [{"file": "src/b.ts"}]}, +]} + + +def recordings(): + return {"source_sha": "a" * 40, "runtime_sha": "b" * 40, "index_complete": True, + "index_staleness": "current", "sanitized_graph": True, + "query_options": CASES["query_options"], + "cases": [{"id": "case", "query": "find handler", "response": { + "process_symbols": [{"filePath": "src/unjudged.ts", "name": "other"}], + "definitions": [{"filePath": "src/a.ts", "name": "handle"}], + }}]} + + +def test_partial_relevance_and_missing_cases_do_not_become_precision_or_zero(): + report = score_retrieval(CASES, recordings()) + assert report["cases"][0]["anchor_recall"] == 1 + assert report["cases"][0]["first_anchor_position"] == 2 + assert report["cases"][1]["anchor_recall"] is None + assert not report["complete"] + assert "precision" not in report["cases"][0] + + +@pytest.mark.parametrize("change", [{"source_sha": "c" * 40}, {"index_staleness": "behind"}, + {"index_complete": False}, {"sanitized_graph": False}, + {"sanitized_graph": "false"}, {"runtime_sha": "main"}, {"query_options": {"limit": 1}}]) +def test_unmatched_or_incomplete_index_is_not_scored(change): + with pytest.raises(ValueError): + score_retrieval(CASES, recordings() | change) + + +def test_changed_query_duplicate_case_and_missing_symbol_are_distinct(): + recorded = recordings() + recorded["cases"][0]["query"] = "different question" + with pytest.raises(ValueError, match="query differs"): + score_retrieval(CASES, recorded) + recorded = recordings() + recorded["cases"].append(copy.deepcopy(recorded["cases"][0])) + with pytest.raises(ValueError, match="duplicate"): + score_retrieval(CASES, recorded) + recorded = recordings() + recorded["cases"][0]["response"] = {"process_symbols": [], "definitions": []} + assert score_retrieval(CASES, recorded)["cases"][0]["anchor_recall"] == 0 + + +def test_committed_cases_cover_each_current_scenario_with_pinned_source(): + import yaml + + root = Path(__file__).resolve().parents[1] / "workflow_bench" + cases = json.loads((root / "retrieval_cases.json").read_text()) + tasks = yaml.safe_load((root / "tasks.scenarios.yaml").read_text())["tasks"] + assert {c["id"] for c in cases["cases"]} == {t["id"] for t in tasks} + assert {t["ref"] for t in tasks} == {cases["source_sha"]} diff --git a/eval/tests/test_session_diagnostics.py b/eval/tests/test_session_diagnostics.py new file mode 100644 index 000000000..89b71883e --- /dev/null +++ b/eval/tests/test_session_diagnostics.py @@ -0,0 +1,302 @@ +"""Diagnostics must explain overhead without changing credit or exposing content.""" + +import json +from collections import Counter + +import pytest + +from tests.test_release_report import DIGEST, NOW, SHA, metadata, rows +from tests.test_workflow_bench_sessions import VALID_REPORT, fake_cli_result, session_record +from workflow_bench import runner_sessions, session_diagnostics as diagnostics +from workflow_bench.release_report import build_report, render_markdown, validate_report + + +def use(tool="mcp__gitnexus__query", *, tool_id="one", args=None): + return {"type": "assistant", "message": {"content": [ + {"type": "tool_use", "id": tool_id, "name": tool, "input": args or {"query": "private source"}}, + ]}} + + +def result(body, *, tool_id="one", error=False): + content = body if isinstance(body, str) else [{"type": "text", "text": json.dumps(body) + "\n\n---\nNext"}] + return {"type": "user", "message": {"content": [ + {"type": "tool_result", "tool_use_id": tool_id, "is_error": error, "content": content}, + ]}} + + +@pytest.mark.parametrize("message", [None, {}, {"id": "message"}, {"content": None}, [], "invalid"]) +def test_top_level_content_preserves_tool_results_and_source_followups(message): + events = [use(), result({"definitions": []}), use("Read", tool_id="read"), result("source", tool_id="read")] + for event in events: + event["content"] = event["message"]["content"] + event["message"] = message + report = diagnostics.diagnose_session(events, [1, 2, 3, 4], 5) + assert report["timing_complete"] is True + assert report["tool_wait_s"] == 2 + query = report["tools"]["mcp__gitnexus__query"] + assert query["calls"] == query["empty"] == query["source_followups"] == 1 + assert query["unfinished"] == 0 + assert report["tools"]["Read"]["calls"] == 1 + + +@pytest.mark.parametrize("nested", [[], "invalid"]) +def test_present_non_null_nested_content_prevents_top_level_fallback(nested): + events = [use(), result("done")] + for event in events: + event["content"] = event["message"]["content"] + event["message"]["content"] = nested + report = diagnostics.diagnose_session(events, [1, 2], 3) + assert report["calls"] == [] + assert report["tools"] == {} + + +def test_nested_tool_content_wins_over_conflicting_top_level_blocks(): + events = [use(), result("done")] + events[0]["content"] = use("Read", tool_id="shadow")["message"]["content"] + events[1]["content"] = result("shadow", tool_id="shadow")["message"]["content"] + report = diagnostics.diagnose_session(events, [1, 2], 3) + assert list(report["tools"]) == ["mcp__gitnexus__query"] + assert report["tools"]["mcp__gitnexus__query"]["calls"] == 1 + assert report["timing_complete"] is True + assert report["tool_wait_s"] == 1 + + +@pytest.mark.parametrize(("wall", "expected"), [ + (0, 0), (500, 0.5), (123.5, 0.1235), + (None, None), (True, None), ("1", None), (-1, None), + (float("nan"), None), (float("inf"), None), (10**400, None), +]) +def test_server_timing_values_do_not_fail_or_reclassify_completed_tool_calls(wall, expected): + events = [use(), result({"timing": {"wall": wall}})] + report = diagnostics.diagnose_session(events, [1, 2], 3) + tool = report["tools"]["mcp__gitnexus__query"] + assert tool["calls"] == 1 + assert tool["errors"] == tool["unfinished"] == 0 + assert tool["server_wall_s"] == expected + assert tool["server_timed_calls"] == int(expected is not None) + assert report["timing_complete"] is True + assert report["tool_wait_s"] == 1 + + +def test_event_clock_handles_chunk_boundaries_blank_lines_and_unterminated_event(monkeypatch): + ticks = iter([10, 11, 13, 15]) + monkeypatch.setattr(diagnostics.time, "monotonic", lambda: next(ticks)) + clock = diagnostics.EventClock() + clock.observe(b'\n {"message":"') + clock.observe('é"}\r\n\n'.encode()) + clock.observe(b'{"last":true}') + assert clock.finish(2) == [3, 5] + assert clock.finish(3) is None + + +def test_event_clock_bounds_memory_and_never_claims_complete_overflow(monkeypatch): + monkeypatch.setattr(diagnostics, "MAX_EVENTS", 2) + clock = diagnostics.EventClock() + clock.observe(b'{}\n{}\n{}\n') + assert len(clock.offsets) == 2 + assert clock.finish(3) is None + + +def test_overlapping_tools_are_not_double_counted_as_wall_time(): + events = [use(), use("Bash", tool_id="two"), result({"status": "found"}), result("done", tool_id="two")] + report = diagnostics.diagnose_session(events, [1, 2, 5, 7], 10) + assert report["timing_complete"] is True + assert report["tool_wait_s"] == 6 + assert report["mcp_wait_s"] == 4 + assert report["other_wall_s"] == 4 + assert [call["duration_s"] for call in report["calls"]] == [4, 5] + assert report["tools"]["Bash"]["elapsed_s"] == 5 + + +@pytest.mark.parametrize("offsets", [None, [], [1], [3, 2], [-1, 2], [1, 20], [True, 2], [1, float("nan")]]) +def test_unavailable_or_invalid_timing_is_unknown_not_zero(offsets): + report = diagnostics.diagnose_session([use(), result("done")], offsets, 10) + assert report["timing_complete"] is False + assert report["tool_wait_s"] is report["mcp_wait_s"] is report["other_wall_s"] is None + assert report["calls"][0]["duration_s"] is None + + +@pytest.mark.parametrize(("body", "signal"), [ + ({"error": "Symbol 'privateName' not found"}, "not_found"), + ({"status": "ambiguous", "candidates": [{"name": "secret"}]}, "ambiguous"), + ({"risk": "UNKNOWN"}, "unknown_risk"), + ({"processes": [], "process_symbols": [], "definitions": []}, "empty"), + ("Error: result (108,197 characters) exceeds maximum allowed tokens", "output_limit_exceeded"), +]) +def test_resolution_problems_and_followup_source_reads_are_counted(body, signal): + events = [use(), result(body), use("Read", tool_id="read"), result("source", tool_id="read")] + report = diagnostics.diagnose_session(events, [1, 2, 3, 4], 5) + tally = report["tools"]["mcp__gitnexus__query"] + assert tally[signal] == 1 + assert tally["source_followups"] == 1 + assert report["calls"][0]["next_tools"] == ["Read"] + assert "privateName" not in json.dumps(report) + + +def test_transport_errors_are_distinct_from_successful_resolution(): + bad = diagnostics.diagnose_session([use(), result("failure", error=True)], [1, 2], 3) + good = diagnostics.diagnose_session([use(), result({"status": "found"})], [1, 2], 3) + assert bad["tools"]["mcp__gitnexus__query"]["errors"] == 1 + assert good["tools"]["mcp__gitnexus__query"]["errors"] == 0 + + +def test_plain_text_mcp_errors_are_not_counted_as_success(): + report = diagnostics.diagnose_session([use(), result("Error: database unavailable")], [1, 2], 3) + assert report["tools"]["mcp__gitnexus__query"]["errors"] == 1 + + +def test_non_ascii_payload_byte_counts_handle_escaped_lone_surrogates(): + report = diagnostics.diagnose_session([use(args={"query": "é\ud800"}), result("é\ud800")], [1, 2], 3) + assert report["calls"][0]["result_bytes"] == len('"é\\ud800"'.encode()) + assert report["timing_complete"] is True + + +def test_identical_arguments_are_counted_without_publishing_arguments_or_hashes(): + events = [use(args={"token": "super-secret"}), result("done"), + use(tool_id="two", args={"token": "super-secret"}), result("done", tool_id="two")] + report = diagnostics.diagnose_session(events, [1, 2, 3, 4], 5) + assert report["calls"][1]["repeat_of"] == 0 + assert report["tools"]["mcp__gitnexus__query"]["repeated"] == 1 + assert "super-secret" not in json.dumps(report) + assert "signature" not in json.dumps(report) + + +def test_unmatched_results_reused_ids_and_call_limit_make_timing_incomplete(monkeypatch): + for events in ([use()], [result("orphan")], [use(), result("ok"), use(), result("ok")]): + assert diagnostics.diagnose_session(events, list(range(len(events))), 10)["timing_complete"] is False + monkeypatch.setattr(diagnostics, "MAX_CALLS", 1) + report = diagnostics.diagnose_session([use(), use(tool_id="two")], [1, 2], 5) + assert report["truncated"] is True and len(report["calls"]) == 1 + assert report["tool_wait_s"] is None + + +@pytest.mark.parametrize(("command", "tool"), [ + ("gitnexus impact parseFile --direction upstream", "cli__gitnexus__impact"), + ("/usr/local/bin/gitnexus query 'some concept'", "cli__gitnexus__query"), + ("gitnexus query 'x;y'", "cli__gitnexus__query"), + ('gitnexus query "x;y"', "cli__gitnexus__query"), + (r"gitnexus query x\;y", "cli__gitnexus__query"), + ("gitnexus query '$(literal) | & > < `literal`'", "cli__gitnexus__query"), + ('gitnexus query "$(cat secret)"', "Bash"), + ('gitnexus query "`cat secret`"', "Bash"), + ("gitnexus query x > result.txt", "Bash"), + ("gitnexus query x < input.txt", "Bash"), + ("gitnexus query x; touch changed", "Bash"), + ("gitnexus query x | cat", "Bash"), + ("gitnexus query x\nnpm test", "Bash"), + ("gitnexus impact x && npm test", "Bash"), + ("echo 'gitnexus query x'", "Bash"), + ("gitnexus query $(cat secret)", "Bash"), + ("gitnexus query 'unterminated", "Bash"), +]) +def test_cli_attribution_only_covers_standalone_shim_commands(command, tool): + report = diagnostics.diagnose_session([use("Bash", args={"command": command}), result("ok")], [1, 2], 3) + assert report["calls"][0]["tool"] == tool + assert command not in json.dumps(report) + + +@pytest.mark.parametrize("mask", range(1, 8)) +@pytest.mark.parametrize("has_hit", [False, True]) +def test_empty_query_signals_cover_every_selected_response_section(mask, has_hit): + sections = [name for bit, name in enumerate(("processes", "process_symbols", "definitions")) if mask & (1 << bit)] + body = dict.fromkeys(sections, []) + if has_hit: + body[sections[0]] = [{"name": "found"}] + events = [use(), result(body), use("Read", tool_id="read", args={"file_path": "src/a.ts"}), + result("source", tool_id="read")] + report = diagnostics.diagnose_session(events, None, 5) + tool = report["tools"]["mcp__gitnexus__query"] + assert tool["empty"] == int(not has_hit) + assert tool["source_followups"] == int(not has_hit) + + +@pytest.mark.parametrize("body", [{}, {"definitions": None}, {"timing": {"wall": 1}}]) +def test_absent_or_invalid_query_sections_are_not_evidence_of_an_empty_result(body): + report = diagnostics.diagnose_session([use(), result(body)], None, 5) + assert report["tools"]["mcp__gitnexus__query"]["empty"] == 0 + + +def test_usage_snapshots_are_deduplicated_and_no_cost_is_fabricated(): + event = use() + event["message"].update(id="message", usage={"input_tokens": 10, "output_tokens": 2, "secret": "hidden"}) + later = {"type": "assistant", "message": {"id": "message", "content": [], "usage": {"output_tokens": 5}}} + report = diagnostics.diagnose_session([event, result("ok"), later], [1, 2, 3], 4) + assert report["assistant_usage"] == [{"event_index": 2, "output_tokens": 5}] + assert "cost_usd" not in json.dumps(report) + assert "hidden" not in json.dumps(report) + + +def test_public_diagnostics_is_idempotent_and_cannot_publish_arbitrary_content(): + raw = diagnostics.diagnose_session([use(), result({"status": "found", "content": "secret source"})], [1, 2], 3) + raw.update(prompt="secret prompt", api_key="secret key") + raw["tools"]["private-tool-name"] = {"calls": 5} + raw["calls"][0]["password"] = "secret password" + safe = diagnostics.public_diagnostics([raw]) + assert diagnostics.public_diagnostics(safe) == safe + assert safe[0]["largest_results"][0]["index"] == 0 + assert safe[0]["slowest_calls"][0]["duration_s"] == 1 + assert "secret" not in json.dumps(safe) + assert "private-tool-name" not in json.dumps(safe) + assert "calls" not in safe[0] + + +def test_missing_status_and_stale_or_truncated_results_are_visible_in_public_diagnostics(): + events = [use(), result({"status": "not_found", "risk": "UNKNOWN", "truncated": True, + "staleness": {"status": "behind"}}), use("Read", tool_id="read")] + report = diagnostics.diagnose_session(events, None, 3) + tally = report["tools"]["mcp__gitnexus__query"] + assert tally["stale"] == tally["truncated_results"] == tally["source_followups"] == 1 + safe = diagnostics.public_diagnostics([report]) + assert safe[0]["problem_calls"][0]["not_found"] is True + assert diagnostics.public_diagnostics(safe) == safe + + +def test_release_report_exports_diagnostics_without_changing_gate_or_accepting_injected_fields(): + measurements = rows() + raw = diagnostics.diagnose_session([use(), result({"error": "private symbol not found"})], [1, 2], 3) + measurements[1]["session_diagnostics"] = [raw] + report = build_report(measurements, metadata(), now=NOW) + assert report["complete"] is True + assert report["paired"]["mean_cost_change_pct"] == 25 + validate_report(report, runtime_sha=SHA, task_set_digest=DIGEST, now=NOW) + text = render_markdown(report) + assert "Diagnostics available for 1/3 cells" in text + assert "not proof of a workaround" in text + assert "private symbol" not in json.dumps(report) + report["per_run"][1]["session_diagnostics"][0]["secret"] = "forged" + with pytest.raises(ValueError, match="outside the published report schema"): + validate_report(report, runtime_sha=SHA, task_set_digest=DIGEST, now=NOW) + + +def test_runner_collects_timings_without_progress_or_changing_session_cost(monkeypatch, tmp_path): + events = [use(), result("ok"), json.loads(VALID_REPORT)] + raw = b"".join((json.dumps(event) + "\n").encode() for event in events) + + def run(*args, **kwargs): + for line in raw.splitlines(keepends=True): + kwargs["stdout_observer"](line) + return fake_cli_result(raw.decode()) + + monkeypatch.setattr(runner_sessions, "run_managed", run) + record = runner_sessions.run_claude("task", tmp_path, claude_bin="claude", timeout=5) + assert record["ok"] is True and record["cost_usd"] == 0.1 + assert record["session_diagnostics"][0]["timing_complete"] is True + assert len(record["session_diagnostics"][0]["calls"]) == 1 + combined = runner_sessions.sum_sessions([record, record]) + assert len(combined["session_diagnostics"]) == 2 + assert combined["cost_usd"] == 0.2 + assert runner_sessions.sum_sessions([session_record()])["session_diagnostics"] == [] + + +def test_diagnostics_tool_whitelist_matches_the_harness_surface(): + assert {f"mcp__gitnexus__{name}" for name in diagnostics.COMMANDS} == set(runner_sessions.GITNEXUS_READ_ONLY_TOOLS) + + +def test_observational_followups_can_count_one_action_for_multiple_problem_results(): + first = use() + second_block = use(tool_id="two")["message"]["content"][0] + first["message"]["content"].append(second_block) + events = [first, result({"risk": "UNKNOWN"}), result({"risk": "UNKNOWN"}, tool_id="two"), use("Read", tool_id="read")] + report = diagnostics.diagnose_session(events, None, 5) + assert report["tools"]["mcp__gitnexus__query"]["source_followups"] == 2 + assert Counter(call["tool"] for call in report["calls"])["Read"] == 1 diff --git a/eval/workflow_bench/README.md b/eval/workflow_bench/README.md index 7b64b6729..3ca916555 100644 --- a/eval/workflow_bench/README.md +++ b/eval/workflow_bench/README.md @@ -930,6 +930,207 @@ measurements unchanged and compare only runs with matching task contracts. ## MCP response cost +### Session diagnostics + +Every new headless session records `session_diagnostics` in `results.jsonl`. +The parent timestamps complete stdout event lines without changing the agent +prompt, tool responses, sandbox, or release criteria. No additional model calls +are made. Existing results without diagnostics remain valid and have unknown +timing, rather than measured zero time. + +The release evaluation's GitHub Actions summary includes a **Detailed +measurements** section for each measured runtime or optional arm. It shows cost +and wall-time distributions, setup/cleanup phases, CLI token counters, tool +outcomes and outliers, context/cache signals, repeated verification, and provider +request sizes and usage. Matching `*-diagnostics.json` and `*-diagnostics.md` +files are included in the sanitized evaluation artifacts. PR comparison +summaries retain the candidate's safe diagnostics too. + +Each sweep writes its own provider ledger. Only aggregate numeric provider +measurements reach the summary and artifacts; raw request identities and usage +ledgers remain on the runner. Missing or torn inputs are marked as partial, +and provider token totals are withheld when ledger records are unreadable. +Concurrent requests are not assigned to individual cells. Historical runs +without this instrumentation show missing telemetry; regenerating a report +cannot recover tool timings that were never recorded. + +Each session contains: + +- `calls`: tool name, event indexes, input/result JSON bytes, receipt offsets, + elapsed time, repeated-argument references, result status, staleness, + truncation, and the next assistant action's tool names. MCP errors, missing + symbols, ambiguity, empty query results, UNKNOWN risk, and output-limit + failures are counted separately. Standalone `gitnexus` shim commands are + identified; compound shell commands remain `Bash` because their elapsed time + includes other work. +- `assistant_usage`: the last available usage snapshot for each assistant + message, keyed by event index. Missing fields stay missing. These are + main-session CLI counters, not a complete provider or subagent ledger. +- `workload`: repeated reads, reads of files referenced by graph results, + query/context/read sequences, graph calls before the first observed edit, + verification commands and repeats, edits after verification, compactions, + and input-context growth. Shell commands are classified conservatively; + compound commands remain mixed. Unknown shell actions invalidate the + assumption that the checkout stayed unchanged between verification calls. + These are action counts, including failed edit attempts, not proven waste. +- `tools`: per-tool counts, elapsed sums, response bytes, and source-read + follow-ups after problematic results. One source action can follow multiple + problem results, so follow-ups are associations, not unique extra reads. +- `tool_wait_s`: the union of observed call intervals; parallel tools count + once. `mcp_wait_s` is a subset. `other_wall_s` includes model requests, + startup, transport, retries and scheduling; it is not a measurement of model + reasoning time. Missing results, unmatched IDs, clock gaps or diagnostics + limits make complete wait totals unavailable. +- `server_wall_s`: the sum of tool-reported `timing.wall` milliseconds, converted + to seconds, with a separate coverage count. Missing server timing stays + unknown. This does not equal the host's tool wait or model request time. + +The normal release JSON and Markdown include numeric per-tool summaries. The +JSON also keeps each session's five largest results, five slowest calls, and +five problematic calls with the largest results, +with event indexes for locating the detail in local `results.jsonl`. Raw tool +arguments, result text, source, prompts, IDs and argument hashes are excluded +from these diagnostics. Call and assistant-usage details are bounded at 2,048 +entries per session; `truncated` exposes that limit. Receipt timing is bounded +at 100,000 event lines. Diagnostics do not make incomplete task evidence valid. + +Use the counters to investigate explanations separately: large successful +responses suggest payload overhead; missing/ambiguous targets followed by +source reads suggest a possible workaround; repeated calls suggest a possible +navigation loop. These are signals, not proof of poor resolution or causation. +An empty result is not proof that a symbol is absent. Repeated arguments can be +justified after edits, and smaller output can require more follow-up reads. +Graph precision/recall requires expected-symbol labels or source verification. +The reported session cost is the money measurement; diagnostics never invent +per-tool dollar costs from bytes or elapsed time. + +### Cost investigations and controlled experiments + +No paid rerun is needed to prepare experiments or summarize existing evidence. +The commands below only read local files and write diagnostic artifacts. +Run them from `eval/` with the project's locked environment. + +For the first trial, prepare two cells for one task, with a $5 CLI-reported +budget stop per cell. This checks telemetry delivery before a larger comparison: + +```bash +uv run --locked python -m workflow_bench.diagnostic_experiments \ + --tasks workflow_bench/tasks.scenarios.yaml \ + --repository /path/to/GitNexus --runtime /path/to/built/GitNexus \ + --harness-sha FULL_HARNESS_SHA --runtime-sha FULL_RUNTIME_SHA \ + --model gpt-5.6-sol --effort xhigh --seed 3762 --repeats 1 \ + --variants no_graph mcp_eager --task-ids trivial-status-json-alias \ + --max-budget-usd 5 --out /tmp/gitnexus-diagnostics-trial +``` + +The runner forwards `--max-budget-usd` to Claude Code and accepts it only for +single-session baseline arms. The default remains uncapped. Budget-stopped +sessions stay failed and retain reported spend; they cannot establish savings +or successful task completion. The $10 planned total uses CLI accounting, not +a provider invoice cap: an in-flight request or different provider accounting +can exceed it. Preparing the plan does not authorize or execute the trial. + +| Possible cause | Measurement | What still needs an experiment | +| --- | --- | --- | +| Missing, ambiguous or incomplete resolution | Tool outcomes, UNKNOWN risk, lower-bound/stale/truncated responses, next source actions | Verify the intended target exists at the indexed source revision | +| Successful but irrelevant retrieval | Fixed source anchors in `retrieval_cases.json`, anchor recall and first anchor position | Capture the same queries against the sanitized task graph; unjudged results are not false positives | +| Too little detail or repeated investigation | Query/context/read chains, repeated arguments/reads, reads after graph references with and without source | A follow-up read may be necessary; compare matched tasks before calling it waste | +| Large responses and context growth | Result bytes, complete input usage snapshots, first/last/peak context, cache reads/writes, compactions | Compare response-size treatments at equal quality; bytes are not tokens | +| Mandatory graph workflow | Graph calls before edits; eager MCP versus optional-use prompt arm | Prompt ablation measures behavior; it does not remove repository instruction bytes | +| Tool catalog and transport | Eager/deferred/CLI arms; proxy tools/messages bytes and counts | Request sizes are available only when the proxy callback receives those fields | +| Verification and scope growth | Test/typecheck/lint/build counts, repeated checks, edits after verification, diff insertions/deletions/files | Larger diffs and repeated tests are not inherently unnecessary | +| Task fit and quality ceiling | Per-task cost distribution, solved/failed/missing counts, cost per solved task | Add harder independently validated tasks before claiming a quality improvement at 100% success | +| Order, concurrency, warm caches | Seeded rotated order, one worker and separate ledger per cell | Run matched worker-count/cold-start studies separately; fresh processes do not guarantee cold provider/filesystem caches | +| Retries, provider translation and accounting | CLI retry events; deduplicated success/failure callbacks, native normalized usage, reasoning tokens, proxy durations and estimates | Reconcile with provider billing; do not join concurrent requests to cells by timestamp | +| Setup, indexing and infrastructure | Cell start/finish timestamps, total duration, clone/index-materialization/dependency/MCP-startup/sandbox/agent-and-verification/artifact/cleanup spans | Shared index-build preparation, CPU/RSS, isolated oracle spans and infrastructure prices need separate evidence; do not fold them into model dollars | + +Summarize a runner result file (including failed cells) and optionally its proxy +ledger. Historical release reports with `per_run` rows are also accepted: + +```bash +uv run --locked python -m workflow_bench.cost_diagnostics \ + --results /path/to/results.jsonl \ + --provider-usage /path/to/provider_usage.jsonl \ + --out /tmp/cost-diagnostics.json --markdown-out /tmp/cost-diagnostics.md +``` + +Unknown fields remain null. Distinct model/effort/source/prompt/runtime/loading +cohorts are kept separate. Duplicate exported cell identities are rejected. +Provider response-ID conflicts withhold token totals; missing IDs prevent a +deduplicated total. Failed requests remain in the accounting. Proxy cost is an +estimate, not a provider invoice. The report is descriptive and does not change +release gates. It cannot infer cells omitted from its input: compare every +expected cell in the experiment plan before computing a treatment effect. +`cell_duration_s` includes setup and cleanup; `duration_s` retains its existing +session meaning. Phase spans partition the cell, including a failed active +phase, and unvisited phases stay missing. Shared snapshot construction happens +before cells and is not included. Agent and harness verification share one +phase; it must not be presented as isolated LLM reasoning time. Raw rows retain +UTC start/finish timestamps for order analysis; public release JSON retains +only numeric durations and whitelisted phase names. + +Prepare a pilot focused on the overhead floor and the expensive retry task: + +```bash +uv run --locked python -m workflow_bench.diagnostic_experiments \ + --tasks workflow_bench/tasks.scenarios.yaml \ + --repository /path/to/GitNexus --runtime /path/to/built/GitNexus \ + --harness-sha FULL_HARNESS_SHA --runtime-sha FULL_RUNTIME_SHA \ + --model gpt-5.6-sol --effort xhigh --seed 3762 --repeats 5 \ + --task-ids trivial-status-json-alias cross-module-parse-retry \ + --out /tmp/gitnexus-cost-experiment +``` + +This creates `plan.json` and one task manifest per cell. It has **no execution +mode** and never starts an agent. Five arms × two tasks × five repeats prepares +50 cells; this is a design, not spend authorization. Use `--variants` to prepare +a narrower comparison. Position balance requires a repeat count divisible by +the number of arms. The plan records argv arrays, environment paths, task class, +block/position, source pins and prompt/manifest digests. It refuses to overwrite +an existing directory. Pin the final instrumentation commit before execution; +an uncommitted implementation is not reproduced by its parent SHA alone. + +After separately authorizing spend, run cells serially in the listed sequence, +from the recorded `cwd`, with the recorded environment and a fresh exclusive +proxy ledger. Each subprocess runs one task/arm/repetition. Its raw `run` is +zero; retain the plan's cell ID and block when assembling comparisons. Do not +mix the optional-prompt arm with ordinary release evidence or reuse its results +for a required-guidance arm. Retain failures and missing cells in the denominator. +Record machine/container, dependency bindings, cache policy, start/finish times +and setup/oracle timing alongside the existing runner output. Reconcile the +ledger sum with CLI-reported session spend, rather than estimating dollars per +tool from duration or payload size. + +The four source-verified retrieval cases cover the same pinned scenario source. +Capture each fixed query from the **sanitized** task index used by the runner, +with the same query limits and content setting. Save parsed response objects as: + +```json +{ + "source_sha": "c4ecf398de637cf779ab2238e69958cfd5d601ce", + "runtime_sha": "FULL_RUNTIME_SHA", + "sanitized_graph": true, + "index_complete": true, + "index_staleness": "current", + "query_options": {"limit": 5, "max_symbols": 10, "include_content": false}, + "cases": [{"id": "CASE_ID", "query": "EXACT_FIXTURE_QUERY", "response": {}}] +} +``` + +Verify those provenance fields against the index receipt; the offline scorer +checks declarations, not the database itself. Keep capture metadata and raw +responses private, since they contain repository source. Score without an LLM: + +```bash +uv run --locked python -m workflow_bench.retrieval_diagnostics \ + --recordings /path/to/retrieval-recordings.json --out /tmp/retrieval-scores.json +``` + +Missing cases remain missing, errors are not successful empty queries, and stale +or mismatched source recordings are rejected. These are partial anchor labels; +there is no precision score or claim that every other result is irrelevant. +Anchor position follows flattened response order, not a global search rank. + With `include_content: true`, queries share a 16,000-character preview allowance equally across unique symbols on the returned page. The database returns only those prefixes, after ranking and pagination. `content_truncated: true` marks diff --git a/eval/workflow_bench/cli_arm_report.py b/eval/workflow_bench/cli_arm_report.py index dd33ea6a8..6a5850c52 100644 --- a/eval/workflow_bench/cli_arm_report.py +++ b/eval/workflow_bench/cli_arm_report.py @@ -15,6 +15,8 @@ from collections import Counter from pathlib import Path from typing import Any +from .session_diagnostics import public_cell_diagnostics, render_diagnostics + ARMS = ("baseline_nomcp", "baseline", "baseline_cli") EXPOSURE = {"baseline_nomcp": "none", "baseline": "stdio-only", "baseline_cli": "stdio-only"} METRICS = ("cost_usd", "duration_s", "num_turns", "cache_read_input_tokens", "output_tokens") @@ -141,6 +143,10 @@ def build_report(paired: Path, cli: Path) -> dict[str, Any]: "pairs": len(matched), "expected_pairs": len(pairs), } + report["diagnostic_rows"] = [ + {**{field: row.get(field) for field in ("task", "run", "arm")}, **public_cell_diagnostics(row)} + for row, _ in rows + ] return report @@ -180,6 +186,7 @@ def render_markdown(report: dict[str, Any]) -> str: if report["problems"]: lines += ["", "**Problems:**", *[f"- {problem}" for problem in report["problems"]]] lines += ["", "Small pinned task set; descriptive evidence only, not statistical significance.", ""] + lines += render_diagnostics(report.get("diagnostic_rows", [])) return "\n".join(lines) diff --git a/eval/workflow_bench/cost_diagnostics.py b/eval/workflow_bench/cost_diagnostics.py new file mode 100644 index 000000000..546685fb3 --- /dev/null +++ b/eval/workflow_bench/cost_diagnostics.py @@ -0,0 +1,283 @@ +"""Offline cost evidence report; no API calls, pricing guesses, or release gates.""" + +from __future__ import annotations + +import argparse +import html +import json +import math +import statistics +from dataclasses import asdict +from pathlib import Path +from typing import Any + +from .provider_usage import NormalizedUsage, UsageSemanticsError, normalize_usage +from .session_diagnostics import CELL_PHASES, public_diagnostics, render_diagnostics + +NUMERIC = ("cost_usd", "duration_s", "cell_duration_s", "num_turns", "diff_insertions", "diff_deletions", "diff_files", + "input_tokens", "cache_creation_input_tokens", "cache_read_input_tokens", "output_tokens") +COHORT = ("model", "effort", "task_base_sha", "task_prompt_digest", "runtime_digest", "mcp_tool_loading", + "sandbox_dependency_content_digest", "oracle_digest", "max_budget_usd") +REQUEST_METRICS = ("messages_bytes", "tools_bytes", "system_bytes", "messages_count", "tools_count", + "duration_s", "proxy_estimated_cost_usd") + + +def _number(value: Any) -> bool: + if type(value) not in (int, float) or value < 0: + return False + try: + return math.isfinite(value) + except OverflowError: + return False + + +def distribution(values: list[Any]) -> dict[str, Any]: + measured = [v for v in values if _number(v)] + return { + "observed": len(measured), "missing": len(values) - len(measured), + "sum_observed": sum(measured) if measured else None, + "mean": statistics.mean(measured) if measured else None, + "median": statistics.median(measured) if measured else None, + "min": min(measured) if measured else None, "max": max(measured) if measured else None, + "stdev": statistics.stdev(measured) if len(measured) > 1 else None, + } + + +def provider_summary(events: list[dict[str, Any]]) -> dict[str, Any]: + """Deduplicate identical callbacks, withhold totals on identity conflicts. + + Never join to cells by wall-clock proximity. Concurrent sweeps require a + session ID mapping; serial experiments can use one exclusive ledger/cell. + """ + seen: dict[tuple, dict] = {} + unique, duplicates, conflicts, unidentified = [], 0, set(), 0 + for event in events: + response_id = event.get("response_id") + if not isinstance(response_id, str) or not response_id: + unique.append(event) + unidentified += 1 + continue + key = (event.get("sweep_id"), event.get("provider"), response_id) + if key in seen: + if seen[key] == event: + duplicates += 1 + else: + conflicts.add(key) + continue + seen[key] = event + unique.append(event) + normalized = [] + invalid = 0 + for event in unique: + try: + normalized.append(asdict(normalize_usage(event.get("provider"), event.get("native_usage")))) + except (UsageSemanticsError, TypeError, ValueError): + invalid += 1 + normalized.append({}) + token_fields = NormalizedUsage.__dataclass_fields__ + token_totals = { + field: sum(row[field] for row in normalized) + if normalized and not conflicts and not unidentified and all(_number(row.get(field)) for row in normalized) else None + for field in token_fields + } + metrics = [e.get("request_metrics") if not conflicts and isinstance(e.get("request_metrics"), dict) else {} for e in unique] + return { + "records": len(events), "observed_requests": len(unique), "duplicate_callbacks": duplicates, + "conflicting_response_ids": len(conflicts), "requests_without_response_id": unidentified, + "invalid_usage_records": invalid, + "successes": sum(e.get("status") == "success" for e in unique), + "failures": sum(e.get("status") == "failure" for e in unique), + "requests_with_session_id": sum(isinstance(e.get("session_id"), str) and bool(e["session_id"]) for e in unique), + "unique_actual_models": len({e["actual_model"] for e in unique if isinstance(e.get("actual_model"), str)}), + "token_totals": token_totals, + "request_metrics": {field: distribution([m.get(field) for m in metrics]) for field in REQUEST_METRICS}, + "identity_complete": bool(unique) and not conflicts and unidentified == 0, + "attribution": "ledger_only_no_cell_join", + "limitations": [ + "No response ID means callbacks cannot be deduplicated; observed totals may overcount.", + "Proxy cost is an estimate, not an invoice. Failures without usage may still incur cost.", + "Callback duration includes proxy/provider work; it is not isolated model thinking time.", + "Request payload bytes are serialized sizes, not token counts.", + ], + } + + +def build_cost_report(rows: list[dict[str, Any]], provider_events: list[dict[str, Any]] | None = None) -> dict[str, Any]: + groups: dict[tuple, list[dict]] = {} + seen_cells = set() + for row in rows: + arm, task = row.get("arm"), row.get("task") + if not isinstance(arm, str) or not isinstance(task, str): + raise ValueError("each cell needs arm and task labels") + identity_fields = {field: row.get(field) if isinstance(row.get(field), str) else "" for field in COHORT} + if arm == "baseline_cli" and "mcp_tool_loading" in row and row["mcp_tool_loading"] is None: + identity_fields["mcp_tool_loading"] = "not_applicable" + if "max_budget_usd" in row: + budget = row["max_budget_usd"] + identity_fields["max_budget_usd"] = "uncapped" if budget is None else str(float(budget)) if _number(budget) and budget > 0 else "" + cohort = tuple(identity_fields[field] for field in COHORT) + identity = (arm, task, row.get("run"), cohort) + if type(row.get("run")) is int: + if identity in seen_cells: + raise ValueError("duplicate cell identity; do not concatenate repeated exports") + seen_cells.add(identity) + groups.setdefault((arm, task, cohort), []).append(row) + summaries = [] + tool_sections = [] + for (arm, task, cohort), cells in sorted(groups.items()): + costs = [row.get("cost_usd") for row in cells] + solved = sum(row.get("resolved") is True for row in cells) + summaries.append({ + "diagnostic_group": len(summaries) + 1, + "arm": arm, "task": task, "cells": len(cells), "solved": solved, + "cohort": dict(zip(COHORT, cohort, strict=True)), + "provenance_complete": all(cohort), + "quality_missing": sum(type(row.get("resolved")) is not bool for row in cells), + "failed_cells": sum(row.get("ok") is False for row in cells), + "diagnostic_cells": sum(bool(public_diagnostics(row.get("session_diagnostics"))) for row in cells), + "metrics": {field: distribution([row.get(field) for row in cells]) for field in NUMERIC}, + "cell_phases": {phase: distribution([ + row["cell_phase_seconds"].get(phase) if isinstance(row.get("cell_phase_seconds"), dict) else None + for row in cells + ]) for phase in CELL_PHASES}, + "cost_per_solved": sum(costs) / solved if solved and all(_number(c) for c in costs) else None, + }) + details = render_diagnostics(cells) + if details: + tool_sections += [f"### Diagnostic group {len(summaries)}", *details] + return { + "schema": "gitnexus.cost-diagnostics/v1", "cells": len(rows), "groups": summaries, + "provider": provider_summary(provider_events) if provider_events is not None else None, + "tool_markdown": "\n".join(tool_sections), + "limitations": [ + "Descriptive observations only; no causal attribution, significance claim or release verdict.", + "Groups include failed cells and cost per solved includes their reported spend.", + "Missing cells require comparison to the experiment manifest; this report cannot infer omitted rows.", + "Legacy rows without diagnostics cannot reveal retrieval relevance, instruction overhead or tool timings.", + "Main-loop session telemetry excludes subagent internals; session cost may include subagents.", + "Task labels, model/effort, runtime/source pins and prompt digests must match the intended treatment.", + ], + } + + +def render_markdown(report: dict[str, Any]) -> str: + """Render measured distributions and coverage, without guessing missing spend.""" + def number(value: Any) -> str: + return f"{value:,.4f}".rstrip("0").rstrip(".") if _number(value) else "n/a" + + def label(value: Any) -> str: + return html.escape(str(value)).replace("\n", " ").translate( + {ord(char): f"&#{ord(char)};" for char in "\\`*_[]()|!"} + ) + + def table(metrics: dict[str, dict]) -> list[str]: + lines = ["| Measurement | Observed / missing | Observed total | Mean | Median | Min | Max | Std. dev. |", + "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: |"] + for name, values in metrics.items(): + lines.append(f"| {name} | {values['observed']} / {values['missing']} | " + " | ".join( + number(values[field]) for field in ("sum_observed", "mean", "median", "min", "max", "stdev") + ) + " |") + return lines + + lines = ["## Time, cost, and tool measurements", "", + f"Observed cells: {report['cells']}. Missing measurements are shown as `n/a`, not zero.", + "Cell token counters come from the CLI. Provider tokens below use provider-specific accounting; do not add the two together.", ""] + if report.get("input_problems"): + lines += ["**Partial input: totals cover only the readable records.**", "", + *[f"- {problem}" for problem in report["input_problems"]], ""] + for group in report["groups"]: + lines += [f"### Group {group['diagnostic_group']}: {label(group['arm'])} / {label(group['task'])}", "", + (f"Solved: {group['solved']}/{group['cells']}; failed sessions: {group['failed_cells']}; " + f"quality unknown: {group['quality_missing']}; diagnostics: {group['diagnostic_cells']}/{group['cells']}."), + f"Cost per solved cell (USD, including failed-session spend): {number(group['cost_per_solved'])}.", + f"Complete cohort provenance: {'yes' if group['provenance_complete'] else 'no'}.", "", + "
Measurements and cohort", "", + "| Cohort field | Value |", "| --- | --- |", + *[f"| {field} | {label(value) if value else 'unknown'} |" + for field, value in group["cohort"].items()], "", + *table(group["metrics"]), "", "Cell phases (seconds; agent_and_verification includes grading):", "", + *table(group["cell_phases"]), "", "
", ""] + provider = report["provider"] + lines += ["### Provider requests", ""] + if provider is None: + lines += ["Provider ledger unavailable. Request counts, provider tokens, and proxy cost are unknown.", ""] + else: + lines += [(f"Observed requests: {provider['observed_requests']}; successes: {provider['successes']}; " + f"failures: {provider['failures']}; duplicate callbacks removed: {provider['duplicate_callbacks']}."), + (f"Conflicting response identities: {provider['conflicting_response_ids']}; " + f"requests without identity: {provider['requests_without_response_id']}; " + f"invalid usage records: {provider['invalid_usage_records']}."), + "Attribution is to this sweep's ledger. Requests are not assigned to individual concurrent cells.", "", + "| Provider token field | Observed total |", "| --- | ---: |", + *[f"| {field} | {number(value)} |" for field, value in provider["token_totals"].items()], "", + *table(provider["request_metrics"]), "", + *provider["limitations"], ""] + lines += [report["tool_markdown"], "", *report["limitations"], ""] + return "\n".join(lines) + + +def _read_rows(path: Path, *, allow_incomplete: bool, name: str, problems: list[str]) -> list[dict]: + if not path.is_file() and allow_incomplete: + problems.append(f"{name} file is missing.") + return [] + raw = path.read_text() + try: + doc = json.loads(raw) + except json.JSONDecodeError: + rows = [] + for line in raw.splitlines(): + if not line.strip(): + continue + try: + rows.append(json.loads(line)) + except json.JSONDecodeError: + if not allow_incomplete: + raise + rows.append(None) + else: + rows = doc.get("per_run", [doc]) if isinstance(doc, dict) else doc + if not isinstance(rows, list): + raise ValueError(f"expected {name} rows") + invalid = sum(not isinstance(row, dict) for row in rows) + if invalid: + if not allow_incomplete: + raise ValueError(f"expected {name} rows") + problems.append(f"{name}: {invalid} unreadable record(s).") + return [row for row in rows if isinstance(row, dict)] + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--results", type=Path, required=True, help="runner results.jsonl or release report JSON") + parser.add_argument("--provider-usage", type=Path) + parser.add_argument("--out", type=Path, required=True) + parser.add_argument("--markdown-out", type=Path) + parser.add_argument("--allow-incomplete", action="store_true", help="report missing/torn inputs after an interrupted run") + args = parser.parse_args() + problems: list[str] = [] + rows = _read_rows(args.results, allow_incomplete=args.allow_incomplete, name="Results", problems=problems) + provider = None + provider_problems: list[str] = [] + if args.provider_usage: + provider = _read_rows(args.provider_usage, allow_incomplete=args.allow_incomplete, + name="Provider ledger", problems=provider_problems) + if provider_problems and not provider: + provider = None + problems += provider_problems + report = build_cost_report(rows, provider) + if report["provider"] is not None: + report["provider"]["ledger_complete"] = not provider_problems + if provider_problems: + report["provider"]["identity_complete"] = False + report["provider"]["token_totals"] = dict.fromkeys(report["provider"]["token_totals"]) + report["input_problems"] = problems + with args.out.open("x") as handle: + json.dump(report, handle, indent=2, allow_nan=False) + handle.write("\n") + if args.markdown_out: + with args.markdown_out.open("x") as handle: + handle.write(render_markdown(report)) + + +if __name__ == "__main__": + main() diff --git a/eval/workflow_bench/diagnostic_experiments.py b/eval/workflow_bench/diagnostic_experiments.py new file mode 100644 index 000000000..80fdd6f75 --- /dev/null +++ b/eval/workflow_bench/diagnostic_experiments.py @@ -0,0 +1,162 @@ +"""Prepare counterbalanced experiment inputs. This module NEVER runs an agent.""" + +from __future__ import annotations + +import argparse +import copy +import hashlib +import json +import random +import re +from pathlib import Path +from typing import Any + +import yaml + +from .runner_tasks import normalized_model_identifier, select_tasks +from .session_budget import session_budget + +VARIANTS = { + "no_graph": ("baseline_nomcp", "eager"), + "mcp_eager": ("baseline", "eager"), + "mcp_deferred": ("baseline", "deferred"), + "cli": ("baseline_cli", "eager"), + "mcp_optional_prompt": ("baseline", "eager"), +} +OPTIONAL_PROMPT = """ + +Experiment-specific instruction: GitNexus is available, but using it is optional +for this task. This overrides repository instructions that require graph-first +navigation, impact analysis before every edit, or detect_changes before commit. +Use the tools that help solve the task. Keep all other repository requirements, +tests, and safety constraints. Do not install or bootstrap additional tools. +""" + + +def prepare_plan( + document: dict[str, Any], *, repository: str, runtime: str, runtime_sha: str, + harness_sha: str, model: str, effort: str, repeats: int, seed: int, + variants: list[str], task_ids: list[str] | None = None, max_budget_usd: float | None = None, +) -> dict[str, Any]: + """A seeded Latin-square order within each task; no execution side effects. + + One runner process per cell makes its provider ledger attributable without + guessing session IDs. Run indices in runner output are always zero; the + experiment's block/sequence fields carry repetition identity instead. + """ + max_budget_usd = session_budget(max_budget_usd) + if not variants or len(set(variants)) != len(variants) or any(v not in VARIANTS for v in variants): + raise ValueError("variants must be unique known experiment variants") + if not 1 <= repeats <= 100: + raise ValueError("repeats must be between 1 and 100") + if not all(re.fullmatch(r"[0-9a-f]{40}", sha) for sha in (runtime_sha, harness_sha)): + raise ValueError("runtime and harness require full commit SHAs") + if not isinstance(model, str): + raise ValueError("use a named model, not auto/latest") + model = normalized_model_identifier(model) + if effort not in ("low", "medium", "high", "xhigh", "max"): + raise ValueError("invalid effort") + if not isinstance(document, dict): + raise ValueError("task document must contain tasks") + tasks = document.get("tasks") + if not isinstance(tasks, list) or not tasks or not all(isinstance(t, dict) for t in tasks): + raise ValueError("task document must contain tasks") + tasks, _ = select_tasks([{**task, "repo": repository} for task in tasks], include_expensive=True) + ids = [t["id"] for t in tasks] + if task_ids is not None and (not task_ids or not set(task_ids) <= set(ids)): + raise ValueError("unknown or empty task selection") + tasks = [t for t in tasks if task_ids is None or t["id"] in task_ids] + if any(not re.fullmatch(r"[0-9a-f]{40}", str(t.get("ref", ""))) or not isinstance(t.get("prompt"), str) for t in tasks): + raise ValueError("each task needs a pinned ref and prompt") + rng = random.Random(seed) + order = variants.copy() + rng.shuffle(order) + rng.shuffle(tasks) + cells = [] + for block in range(repeats): + for task_index, task in enumerate(tasks): + rotation = (block + task_index) % len(order) + for position, variant in enumerate(order[rotation:] + order[:rotation]): + arm, loading = VARIANTS[variant] + prepared = copy.deepcopy(task) + prepared["repo"] = repository + if variant == "mcp_optional_prompt": + prepared["prompt"] += OPTIONAL_PROMPT + sequence = len(cells) + cell_id = f"cell-{sequence:04d}" + cells.append({ + "cell_id": cell_id, "sequence": sequence, "block": block, + "position": position, "task": task["id"], "task_class": task.get("class"), + "variant": variant, "arm": arm, "tool_loading": loading, + "base_prompt_sha256": hashlib.sha256(task["prompt"].encode()).hexdigest(), + "prompt_sha256": hashlib.sha256(prepared["prompt"].encode()).hexdigest(), + "task_document": {"tasks": [prepared]}, + "argv": ["uv", "run", "--locked", "python", "-m", "workflow_bench.runner", + "--tasks", f"{cell_id}.tasks.json", "--runs", "1", "--workers", "1", + "--arms", arm, "--mcp-tool-loading", loading, + "--gitnexus-root", runtime, "--model", model, "--effort", effort, + "--include-expensive", *(["--max-budget-usd", str(max_budget_usd)] if max_budget_usd is not None else []), + "--out", f"results/{cell_id}"], + "env": {"GITNEXUS_BENCH_SWEEP_ID": cell_id, + "GITNEXUS_BENCH_PROVIDER_USAGE": f"results/{cell_id}.provider.jsonl"}, + }) + return { + "schema": "gitnexus.diagnostic-experiments/v1", "execution": "not_started", + "harness_sha": harness_sha, "runtime_sha": runtime_sha, "seed": seed, + "model": model, "effort": effort, "repeats": repeats, "workers": 1, + "max_budget_usd_per_cell": max_budget_usd, + "planned_cli_budget_usd": len(cells) * max_budget_usd if max_budget_usd is not None else None, + "position_balanced": repeats % len(variants) == 0, + "release_gate_eligible": False, "cells": cells, + "limitations": [ + "No model-spend authorization; this plan does not execute commands.", + "Verify checkout SHAs, instrumentation diff, dependencies and oracle bindings before execution.", + "Run serially in listed sequence, with an exclusive proxy/ledger per cell; never reuse ledgers.", + "Optional-prompt treatment changes tool-use policy, not repository guidance bytes or tool schemas.", + "This plan controls concurrency at one; a separate matched worker-count study is needed for contention.", + "Fresh processes do not guarantee cold filesystem/provider caches; record machine and cache policy.", + "CLI budget stops use reported model cost; in-flight requests and provider billing can exceed them.", + ], + } + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--tasks", type=Path, required=True) + parser.add_argument("--repository", type=Path, required=True) + parser.add_argument("--runtime", type=Path, required=True) + parser.add_argument("--runtime-sha", required=True) + parser.add_argument("--harness-sha", required=True) + parser.add_argument("--model", required=True) + parser.add_argument("--effort", default="xhigh") + parser.add_argument("--repeats", type=int, default=5) + parser.add_argument("--seed", type=int, default=3762) + parser.add_argument("--max-budget-usd", type=float, help="CLI-reported budget stop per cell, not an invoice cap") + parser.add_argument("--variants", nargs="+", choices=VARIANTS, default=list(VARIANTS)) + parser.add_argument("--task-ids", nargs="+") + parser.add_argument("--out", type=Path, required=True) + args = parser.parse_args() + plan = prepare_plan( + yaml.safe_load(args.tasks.read_text()), repository=str(args.repository.resolve()), + runtime=str(args.runtime.resolve()), runtime_sha=args.runtime_sha, harness_sha=args.harness_sha, + model=args.model, effort=args.effort, repeats=args.repeats, seed=args.seed, + variants=args.variants, task_ids=args.task_ids, max_budget_usd=args.max_budget_usd, + ) + # Refuse overwrite: a previous plan/ledger may already be evidence. + args.out.mkdir(parents=True, exist_ok=False) + root = args.out.resolve() + (root / "results").mkdir() + for cell in plan["cells"]: + manifest = root / f"{cell['cell_id']}.tasks.json" + manifest.write_text(json.dumps(cell.pop("task_document"), indent=2) + "\n") + cell["task_manifest_sha256"] = hashlib.sha256(manifest.read_bytes()).hexdigest() + cell["argv"][cell["argv"].index("--tasks") + 1] = str(manifest) + cell["argv"][-1] = str(root / "results" / cell["cell_id"]) + cell["env"]["GITNEXUS_BENCH_PROVIDER_USAGE"] = str(root / "results" / f"{cell['cell_id']}.provider.jsonl") + cell["cwd"] = str(Path(__file__).resolve().parents[1]) + (root / "plan.json").write_text(json.dumps(plan, indent=2) + "\n") + print(f"Prepared {len(plan['cells'])} cells at {root}; no agents started.") + + +if __name__ == "__main__": + main() diff --git a/eval/workflow_bench/litellm_usage_callback.py b/eval/workflow_bench/litellm_usage_callback.py index 8277d0102..3e3585c81 100644 --- a/eval/workflow_bench/litellm_usage_callback.py +++ b/eval/workflow_bench/litellm_usage_callback.py @@ -19,6 +19,7 @@ accounting because the log write failed would be the worse outcome. from __future__ import annotations import json +import math import os import threading from typing import Any @@ -67,6 +68,36 @@ def _plain(value: Any) -> Any: return None +def _request_metrics(kwargs: dict, start_time: Any, end_time: Any) -> dict: + """Numeric envelope evidence only; never log request content or headers.""" + optional = kwargs.get("optional_params") + optional = optional if isinstance(optional, dict) else {} + metrics = {} + for field in ("messages", "tools", "system"): + value = kwargs.get(field, optional.get(field)) + metrics[f"{field}_bytes"] = None + if isinstance(value, (str, dict, list)): + try: + metrics[f"{field}_bytes"] = len(json.dumps( + value, ensure_ascii=False, separators=(",", ":"), + ).encode(errors="backslashreplace")) + except (TypeError, ValueError, RecursionError): + # Keep size unknown when serialization fails; usage accounting + # must still be recorded even without this optional metric. + pass + if field != "system": + metrics[f"{field}_count"] = len(value) if isinstance(value, list) else None + try: + elapsed = end_time - start_time + elapsed = elapsed.total_seconds() if hasattr(elapsed, "total_seconds") else elapsed + except (TypeError, ValueError): + elapsed = None + metrics["duration_s"] = elapsed if type(elapsed) in (int, float) and math.isfinite(elapsed) and elapsed >= 0 else None + cost = kwargs.get("response_cost") + metrics["proxy_estimated_cost_usd"] = cost if type(cost) in (int, float) and math.isfinite(cost) and cost >= 0 else None + return metrics + + class ProviderUsageLogger(CustomLogger): async def async_log_success_event(self, kwargs, response_obj, start_time, end_time) -> None: # noqa: ANN001 self._append("success", kwargs, response_obj, start_time, end_time) @@ -123,6 +154,7 @@ class ProviderUsageLogger(CustomLogger): "completed_at": str(end_time), # Verbatim. Not flattened, not renamed, not summed. "native_usage": _plain(getattr(response_obj, "usage", None)), + "request_metrics": _request_metrics(kwargs, start_time, end_time), } line = json.dumps(event, default=str) + "\n" with _LOCK, open(path, "a", encoding="utf-8") as handle: diff --git a/eval/workflow_bench/mcp_loading_report.py b/eval/workflow_bench/mcp_loading_report.py index a66a4d153..029ead208 100644 --- a/eval/workflow_bench/mcp_loading_report.py +++ b/eval/workflow_bench/mcp_loading_report.py @@ -10,6 +10,7 @@ from pathlib import Path from typing import Any from .release_report import META_FIELDS, build_report +from .session_diagnostics import render_diagnostics COUNTERS = ("num_turns", "input_tokens", "cache_creation_input_tokens", "cache_read_input_tokens", "output_tokens") BINDINGS = ("task_prompt_digest", "sandbox_dependency_content_digest") @@ -105,6 +106,10 @@ def render_markdown(report: dict[str, Any]) -> str: "Hooks disabled; prompts, MCP backend and outputs unchanged. Descriptive evidence only; no release approval.", "", ] + for mode in ("none", "eager", "deferred"): + details = render_diagnostics([row for row in report.get("per_run", []) if row.get("mcp_tool_loading") == mode]) + if details: + lines += [f"### Diagnostics: {mode}", *details, ""] if not report["complete"]: return "\n".join([*lines, "**Incomplete or mismatched evidence; no savings estimate.**", "", *[f"- {p}" for p in report["problems"]], ""]) diff --git a/eval/workflow_bench/pr_evaluation.py b/eval/workflow_bench/pr_evaluation.py index b106aa236..9df44c806 100644 --- a/eval/workflow_bench/pr_evaluation.py +++ b/eval/workflow_bench/pr_evaluation.py @@ -18,6 +18,8 @@ from datetime import UTC, datetime from pathlib import Path from typing import Any +from .session_diagnostics import public_cell_diagnostics, render_diagnostics + SHA = re.compile(r"[0-9a-f]{40}") REPO = re.compile(r"[\w.-]+/[\w.-]+") COST_LABEL = "run-cost-evaluation" @@ -349,6 +351,7 @@ def compare_candidate( and _number(row.get("duration_s")) ) safe = {field: row[field] for field in ROW_FIELDS if field in row} + safe.update(public_cell_diagnostics(row)) try: json.dumps(safe, allow_nan=False) except (ValueError, TypeError, OverflowError): @@ -414,7 +417,8 @@ def render_markdown(report: dict[str, Any]) -> str: "", ] if not report["complete"]: - return "\n".join([*lines, "**Incomplete evidence; no savings conclusion.**", "", *report["problems"], ""]) + return "\n".join([*lines, "**Incomplete evidence; no savings conclusion.**", "", *report["problems"], + *render_diagnostics(report.get("per_run", [])), ""]) lines += [ "| MCP sessions | Solved | Total model cost | Mean model cost |", "| --- | --- | --- | --- |", @@ -446,6 +450,7 @@ def render_markdown(report: dict[str, Any]) -> str: "", "Only the PR MCP cells are newly measured; no-MCP and stable sessions are not rerun.", "Historical comparisons are descriptive and do not establish statistical significance or isolate this PR from other runtime changes.", + *render_diagnostics(report.get("per_run", [])), "", ] ) diff --git a/eval/workflow_bench/release_report.py b/eval/workflow_bench/release_report.py index 3ceeb089a..668282727 100644 --- a/eval/workflow_bench/release_report.py +++ b/eval/workflow_bench/release_report.py @@ -17,6 +17,7 @@ import yaml from .oracle_assets import capture_task_oracles from .runner_tasks import normalized_model_identifier, select_tasks +from .session_diagnostics import public_cell_diagnostics, render_diagnostics from .task_assets import DEPENDENCY_CONTENT_BINDING_FIELD SCHEMA = "gitnexus.release-evaluation/v1" @@ -110,7 +111,9 @@ def build_report(rows: list[Any], metadata: dict[str, Any], *, now: datetime | N problems.append(f"row {index}: invalid measurement") continue # Invalid rows still need a public, JSON-serializable failure receipt. - safe_rows.append({key: _json_safe(row[key]) for key in ROW_FIELDS if key in row}) + safe_row = {key: _json_safe(row[key]) for key in ROW_FIELDS if key in row} + safe_row.update(public_cell_diagnostics(row)) + safe_rows.append(safe_row) task, run, arm = row.get("task"), row.get("run"), row.get("arm") if not isinstance(task, str) or type(run) is not int or not isinstance(arm, str): problems.append(f"row {index}: invalid cell identity") @@ -205,6 +208,7 @@ def render_markdown(report: dict[str, Any]) -> str: "This small pinned task set is release evidence; it does not establish general gains or statistical significance.", "", ] + lines += render_diagnostics(report.get("per_run", [])) return "\n".join(lines) diff --git a/eval/workflow_bench/retrieval_cases.json b/eval/workflow_bench/retrieval_cases.json new file mode 100644 index 000000000..256f9b323 --- /dev/null +++ b/eval/workflow_bench/retrieval_cases.json @@ -0,0 +1,37 @@ +{ + "source_sha": "c4ecf398de637cf779ab2238e69958cfd5d601ce", + "judgment": "partial source-verified anchors; no precision labels", + "query_options": { "limit": 5, "max_symbols": 10, "include_content": false }, + "cases": [ + { + "id": "trivial-status-json-alias", + "query": "status CLI command JSON output option registration", + "anchors": [{ "file": "gitnexus/src/cli/index.ts" }] + }, + { + "id": "inv-feature-c-include-provenance", + "query": "C include capture quoted and system header provenance", + "anchors": [ + { + "file": "gitnexus/src/core/ingestion/languages/c/import-decomposer.ts", + "name": "splitCInclude" + } + ] + }, + { + "id": "inv-feature-list-repos-filter", + "query": "list_repos MCP repository listings filter by name", + "anchors": [{ "file": "gitnexus/src/mcp/local/local-backend.ts", "name": "listRepos" }] + }, + { + "id": "cross-module-parse-retry", + "query": "parse worker transient failure error exit retry quarantine dispatch", + "anchors": [ + { + "file": "gitnexus/src/core/ingestion/workers/worker-pool.ts", + "name": "handleWorkerDeath" + } + ] + } + ] +} diff --git a/eval/workflow_bench/retrieval_diagnostics.py b/eval/workflow_bench/retrieval_diagnostics.py new file mode 100644 index 000000000..ee13b2c5d --- /dev/null +++ b/eval/workflow_bench/retrieval_diagnostics.py @@ -0,0 +1,92 @@ +"""Score fixed source anchors against recorded retrievals, without an LLM judge. + +These are partial relevance judgments: unjudged results are not false positives. +The score is anchor coverage, not end-to-end task success or search precision. +""" + +from __future__ import annotations + +import argparse +import json +import re +from pathlib import Path +from typing import Any + + +def score_retrieval(cases: dict[str, Any], recordings: dict[str, Any]) -> dict[str, Any]: + if recordings.get("source_sha") != cases.get("source_sha") or not cases.get("source_sha"): + raise ValueError("retrieval recordings must match the fixture source SHA") + if recordings.get("index_complete") is not True or recordings.get("index_staleness") != "current": + raise ValueError("retrieval recordings require a complete current index receipt") + if recordings.get("sanitized_graph") is not True or not re.fullmatch(r"[0-9a-f]{40}", str(recordings.get("runtime_sha", ""))): + raise ValueError("record the sanitized graph and runtime identity") + if recordings.get("query_options") != cases.get("query_options") or not cases.get("query_options"): + raise ValueError("record the fixed query options from the fixture") + rows = recordings.get("cases", []) + if not isinstance(rows, list) or not all(isinstance(row, dict) for row in rows): + raise ValueError("recorded cases must be a list") + by_id = {row.get("id"): row for row in rows} + if len(by_id) != len(rows): + raise ValueError("duplicate retrieval case ID") + expected_ids = {case["id"] for case in cases["cases"]} + if not set(by_id) <= expected_ids: + raise ValueError("unknown retrieval case ID") + scores = [] + for case in cases["cases"]: + row = by_id.get(case["id"]) + if row is None: + scores.append({"id": case["id"], "status": "missing", "anchor_recall": None}) + continue + if row.get("query") != case["query"]: + raise ValueError(f"query differs for case {case['id']}") + body = row.get("response") + if not isinstance(body, dict): + raise ValueError("record parsed query response objects") + symbols = [*body.get("process_symbols", []), *body.get("definitions", [])] + anchors = [] + for symbol in symbols: + if isinstance(symbol, dict) and isinstance(symbol.get("filePath"), str): + anchor = (symbol["filePath"], symbol.get("name")) + if anchor not in anchors: + anchors.append(anchor) + matched = [] + for expected in case["anchors"]: + positions = [i + 1 for i, (path, name) in enumerate(anchors) + if path == expected["file"] and ("name" not in expected or name == expected["name"])] + matched.append(min(positions) if positions else None) + failed = bool(body.get("error")) + scores.append({ + "id": case["id"], "status": "error" if failed else "observed", + "expected_anchors": len(matched), "found_anchors": sum(p is not None for p in matched), + "anchor_recall": sum(p is not None for p in matched) / len(matched) if matched and not failed else None, + "first_anchor_position": min((p for p in matched if p is not None), default=None), + "returned_anchors": len(anchors), "response_bytes": len(json.dumps(body, ensure_ascii=True).encode()), + "truncated": body.get("truncated") is True or body.get("partial") is True, + }) + return { + "schema": "gitnexus.retrieval-diagnostics/v1", "source_sha": cases["source_sha"], + "runtime_sha": recordings["runtime_sha"], "cases": scores, + "complete": all(row["status"] == "observed" and not row.get("truncated") for row in scores), + "limitations": [ + "Source anchors are partial judgments; no precision or false-positive rate is inferred.", + "Position is flattened process_symbols then definitions order, not a calibrated relevance rank.", + "Coverage is conditional on the recorded query limits and index layers; preserve the capture receipt.", + "The scorer checks declared provenance; the capture must verify it against the sanitized index.", + ], + } + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--cases", type=Path, default=Path(__file__).with_name("retrieval_cases.json")) + parser.add_argument("--recordings", type=Path, required=True) + parser.add_argument("--out", type=Path, required=True) + args = parser.parse_args() + report = score_retrieval(json.loads(args.cases.read_text()), json.loads(args.recordings.read_text())) + with args.out.open("x") as handle: + json.dump(report, handle, indent=2, allow_nan=False) + handle.write("\n") + + +if __name__ == "__main__": + main() diff --git a/eval/workflow_bench/runner.py b/eval/workflow_bench/runner.py index cf8748271..6d37446b1 100644 --- a/eval/workflow_bench/runner.py +++ b/eval/workflow_bench/runner.py @@ -56,6 +56,8 @@ from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any +from .session_budget import session_budget + import yaml from .mcp_transport import connector_config, isolated_mcp @@ -492,6 +494,7 @@ def run_arm( "progress_label": progress_label, "claude_bin": sandbox.claude_bin, "timeout": args.timeout, + "max_budget_usd": getattr(args, "max_budget_usd", None), "model": args.model, "effort": args.effort, "env": env, @@ -1212,6 +1215,18 @@ def run_cell(ctx: TaskCellContext, run_idx: int, arm: str) -> dict[str, Any]: expected kinds is deliberately left to propagate — a harness bug must not be recorded as an ordinary infra-error and averaged into the evidence. """ + cell_started_at = datetime.now(UTC).isoformat() + cell_started = time.monotonic() + phase_started = cell_started + active_phase = "validation" + phase_seconds: dict[str, float] = {} + + def enter_phase(name: str) -> None: + nonlocal active_phase, phase_started + now = time.monotonic() + phase_seconds[active_phase] = phase_seconds.get(active_phase, 0.0) + now - phase_started + active_phase, phase_started = name, now + args = ctx.args task = ctx.task worktree: Path | None = None @@ -1235,6 +1250,7 @@ def run_cell(ctx: TaskCellContext, run_idx: int, arm: str) -> dict[str, Any]: raise RuntimeError("sanitized graph snapshot is unavailable") if ctx.asset_snapshot is None: raise RuntimeError("task asset snapshot is unavailable") + enter_phase("clone") if ctx.clone_template is not None: if not ctx.sanitized_head: raise RuntimeError("clone template is missing its sanitized HEAD") @@ -1243,9 +1259,11 @@ def run_cell(ctx: TaskCellContext, run_idx: int, arm: str) -> dict[str, Any]: else: worktree = make_worktree(ctx.repo, ctx.task_sha, ctx.trees_dir) sanitized_head = sanitize_clone_for_hidden_oracles(worktree) + enter_phase("graph_materialize") if gitnexus_available: assert ctx.graph_snapshot is not None ctx.graph_snapshot.materialize(worktree, sanitized_head=sanitized_head) + enter_phase("dependencies") dependency_mounts = stage_task_assets( task, repo=ctx.repo, @@ -1258,12 +1276,14 @@ def run_cell(ctx: TaskCellContext, run_idx: int, arm: str) -> dict[str, Any]: else [] ) ce_mounts = ce_plugin_mounts_for_arm(execution_arm, ctx.ce_plugin_snapshot) + enter_phase("mcp_startup") if isolated_baseline: endpoint = resources.enter_context(isolated_mcp( clone=worktree, mounts=[*dependency_mounts, *gitnexus_mounts], bwrap_bin=ctx.bwrap_bin, )) gitnexus_mounts = [endpoint] runtime_exposure = "stdio-only" + enter_phase("sandbox_setup") with prepare_sandbox( clone=worktree, claude_bin=args.claude_bin, @@ -1346,6 +1366,7 @@ def run_cell(ctx: TaskCellContext, run_idx: int, arm: str) -> dict[str, Any]: before_work_digest = ( implementation_diff_digest(sandbox, orig_sha) if execution_arm in IMPLEMENTATION_ARMS else "" ) + enter_phase("agent_and_verification") record = run_arm( execution_arm, task, @@ -1360,6 +1381,7 @@ def run_cell(ctx: TaskCellContext, run_idx: int, arm: str) -> dict[str, Any]: oracle_snapshot=ctx.oracle_snapshot, mcp_config_json=connector_config() if isolated_baseline else None, ) + enter_phase("artifacts") if execution_arm in ("review", "ce_review"): review_source = review_output_path(sandbox, REVIEW_OUTPUT) if review_source.is_file() and not review_source.is_symlink(): @@ -1433,6 +1455,7 @@ def run_cell(ctx: TaskCellContext, run_idx: int, arm: str) -> dict[str, Any]: detail = redact_text(str(exc), credential_secrets(args)) print(f"[{task['id']}][{arm}][run {run_idx}] infra-error: {detail}") finally: + enter_phase("cleanup") try: resources.close() except (SandboxError, OSError, RuntimeError) as exc: @@ -1453,8 +1476,14 @@ def run_cell(ctx: TaskCellContext, run_idx: int, arm: str) -> dict[str, Any]: record["error_detail"] = ( f"primary={primary_kind}: {primary_detail}; cleanup: {type(cleanup_error).__name__}: {cleanup_error}" )[:2000] + cell_finished = time.monotonic() + phase_seconds["cleanup"] = cell_finished - phase_started record.update( { + "cell_started_at": cell_started_at, + "cell_finished_at": datetime.now(UTC).isoformat(), + "cell_duration_s": cell_finished - cell_started, + "cell_phase_seconds": phase_seconds, "task": task["id"], "class": task.get("class", ""), "run": run_idx, @@ -2105,6 +2134,10 @@ def build_parser() -> argparse.ArgumentParser: "subagent spend — token metrics count only the main loop", ) parser.add_argument("--promotion-min-runs", type=int, default=3) + parser.add_argument( + "--max-budget-usd", type=float, + help="opt-in CLI-reported cost stop per baseline session; provider billing may exceed it", + ) parser.add_argument("--promotion-min-improvement", type=float, default=5.0) parser.add_argument("--promotion-max-task-regression", type=float, default=20.0) parser.add_argument("--task-bindings-json", default=None, help=argparse.SUPPRESS) @@ -2126,6 +2159,9 @@ def main() -> None: parser = build_parser() args = parser.parse_args() try: + args.max_budget_usd = session_budget(args.max_budget_usd) + if args.max_budget_usd is not None and any(arm not in ("baseline", "baseline_nomcp", "baseline_cli") for arm in args.arms): + raise ValueError("--max-budget-usd is limited to single-session baseline arms") args.model = normalized_model_identifier(args.model) args.proposer_model = ( normalized_model_identifier(args.proposer_model, flag="--proposer-model") diff --git a/eval/workflow_bench/runner_sessions.py b/eval/workflow_bench/runner_sessions.py index 808629ccf..309178ef8 100644 --- a/eval/workflow_bench/runner_sessions.py +++ b/eval/workflow_bench/runner_sessions.py @@ -18,6 +18,8 @@ from pathlib import Path, PurePosixPath from typing import Any from .process_control import run_managed +from .session_budget import session_budget +from .session_diagnostics import EventClock, diagnose_session from .proposer_sandbox import ( SANDBOX_GITNEXUS, SANDBOX_GITNEXUS_REGISTRY, @@ -728,6 +730,7 @@ def run_claude( transcript_secrets: tuple[str, ...] = (), plugin_dirs: Sequence[str] = (), progress_label: str | None = None, + max_budget_usd: float | None = None, ) -> dict[str, Any]: """Run one headless session and return its usage record.""" @@ -735,6 +738,7 @@ def run_claude( # every file under the sandbox HOME is writable by agent tools and cannot # serve as trusted evidence. del transcript_projects, transcript_cwd, transcript_wait_seconds + max_budget_usd = session_budget(max_budget_usd) cmd = [ claude_bin, @@ -766,10 +770,13 @@ def run_claude( cmd += ["--model", model] if effort: cmd += ["--effort", effort] + if max_budget_usd is not None: + cmd += ["--max-budget-usd", str(max_budget_usd)] for tool in disallowed_tools or []: cmd += ["--disallowedTools", tool] managed_cmd = [*(command_prefix or []), *cmd] started = time.monotonic() + event_clock = EventClock() with contextlib.ExitStack() as progress_stack: progress = ( progress_stack.enter_context( @@ -781,6 +788,11 @@ def run_claude( if progress_label else None ) + def observe(chunk: bytes) -> None: + event_clock.observe(chunk) + if progress is not None: + progress.observe(chunk) + proc = run_managed( managed_cmd, cwd=None if command_prefix else cwd, @@ -789,17 +801,19 @@ def run_claude( require_pid_namespace=require_pid_namespace, stdin_data=prompt.encode(), capture_stdout_bytes=MAX_TRANSCRIPT_BYTES, - stdout_observer=progress.observe if progress is not None else None, + stdout_observer=observe, ) wall_s = time.monotonic() - started event_stream_error: str | None = None events: list[dict[str, Any]] = [] + event_offsets: list[float] | None = None try: if proc.stdout_capture is None: raise ValueError("parent process did not capture Claude stdout") if proc.stdout_capture_overflow: raise ValueError(f"parent-captured event stream exceeds {MAX_TRANSCRIPT_BYTES} bytes") events = _parse_parent_event_stream(proc.stdout_capture) + event_offsets = event_clock.finish(len(events)) data, result_index = _session_result(events, proc.duration_s) # A session that used a background task drains its bookkeeping after # the final result event (`background_tasks_changed`, `task_updated`, @@ -811,9 +825,12 @@ def run_claude( # a property of what the evidence readers below can see, rather than an # assumption that a `system` event never carries a tool_use block. events = events[: result_index + 1] + if event_offsets is not None: + event_offsets = event_offsets[: result_index + 1] except (UnicodeError, ValueError) as exc: event_stream_error = str(exc) data = {} + event_offsets = None usage = data.get("usage") or {} subtype = data.get("subtype") well_formed = all(field in usage for field in USAGE_FIELDS) @@ -860,6 +877,8 @@ def run_claude( "transcript_missing": False, "tool_use_counts": tool_use_counts, "gitnexus_tool_uses": sum(count for name, count in tool_use_counts.items() if name.startswith("mcp__gitnexus")), + "session_diagnostics": [diagnose_session(events, event_offsets, wall_s)], + "max_budget_usd": max_budget_usd, **{field: usage.get(field, 0) for field in USAGE_FIELDS}, } needs_evidence = expected_skill is not None or transcript_output_dir is not None @@ -955,4 +974,7 @@ def sum_sessions(sessions: list[dict[str, Any]]) -> dict[str, Any]: for name, count in session.get("tool_use_counts", {}).items(): tool_use_counts[name] = tool_use_counts.get(name, 0) + int(count) total["tool_use_counts"] = tool_use_counts + total["session_diagnostics"] = [ + diagnostic for session in sessions for diagnostic in session.get("session_diagnostics", []) + ] return total diff --git a/eval/workflow_bench/session_budget.py b/eval/workflow_bench/session_budget.py new file mode 100644 index 000000000..aa1783fb4 --- /dev/null +++ b/eval/workflow_bench/session_budget.py @@ -0,0 +1,20 @@ +"""Validate opt-in CLI-reported session budgets, not provider billing caps.""" + +from __future__ import annotations + +import math + + +def session_budget(value: float | None) -> float | None: + if value is None: + return None + message = "--max-budget-usd must be finite and greater than zero" + if type(value) not in (int, float): + raise ValueError(message) + try: + budget = float(value) + except OverflowError as exc: + raise ValueError(message) from exc + if not math.isfinite(budget) or budget <= 0: + raise ValueError(message) + return budget diff --git a/eval/workflow_bench/session_diagnostics.py b/eval/workflow_bench/session_diagnostics.py new file mode 100644 index 000000000..387390e3b --- /dev/null +++ b/eval/workflow_bench/session_diagnostics.py @@ -0,0 +1,471 @@ +"""Content-free diagnostics from parent-captured agent events. + +Tool intervals are observed at the stdout boundary, not inside the server. +The next tool action is a correlation, never proof of a workaround or its cost. +""" + +from __future__ import annotations + +import hashlib +import json +import math +import shlex +import time +from collections import Counter +from pathlib import PurePosixPath +from typing import Any + +from .workload_diagnostics import COUNT_FIELDS, TOKEN_FIELDS, public_workload, workload_metrics + +SCHEMA = "gitnexus.session-diagnostics/v1" +MAX_EVENTS = 100_000 +MAX_CALLS = 2048 +CELL_PHASES = ("validation", "clone", "graph_materialize", "dependencies", "mcp_startup", + "sandbox_setup", "agent_and_verification", "artifacts", "cleanup") +COMMANDS = frozenset({ + "list_repos", "query", "context", "check", "impact", "explain", "pdg_query", + "route_map", "tool_map", "shape_check", "api_impact", "trace", "detect_changes", +}) +TOOLS = frozenset({ + "Read", "Grep", "Glob", "Edit", "Write", "Bash", "Skill", "ToolSearch", "other", + *(f"mcp__gitnexus__{name}" for name in COMMANDS), + *(f"cli__gitnexus__{name}" for name in COMMANDS), +}) +USAGE = ("input_tokens", "cache_creation_input_tokens", "cache_read_input_tokens", "output_tokens") +COUNTERS = ( + "calls", "errors", "ambiguous", "not_found", "empty", "unknown_risk", "repeated", + "result_bytes", "max_result_bytes", "source_followups", "unfinished", "output_limit_exceeded", "timed_calls", + "stale", "truncated_results", + "lower_bound", "server_timed_calls", +) + + +class EventClock: + """Record receipt offsets without retaining or parsing event content. + + Called on the stdout reader thread; no I/O, unbounded buffers, or locks. + Several events in one chunk share a receipt time. Missing timing stays unknown. + """ + + def __init__(self) -> None: + self.started = time.monotonic() + self.offsets: list[float] = [] + self.partial = False + self.last = 0.0 + self.overflow = False + + def observe(self, chunk: bytes) -> None: + self.last = time.monotonic() - self.started + parts = chunk.split(b"\n") + for part in parts[:-1]: + if self.partial or part.strip(): + if len(self.offsets) < MAX_EVENTS: + self.offsets.append(self.last) + else: + self.overflow = True + self.partial = False + self.partial = self.partial or bool(parts[-1].strip()) + + def finish(self, expected: int) -> list[float] | None: + offsets = [*self.offsets, *([self.last] if self.partial else [])] + return offsets if not self.overflow and len(offsets) <= MAX_EVENTS and len(offsets) == expected else None + + +def _content(event: dict[str, Any]) -> list[Any]: + message = event.get("message") + content = message.get("content") if isinstance(message, dict) else None + if content is None: + content = event.get("content") + return content if isinstance(content, list) else [] + + +def _has_shell_control(command: str) -> bool: + """Reject operators/expansions while allowing quoted or escaped literals.""" + quote = None + escaped = False + for char in command: + if escaped: + escaped = False + continue + if quote == "'": + if char == "'": + quote = None + continue + if char == "\\": + escaped = True + elif char == '"': + quote = None if quote == '"' else '"' + elif char == "'" and quote is None: + quote = "'" + elif char in "`$" or (quote is None and char in ";|&\n<>()"): + return True + return False + + +def _tool_name(block: dict[str, Any]) -> str: + name = block.get("name") + if name == "Bash" and isinstance(block.get("input"), dict): + # Only a standalone shim command. Compound shell commands retain Bash + # attribution: their elapsed time cannot be called GitNexus server time. + command = block["input"].get("command") + try: + words = shlex.split(command) if isinstance(command, str) else [] + except ValueError: + words = [] + if (len(words) >= 2 and PurePosixPath(words[0]).name == "gitnexus" + and words[1] in COMMANDS and not _has_shell_control(command)): + return f"cli__gitnexus__{words[1]}" + return name if isinstance(name, str) and name in TOOLS else "other" + + +def _result_signals(content: Any) -> dict[str, Any]: + text = content if isinstance(content, str) else "\n".join( + block["text"] for block in content + if isinstance(block, dict) and block.get("type") == "text" and isinstance(block.get("text"), str) + ) if isinstance(content, list) else "" + signals = { + "output_limit_exceeded": "exceeds maximum allowed tokens" in text, + "semantic_error": text.lstrip().lower().startswith("error:"), + } + try: + body = json.loads(text.split("\n\n---", 1)[0]) + except (ValueError, TypeError, RecursionError): + return signals + if not isinstance(body, dict): + return signals + error = body.get("error") + signals["semantic_error"] = isinstance(error, str) + signals["not_found"] = body.get("status") == "not_found" or isinstance(error, str) and "not found" in error.lower() + signals["ambiguous"] = body.get("status") == "ambiguous" + signals["unknown_risk"] = body.get("risk") == "UNKNOWN" + sections = [body[key] for key in ("processes", "process_symbols", "definitions") if key in body] + signals["empty"] = bool(sections) and all(section == [] for section in sections) + signals["truncated"] = body.get("truncated") is True or body.get("partial") is True + signals["lower_bound"] = body.get("epistemic") == "lower-bound" + timing = body.get("timing") + wall = timing.get("wall") if isinstance(timing, dict) else None + if type(wall) in (int, float) and 0 <= wall <= 10**12: + signals["server_wall_ms"] = wall + staleness = body.get("staleness") + if isinstance(staleness, dict) and staleness.get("status") in ("current", "behind", "diverged", "unknown"): + signals["staleness"] = staleness["status"] + return signals + + +def _union_seconds(intervals: list[tuple[float, float]]) -> float: + total, end = 0.0, 0.0 + for start, stop in sorted(intervals): + total += max(0.0, stop - max(end, start)) + end = max(end, stop) + return round(total, 6) + + +def diagnose_session( + events: list[dict[str, Any]], offsets: list[float] | None, duration_s: float, +) -> dict[str, Any]: + """Explain observed tool work; never estimate tool dollars or causal savings.""" + timing = offsets is not None and len(offsets) == len(events) and all( + type(value) in (int, float) and math.isfinite(value) and 0 <= value <= duration_s + and (index == 0 or value >= offsets[index - 1]) for index, value in enumerate(offsets) + ) + calls: list[dict[str, Any]] = [] + inputs: dict[int, Any] = {} + results: dict[int, Any] = {} + pending: dict[str, dict[str, Any]] = {} + signatures: dict[str, int] = {} + turns: dict[str, dict[str, Any]] = {} + retries = 0 + truncated = False + duplicate_ids = False + seen_ids: set[str] = set() + for index, event in enumerate(events): + if event.get("type") == "system" and event.get("subtype") == "api_retry": + retries += 1 + if event.get("type") == "assistant": + message = event.get("message") or {} + # Streaming can repeat a message id. Retain its last usage snapshot + # rather than adding cumulative snapshots as separate model requests. + if isinstance(message, dict) and isinstance(message.get("usage"), dict): + key = str(message.get("id", index)) + if key in turns or len(turns) < MAX_CALLS: + turns[key] = {"event_index": index, **{ + field: value for field in USAGE + if type(value := message["usage"].get(field)) is int and value >= 0 + }} + else: + truncated = True + for block in _content(event): + if not isinstance(block, dict) or block.get("type") != "tool_use": + continue + if len(calls) >= MAX_CALLS: + truncated = True + continue + tool_id = block.get("id") + if not isinstance(tool_id, str) or not 0 < len(tool_id) <= 256: + duplicate_ids = True + continue + if tool_id in seen_ids: + duplicate_ids = True + continue + seen_ids.add(tool_id) + name = _tool_name(block) + encoded = json.dumps(block.get("input"), sort_keys=True, ensure_ascii=False, separators=(",", ":")) + signature = hashlib.sha256((name + encoded).encode(errors="backslashreplace")).hexdigest() + call = { + "index": len(calls), "tool": name, "request_event": index, + "started_s": offsets[index] if timing else None, + "completed_s": None, "duration_s": None, "result_event": None, + "input_bytes": len(encoded.encode(errors="backslashreplace")), "result_bytes": None, + "repeat_of": signatures.get(signature), "outcome": "unfinished", + "source_followups": 0, + } + signatures[signature] = call["index"] + pending[tool_id] = call + calls.append(call) + inputs[call["index"]] = block.get("input") + elif event.get("type") == "user": + for block in _content(event): + if not isinstance(block, dict) or block.get("type") != "tool_result": + continue + tool_id = block.get("tool_use_id") + call = pending.pop(tool_id, None) if isinstance(tool_id, str) else None + if call is None: + duplicate_ids = True + continue + content = block.get("content") + results[call["index"]] = content + call["result_bytes"] = len(json.dumps( + content, ensure_ascii=False, separators=(",", ":"), + ).encode(errors="backslashreplace")) + call["result_event"] = index + if timing: + call["completed_s"] = offsets[index] + call["duration_s"] = round(offsets[index] - call["started_s"], 6) + call["outcome"] = "error" if block.get("is_error") is True else "ok" + if "__gitnexus__" in call["tool"]: + call.update(_result_signals(content)) + if call.get("semantic_error") or call.get("output_limit_exceeded"): + call["outcome"] = "error" + # Only inspect the immediately following assistant action; this is a + # temporal signal, not evidence that GitNexus caused the source read. + next_actions: dict[int, list[str]] = {} + next_tools: list[str] = [] + for index in range(len(events) - 1, -1, -1): + next_actions[index] = next_tools + if events[index].get("type") == "assistant": + next_tools = [_tool_name(b) for b in _content(events[index]) + if isinstance(b, dict) and b.get("type") == "tool_use"] + tools: dict[str, Counter] = {} + intervals, mcp_intervals = [], [] + for call in calls: + following = next_actions.get(call["result_event"], []) + call["next_tools"] = following + call["source_followups"] = sum(name in ("Read", "Grep", "Glob") for name in following) + totals = tools.setdefault(call["tool"], Counter({**dict.fromkeys(COUNTERS, 0), "elapsed_s": 0.0, "server_wall_s": 0.0})) + totals["calls"] += 1 + totals["errors"] += call["outcome"] == "error" + totals["unfinished"] += call["outcome"] == "unfinished" + totals["repeated"] += call["repeat_of"] is not None + totals["result_bytes"] += call["result_bytes"] or 0 + totals["max_result_bytes"] = max(totals["max_result_bytes"], call["result_bytes"] or 0) + for field in ("ambiguous", "not_found", "empty", "unknown_risk", "output_limit_exceeded", "lower_bound"): + totals[field] += call.get(field) is True + if "server_wall_ms" in call: + totals["server_timed_calls"] += 1 + totals["server_wall_s"] += call["server_wall_ms"] / 1000 + totals["stale"] += call.get("staleness") in ("behind", "diverged") + totals["truncated_results"] += call.get("truncated") is True + call["problematic"] = call["outcome"] == "error" or any( + call.get(field) for field in ("not_found", "ambiguous", "empty", "unknown_risk") + ) + totals["source_followups"] += call["source_followups"] if call["problematic"] else 0 + if call["duration_s"] is not None: + totals["timed_calls"] += 1 + totals["elapsed_s"] += call["duration_s"] + interval = (call["started_s"], call["completed_s"]) + intervals.append(interval) + if call["tool"].startswith("mcp__gitnexus__"): + mcp_intervals.append(interval) + complete = bool(events) and timing and not (pending or truncated or duplicate_ids) + wait = _union_seconds(intervals) if complete else None + return { + "schema": SCHEMA, "timing_complete": complete, "truncated": truncated, + "tool_wait_s": wait, "mcp_wait_s": _union_seconds(mcp_intervals) if complete else None, + "other_wall_s": round(max(0.0, duration_s - wait), 6) if wait is not None else None, + "api_retries": retries, "tools": { + name: {**values, "elapsed_s": round(values["elapsed_s"], 6) if values["timed_calls"] else None, + "server_wall_s": round(values["server_wall_s"], 6) if values["server_timed_calls"] else None} + for name, values in tools.items() + }, "calls": calls, "assistant_usage": list(turns.values()), + "workload": workload_metrics( + [(call, inputs.get(call["index"]), results.get(call["index"])) for call in calls], + sorted(turns.values(), key=lambda turn: turn["event_index"]), events, + ), + } + + +def public_diagnostics(value: Any) -> list[dict[str, Any]]: + """Idempotent metadata whitelist for public summaries and bounded outliers.""" + output = [] + for session in value[:16] if isinstance(value, list) else []: + if not isinstance(session, dict) or session.get("schema") != SCHEMA: + continue + safe: dict[str, Any] = {"schema": SCHEMA} + for field in ("timing_complete", "truncated"): + safe[field] = session.get(field) is True + for field in ("tool_wait_s", "mcp_wait_s", "other_wall_s", "api_retries"): + number = session.get(field) + safe[field] = number if type(number) in (int, float) and 0 <= number <= 10**12 else None + safe["tools"] = {} + if "workload" in session: + safe["workload"] = public_workload(session["workload"]) + tools = session.get("tools") + if isinstance(tools, dict): + for name in sorted(TOOLS): + row = tools.get(name) + if not isinstance(row, dict): + continue + safe["tools"][name] = { + field: number if type(number := row.get(field)) is int and 0 <= number <= 10**12 else None + for field in COUNTERS + } + for field in ("elapsed_s", "server_wall_s"): + elapsed = row.get(field) + safe["tools"][name][field] = elapsed if type(elapsed) in (int, float) and 0 <= elapsed <= 10**12 else None + # Keep a few inspectable outliers in the already-uploaded release JSON. + # Full calls and assistant usage remain in local results.jsonl. + for field, sort_key in ( + ("largest_results", "result_bytes"), ("slowest_calls", "duration_s"), ("problem_calls", "result_bytes"), + ): + candidates = session.get(field, session.get("calls", [])) + safe[field] = [] + if not isinstance(candidates, list): + continue + for call in sorted( + [c for c in candidates[:MAX_CALLS] if isinstance(c, dict) + and (field != "problem_calls" or c.get("problematic") is True) + and type(c.get(sort_key)) in (int, float) and 0 <= c[sort_key] <= 10**12], + key=lambda c: c[sort_key], reverse=True, + )[:5]: + tool = call.get("tool") + item = {"tool": tool if isinstance(tool, str) and tool in TOOLS else "other"} + for name in ("index", "request_event", "result_event", "input_bytes", "result_bytes", "duration_s", "source_followups", "server_wall_ms"): + number = call.get(name) + item[name] = number if type(number) in (int, float) and 0 <= number <= 10**12 else None + outcome = call.get("outcome") + item["outcome"] = outcome if outcome in ("ok", "error", "unfinished") else "unknown" + for name in ("problematic", "not_found", "ambiguous", "empty", "unknown_risk", "output_limit_exceeded"): + item[name] = call.get(name) is True + safe[field].append(item) + output.append(safe) + return output + + +def public_cell_diagnostics(row: dict[str, Any]) -> dict[str, Any]: + """Shared publication boundary for release and optional-arm artifacts.""" + safe = {} + if "session_diagnostics" in row: + safe["session_diagnostics"] = public_diagnostics(row["session_diagnostics"]) + if "cell_duration_s" in row: + value = row["cell_duration_s"] + safe["cell_duration_s"] = value if type(value) in (int, float) and 0 <= value <= 10**12 else None + phases = row.get("cell_phase_seconds") + if isinstance(phases, dict): + safe["cell_phase_seconds"] = { + field: value if type(value := phases[field]) in (int, float) and 0 <= value <= 10**12 else None + for field in CELL_PHASES if field in phases + } + return safe + + +def render_diagnostics(rows: list[dict[str, Any]]) -> list[str]: + """Add tool diagnostics to the existing public report without new CI wiring.""" + lines = [] + for arm in ("baseline_nomcp", "baseline", "baseline_cli"): + cells = [row for row in rows if row.get("arm") == arm] + sessions = [s for row in cells for s in public_diagnostics(row.get("session_diagnostics"))] + if not sessions: + continue + measured = sum(bool(public_diagnostics(row.get("session_diagnostics"))) for row in cells) + lines += ["", f"### Tool diagnostics: {arm}", "", + f"Diagnostics available for {measured}/{len(cells)} cells.", "", + "| Tool | Calls | Errors | Ambiguous | Not found | Empty | UNKNOWN risk | Repeated args | Result bytes | Largest result | Source follow-ups after problems | Timed calls | Elapsed seconds |", + "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |"] + totals: dict[str, Counter] = {} + coverage: dict[str, Counter] = {} + tool_sessions: Counter = Counter() + for session in sessions: + for name, values in session["tools"].items(): + tally = totals.setdefault(name, Counter()) + known = coverage.setdefault(name, Counter()) + tool_sessions[name] += 1 + for field, number in values.items(): + if number is not None: + known[field] += 1 + tally[field] = max(tally[field], number) if field == "max_result_bytes" else tally[field] + number + for name, tally in sorted(totals.items()): + lines.append("| " + " | ".join(str(v) for v in ( + name, *(tally[f] if coverage[name][f] == tool_sessions[name] else "n/a" for f in + ("calls", "errors", "ambiguous", "not_found", "empty", "unknown_risk", + "repeated", "result_bytes", "max_result_bytes", "source_followups", "timed_calls")), + f"{tally['elapsed_s']:.1f}" if tally["timed_calls"] else "n/a", + )) + " |") + lines += ["", "| Tool | Stale | Truncated results | Lower bound | Output limit | Unfinished | Server-timed calls | Server seconds |", + "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: |"] + for name, tally in sorted(totals.items()): + lines.append("| " + " | ".join(str(v) for v in ( + name, *(tally[f] if coverage[name][f] == tool_sessions[name] else "n/a" for f in + ("stale", "truncated_results", "lower_bound", "output_limit_exceeded", "unfinished", "server_timed_calls")), + f"{tally['server_wall_s']:.3f}" if tally["server_timed_calls"] else "n/a", + )) + " |") + complete = [s for s in sessions if s["timing_complete"] and s["tool_wait_s"] is not None] + lines += ["", f"Complete host timing: {len(complete)}/{len(sessions)} sessions."] + limited = sum(s["truncated"] for s in sessions) + if limited: + lines.append(f"Diagnostics truncated in {limited} session(s); tool counts are lower bounds.") + if complete: + lines.append(f"Observed tool wait: {sum(s['tool_wait_s'] for s in complete):.1f}s; " + f"MCP wait: {sum(s['mcp_wait_s'] or 0 for s in complete):.1f}s " + "(a subset of tool wait). Parallel intervals are counted once per session.") + for field, label in (("other_wall_s", "Other session wall time (seconds)"), ("api_retries", "Observed API retries")): + values = [s[field] for s in sessions if s.get(field) is not None] + lines.append(f"{label}: {sum(values):.1f} across {len(values)}/{len(sessions)} sessions." + if values else f"{label}: n/a.") + workloads = [s["workload"] for s in sessions if s.get("workload")] + if workloads: + lines += ["", f"Workload signals available for {len(workloads)}/{len(sessions)} sessions.", "", + "| Workload signal | Observed count | Sessions reporting |", "| --- | ---: | ---: |"] + for field in COUNT_FIELDS: + values = [w[field] for w in workloads if w.get(field) is not None] + lines.append(f"| {field} | {sum(values) if values else 'n/a'} | {len(values)}/{len(sessions)} |") + peaks = [w["max_input_tokens"] for w in workloads if w.get("max_input_tokens") is not None] + if peaks: + lines.append(f"\nLargest observed CLI input context: {max(peaks):,} tokens. " + "Uses complete Anthropic-shaped usage snapshots; provider-native usage may differ.") + lines += ["", "| Context/cache signal | Session minimum | Session maximum | Sessions reporting |", + "| --- | ---: | ---: | ---: |"] + for field in TOKEN_FIELDS: + values = [w[field] for w in workloads if w.get(field) is not None] + lines.append(f"| {field} | {min(values) if values else 'n/a'} | {max(values) if values else 'n/a'} | " + f"{len(values)}/{len(sessions)} |") + lines += ["", "
Largest, slowest, and problematic tool calls", "", + "Session and call indexes identify observations; arguments and source content are omitted.", "", + "| View | Session | Call | Tool | Result bytes | Host seconds | Outcome |", + "| --- | ---: | ---: | --- | ---: | ---: | --- |"] + for view, sort_key in (("largest_results", "result_bytes"), ("slowest_calls", "duration_s"), + ("problem_calls", "result_bytes")): + calls = [(index, call) for index, session in enumerate(sessions) for call in session[view]] + for index, call in sorted(calls, key=lambda item: item[1][sort_key], reverse=True)[:5]: + duration = call["duration_s"] + lines.append(f"| {view} | {index} | {call['index']} | {call['tool']} | {call['result_bytes']} | " + f"{f'{duration:.3f}' if duration is not None else 'n/a'} | {call['outcome']} |") + lines += ["", "
"] + if lines: + lines += ["", ("Receipt timing includes transport and scheduling, not just server execution. Other wall time is not model time. " + "Bytes are serialized payload bytes, not tokens. Repeated arguments may be justified after edits. " + "Source follow-ups are the next assistant action after a problematic result, not proof of a workaround. " + "Per-tool elapsed sums may overlap. Session cost includes all work; no per-tool dollar attribution is available.")] + lines += [("A file read after a graph reference is an association, not duplicated content or low relevance. " + "Repeated verification can be necessary. Bash classification is conservative; mixed commands stay unclassified. " + "Missing usage fields remain unknown. Context totals describe observed snapshots only.")] + return lines diff --git a/eval/workflow_bench/workload_diagnostics.py b/eval/workflow_bench/workload_diagnostics.py new file mode 100644 index 000000000..7d2b56fc9 --- /dev/null +++ b/eval/workflow_bench/workload_diagnostics.py @@ -0,0 +1,247 @@ +"""Content-free workload signals. Associations are not causal cost attribution.""" + +from __future__ import annotations + +import json +import shlex +from collections import Counter +from pathlib import PurePosixPath +from typing import Any + +# Also the publication allowlist. No arguments, paths, source, or hashes leave +# this module. Unknown/missing observations must not become measured zeros. +COUNT_FIELDS = ( + "source_reads", "repeated_source_reads", "reads_after_graph_reference", + "reads_after_graph_content_reference", "query_context_read_chains", + "graph_calls", "graph_calls_before_first_edit", "edits", "edits_after_verification", + "verification_calls", "verification_repeated_without_observed_change", + "verification_repeated_after_possible_change", "verification_errors", + "bash_test", "bash_typecheck", "bash_lint", "bash_build", "bash_source", + "bash_mutation", "bash_mixed", "bash_other", "compactions", + "usage_snapshots", "complete_input_snapshots", "usage_after_graph_snapshots", + "reference_scan_limited", +) +TOKEN_FIELDS = ( + "first_input_tokens", "last_input_tokens", "max_input_tokens", "sum_input_tokens", + "sum_cache_read_tokens", "sum_cache_write_tokens", "sum_output_tokens", + "input_tokens_after_graph", +) +FIELDS = (*COUNT_FIELDS, *TOKEN_FIELDS) +VERIFY = frozenset({"test", "typecheck", "lint", "build"}) +_UV_RUN_FLAGS = frozenset({"--locked", "--frozen", "--no-sync", "--offline", "--all-extras", + "--no-dev", "--only-dev", "--isolated", "--active", "--no-project"}) +_UV_RUN_VALUES = frozenset({"--extra", "--group", "--python", "--project", "--directory", + "--package", "--with"}) + + +def classify_bash(command: Any) -> str: + """Conservative lexical classification, never shell evaluation. + + Compound commands are explicitly mixed; their time cannot be assigned to + one test/build. A common `cd dir && command` prefix is the sole exception. + """ + if not isinstance(command, str): + return "other" + try: + words = shlex.split(command) + except ValueError: + return "other" + if len(words) >= 4 and words[0] == "cd" and words[2] == "&&": + words = words[3:] + # Inspect the remainder too: shell substitutions and pipes stay mixed. + command = shlex.join(words) + if any(op in command for op in (";", "|", "&", "\n", "`", "$(", ">", "<")): + return "mixed" + if not words: + return "other" + executable = PurePosixPath(words[0]).name + args = words[1:] + if executable == "uv" and args[:1] == ["run"]: + args = args[1:] + while args and args[0].startswith("-"): + option = args.pop(0) + if option == "--": + break + name, separator, value = option.partition("=") + if name in _UV_RUN_FLAGS and not separator: + continue + if name in _UV_RUN_VALUES: + if separator and value: + continue + if not separator and args and args[0] and not args[0].startswith("-"): + args.pop(0) + continue + # Unknown or malformed options may change what uv executes. + return "other" + return classify_bash(shlex.join(args)) + if executable in ("python", "python3") and args[:2] == ["-m", "pytest"]: + return "test" + if executable in ("npm", "pnpm", "yarn", "bun", "npx") and args: + if args[0] in ("run", "exec", "x"): + args = args[1:] + executable = args[0] if args else "" + if executable in ("test", "pytest", "vitest", "jest") or executable.startswith("test:"): + return "test" + if executable in ("tsc", "mypy", "pyright", "typecheck"): + return "typecheck" + if executable in ("eslint", "ruff", "prettier", "lint"): + mutates = any(arg in ("--fix", "--write", "-w") for arg in args) + if executable == "ruff" and "format" in args and not any(arg in ("--check", "--diff") for arg in args): + mutates = True + return "mutation" if mutates else "lint" + if executable in ("build", "vite", "tsup"): + return "build" + if executable in ("rg", "grep", "cat", "head", "tail", "sed"): + return "mutation" if executable == "sed" and any( + arg.startswith(("-i", "--in-place")) for arg in args + ) else "source" + if executable in ("apply_patch", "patch", "touch", "rm", "mv", "cp", "tee"): + return "mutation" + return "other" + + +def _references(content: Any) -> tuple[dict[str, bool], bool]: + """Structured result references only; cap traversal of large tool results.""" + if isinstance(content, list): + text = "\n".join(b["text"] for b in content + if isinstance(b, dict) and isinstance(b.get("text"), str)) + else: + text = content if isinstance(content, str) else "" + if len(text) > 2_000_000: + return {}, True + try: + body = json.loads(text.split("\n\n---", 1)[0]) + except (ValueError, RecursionError): + return {}, False + stack, paths, visited = [body], {}, 0 + while stack and visited < 10_000 and len(paths) < 1000: + item = stack.pop() + visited += 1 + if isinstance(item, dict): + path = item.get("filePath", item.get("file_path")) + if isinstance(path, str) and path and len(path) <= 4096: + # Same-object source only. Descriptions and arbitrary text are + # not evidence that the response supplied the file's contents. + supplied = any(isinstance(item.get(key), str) and bool(item[key].strip()) + for key in ("content", "source", "code")) + paths[path] = paths.get(path, False) or supplied + stack.extend(value for value in item.values() if isinstance(value, (dict, list))) + elif isinstance(item, list): + stack.extend(item) + return paths, bool(stack) + + +def _same_path(read: str, reference: str) -> bool: + read_parts, ref_parts = PurePosixPath(read).parts, PurePosixPath(reference).parts + if ".." in read_parts or ".." in ref_parts: + return False + if read_parts == ref_parts: + return True + # Absolute sandbox paths can match repo-relative graph paths. Never match + # unrelated absolute roots or a basename-only relative reference. + return (len(ref_parts) > 1 and not reference.startswith("/") + and len(read_parts) >= len(ref_parts) and read_parts[-len(ref_parts):] == ref_parts) + + +def workload_metrics( + observations: list[tuple[dict[str, Any], Any, Any]], + usage: list[dict[str, Any]], events: list[dict[str, Any]], +) -> dict[str, int | None]: + """Compute bounded numeric signals from accepted, deduplicated calls. + + `observations` holds (call metadata, input, result content) only in memory. + No assumption that an unknown Bash action left the checkout unchanged. + """ + counts = Counter(dict.fromkeys(COUNT_FIELDS, 0)) + counts["compactions"] = sum(e.get("type") == "system" and e.get("subtype") == "compact_boundary" for e in events) + reads: set[str] = set() + checks: dict[str, int] = {} + mutation_epoch = 0 + verified = False + graph_results: list[tuple[int, str, dict[str, bool]]] = [] + reference_budget = 5000 + for call, args, content in observations: + args = args if isinstance(args, dict) else {} + tool = call["tool"] + if "__gitnexus__" in tool: + counts["graph_calls"] += 1 + counts["graph_calls_before_first_edit"] += counts["edits"] == 0 + if call.get("result_event") is not None: + paths, limited = _references(content) + if len(paths) > reference_budget: + paths = dict(list(paths.items())[:reference_budget]) + limited = True + reference_budget -= len(paths) + counts["reference_scan_limited"] += limited + graph_results.append((call["result_event"], tool.rsplit("__", 1)[-1], paths)) + elif tool == "Read": + counts["source_reads"] += 1 + signature = json.dumps(args, sort_keys=True, ensure_ascii=True) + counts["repeated_source_reads"] += signature in reads + reads.add(signature) + path = args.get("file_path") + prior = sorted((g for g in graph_results if g[0] < call["request_event"]), key=lambda g: g[0]) + if isinstance(path, str): + matching = [supplied for _, _, paths in prior for ref, supplied in paths.items() if _same_path(path, ref)] + counts["reads_after_graph_reference"] += bool(matching) + counts["reads_after_graph_content_reference"] += any(matching) + counts["query_context_read_chains"] += ( + len(prior) >= 2 and [g[1] for g in prior[-2:]] == ["query", "context"] + and any(_same_path(path, ref) for ref in prior[-1][2]) + ) + elif tool in ("Edit", "Write"): + counts["edits"] += 1 + counts["edits_after_verification"] += verified + mutation_epoch += 1 + elif tool == "Bash": + category = classify_bash(args.get("command")) + counts[f"bash_{category}"] += 1 + if category in VERIFY: + counts["verification_calls"] += 1 + counts["verification_errors"] += call["outcome"] == "error" + signature = json.dumps(args, sort_keys=True, ensure_ascii=True) + if signature in checks: + field = "verification_repeated_without_observed_change" if checks[signature] == mutation_epoch else "verification_repeated_after_possible_change" + counts[field] += 1 + checks[signature] = mutation_epoch + verified = True + elif category in ("mutation", "mixed", "other"): + mutation_epoch += 1 + if category == "mutation": + counts["edits"] += 1 + counts["edits_after_verification"] += verified + counts["usage_snapshots"] = len(usage) + input_fields = ("input_tokens", "cache_read_input_tokens", "cache_creation_input_tokens") + totals = [sum(u[f] for f in input_fields) + if all(type(u.get(f)) is int and u[f] >= 0 for f in input_fields) else None for u in usage] + complete = [total for total in totals if total is not None] + counts["complete_input_snapshots"] = len(complete) + after_graph = [] + for u, total in zip(usage, totals, strict=True): + preceding = [call for call, _, _ in observations + if call.get("result_event") is not None and call["result_event"] < u["event_index"]] + latest = max((call["result_event"] for call in preceding), default=-1) + if any("__gitnexus__" in c["tool"] and c["result_event"] == latest for c in preceding): + after_graph.append(total) + counts["usage_after_graph_snapshots"] = len(after_graph) + tokens = { + "first_input_tokens": totals[0] if totals else None, + "last_input_tokens": totals[-1] if totals else None, + "max_input_tokens": max(complete) if complete else None, + "sum_input_tokens": sum(complete) if totals and len(complete) == len(totals) else None, + "input_tokens_after_graph": sum(after_graph) if after_graph and all(t is not None for t in after_graph) else None, + } + for dest, source in (("sum_cache_read_tokens", "cache_read_input_tokens"), + ("sum_cache_write_tokens", "cache_creation_input_tokens"), ("sum_output_tokens", "output_tokens")): + # A partial sum labelled as total would understate spend. These fields + # stay unknown unless every observed usage snapshot supplies the field. + tokens[dest] = sum(u[source] for u in usage) if usage and all( + type(u.get(source)) is int and u[source] >= 0 for u in usage + ) else None + return {**counts, **tokens} + + +def public_workload(value: Any) -> dict[str, int | None]: + if not isinstance(value, dict): + return {} + return {field: n if type(n := value.get(field)) is int and 0 <= n <= 10**15 else None for field in FIELDS} diff --git a/gitnexus/test/unit/staleness.test.ts b/gitnexus/test/unit/staleness.test.ts index c54d53d02..8e856d582 100644 --- a/gitnexus/test/unit/staleness.test.ts +++ b/gitnexus/test/unit/staleness.test.ts @@ -123,7 +123,7 @@ describe('checkStalenessAsync', () => { expect(result.commitsBehind).toBe(0); }); - it('parallel calls complete faster than sequential', async () => { + it('concurrent calls agree with the synchronous lookup', async () => { let headCommit: string; try { headCommit = execFileSync('git', ['rev-parse', 'HEAD'], { @@ -137,19 +137,11 @@ describe('checkStalenessAsync', () => { const cwd = process.cwd(); const N = 10; - // Parallel - const t0 = performance.now(); - await Promise.all(Array.from({ length: N }, () => checkStalenessAsync(cwd, headCommit))); - const parallelMs = performance.now() - t0; - - // Sequential sync - const t1 = performance.now(); - for (let i = 0; i < N; i++) checkStaleness(cwd, headCommit); - const sequentialMs = performance.now() - t1; - - // Parallel should be meaningfully faster than sequential. - // Use a generous ratio to avoid flakiness on slow CI machines. - expect(parallelMs).toBeLessThan(sequentialMs * 1.5); + const expected = checkStaleness(cwd, headCommit); + const results = await Promise.all( + Array.from({ length: N }, () => checkStalenessAsync(cwd, headCommit)), + ); + expect(results).toEqual(Array.from({ length: N }, () => expected)); }); });