GitNexus/eval/tests/test_workflow_bench.py

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