"""Unit tests for workflow benchmark aggregation, reporting, task, and CI contracts.""" import json import os import re import shlex import subprocess from pathlib import Path import pytest import yaml from workflow_bench.runner import ( aggregate, broken_incumbent_arms, build_parser, infra_error_record, normalized_model_identifier, parse_shortstat, render_report, savings, select_tasks, systemic_outage_streak, ) def record(**overrides): base = { "input_tokens": 1000, "cache_creation_input_tokens": 200, "cache_read_input_tokens": 5000, "output_tokens": 400, "cost_usd": 0.5, "duration_s": 60.0, "num_turns": 10, "diff_files": 2, "diff_insertions": 30, "diff_deletions": 5, "class": "demo", "resolved": True, } base.update(overrides) return base def test_aggregate_takes_medians_and_counts_resolved(): records = [ record(input_tokens=1000, resolved=True), record(input_tokens=3000, resolved=False), record(input_tokens=2000, resolved=True), ] agg = aggregate(records) assert agg == { "input_tokens": 2000, "cache_creation_input_tokens": 200, "cache_read_input_tokens": 5000, "output_tokens": 400, "cost_usd": 0.5, "duration_s": 60.0, "num_turns": 10, "diff_files": 2, "diff_insertions": 30, "diff_deletions": 5, "class": "demo", "resolved": 2, "runs": 3, "valid_runs": 3, "excluded_runs": 0, "transcripts_missing": 0, "error_kinds": {}, } def test_savings_is_positive_when_workflow_is_cheaper(): baseline = aggregate([record(input_tokens=2000, output_tokens=800, cost_usd=1.0)]) workflow = aggregate([record(input_tokens=1000, output_tokens=400, cost_usd=0.4)]) s = savings(baseline, workflow) assert s["input_tokens"] == 50.0 assert s["output_tokens"] == 50.0 assert s["cost_usd"] == 60.0 def task_row(task_id: str, **overrides): task = { "id": task_id, "class": "demo", "repo": "/repo", "prompt": "do it", "verify": "true", "oracle": { "command": "true", "files": [ { "source": "trivial-status-json-alias.oracle.test.ts", "target": "oracle.test.ts", } ], }, } task.update(overrides) return task def test_expensive_tasks_are_opt_in_and_reported_as_skipped(): tasks = [task_row("default"), task_row("large", expensive=True)] selected, skipped = select_tasks(tasks, include_expensive=False) assert [task["id"] for task in selected] == ["default"] assert skipped == ["large"] selected, skipped = select_tasks(tasks, include_expensive=True) assert [task["id"] for task in selected] == ["default", "large"] assert skipped == [] @pytest.mark.parametrize("value", ["true", 1, None, [], {}]) def test_expensive_metadata_must_be_boolean(value): with pytest.raises(ValueError, match="expensive.*boolean"): select_tasks([task_row("bad", expensive=value)], include_expensive=False) def test_task_selection_rejects_duplicate_ids_and_empty_selection(): with pytest.raises(ValueError, match="duplicate task id"): select_tasks([task_row("same"), task_row("same")], include_expensive=True) with pytest.raises(ValueError, match="no tasks selected"): select_tasks([task_row("large", expensive=True)], include_expensive=False) def test_runner_requires_a_named_model_and_supports_expensive_opt_in(): with pytest.raises(SystemExit): build_parser().parse_args(["--tasks", "tasks.yaml"]) args = build_parser().parse_args( [ "--tasks", "tasks.yaml", "--model", "claude-sonnet-4-20250514", "--include-expensive", ] ) assert args.include_expensive is True with pytest.raises(ValueError, match="nonblank"): normalized_model_identifier(" ") @pytest.mark.parametrize( "alias", ["Auto", "AUTO", "latest", "provider/latest", "provider:Latest", "provider@LATEST"], ) def test_runner_rejects_mutable_model_aliases(alias): with pytest.raises(ValueError, match="mutable auto/latest"): normalized_model_identifier(alias) assert normalized_model_identifier("free-coder") == "free-coder" assert normalized_model_identifier("claude-sonnet-4-20250514") == "claude-sonnet-4-20250514" def test_eval_ci_uses_locked_uv_and_blocking_native_containment_jobs(): repo_root = Path(__file__).resolve().parents[2] workflow = (repo_root / ".github" / "workflows" / "ci-tests.yml").read_text() workflow_document = yaml.safe_load(workflow) containment = workflow_document["jobs"]["eval-containment-linux"] containment_steps = {step.get("name"): step for step in containment["steps"] if "name" in step} containment_node_setup = next( step for step in containment["steps"] if str(step.get("uses", "")).startswith("actions/setup-node@") ) claude_lock = json.loads((repo_root / ".github" / "claude-canary-runtime" / "package-lock.json").read_text()) setup_uv = "astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990" assert workflow.count(setup_uv) >= 3 assert workflow.count("version: '0.11.23'") >= 3 assert workflow.count("uv run --locked --extra dev python -m pytest") >= 3 assert "eval-containment-linux:" in workflow assert "GITNEXUS_REQUIRE_BWRAP_CANARY: '1'" in workflow assert "GITNEXUS_REQUIRE_CLAUDE_CANARY: '1'" in workflow assert containment["env"] == { "GITNEXUS_REQUIRE_BWRAP_CANARY": "1", "GITNEXUS_REQUIRE_CLAUDE_CANARY": "1", } assert containment["timeout-minutes"] == 20 assert containment_node_setup["with"] == { "node-version": "22.18.0", "cache": "npm", "cache-dependency-path": "gitnexus/package-lock.json\ngitnexus-shared/package-lock.json\n", } assert ( "CLAUDE_CANARY_BIN: ${{ runner.temp }}/claude-canary/node_modules/@anthropic-ai/claude-code-linux-x64/claude" in workflow ) assert ".github/claude-canary-runtime/package-lock.json" in workflow assert "npm ci" in workflow assert "--package-lock=false" not in workflow assert claude_lock["packages"]["node_modules/@anthropic-ai/claude-code"]["version"] == "2.1.214" assert claude_lock["packages"]["node_modules/@anthropic-ai/claude-code"]["integrity"].startswith("sha512-") assert "if(p.version!=='2.1.214') process.exit(1)" in workflow assert "'2.1.214 (Claude Code)'" in workflow # Shared is compiled by gitnexus `npm run build` (scripts/build.js runTsc). # A dedicated npm ci in gitnexus-shared pulls TypeScript 7 and stalls CI. assert "Build pinned shared runtime" not in containment_steps assert not any( step.get("working-directory") == "gitnexus-shared" and "npm ci" in str(step.get("run", "")) for step in containment["steps"] ) assert containment_steps["Install and build pinned GitNexus runtime"]["working-directory"] == "gitnexus" assert containment_steps["Install and build pinned GitNexus runtime"]["run"].splitlines() == [ "npm ci", "npm run build", ] selected_containment_tests = containment_steps["Prove process-tree and sandbox containment"]["run"].split() assert selected_containment_tests == [ "uv", "run", "--locked", "--extra", "dev", "python", "-m", "pytest", "tests/test_process_control.py", "tests/test_proposer_sandbox.py", "tests/test_workflow_bench_sessions.py", "tests/test_ce_plugin_runtime.py", "-q", ] bwrap_canary_marker = re.compile( r'@pytest\.mark\.skipif\(\s*os\.environ\.get\("GITNEXUS_REQUIRE_BWRAP_CANARY"\)', re.MULTILINE, ) bwrap_canary_files = sorted( path.name for path in (repo_root / "eval" / "tests").glob("test_*.py") if bwrap_canary_marker.search(path.read_text()) ) assert bwrap_canary_files == ["test_proposer_sandbox.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 def test_shipped_scenarios_opt_out_the_cross_module_cell_and_rebuild_graph_assets(): task_file = Path(__file__).resolve().parents[1] / "workflow_bench" / "tasks.scenarios.yaml" tasks = yaml.safe_load(task_file.read_text())["tasks"] selected, skipped = select_tasks(tasks, include_expensive=False) assert [task["id"] for task in selected] == [ "trivial-status-json-alias", "inv-bug-c-system-include", "inv-feature-list-repos-filter", ] assert skipped == ["cross-module-parse-retry"] assert all(not task.get("sandbox_copy") for task in tasks) assert all(task["sandbox_dependencies"] for task in tasks) assert all(task["oracle"]["command"] and task["oracle"]["files"] for task in tasks) assert all("./node_modules/.bin/vitest run" in task["oracle"]["command"] for task in tasks) assert all("npx vitest" not in task["oracle"]["command"] for task in tasks) assert all( '--config "$GITNEXUS_BENCH_ORACLE_ROOT/vitest.config.mts"' in task["oracle"]["command"] for task in tasks ) assert all({item["target"] for item in task["oracle"]["files"]} >= {"vitest.config.mts"} for task in tasks) def test_savings_handles_zero_baseline_without_dividing(): baseline = aggregate([record(cost_usd=0.0)]) workflow = aggregate([record(cost_usd=0.0)]) assert savings(baseline, workflow)["cost_usd"] == 0.0 def test_parse_shortstat_full_and_empty(): full = parse_shortstat(" 3 files changed, 120 insertions(+), 7 deletions(-)") assert full == {"diff_files": 3, "diff_insertions": 120, "diff_deletions": 7} assert parse_shortstat("") == { "diff_files": 0, "diff_insertions": 0, "diff_deletions": 0, } singular = parse_shortstat(" 1 file changed, 1 insertion(+)") assert singular == {"diff_files": 1, "diff_insertions": 1, "diff_deletions": 0} def test_render_report_emits_arm_rows_and_per_arm_savings_rows(): results = { "demo-task": { "workflow": aggregate([record(input_tokens=1000)]), "workflow_direct": aggregate([record(input_tokens=1500)]), "baseline": aggregate([record(input_tokens=2000)]), } } report = render_report(results) assert "| demo-task | demo | workflow | 1/1 | 1000 |" in report assert "| demo-task | demo | baseline | 1/1 | 2000 |" in report assert "| demo-task | demo | **workflow savings %** | — | 50.0 |" in report assert "| demo-task | demo | **workflow_direct savings %** | — | 25.0 |" in report assert "2/+30/−5" in report assert "results.jsonl" in report assert "subagent spend" in report # token columns are main-loop-only def test_aggregate_excludes_session_error_rows_from_medians(): records = [ record(cost_usd=1.0), record(cost_usd=3.0, transcript_missing=True), record(cost_usd=100.0, resolved=False, error_kind="session-error"), ] agg = aggregate(records) assert agg["cost_usd"] == 2.0 assert agg["runs"] == 3 assert agg["valid_runs"] == 2 assert agg["excluded_runs"] == 1 assert agg["transcripts_missing"] == 1 assert agg["resolved"] == 2 def test_aggregate_excludes_unverified_transcript_evidence(): agg = aggregate( [ record(cost_usd=1.0), record( cost_usd=100.0, resolved=False, error_kind="evidence-unverified", transcript_missing=True, ), ] ) assert agg["cost_usd"] == 1.0 assert agg["valid_runs"] == 1 assert agg["excluded_runs"] == 1 def test_aggregate_excludes_invalid_review_artifacts_from_quality_metrics(): scored = record( cost_usd=1.0, review_weighted_f1=0.8, review_true_positives=2, review_false_positives=0, review_false_negatives=1, review_precision=1.0, review_recall=0.67, review_f1=0.8, review_weighted_precision=0.8, review_weighted_recall=0.8, review_blocker_recall=1.0, review_severity_accuracy=1.0, review_category_accuracy=1.0, review_grounded_evidence=1.0, review_clean_control=False, ) agg = aggregate( [ scored, record(cost_usd=2.0, resolved=False, error_kind="review-evidence-invalid"), ] ) assert agg["valid_runs"] == 1 assert agg["excluded_runs"] == 1 assert agg["review_weighted_f1"] == 0.8 assert agg["review_true_positives"] == 2 def test_render_report_surfaces_excluded_and_unverified_runs(): results = { "t": { "workflow": aggregate( [ record(transcript_missing=True), record(resolved=False, error_kind="session-error"), ] ) } } report = render_report(results) assert "| t | demo | workflow | 1/1 (1 excluded) |" in report assert "session/infra errors" in report assert "no locatable session transcript" in report def test_render_report_surfaces_why_each_row_failed(): results = { "t": { "workflow": aggregate( [record(resolved=False, error_kind="plan-evidence-invalid")], ), } } report = render_report(results) assert "plan-evidence-invalid×1" in report def test_broken_incumbent_arms_flags_an_incumbent_that_resolved_nothing(): results = { "t1": {"workflow": aggregate([record(resolved=False, error_kind="plan-evidence-invalid")])}, "t2": {"workflow": aggregate([record(resolved=False, error_kind="plan-evidence-invalid")])}, } assert broken_incumbent_arms(results, {"workflow"}) == ["workflow"] def test_broken_incumbent_arms_ignores_a_merely_underperforming_candidate(): # The incumbent works fine; only the candidate arm fails. That's a normal, # expected "bad candidate" outcome and must not read as a broken harness. results = { "t1": { "workflow": aggregate([record(resolved=True)]), "candidate_workflow": aggregate([record(resolved=False, error_kind="verify-failed")]), }, } assert broken_incumbent_arms(results, {"workflow"}) == [] def test_broken_incumbent_arms_flags_an_incumbent_with_zero_valid_runs(): # Every run excluded via an excluded-but-non-systemic error_kind # ("evidence-unverified"): valid_runs == 0 for every task, which the old # `valid_runs > 0` guard let sail through silently, and which the outage # streak breaker also doesn't catch (it resets rather than accumulates # on this exact error_kind -- see test_systemic_outage_streak_resets_on_non_outage). results = { "t1": {"workflow": aggregate([record(resolved=False, error_kind="evidence-unverified")])}, "t2": {"workflow": aggregate([record(resolved=False, error_kind="evidence-unverified")])}, } assert results["t1"]["workflow"]["valid_runs"] == 0 assert broken_incumbent_arms(results, {"workflow"}) == ["workflow"] def test_broken_incumbent_arms_ignores_partial_incumbent_failure(): # Resolved in at least one task — struggling, not broken. results = { "t1": {"workflow": aggregate([record(resolved=False, error_kind="verify-failed")])}, "t2": {"workflow": aggregate([record(resolved=True)])}, } assert broken_incumbent_arms(results, {"workflow"}) == [] def test_infra_error_record_captures_the_failure_and_is_excluded(): exc = subprocess.TimeoutExpired(cmd="claude -p", timeout=5) rec = infra_error_record(exc) assert rec["ok"] is False assert rec["resolved"] is False assert rec["error_kind"] == "infra-error" assert "TimeoutExpired" in rec["error_detail"] assert rec["output_tokens"] == 0 agg = aggregate([record(cost_usd=2.0), rec]) assert agg["cost_usd"] == 2.0 assert agg["valid_runs"] == 1 assert agg["excluded_runs"] == 1 def test_systemic_outage_streak_counts_consecutive_systemic_failures(): # session/infra/cleanup failures accumulate; a cleanup-failure that masked a # session-error still counts toward the streak. streak = 0 for kind in ("session-error", "infra-error", "cleanup-failure"): streak = systemic_outage_streak(kind, streak) assert streak == 3 assert systemic_outage_streak("cleanup-failure", 4) == 5 def test_systemic_outage_streak_resets_on_non_outage(): # A real task failure (resolved=False → error_kind None) or an unverifiable # evidence run is not an outage and resets the streak. assert systemic_outage_streak(None, 4) == 0 assert systemic_outage_streak("evidence-unverified", 4) == 0 def test_outage_streak_flag_defaults_and_disables(): base = ["--tasks", "tasks.yaml", "--model", "claude-sonnet-4-20250514"] assert build_parser().parse_args(base).outage_streak == 5 assert build_parser().parse_args([*base, "--outage-streak", "0"]).outage_streak == 0 def test_run_evolution_script_is_the_shared_ci_and_local_entrypoint(): eval_dir = Path(__file__).resolve().parents[1] script = eval_dir / "workflow_bench" / "run-evolution.sh" workflow = eval_dir.parent / ".github" / "workflows" / "gitnexus-skill-evolution.yml" assert script.is_file() assert script.stat().st_mode & 0o111 workflow_text = workflow.read_text() assert "./workflow_bench/run-evolution.sh --apply" in workflow_text assert "python -m workflow_bench.evolve" not in workflow_text env = { "PATH": os.environ.get("PATH", "/usr/bin"), "MODEL": "claude-sonnet-5", "PROPOSER_MODEL": "claude-opus-4-8", "EFFORT": "xhigh", "GENERATIONS": "1", "RUNS": "3", "WORKERS": "2", "PROVIDER": "openai", "INCLUDE_EXPENSIVE": "1", "SEED_RESULTS": "/tmp/seed-bench", "CLAUDE_BIN": "/opt/claude", "OUT_ROOT": "/tmp/wfevolve", "CE_PLUGIN_DIR": "/tmp/ce-plugin", "CE_PLUGIN_VERSION": "3.24.0", "HOME": os.environ.get("HOME", "/tmp"), } printed = subprocess.run( [str(script), "--dry-run", "--apply"], check=True, capture_output=True, text=True, env=env, ) argv = shlex.split(printed.stdout) assert argv[:7] == ["uv", "run", "--locked", "--extra", "dev", "python", "-m"] assert argv[7] == "workflow_bench.evolve" assert argv[argv.index("--tasks") + 1] == "workflow_bench/tasks.review.scenarios.yaml" assert argv[argv.index("--arms") + 1] == "review" assert argv[argv.index("--ce-plugin-version") + 1] == "3.24.0" assert argv[argv.index("--model") + 1] == "gpt-5.6-sol" assert argv[argv.index("--proposer-model") + 1] == "gpt-5.6-sol" assert argv[argv.index("--effort") + 1] == "xhigh" assert argv[argv.index("--workers") + 1] == "2" assert argv[argv.index("--claude-bin") + 1] == "/opt/claude" assert argv[argv.index("--out-root") + 1] == "/tmp/wfevolve" assert argv[argv.index("--seed-results") + 1] == "/tmp/seed-bench" assert "--apply" in argv assert "--include-expensive" in argv assert "claude-sonnet-5" not in argv assert printed.stderr # rewrite notice goes to stderr