feat(eval): measure cost drivers in evaluation summaries (#3773)

* feat(eval): add cost diagnostics and trials

* style(eval): format retrieval case fixtures

* fix(eval): address review and CI failures

Refs #3773. Validate locked uv commands, model aliases and oversized budgets. Stop the fake MCP server before releasing its listener and replace the wall-clock staleness assertion with a concurrent result check.

Note: pre-existing failure in test_oracle_assets.py reference-checkout type resolution not addressed by this PR.

* feat(eval): publish detailed run measurements

Collect per-sweep provider ledgers and publish sanitized measurements in the Actions summary and artifacts. Validate experiment tasks with the runner and withhold incomplete token totals. Refs #3773.

Note: pre-existing failure in test_oracle_assets.py reference-checkout type resolution not addressed by this PR.

* fix(eval): validate inputs and quoted commands

Reject non-mapping task YAML before reading fields. Attribute literal quoted CLI queries while preserving Bash attribution for shell operators and expansions. Refs #3773.

* fix(eval): count empty filtered query results

* fix(eval): handle alternate event content

---------

Co-authored-by: Gergo Magyar <gergomagyar0@gmail.com>
This commit is contained in:
Gergő Magyar 2026-10-30 12:58:34 +00:00 • committed by GitHub
parent cfe418cca0
commit f6127c5aee
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
23 changed files with 2628 additions and 20 deletions

View file

@ -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<details><summary>Detailed measurements: %s</summary>\n\n' "$variant"
cat "$public/$variant-diagnostics.md"
printf '\n</details>\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

View file

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

View file

@ -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": "</code><img src=x>|bad![x](https://example.test)"}])
markdown = render_markdown(report)
assert "<img" not in markdown and "&#124;" in markdown
assert "![x]" not in markdown
def test_summary_does_not_turn_missing_tool_counters_into_zero():
from workflow_bench.session_diagnostics import SCHEMA, render_diagnostics
body = "\n".join(render_diagnostics([{"arm": "baseline", "session_diagnostics": [
{"schema": SCHEMA, "tools": {"Read": {"calls": 1}}},
]}]))
assert "| Read | 1 | n/a |" in body
assert "Other session wall time (seconds): n/a." in body
def test_summary_treats_unrepresentable_numbers_as_unknown():
report = build_cost_report([{"arm": "baseline", "task": "one", "cost_usd": 10 ** 1000}])
assert report["groups"][0]["metrics"]["cost_usd"]["observed"] == 0
assert report["groups"][0]["metrics"]["cost_usd"]["missing"] == 1
@pytest.mark.skipif(os.name == "nt", reason="release summary executes in bash on the Linux runner")
@pytest.mark.parametrize(("pr", "expected"), [("", {"stable", "candidate", "cli", "deferred"}),
("3773", {"candidate", "cli"})])
def test_actual_workflow_summary_step_publishes_each_enabled_sweep(tmp_path, pr, expected):
repository = Path(__file__).resolve().parents[2]
workflow = yaml.safe_load((repository / ".github/workflows/release-evaluation.yml").read_text())
step = next(step for job in workflow["jobs"].values() for step in job.get("steps", [])
if step.get("name") == "Summarize time, cost, and tool measurements")
# Run the workflow's actual shell and report command with this test's Python;
# dependency provisioning and paid model commands are outside this step.
bin_dir = tmp_path / "bin"
bin_dir.mkdir()
uv = bin_dir / "uv"
uv.write_text('#!/bin/sh\n[ "$1 $2 $3 $4 $5" = "run --locked --extra dev python" ] || exit 90\n'
'shift 5\nexec "$SUMMARY_TEST_PYTHON" "$@"\n')
uv.chmod(0o755)
root = tmp_path / "release-evaluation"
for runtime, directory, ledger in (("stable", "raw", "provider-usage.jsonl"),
("candidate", "raw", "provider-usage.jsonl"),
("candidate", "cli-raw", "cli-provider-usage.jsonl"),
("candidate", "deferred", "deferred-provider-usage.jsonl")):
out = root / runtime / directory
out.mkdir(parents=True)
out.joinpath("results.jsonl").write_text(json.dumps({
"arm": "baseline_cli" if directory == "cli-raw" else "baseline", "task": "one",
"run": 0, "cost_usd": 2, "duration_s": 5, "resolved": True, "ok": True,
}) + "\n")
(root / runtime / ledger).write_text(json.dumps(provider_event() | {"request": "SECRET"}) + "\n")
summary = tmp_path / "summary.md"
completed = subprocess.run(["bash", "-euo", "pipefail", "-c", step["run"]], cwd=repository / "eval",
env={**os.environ, "PATH": f"{bin_dir}:{os.environ['PATH']}",
"SUMMARY_TEST_PYTHON": sys.executable, "RUNNER_TEMP": str(tmp_path),
"GITHUB_STEP_SUMMARY": str(summary), "INPUT_PR": pr,
"CLI_ARM": "true", "DEFERRED_ARM": "true"},
capture_output=True, text=True, timeout=60)
assert completed.returncode == 0, completed.stderr
assert {p.name.removesuffix("-diagnostics.json") for p in (root / "public").glob("*.json")} == expected
body = summary.read_text()
for variant in expected:
assert f"Detailed measurements: {variant}" in body
assert json.loads((root / "public" / f"{variant}-diagnostics.json").read_text())["provider"]["observed_requests"] == 1
assert "SECRET" not in body

View file

@ -170,7 +170,12 @@ def test_a_request_through_the_real_gateway_records_native_usage(tmp_path, monke
) as gateway:
request = urllib.request.Request(
gateway.base_url + "/v1/messages",
data=json.dumps({"model": "gpt-4.1", "max_tokens": 32, "messages": [{"role": "user", "content": "ping"}]}).encode(),
data=json.dumps({
"model": "gpt-4.1", "max_tokens": 32,
"messages": [{"role": "user", "content": "PRIVATE REQUEST"}],
"tools": [{"name": "inspect_file", "description": "PRIVATE TOOL DESCRIPTION",
"input_schema": {"type": "object", "properties": {"path": {"type": "string"}}}}],
}).encode(),
headers={"Content-Type": "application/json", "x-api-key": gateway.auth_token, "anthropic-version": "2023-06-01"},
)
with urllib.request.urlopen(request, timeout=60):
@ -189,6 +194,11 @@ def test_a_request_through_the_real_gateway_records_native_usage(tmp_path, monke
assert native["prompt_tokens_details"]["cache_write_tokens"] == 1_000
assert event["provider"] == LITELLM_NORMALIZED
assert event["call_type"] == "anthropic_messages", "the observed call type, not a Responses one"
metrics = event["request_metrics"]
assert metrics["messages_count"] == metrics["tools_count"] == 1
assert metrics["messages_bytes"] > 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

View file

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

View file

@ -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"]}

View file

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

View file

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

View file

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

View file

@ -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']}: <code>{label(group['arm'])}</code> / <code>{label(group['task'])}</code>", "",
(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'}.", "",
"<details><summary>Measurements and cohort</summary>", "",
"| Cohort field | Value |", "| --- | --- |",
*[f"| {field} | <code>{label(value) if value else 'unknown'}</code> |"
for field, value in group["cohort"].items()], "",
*table(group["metrics"]), "", "Cell phases (seconds; agent_and_verification includes grading):", "",
*table(group["cell_phases"]), "", "</details>", ""]
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()

View file

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

View file

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

View file

@ -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"]], ""])

View file

@ -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", [])),
"",
]
)

View file

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

View file

@ -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"
}
]
}
]
}

View file

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

View file

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

View file

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

View file

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

View file

@ -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 += ["", "<details><summary>Largest, slowest, and problematic tool calls</summary>", "",
"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 += ["", "</details>"]
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

View file

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

View file

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