scriptoriumadmin/GitNexus
1
0
Fork 0
mirror of https://github.com/abhigyanpatwari/GitNexus.git synced 2026-11-11 08:37:35 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

fix(eval): validate trial setup and report missing measurements (#3779)

Browse source 
* fix(eval): validate trials and report measurements

Keep pinned task dependencies separate from the candidate runtime. Gate paid workflow sessions on contained reference grading and retain private-safe setup, verification, provenance and provider-token measurements.

* fix(eval): align CI with the release task build

* fix(eval): address review feedback (#3779)

Bind trial dependency bytes before execution and include MCP environment instructions in comparator identity. Preserve verifier states and strengthen preflight coverage.

---------

Co-authored-by: Gergo Magyar <gergomagyar0@gmail.com>
This commit is contained in:
Gergő Magyar 2026-10-31 04:40:37 +00:00 • committed by GitHub
parent 7af4a274e1
commit c06f3ee32d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
21 changed files with 745 additions and 51 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
.github/workflows/ci-tests.yml vendored
View file

@ -1534,10 +1534,8 @@ jobs:
sha="$(uv run --locked --extra dev python -m workflow_bench.release_report task-sha)"
[[ "$sha" =~ ^[0-9a-f]{40}$ ]]
git -C .. worktree add --detach "${RUNNER_TEMP}/tasks-base" "$sha"
cd "${RUNNER_TEMP}/tasks-base"
npm ci --audit=false --fund=false
npm ci --prefix gitnexus --audit=false --fund=false
npm run build --prefix gitnexus
uv run --locked --extra dev python -m workflow_bench.release_build \
--repo "${RUNNER_TEMP}/tasks-base"
- name: Prove process-tree and sandbox containment
env:
CLAUDE_CANARY_BIN: ${{ runner.temp }}/claude-canary/node_modules/@anthropic-ai/claude-code-linux-x64/claude
@ -1553,6 +1551,7 @@ jobs:
tests/test_mock_provider.py
tests/test_mcp_tool_loading.py
tests/test_release_evaluation_smoke.py
tests/test_release_controls.py
tests/test_evolve.py::test_outer_runner_pid_namespace_kills_setsid_descendant
tests/test_oracle_assets.py::test_hidden_vitest_config_executes_sibling_oracle_against_candidate_checkout
tests/test_oracle_assets.py::test_shipped_oracles_reject_task_base_and_accept_reference_fixes

6
.github/workflows/release-evaluation.yml vendored
View file

@ -394,11 +394,11 @@ jobs:
run: |
uv run --locked --extra dev python -m workflow_bench.diagnostic_experiments \
--tasks "$RUNNER_TEMP/release-evaluation/candidate/tasks.yaml" \
--repository "$GITHUB_WORKSPACE/candidate" --runtime "$GITHUB_WORKSPACE/candidate" \
--repository "$GITHUB_WORKSPACE/tasks-base" --runtime "$GITHUB_WORKSPACE/candidate" \
--harness-sha "$GITHUB_SHA" --runtime-sha "$RUNTIME_SHA" \
--model "$MODEL" --effort "$EFFORT" --seed 3762 --repeats 1 \
--variants no_graph mcp_eager --task-ids trivial-status-json-alias \
--max-budget-usd 5 --out "$RUNNER_TEMP/release-evaluation/trial"
--max-budget-usd 5 --bind-task-dependencies --out "$RUNNER_TEMP/release-evaluation/trial"
- name: Require historical task dependencies before paid PR sessions
if: inputs.candidate_pr != ''
working-directory: eval
@ -425,6 +425,7 @@ jobs:
if [[ "${DIAGNOSTIC_TRIAL:-false}" == true ]]; then
uv run --locked --extra dev python -m workflow_bench.release_preflight \
--tasks "$RUNNER_TEMP/release-evaluation/trial/cell-0000.tasks.json" \
--grading-controls \
--gitnexus-root "$GITHUB_WORKSPACE/candidate" \
--claude-bin "$RUNNER_TEMP/claude-canary/node_modules/@anthropic-ai/claude-code-linux-x64/claude"
exit 0
@ -433,6 +434,7 @@ jobs:
if [[ -n "$INPUT_PR" && "$runtime" == stable ]]; then continue; fi
uv run --locked --extra dev python -m workflow_bench.release_preflight \
--tasks "$RUNNER_TEMP/release-evaluation/$runtime/tasks.yaml" \
--grading-controls \
--gitnexus-root "$GITHUB_WORKSPACE/$runtime" \
--claude-bin "$RUNNER_TEMP/claude-canary/node_modules/@anthropic-ai/claude-code-linux-x64/claude"
done

10
eval/tests/test_diagnostic_trial.py
View file

@ -24,16 +24,20 @@ SHA = "a" * 40
def trial_plan(tmp_path, monkeypatch):
runtime = tmp_path / "candidate"
runtime.mkdir()
task_repo = tmp_path / "tasks-base"
task_repo.mkdir()
monkeypatch.setattr(diagnostic_experiments, "trial_task_bindings", lambda tasks, expected=None: [{"id": tasks[0]["id"]}])
output = tmp_path / "release-evaluation" / "trial"
monkeypatch.setattr(sys, "argv", [
"prepare", "--tasks", str(ROOT / "eval/workflow_bench/tasks.scenarios.yaml"),
"--repository", str(runtime), "--runtime", str(runtime), "--runtime-sha", SHA,
"--repository", str(task_repo), "--runtime", str(runtime), "--runtime-sha", SHA,
"--harness-sha", SHA, "--model", diagnostic_trial.MODEL, "--repeats", "1",
"--variants", "no_graph", "mcp_eager", "--task-ids", diagnostic_trial.TASK,
"--max-budget-usd", "5", "--out", str(output),
"--max-budget-usd", "5", "--bind-task-dependencies", "--out", str(output),
])
diagnostic_experiments.main()
monkeypatch.setattr(diagnostic_trial, "_head", lambda repository: SHA)
monkeypatch.setattr(diagnostic_trial, "_head", lambda repository:
"c4ecf398de637cf779ab2238e69958cfd5d601ce" if repository == task_repo else SHA)
return output / "plan.json", runtime

71
eval/tests/test_release_controls.py Normal file
View file

@ -0,0 +1,71 @@
"""Reference grading must work before a paid release or telemetry session starts."""
from contextlib import nullcontext
import os
from pathlib import Path
from types import SimpleNamespace
import pytest
import yaml
from workflow_bench import release_controls
from workflow_bench.proposer_sandbox import SandboxError, preflight_bubblewrap
from workflow_bench.task_assets import TaskAssetCache
@pytest.mark.parametrize("failure", [None, "negative_passes", "startup", "permission", "patch", "typecheck", "reference"])
def test_controls_require_behavioral_failure_then_a_typed_reference(monkeypatch, tmp_path, failure):
task = {"id": "trivial-status-json-alias"}
clone = tmp_path / "clone"
clone.mkdir()
mounts, calls = [], []
sandbox = SimpleNamespace(clone=clone, environment=lambda: {},
command_prefix_for=lambda **kw: mounts.append(kw) or ["sandbox"])
monkeypatch.setattr(release_controls, "make_worktree", lambda *_: clone)
monkeypatch.setattr(release_controls, "sanitize_clone_for_hidden_oracles", lambda *_: None)
monkeypatch.setattr(release_controls, "stage_task_assets", lambda *_a, **_kw: [])
monkeypatch.setattr(release_controls, "capture_task_oracle", lambda *_: object())
monkeypatch.setattr(release_controls, "prepare_sandbox", lambda **_: nullcontext(sandbox))
def oracle(*_args, diagnostics=None):
calls.append("negative" if diagnostics is not None else "reference")
if diagnostics is not None:
diagnostics.update(exit_code=1, assertion_failure=True, startup_failure=failure == "startup",
permission_failure=failure == "permission")
return failure == "negative_passes", "hidden output withheld"
return failure != "reference", "hidden output withheld"
def verify(command, *_args, **_kwargs):
step = "patch" if command.startswith("git apply") else "typecheck"
calls.append(step)
return failure != step, "PRIVATE OUTPUT"
monkeypatch.setattr(release_controls, "_run_hidden_oracle", oracle)
monkeypatch.setattr(release_controls, "run_verify", verify)
if failure:
with pytest.raises(SandboxError) as error:
release_controls.check_task_controls(task, repo=tmp_path, sha="a" * 40, parent=tmp_path,
assets=object(), claude_bin=Path("claude"), bwrap_bin=Path("bwrap"))
assert "PRIVATE" not in str(error.value)
else:
release_controls.check_task_controls(task, repo=tmp_path, sha="a" * 40, parent=tmp_path,
assets=object(), claude_bin=Path("claude"), bwrap_bin=Path("bwrap"))
assert calls == ["negative", "patch", "typecheck", "reference"]
assert mounts[-1] == {"read_only_workspace": True, "unshare_network": True}
if failure in ("negative_passes", "startup", "permission"):
assert calls == ["negative"]
@pytest.mark.skipif(os.environ.get("GITNEXUS_REQUIRE_BWRAP_CANARY") != "1",
reason="requires the provisioned Linux containment runner")
def test_real_release_reference_controls_use_pinned_dependencies_and_read_only_grading(tmp_path):
# No provider, proxy, or agent process. The pinned CLI executable is only a
# sandbox mount; grading invokes the historical TypeScript/Vitest toolchain.
repo = Path(os.environ["GITNEXUS_ORACLE_TASK_DEPENDENCIES"])
claude = Path(os.environ["CLAUDE_CANARY_BIN"])
tasks = yaml.safe_load((Path(__file__).parents[1] / "workflow_bench/tasks.scenarios.yaml").read_text())["tasks"]
task = next(t for t in tasks if t["id"] == "trivial-status-json-alias")
with TaskAssetCache(tmp_path / "assets") as cache:
snapshot = cache.prepare(task, repo=repo, resolved_sha=task["ref"])
release_controls.check_task_controls(task, repo=repo, sha=task["ref"], parent=tmp_path,
assets=snapshot, claude_bin=claude, bwrap_bin=preflight_bubblewrap())

25
eval/tests/test_release_preflight.py
View file

@ -39,6 +39,7 @@ def prepared(monkeypatch, tmp_path):
monkeypatch.setattr(release_preflight, "preflight_bubblewrap", lambda: Path("/fake/bwrap"))
monkeypatch.setattr(release_preflight, "require_claude_sandbox_helpers", lambda: None)
monkeypatch.setattr(release_preflight, "trusted_gitnexus_runtime_mounts", lambda **_: ())
monkeypatch.setattr(release_preflight, "check_task_controls", lambda *_a, **_kw: None)
monkeypatch.setattr(task_assets, "_try_reflink", lambda *_: False)
# Only graph construction/containment are substituted. Binding resolution,
@ -101,3 +102,27 @@ def test_preflight_rejects_task_assets_that_replace_the_graph(prepared, tmp_path
release_preflight.preflight_release(
[{**task, "sandbox_copy": [".gitnexus/meta.json"]}], gitnexus_root=tmp_path, claude_bin=Path("claude"),
)
def test_grading_control_failure_stops_preflight_with_captured_task_dependencies(prepared, monkeypatch, tmp_path):
task, _, _ = prepared
dependency = Path(task["repo"]) / "node_modules" / "fixture"
dependency.mkdir(parents=True)
(dependency / "index.js").write_bytes(b"module.exports = 'pinned dependency';\n")
task["sandbox_dependencies"] = [{"source": "node_modules", "target": "node_modules"}]
calls = []
def reject(task_arg, **kwargs):
calls.append(kwargs)
assets = kwargs["assets"]
assert task_arg == task
assert assets.dependency_declarations == (("node_modules", "node_modules"),)
assert len(assets.dependencies) == 1
captured = assets.root / assets.dependencies[0].snapshot_path / "fixture/index.js"
assert captured.read_bytes() == b"module.exports = 'pinned dependency';\n"
raise SandboxError("reference grading control failed")
monkeypatch.setattr(release_preflight, "check_task_controls", reject)
with pytest.raises(SandboxError, match="reference grading control failed"):
release_preflight.preflight_release([task], gitnexus_root=tmp_path, claude_bin=Path("claude"), grading_controls=True)
assert len(calls) == 1 and calls[0]["sha"] == task["ref"]

15
eval/tests/test_reuse_round_trip.py
View file

@ -214,7 +214,7 @@ def _expectation(**overrides: Any) -> ComparatorReuseExpectation:
binding = TaskReuseBinding(
task_base_sha=SHA,
task_prompt_digest=runner.hashlib.sha256(b"review it").hexdigest(),
task_prompt_digest=runner.task_prompt_digest({"prompt": "review it"}),
oracle_digest="oracle-content",
oracle_command_digest="oracle-command",
oracle_manifest_digest="oracle-manifest",
@ -258,7 +258,7 @@ def test_a_changed_binding_rejects_the_same_emitted_row(
binding = TaskReuseBinding(
task_base_sha=SHA,
task_prompt_digest=runner.hashlib.sha256(b"review it").hexdigest(),
task_prompt_digest=runner.task_prompt_digest({"prompt": "review it"}),
oracle_digest="oracle-content",
oracle_command_digest="oracle-command",
oracle_manifest_digest="oracle-manifest",
@ -266,3 +266,14 @@ def test_a_changed_binding_rejects_the_same_emitted_row(
sandbox_dependency_manifest_digest="DIFFERENT-dependencies",
)
assert row_is_reusable_comparator(row, _expectation(tasks={TASK_ID: binding})) is False
def test_old_prompt_identity_cannot_reuse_a_row_without_the_environment_instruction(emitted_row):
legacy = {**emitted_row, "task_prompt_digest": runner.hashlib.sha256(b"review it").hexdigest()}
assert not row_is_reusable_comparator(legacy, _expectation())
def test_changed_environment_instruction_invalidates_a_previously_matching_row(emitted_row, monkeypatch):
assert row_is_reusable_comparator(emitted_row, _expectation())
monkeypatch.setattr(runner, "MCP_REPOSITORY_PROMPT", runner.MCP_REPOSITORY_PROMPT + " Changed instruction.")
assert not row_is_reusable_comparator(emitted_row, _expectation())

101
eval/tests/test_trial_dependency_binding.py Normal file
View file

@ -0,0 +1,101 @@
"""Prepared trials must reject dependency drift before launching paid cells."""
import json
from pathlib import Path
import subprocess
import sys
from typing import get_args
import pytest
import yaml
from tests import test_diagnostic_trial
from workflow_bench import diagnostic_experiments, diagnostic_trial
from workflow_bench.process_control import ProcessState
from workflow_bench.verification_diagnostics import public_verification
@pytest.fixture
def bound_trial(tmp_path, monkeypatch):
repo, runtime = tmp_path / "tasks-base", tmp_path / "runtime"
repo.mkdir()
runtime.mkdir()
(repo / ".gitignore").write_text("node_modules/\n")
(repo / "package-lock.json").write_text('{"lockfileVersion": 3}\n')
for args in (
["init", "-q"], ["add", "."],
["-c", "user.name=Test", "-c", "user.email=test@example.test", "commit", "-qm", "fixture"],
):
subprocess.run(["git", "-C", str(repo), *args], check=True, capture_output=True)
sha = subprocess.check_output(["git", "-C", str(repo), "rev-parse", "HEAD"], text=True).strip()
dependency = repo / "node_modules" / "fixture.js"
dependency.parent.mkdir()
dependency.write_text("module.exports = 'pinned';\n")
corpus = yaml.safe_load((Path(__file__).parents[1] / "workflow_bench/tasks.scenarios.yaml").read_text())
task = next(task for task in corpus["tasks"] if task["id"] == diagnostic_trial.TASK)
task.update(ref=sha, sandbox_dependencies=[{"source": "node_modules", "target": "node_modules"}])
manifest = tmp_path / "tasks.yaml"
manifest.write_text(yaml.safe_dump({"tasks": [task]}))
output = tmp_path / "trial"
monkeypatch.setattr(sys, "argv", [
"prepare", "--tasks", str(manifest), "--repository", str(repo), "--runtime", str(runtime),
"--runtime-sha", test_diagnostic_trial.SHA, "--harness-sha", test_diagnostic_trial.SHA,
"--model", diagnostic_trial.MODEL, "--repeats", "1", "--variants", "no_graph", "mcp_eager",
"--max-budget-usd", "5", "--bind-task-dependencies", "--out", str(output),
])
diagnostic_experiments.main()
monkeypatch.setattr(diagnostic_trial, "_head", lambda path: sha if path == repo else test_diagnostic_trial.SHA)
return output / "plan.json", runtime, repo, dependency
def test_prepared_bindings_reach_both_runner_processes(bound_trial, monkeypatch):
path, runtime, _, _ = bound_trial
plan = json.loads(path.read_text())
calls = []
monkeypatch.setattr(diagnostic_trial.subprocess, "run", test_diagnostic_trial.fake_runner(calls))
assert diagnostic_trial.execute_trial(path, runtime, "claude") == 0
assert len(calls) == 2
for (command, _), cell in zip(calls, plan["cells"], strict=True):
bound = json.loads(command[command.index("--task-bindings-json") + 1])
assert bound == cell["task_bindings"]
assert bound[0]["sandbox_dependency_content_digest"]
@pytest.mark.parametrize("change", ["tracked", "staged", "modified_dependency", "new_dependency", "deleted_dependency", "missing_binding"])
def test_trial_rejects_checkout_or_dependency_drift_before_execution(bound_trial, monkeypatch, change):
path, runtime, repo, dependency = bound_trial
if change in ("tracked", "staged"):
(repo / "package-lock.json").write_text('{"lockfileVersion": 2}\n')
if change == "staged":
subprocess.run(["git", "-C", str(repo), "add", "package-lock.json"], check=True)
elif change == "modified_dependency":
dependency.write_text("module.exports = 'changed';\n")
elif change == "new_dependency":
(dependency.parent / "extra.js").write_text("unexpected dependency\n")
elif change == "deleted_dependency":
dependency.unlink()
else:
plan = json.loads(path.read_text())
del plan["cells"][0]["task_bindings"]
path.write_text(json.dumps(plan))
monkeypatch.setattr(diagnostic_trial.subprocess, "run", lambda *_a, **_kw: pytest.fail("paid executor reached"))
with pytest.raises(ValueError, match="clean pinned|definition drifted|pin task dependency"):
diagnostic_trial.execute_trial(path, runtime, "claude")
assert not (path.parent / "execution.json").exists()
def test_planning_rejects_dirty_dependency_source_before_pinning(bound_trial):
path, _, repo, _ = bound_trial
(repo / "package-lock.json").write_text("dirty\n")
tasks = json.loads((path.parent / "cell-0000.tasks.json").read_text())["tasks"]
with pytest.raises(ValueError, match="clean pinned"):
diagnostic_experiments.trial_task_bindings(tasks)
@pytest.mark.parametrize("state", get_args(ProcessState))
def test_public_verification_preserves_every_terminal_process_state(state):
assert public_verification({"state": state})["state"] == state
def test_public_verification_withholds_unknown_process_state():
assert public_verification({"state": "private unrecognized output"})["state"] == "unknown"

190
eval/tests/test_trial_measurement_fixes.py Normal file
View file

@ -0,0 +1,190 @@
"""Regressions from the first paid two-cell telemetry trial. No model calls."""
import hashlib
import json
import os
from pathlib import Path
import subprocess
import sys
import pytest
from tests import test_diagnostic_trial
from tests.test_workflow_bench_sessions import _probe_cell_provisioning
from workflow_bench import cost_diagnostics, diagnostic_experiments, diagnostic_trial, runner
from workflow_bench.process_control import ManagedProcessResult
from workflow_bench.runner_artifacts import VerificationResult
from workflow_bench.runner_sessions import allowed_agent_tools
from workflow_bench.session_diagnostics import diagnose_session, public_cell_diagnostics
from workflow_bench.verification_diagnostics import public_verification, verification_metrics
trial_plan = test_diagnostic_trial.trial_plan
def test_real_trial_planner_command_keeps_task_dependencies_separate_from_runtime(tmp_path, monkeypatch):
step = next(s for s in test_diagnostic_trial.workflow()["jobs"]["evaluate"]["steps"] if s.get("name") == "Prepare the capped diagnostic trial")
capture = tmp_path / "argv.json"
uv = tmp_path / "uv"
uv.write_text(f"#!{sys.executable}\nimport json,os,sys\nopen(os.environ['CAPTURE'],'w').write(json.dumps(sys.argv[1:]))\n")
uv.chmod(0o755)
subprocess.run(["bash", "-c", step["run"]], check=True, env={
**os.environ, "PATH": f"{tmp_path}:{os.defpath}", "CAPTURE": str(capture),
"RUNNER_TEMP": str(tmp_path), "GITHUB_WORKSPACE": str(tmp_path),
"GITHUB_SHA": "a" * 40, "RUNTIME_SHA": "b" * 40, "MODEL": diagnostic_trial.MODEL, "EFFORT": "xhigh",
})
argv = json.loads(capture.read_text())
assert "--bind-task-dependencies" in argv
monkeypatch.setattr(diagnostic_experiments, "trial_task_bindings", lambda tasks: [{"id": tasks[0]["id"]}])
manifest = Path(argv[argv.index("--tasks") + 1])
manifest.parent.mkdir(parents=True)
source = Path(__file__).parents[1] / "workflow_bench/tasks.scenarios.yaml"
manifest.write_bytes(source.read_bytes())
monkeypatch.setattr(sys, "argv", ["prepare", *argv[argv.index("workflow_bench.diagnostic_experiments") + 1:]])
diagnostic_experiments.main()
plan_path = tmp_path / "release-evaluation/trial/plan.json"
plan = json.loads(plan_path.read_text())
for cell in plan["cells"]:
task = json.loads((plan_path.parent / f"{cell['cell_id']}.tasks.json").read_text())["tasks"][0]
assert task["repo"] == str(tmp_path / "tasks-base")
assert task["repo"] == plan["task_repository"]
assert task["ref"] == "c4ecf398de637cf779ab2238e69958cfd5d601ce"
assert cell["argv"][cell["argv"].index("--gitnexus-root") + 1] == str(tmp_path / "candidate")
def test_old_candidate_dependency_override_fails_before_paid_executor(trial_plan, monkeypatch):
path, runtime = trial_plan
plan = json.loads(path.read_text())
plan["task_repository"] = str(runtime)
for cell in plan["cells"]:
manifest = path.parent / f"{cell['cell_id']}.tasks.json"
doc = json.loads(manifest.read_text())
doc["tasks"][0]["repo"] = str(runtime)
manifest.write_text(json.dumps(doc))
cell["task_manifest_sha256"] = hashlib.sha256(manifest.read_bytes()).hexdigest()
path.write_text(json.dumps(plan))
monkeypatch.setattr(diagnostic_trial.subprocess, "run", lambda *_a, **_k: pytest.fail("paid executor reached"))
with pytest.raises(ValueError, match="pinned task checkout"):
diagnostic_trial.execute_trial(path, runtime, "claude")
assert not (path.parent / "execution.json").exists()
def test_placeholder_cli_usage_does_not_claim_zero_context():
fields = ("input_tokens", "cache_read_input_tokens", "cache_creation_input_tokens", "output_tokens")
work = diagnose_session([{"type": "assistant", "message": {
"id": "placeholder", "usage": dict.fromkeys(fields, 0), "content": [{"type": "text", "text": "done"}],
}}], [1], 2)["workload"]
assert work["usage_snapshots"] == 1 and work["complete_input_snapshots"] == 0
for field in ("first_input_tokens", "max_input_tokens", "sum_input_tokens", "sum_output_tokens"):
assert work[field] is None
def test_provider_request_context_uses_native_cache_semantics_and_withholds_conflicts():
event = {"response_id": "r1", "provider": "litellm-normalized", "native_usage": {
"prompt_tokens": 100, "prompt_tokens_details": {"cached_tokens": 80}, "completion_tokens": 10,
}}
report = cost_diagnostics.provider_summary([event, event])
assert report["request_tokens"]["total_input_tokens"]["max"] == 100
assert report["request_tokens"]["cache_read_input_tokens"]["max"] == 80
assert report["request_tokens"]["total_input_tokens"]["observed"] == 1
conflicted = cost_diagnostics.provider_summary([event, {**event, "status": "failure"}])
assert conflicted["request_tokens"]["total_input_tokens"]["max"] is None
def test_grading_evidence_survives_publication_without_hidden_output():
result = VerificationResult("private command", ManagedProcessResult(
state="exited", returncode=1, stdout_tail="PRIVATE hidden source", stderr_tail="EROFS", duration_s=2.5,
), "PRIVATE hidden source\nAssertionError: private test\nEROFS")
evidence = verification_metrics(result, 2.5)
public = public_cell_diagnostics({"verification_diagnostics": {"hidden": evidence | {"output": result.output}}})
assert "PRIVATE" not in json.dumps(public)
assert public["verification_diagnostics"]["hidden"]["exit_code"] == 1
assert public["verification_diagnostics"]["hidden"]["assertion_failure"] is True
assert public_verification({"state": "PRIVATE", "exit_code": True, "duration_s": float("nan")})["exit_code"] is None
def test_failed_cells_keep_task_provenance_and_budget(monkeypatch, tmp_path):
ctx, *_ = _probe_cell_provisioning(monkeypatch, tmp_path)
ctx.args.max_budget_usd = 5
ctx.args.measurement_revisions = {"runtime_sha": "b" * 40, "harness_sha": "c" * 40}
row = runner.run_cell(ctx, 0, "baseline_nomcp")
assert row["error_kind"] == "infra-error"
assert row["runtime_sha"] == "b" * 40 and row["harness_sha"] == "c" * 40
assert row["max_budget_usd"] == 5 and row["task_base_sha"] == ctx.task_sha
assert row["task_prompt_digest"] and row["model"] == ctx.args.model
def test_mcp_read_tools_are_permitted_and_measured_without_mutating_permissions():
tools = allowed_agent_tools(implementation=True)
assert {"mcp__gitnexus__grep", "mcp__gitnexus__read_file"} <= set(tools)
assert "mcp__gitnexus__rename" not in tools and "mcp__gitnexus__cypher" not in tools
events = [{"type": "assistant", "message": {"content": [
{"type": "tool_use", "id": "grep", "name": "mcp__gitnexus__grep", "input": {}},
]}}]
assert diagnose_session(events, [0], 1)["workload"]["graph_calls"] == 1
def test_cost_summary_displays_setup_grading_and_provider_tokens(tmp_path, monkeypatch):
results = tmp_path / "results.jsonl"
results.write_text(json.dumps({"arm": "baseline", "task": "fixture", "verification_diagnostics": {
"hidden": {"passed": False, "state": "exited", "exit_code": 1, "duration_s": 3, "typecheck_failure": True},
}}) + "\n")
timing = {"complete": True, "workers": 1, "total_s": 600, "before_first_cell_s": 500,
"dependency_setup_s": 100, "foreground_graph_setup_s": 400, "private": "PRIVATE"}
(tmp_path / "sweep-timing.json").write_text(json.dumps(timing))
out, markdown = tmp_path / "public.json", tmp_path / "public.md"
monkeypatch.setattr(sys, "argv", ["diagnostics", "--results", str(results), "--out", str(out), "--markdown-out", str(markdown)])
cost_diagnostics.main()
assert json.loads(out.read_text())["sweep_timing"]["before_first_cell_s"] == 500
assert "PRIVATE" not in out.read_text()
text = markdown.read_text()
assert "before_first_cell_s | 500" in text and "typecheck_failure" in text
def test_release_preflight_requires_grading_controls_in_both_workflow_paths():
step = next(s for s in test_diagnostic_trial.workflow()["jobs"]["evaluate"]["steps"] if s.get("name") == "Materialize both release graphs before paid sessions")
assert step["run"].count("workflow_bench.release_preflight") == step["run"].count("--grading-controls") == 2
def test_public_release_rows_keep_comparable_provenance_and_budget():
from tests.test_release_report import metadata, rows
from workflow_bench.release_report import build_report
source = rows()
for row in source:
row.update(runtime_sha="a" * 40, harness_sha="c" * 40, task_prompt_digest="d" * 64,
max_budget_usd=5.0, mcp_tool_loading="eager" if row["arm"] == "baseline" else "none")
exported = build_report(source, metadata())["per_run"]
report = cost_diagnostics.build_cost_report(exported)
assert all(group["provenance_complete"] for group in report["groups"])
assert all(group["cohort"]["max_budget_usd"] == "5.0" for group in report["groups"])
assert all(group["cohort"]["runtime_digest"] == "" for group in report["groups"])
def test_mcp_session_prompt_names_the_isolated_repository(monkeypatch, tmp_path):
from tests.test_workflow_bench_sessions import VALID_REPORT, bench_args, fake_cli_result, fake_sandbox
from workflow_bench import runner_sessions
prompts = []
def execute(command, **kwargs):
prompts.append(kwargs["stdin_data"].decode())
return fake_cli_result(VALID_REPORT)
monkeypatch.setattr(runner_sessions, "run_managed", execute)
monkeypatch.setattr(runner, "run_verify", lambda *_a, **_k: (True, "ok"))
for arm in ("baseline", "baseline_nomcp"):
runner.run_arm(arm, {"prompt": "Repair the task", "verify": "true"}, tmp_path,
bench_args(), sandbox=fake_sandbox(tmp_path))
assert "benchmark-target" in prompts[0]
assert "benchmark-target" not in prompts[1]
def test_interrupted_timing_does_not_discard_cell_evidence(tmp_path, monkeypatch):
results, out = tmp_path / "results.jsonl", tmp_path / "public.json"
results.write_text('{"arm":"baseline_nomcp","task":"fixture"}\n')
(tmp_path / "sweep-timing.json").write_text('{"complete":')
monkeypatch.setattr(sys, "argv", ["diagnostics", "--results", str(results), "--out", str(out), "--allow-incomplete"])
cost_diagnostics.main()
report = json.loads(out.read_text())
assert report["cells"] == 1 and report["sweep_timing"] is None
assert "Sweep timing file is unreadable." in report["input_problems"]

7
eval/tests/test_workflow_bench.py
View file

@ -255,6 +255,9 @@ def test_eval_ci_uses_locked_uv_and_blocking_native_containment_jobs():
"npm ci",
"npm run build",
]
task_build = containment_steps["Build the pinned task toolchain that grades release evaluation"]["run"]
assert "python -m workflow_bench.release_build" in task_build
assert '--repo "${RUNNER_TEMP}/tasks-base"' in task_build
selected_containment_tests = containment_steps["Prove process-tree and sandbox containment"]["run"].split()
assert selected_containment_tests == [
"uv",
@ -281,6 +284,7 @@ def test_eval_ci_uses_locked_uv_and_blocking_native_containment_jobs():
# Paired prepare/session/oracle/report composition uses that same native
# CLI and containment; it must execute here rather than skip everywhere.
"tests/test_release_evaluation_smoke.py",
"tests/test_release_controls.py",
# These also need runtime dependencies absent from the locked pytest job.
"tests/test_evolve.py::test_outer_runner_pid_namespace_kills_setsid_descendant",
"tests/test_oracle_assets.py::test_hidden_vitest_config_executes_sibling_oracle_against_candidate_checkout",
@ -298,7 +302,8 @@ def test_eval_ci_uses_locked_uv_and_blocking_native_containment_jobs():
for path in (repo_root / "eval" / "tests").glob("test_*.py")
if bwrap_canary_marker.search(path.read_text())
)
assert bwrap_canary_files == ["test_cli_shim.py", "test_proposer_sandbox.py", "test_workflow_bench_sessions.py"]
assert bwrap_canary_files == ["test_cli_shim.py", "test_proposer_sandbox.py", "test_release_controls.py",
"test_workflow_bench_sessions.py"]
assert all(f"tests/{name}" in selected_containment_tests for name in bwrap_canary_files)
assert "eval-containment-windows:" in workflow

31
eval/workflow_bench/README.md
View file

@ -1016,13 +1016,29 @@ 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 \
--repository /path/to/tasks-base --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
--max-budget-usd 5 --bind-task-dependencies --out /tmp/gitnexus-diagnostics-trial
```
`--repository` must be the checkout built at the task's pinned commit, including
its dependency trees. `--runtime` is the separately built GitNexus candidate.
`--bind-task-dependencies` requires a clean tracked checkout at the task ref and
records the built dependency contents. The trial executor rejects missing bindings,
dirty tracked files, a different HEAD, or dependency drift. Each runner receives
the saved bindings and checks its captured dependencies again before model calls.
The release workflow also runs offline negative and reference grading controls
with those dependency snapshots and the runner's read-only verifier sandbox.
The unmodified source must fail a behavioral assertion; the reference repair
must pass the hidden oracle and TypeScript. A setup crash is not a negative
control. These checks make no model requests.
The recorded task prompt digest includes the conditional MCP environment
instruction. Comparator reuse rejects older raw-prompt hashes and changes to
that instruction, while keeping the two arms under the same task identity.
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
@ -1064,7 +1080,7 @@ new dispatch after failure; the executor never retries or resumes a used plan.
| 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 |
| Setup, indexing and infrastructure | Sweep wall time, time before the first cell, dependency preparation, foreground graph setup/waits, cell phases, authored/hidden grading durations | Background graph work can overlap cells; CPU/RSS, workflow provisioning, proxy startup 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:
@ -1078,6 +1094,15 @@ uv run --locked python -m workflow_bench.cost_diagnostics \
Unknown fields remain null. Distinct model/effort/source/prompt/runtime/loading
cohorts are kept separate. Duplicate exported cell identities are rejected.
Runtime and harness commit SHAs, task prompt digest, and the per-cell CLI budget
survive public export. The legacy evolution `runtime_digest` is optional when
the required commit provenance is present. All-zero CLI usage snapshots are
treated as unavailable; provider-normalized per-request token distributions
give a separate view of context size without guessing from aggregate usage.
When `sweep-timing.json` exists beside the results, the summary includes its
setup measurements. They overlap and must not be added together. Grading
receipts expose duration, exit code, process state, and fixed failure markers;
they never expose hidden-test output. Failure markers are clues, not diagnoses.
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

56
eval/workflow_bench/cost_diagnostics.py
View file

@ -13,10 +13,11 @@ from typing import Any
from .provider_usage import NormalizedUsage, UsageSemanticsError, normalize_usage
from .session_diagnostics import CELL_PHASES, public_diagnostics, render_diagnostics
from .verification_diagnostics import public_verification
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",
COHORT = ("model", "effort", "task_base_sha", "task_prompt_digest", "runtime_sha", "harness_sha", "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")
@ -90,6 +91,9 @@ def provider_summary(events: list[dict[str, Any]]) -> dict[str, Any]:
"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_tokens": {field: distribution([
row.get(field) if not conflicts and not unidentified else None for row in normalized
]) for field in token_fields},
"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",
@ -131,7 +135,15 @@ def build_cost_report(rows: list[dict[str, Any]], provider_events: list[dict[str
"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),
# runtime_digest is a legacy evolution launcher field. Release
# measurements bind the actual runtime and harness commits instead.
"provenance_complete": all(value for field, value in zip(COHORT, cohort, strict=True)
if field != "runtime_digest"),
"verification": {
phase: [public_verification(row.get("verification_diagnostics", {}).get(phase))
for row in cells if isinstance(row.get("verification_diagnostics"), dict)]
for phase in ("authored", "hidden")
},
"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),
@ -197,6 +209,27 @@ def render_markdown(report: dict[str, Any]) -> str:
for field, value in group["cohort"].items()], "",
*table(group["metrics"]), "", "Cell phases (seconds; agent_and_verification includes grading):", "",
*table(group["cell_phases"]), "", "</details>", ""]
lines += ["Grading checks (failure markers are observations, not proven root causes):", "",
"| Check | Passed | Exit code | State | Seconds | Observed failure markers |",
"| --- | --- | --- | --- | ---: | --- |"]
for phase, checks in group["verification"].items():
for check in checks:
markers = ", ".join(field for field in ("assertion_failure", "startup_failure", "permission_failure", "typecheck_failure")
if check.get(field) is True)
lines.append(f"| {phase} | {check.get('passed', 'unknown')} | {check.get('exit_code')} | "
f"{check.get('state', 'unknown')} | {number(check.get('duration_s'))} | {markers or 'none observed'} |")
lines += [""]
timing = report.get("sweep_timing")
if timing:
lines += ["### Runner setup and wall time", "",
f"Complete sweep: {'yes' if timing['complete'] else 'no'}; workers: {number(timing['workers'])}.",
("Times overlap: before-first-cell includes dependency and foreground graph setup. "
"Graph time includes foreground waits; background indexing can overlap cells. "
"Total excludes CLI/proxy startup and report rendering."), "",
"| Measurement | Seconds |", "| --- | ---: |",
*[f"| {field} | {number(timing[field])} |" for field in (
"total_s", "before_first_cell_s", "dependency_setup_s", "foreground_graph_setup_s",
)], ""]
provider = report["provider"]
lines += ["### Provider requests", ""]
if provider is None:
@ -211,6 +244,8 @@ def render_markdown(report: dict[str, Any]) -> str:
"| Provider token field | Observed total |", "| --- | ---: |",
*[f"| {field} | {number(value)} |" for field, value in provider["token_totals"].items()], "",
*table(provider["request_metrics"]), "",
"Provider-reported tokens per request (includes cached input; independent of CLI context snapshots):", "",
*table(provider["request_tokens"]), "",
*provider["limitations"], ""]
lines += [report["tool_markdown"], "", *report["limitations"], ""]
return "\n".join(lines)
@ -265,6 +300,23 @@ def main() -> None:
provider = None
problems += provider_problems
report = build_cost_report(rows, provider)
timing_path = args.results.parent / "sweep-timing.json"
report["sweep_timing"] = None
if timing_path.is_file():
try:
timing = json.loads(timing_path.read_text())
except (OSError, ValueError):
if not args.allow_incomplete:
raise
timing = None
problems.append("Sweep timing file is unreadable.")
if isinstance(timing, dict):
report["sweep_timing"] = {
"complete": timing.get("complete") is True,
**{field: timing.get(field) if _number(timing.get(field)) else None for field in (
"total_s", "before_first_cell_s", "dependency_setup_s", "foreground_graph_setup_s", "workers",
)},
}
if report["provider"] is not None:
report["provider"]["ledger_complete"] = not provider_problems
if provider_problems:

27
eval/workflow_bench/diagnostic_experiments.py
View file

@ -13,7 +13,8 @@ from typing import Any
import yaml
from .runner_tasks import normalized_model_identifier, select_tasks
from .process_control import run_checked
from .runner_tasks import normalized_model_identifier, resolve_task_bindings, select_tasks
from .session_budget import session_budget
VARIANTS = {
@ -103,6 +104,7 @@ def prepare_plan(
return {
"schema": "gitnexus.diagnostic-experiments/v1", "execution": "not_started",
"harness_sha": harness_sha, "runtime_sha": runtime_sha, "seed": seed,
"task_repository": repository,
"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,
@ -120,6 +122,19 @@ def prepare_plan(
}
def trial_task_bindings(tasks: list[dict[str, Any]], expected: list[dict[str, Any]] | None = None) -> list[dict[str, Any]]:
"""Pin built dependencies only after checking their tracked source checkout."""
for task in tasks:
repo = str(Path(task["repo"]).resolve())
head = run_checked(["git", "-C", repo, "rev-parse", "HEAD"], timeout=60).stdout_tail.strip()
dirty = run_checked(
["git", "-C", repo, "status", "--porcelain", "--untracked-files=no"], timeout=60,
).stdout_tail.strip()
if head != task["ref"] or dirty:
raise ValueError("trial dependencies require a clean pinned task checkout")
return resolve_task_bindings(tasks, expected)
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--tasks", type=Path, required=True)
@ -134,6 +149,8 @@ def main() -> None:
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("--bind-task-dependencies", action="store_true",
help="pin built task dependencies for execution; requires a clean pinned checkout")
parser.add_argument("--out", type=Path, required=True)
args = parser.parse_args()
plan = prepare_plan(
@ -142,6 +159,14 @@ def main() -> None:
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,
)
if args.bind_task_dependencies:
bindings = {}
for cell in plan["cells"]:
tasks = cell["task_document"]["tasks"]
key = json.dumps(tasks, sort_keys=True)
if key not in bindings:
bindings[key] = trial_task_bindings(tasks)
cell["task_bindings"] = bindings[key]
# Refuse overwrite: a previous plan/ledger may already be evidence.
args.out.mkdir(parents=True, exist_ok=False)
root = args.out.resolve()

13
eval/workflow_bench/diagnostic_trial.py
View file

@ -11,6 +11,8 @@ import subprocess
import sys
from typing import Any
from . import diagnostic_experiments
MODEL = "gpt-5.6-sol"
TASK = "trivial-status-json-alias"
BUDGET_USD = 5.0
@ -66,8 +68,16 @@ def _validated_cells(plan_path: Path, runtime: Path) -> tuple[dict[str, Any], li
tasks = document.get("tasks") if isinstance(document, dict) else None
if not isinstance(tasks, list) or len(tasks) != 1 or not isinstance(tasks[0], dict) or tasks[0].get("id") != TASK:
raise ValueError("trial manifest must contain only the authorized task")
task_repo = tasks[0].get("repo")
if (not isinstance(task_repo, str) or task_repo != plan.get("task_repository")
or _head(Path(task_repo)) != tasks[0].get("ref")):
raise ValueError("task source and dependencies must come from the pinned task checkout")
if output.exists() or output.is_symlink() or ledger.exists() or ledger.is_symlink():
raise ValueError("trial results already exist; refusing to repeat paid work")
bindings = cell.get("task_bindings")
if not isinstance(bindings, list) or len(bindings) != 1 or not isinstance(bindings[0], dict):
raise ValueError("trial plan must pin task dependency bindings before execution")
diagnostic_experiments.trial_task_bindings(tasks, bindings)
return plan, cells
@ -86,7 +96,8 @@ def execute_trial(plan_path: Path, runtime: Path, claude_bin: str) -> int:
stream.write(json.dumps(state, indent=2) + "\n")
failed = False
for cell in cells:
command = [sys.executable, *cell["argv"][4:], "--timeout", "900", "--claude-bin", claude_bin]
command = [sys.executable, *cell["argv"][4:], "--timeout", "900", "--claude-bin", claude_bin,
"--task-bindings-json", json.dumps(cell["task_bindings"])]
item: dict[str, Any] = {"cell_id": cell["cell_id"], "variant": cell["variant"], "status": "started"}
state["cells"].append(item)
receipt.write_text(json.dumps(state, indent=2) + "\n")

61
eval/workflow_bench/release_controls.py Normal file
View file

@ -0,0 +1,61 @@
"""Offline grading controls using the runner's dependency and sandbox boundaries."""
from __future__ import annotations
import argparse
from pathlib import Path
from typing import Any
from .oracle_assets import ORACLE_ROOT, capture_task_oracle, sanitize_clone_for_hidden_oracles
from .proposer_sandbox import ReadOnlyMount, SandboxError, prepare_sandbox
from .runner import _run_hidden_oracle, _verification_outcome
from .runner_artifacts import make_worktree, run_verify
from .task_assets import stage_task_assets
REFERENCES = {
"trivial-status-json-alias": "trivial-status-json-alias",
"inv-feature-c-include-provenance": "inv-feature-c-include-provenance-flag",
"inv-feature-list-repos-filter": "inv-feature-list-repos-filter",
"cross-module-parse-retry": "cross-module-parse-retry",
}
def check_task_controls(task: dict[str, Any], *, repo: Path, sha: str, parent: Path,
assets: Any, claude_bin: Path, bwrap_bin: Path) -> None:
"""Reject broken setup before spending: base fails assertions, reference passes.
Reference patches and oracle bytes never enter a paid cell. The probe uses
the same immutable dependency snapshot and read-only verifier as the runner.
"""
reference = REFERENCES.get(task["id"])
if reference is None:
raise ValueError(f"no offline grading control for release task {task['id']}")
clone = make_worktree(repo, sha, parent)
sanitize_clone_for_hidden_oracles(clone)
mounts = stage_task_assets(task, repo=repo, clone=clone, snapshot=assets)
snapshot = capture_task_oracle(task)
args = argparse.Namespace(timeout=180)
with prepare_sandbox(clone=clone, claude_bin=claude_bin, bwrap_bin=bwrap_bin,
read_only_mounts=mounts, preflight=False, gitnexus_available=False) as sandbox:
negative: dict[str, Any] = {}
passed, _ = _run_hidden_oracle(snapshot, clone, args, sandbox, diagnostics=negative)
if (passed or negative.get("exit_code") != 1 or not negative.get("assertion_failure")
or negative.get("startup_failure") or negative.get("permission_failure")):
raise SandboxError("negative grading control did not fail on a behavioral assertion")
patch_mount = ReadOnlyMount(ORACLE_ROOT / "reference" / f"{reference}.patch", "/tmp/reference.patch")
applied = run_verify(
"git apply /tmp/reference.patch", clone, 60,
command_prefix=sandbox.command_prefix_for(unshare_network=True, extra_read_only_mounts=(patch_mount,)),
env=sandbox.environment(), require_pid_namespace=True,
)
if not _verification_outcome(applied)[0]:
raise SandboxError("reference patch could not be applied in grading control")
typed = run_verify(
"cd gitnexus && ./node_modules/.bin/tsc --noEmit", clone, 180,
command_prefix=sandbox.command_prefix_for(read_only_workspace=True, unshare_network=True),
env=sandbox.environment(), require_pid_namespace=True,
)
if not _verification_outcome(typed)[0]:
raise SandboxError("reference grading control failed typecheck")
if not _run_hidden_oracle(snapshot, clone, args, sandbox)[0]:
raise SandboxError("reference grading control failed the hidden oracle")

12
eval/workflow_bench/release_preflight.py
View file

@ -13,11 +13,13 @@ from .process_control import ManagedProcessError, cancellation_scope
from .proposer_sandbox import SandboxError, preflight_bubblewrap, require_claude_sandbox_helpers
from .runner_tasks import resolve_task_bindings, select_tasks
from .runtime_mounts import trusted_gitnexus_runtime_mounts
from .release_controls import check_task_controls
from .sanitized_graph import prepare_sanitized_graph, validate_no_prebuilt_graph_assets
from .task_assets import TaskAssetCache, stage_task_assets
def preflight_release(tasks: list[dict[str, Any]], *, gitnexus_root: Path, claude_bin: Path) -> None:
def preflight_release(tasks: list[dict[str, Any]], *, gitnexus_root: Path, claude_bin: Path,
grading_controls: bool = False) -> None:
"""Exercise the real snapshot/copy path on the same temp filesystem as the runner.
No provider or session code is involved. Graphs are rebuilt by the paid
@ -49,6 +51,10 @@ def preflight_release(tasks: list[dict[str, Any]], *, gitnexus_root: Path, claud
clone = Path(probe)
graph.materialize(clone, sanitized_head=graph.sanitized_head)
stage_task_assets(task, repo=repo, clone=clone, snapshot=assets)
if grading_controls:
check_task_controls(task, repo=repo, sha=sha, parent=Path(directory), assets=assets,
claude_bin=claude_bin, bwrap_bin=bwrap_bin)
print(f"Preflight {task['id']}: negative, reference, and typecheck controls passed", flush=True)
print(f"Preflight {task['id']}: materialization passed", flush=True)
@ -57,6 +63,7 @@ def main() -> None:
parser.add_argument("--tasks", required=True, type=Path)
parser.add_argument("--gitnexus-root", required=True, type=Path)
parser.add_argument("--claude-bin", required=True, type=Path)
parser.add_argument("--grading-controls", action="store_true", help="require offline controls for the release corpus")
args = parser.parse_args()
try:
document = yaml.safe_load(args.tasks.read_text())
@ -64,7 +71,8 @@ def main() -> None:
raise ValueError("task file must contain a tasks list")
tasks, _ = select_tasks(document["tasks"], include_expensive=True)
with cancellation_scope(handle_signals=True):
preflight_release(tasks, gitnexus_root=args.gitnexus_root, claude_bin=args.claude_bin)
preflight_release(tasks, gitnexus_root=args.gitnexus_root, claude_bin=args.claude_bin,
grading_controls=args.grading_controls)
except (ManagedProcessError, OSError, SandboxError, RuntimeError, ValueError, yaml.YAMLError) as exc:
parser.error(str(exc))

6
eval/workflow_bench/release_report.py
View file

@ -31,6 +31,12 @@ ROW_FIELDS = (
"model",
"effort",
"task_base_sha",
"task_prompt_digest",
"runtime_sha",
"harness_sha",
"runtime_digest",
"max_budget_usd",
"mcp_tool_loading",
"oracle_digest",
"sandbox_backend",
"runtime_exposure",

101
eval/workflow_bench/runner.py
View file

@ -172,6 +172,7 @@ from .runner_sessions import (
sandbox_mcp_config,
sum_sessions,
)
from .verification_diagnostics import verification_metrics
from .runner_tasks import (
normalized_model_identifier,
resolve_task_bindings,
@ -333,6 +334,8 @@ def _run_hidden_oracle(
worktree: Path,
args: argparse.Namespace,
sandbox: SandboxSession,
*,
diagnostics: dict[str, Any] | None = None,
) -> tuple[bool, str]:
"""Stage a captured oracle after the model exits, execute it, then erase it."""
@ -355,22 +358,24 @@ def _run_hidden_oracle(
# credited patch is captured.
oracle_mount = f"{SANDBOX_WORKSPACE}/{mount_name}"
oracle_env[ORACLE_ENV_VAR] = str(stage_root) if host_unsafe else oracle_mount
passed, _output = _verification_outcome(
run_verify(
snapshot.command,
sandbox.clone,
args.timeout,
command_prefix=sandbox.command_prefix_for(
read_only_workspace=True,
unshare_network=True,
extra_read_only_mounts=()
if host_unsafe
else (ReadOnlyMount(source=stage_root, target=oracle_mount),),
),
env=oracle_env,
require_pid_namespace=getattr(sandbox, "require_pid_namespace", True),
)
started = time.monotonic()
result = run_verify(
snapshot.command,
sandbox.clone,
args.timeout,
command_prefix=sandbox.command_prefix_for(
read_only_workspace=True,
unshare_network=True,
extra_read_only_mounts=()
if host_unsafe
else (ReadOnlyMount(source=stage_root, target=oracle_mount),),
),
env=oracle_env,
require_pid_namespace=getattr(sandbox, "require_pid_namespace", True),
)
if diagnostics is not None:
diagnostics.update(verification_metrics(result, time.monotonic() - started))
passed, _output = _verification_outcome(result)
# Candidate code executes in this process. Never persist its stdout
# or stderr: it can read the mounted hidden test bytes and print them.
return passed, "hidden oracle passed" if passed else "hidden oracle failed"
@ -451,6 +456,13 @@ def _unchanged(value: Any) -> Any:
return value
MCP_REPOSITORY_PROMPT = (
"\n\nBenchmark environment: the isolated GitNexus repository is registered as "
"benchmark-target. Use that repo name, or omit repo to use the default. "
"Repository aliases in checked-in guidance refer to the original checkout."
)
def run_arm(
arm: str,
task: dict[str, Any],
@ -467,6 +479,8 @@ def run_arm(
mcp_config_json: str | None = None,
) -> dict[str, Any]:
sessions: list[dict[str, Any]] = []
if arm != "baseline_nomcp":
task = {**task, "prompt": task["prompt"] + MCP_REPOSITORY_PROMPT}
environment_builder = getattr(sandbox, "environment", build_sandbox_environment)
host_text = getattr(sandbox, "host_text", _unchanged)
host_path = getattr(sandbox, "host_path", lambda value: str(value))
@ -709,20 +723,23 @@ def run_arm(
verify_mounts = (
ReadOnlyMount(source=review_artifact.parent, target=SANDBOX_REVIEW_OUTPUT),
)
authored_tests_passed, authored_test_output = _verification_outcome(
run_verify(
task["verify"],
worktree,
args.timeout,
command_prefix=sandbox.command_prefix_for(
read_only_workspace=True,
unshare_network=True,
extra_read_only_mounts=verify_mounts,
),
env=verify_env,
require_pid_namespace=getattr(sandbox, "require_pid_namespace", True),
)
verification_started = time.monotonic()
authored_result = run_verify(
task["verify"],
worktree,
args.timeout,
command_prefix=sandbox.command_prefix_for(
read_only_workspace=True,
unshare_network=True,
extra_read_only_mounts=verify_mounts,
),
env=verify_env,
require_pid_namespace=getattr(sandbox, "require_pid_namespace", True),
)
record["verification_diagnostics"] = {
"authored": verification_metrics(authored_result, time.monotonic() - verification_started),
}
authored_tests_passed, authored_test_output = _verification_outcome(authored_result)
review_score: dict[str, Any] | None = None
if arm in ("review", "ce_review"):
try:
@ -751,12 +768,15 @@ def run_arm(
)
oracle_output = "hidden review labels matched" if oracle_passed else "hidden review labels not fully matched"
else:
hidden_diagnostics: dict[str, Any] = {}
oracle_passed, oracle_output = _run_hidden_oracle(
oracle_snapshot,
worktree,
args,
sandbox,
diagnostics=hidden_diagnostics,
)
record["verification_diagnostics"]["hidden"] = hidden_diagnostics
record["authored_tests_passed"] = authored_tests_passed
record["authored_test_output"] = authored_test_output
record["oracle_passed"] = oracle_passed
@ -836,7 +856,10 @@ def task_prompt_digest(task: Mapping[str, Any]) -> str:
stop rows matching, or match rows that should not.
"""
return hashlib.sha256(str(task["prompt"]).encode()).hexdigest()
# Bind the conditional harness instruction as well as the task text. Both
# arms retain the same experiment identity; old raw-prompt hashes expire.
payload = {"task": str(task["prompt"]), "mcp_environment": MCP_REPOSITORY_PROMPT}
return hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()
# A sustained upstream outage shows up as a run of session/infra/cleanup
@ -1489,6 +1512,11 @@ def run_cell(ctx: TaskCellContext, run_idx: int, arm: str) -> dict[str, Any]:
"run": run_idx,
"sandbox_backend": ctx.sandbox_backend,
"runtime_exposure": runtime_exposure,
"max_budget_usd": getattr(args, "max_budget_usd", None),
"model": args.model,
"effort": args.effort,
"task_base_sha": ctx.task_sha,
"task_prompt_digest": task_prompt_digest(task),
**getattr(args, "measurement_revisions", {}),
"mcp_tool_loading": (
getattr(args, "mcp_tool_loading", "eager") if execution_arm == "baseline"
@ -2566,6 +2594,11 @@ def _run_sweep(
total_cells = len(tasks) * args.runs * len(args.arms)
started_cells = 0
sweep_started = time.monotonic()
timing: dict[str, Any] = {"complete": False, "workers": args.workers, "total_s": None,
"before_first_cell_s": None, "dependency_setup_s": None,
"foreground_graph_setup_s": 0.0}
timing_path = out_dir / "sweep-timing.json"
timing_path.write_text(json.dumps(timing) + "\n")
with (
tempfile.TemporaryDirectory(prefix="wfbench-trees-") as trees,
TaskAssetCache(Path(trees) / ".task-assets") as task_asset_cache,
@ -2603,6 +2636,8 @@ def _run_sweep(
except (OSError, SandboxError, ValueError) as exc:
asset_snapshot_errors[str(_task["id"])] = exc
timing["dependency_setup_s"] = time.monotonic() - sweep_started
timing_path.write_text(json.dumps(timing) + "\n")
reuse_source = args.reuse_results.expanduser().resolve() if args.reuse_results is not None else None
reusable_rows: dict[tuple[str, str, int], dict[str, Any]] = {}
if reuse_source is not None:
@ -2695,12 +2730,15 @@ def _run_sweep(
asset_snapshot = asset_snapshots.get(str(task["id"]))
asset_snapshot_error: BaseException | None = asset_snapshot_errors.get(str(task["id"]))
graph_key = (str(repo), task_sha)
graph_started = time.monotonic()
if graph_prefetch is not None and graph_prefetch.key == graph_key:
_join_graph_prefetch()
if paid_cells:
ensure_task_graph(
task=task, repo=repo, task_sha=task_sha, graph_key=graph_key, env=graph_env
)
timing["foreground_graph_setup_s"] += time.monotonic() - graph_started
timing_path.write_text(json.dumps(timing) + "\n")
graph_snapshot = graph_env.graph_snapshots.get(graph_key)
graph_snapshot_error = graph_env.graph_snapshot_errors.get(graph_key)
clone_template, template_head = graph_env.clone_templates.get(graph_key, (None, None))
@ -2728,6 +2766,9 @@ def _run_sweep(
def announce(run_idx: int, arm: str) -> None:
nonlocal started_cells
if timing["before_first_cell_s"] is None:
timing["before_first_cell_s"] = time.monotonic() - sweep_started
timing_path.write_text(json.dumps(timing) + "\n")
started_cells += 1
print(
f"[{task['id']}][{arm}][run {run_idx}] starting "
@ -2792,6 +2833,8 @@ def _run_sweep(
finally:
_join_graph_prefetch()
timing.update(complete=not outage_tripped and not cancel_event.is_set(), total_s=time.monotonic() - sweep_started)
timing_path.write_text(json.dumps(timing) + "\n")
selection_report = [
"## Run provenance",
"",

2
eval/workflow_bench/runner_sessions.py
View file

@ -356,6 +356,8 @@ GITNEXUS_READ_ONLY_TOOLS = (
"mcp__gitnexus__list_repos",
"mcp__gitnexus__query",
"mcp__gitnexus__context",
"mcp__gitnexus__grep",
"mcp__gitnexus__read_file",
"mcp__gitnexus__check",
"mcp__gitnexus__impact",
"mcp__gitnexus__explain",

9
eval/workflow_bench/session_diagnostics.py
View file

@ -16,6 +16,7 @@ from pathlib import PurePosixPath
from typing import Any
from .workload_diagnostics import COUNT_FIELDS, TOKEN_FIELDS, public_workload, workload_metrics
from .verification_diagnostics import public_verification
SCHEMA = "gitnexus.session-diagnostics/v1"
MAX_EVENTS = 100_000
@ -23,7 +24,7 @@ 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",
"list_repos", "query", "context", "grep", "read_file", "check", "impact", "explain", "pdg_query",
"route_map", "tool_map", "shape_check", "api_impact", "trace", "detect_changes",
})
TOOLS = frozenset({
@ -364,6 +365,12 @@ def public_diagnostics(value: Any) -> list[dict[str, Any]]:
def public_cell_diagnostics(row: dict[str, Any]) -> dict[str, Any]:
"""Shared publication boundary for release and optional-arm artifacts."""
safe = {}
verification = row.get("verification_diagnostics")
if isinstance(verification, dict):
safe["verification_diagnostics"] = {
phase: public_verification(verification[phase])
for phase in ("authored", "hidden") if phase in verification
}
if "session_diagnostics" in row:
safe["session_diagnostics"] = public_diagnostics(row["session_diagnostics"])
if "cell_duration_s" in row:

41
eval/workflow_bench/verification_diagnostics.py Normal file
View file

@ -0,0 +1,41 @@
"""Bounded verifier evidence without publishing candidate or hidden-test output."""
from __future__ import annotations
from typing import Any
from .runner_artifacts import VerificationResult
# These are observed text markers, not authoritative diagnoses. Never publish
# matched text: candidate code can print source, credentials, or hidden tests.
MARKERS = {
"assertion_failure": ("AssertionError",),
"startup_failure": ("Failed Suites", "Startup Error", "Unhandled Errors", "Cannot find module"),
"permission_failure": ("EACCES", "EROFS", "Permission denied", "Failed to create Vitest API token"),
"typecheck_failure": ("error TS",),
}
def verification_metrics(result: VerificationResult | tuple[bool, str], elapsed: float) -> dict[str, Any]:
passed, output = result
return {
"passed": passed,
"duration_s": elapsed,
"exit_code": result.process.returncode if isinstance(result, VerificationResult) else None,
"state": result.process.state if isinstance(result, VerificationResult) else "unknown",
**{field: any(marker in output for marker in markers) for field, markers in MARKERS.items()},
}
def public_verification(value: Any) -> dict[str, Any]:
if not isinstance(value, dict):
return {}
code, elapsed, state = value.get("exit_code"), value.get("duration_s"), value.get("state")
return {
"exit_code": code if type(code) is int and -255 <= code <= 255 else None,
"duration_s": elapsed if type(elapsed) in (int, float) and 0 <= elapsed <= 10**12 else None,
"state": state if state in ("exited", "timeout", "cancelled", "input-failure", "spawn-failure", "forced-kill",
"ownership-failure", "reap-failure", "cleanup-failure", "unknown") else "unknown",
**{field: value.get(field) if type(value.get(field)) is bool else None
for field in ("passed", *MARKERS)},
}

5
eval/workflow_bench/workload_diagnostics.py
View file

@ -211,6 +211,11 @@ def workload_metrics(
counts["edits"] += 1
counts["edits_after_verification"] += verified
counts["usage_snapshots"] = len(usage)
# Some CLI/provider adapters emit all-zero placeholders for completed
# assistant turns. They are not evidence of an empty model context.
usage = [u if any(type(u.get(f)) is int and u[f] > 0 for f in (
"input_tokens", "cache_read_input_tokens", "cache_creation_input_tokens", "output_tokens",
)) else {"event_index": u["event_index"]} for u in 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]