GitNexus/eval/tests/test_oracle_assets.py

"""Hidden-oracle capture, staging, and promotion-boundary regressions."""

from __future__ import annotations

import argparse
import hashlib
import os
import re
import shutil
import subprocess
import sys
import tarfile
from pathlib import Path
from types import SimpleNamespace

import pytest
import yaml

from workflow_bench import oracle_assets, runner
from workflow_bench.evolution import evaluate_candidate
from workflow_bench.oracle_assets import (
    capture_task_oracle,
    require_hidden_harness_absent,
    review_case_setup_command,
    staged_task_oracle,
    with_hidden_harness_apply_exclude,
)


def oracle_task(*, command: str = "true", source: str = "oracle.test.ts") -> dict[str, object]:
    return {
        "id": "hidden",
        "oracle": {
            "command": command,
            "files": [{"source": source, "target": "nested/oracle.test.ts"}],
        },
    }


def write_oracle(root: Path, payload: bytes = b"hidden behavior") -> None:
    root.mkdir()
    (root / "oracle.test.ts").write_bytes(payload)


def session_record() -> dict[str, object]:
    return {
        "input_tokens": 1,
        "cache_creation_input_tokens": 0,
        "cache_read_input_tokens": 0,
        "output_tokens": 1,
        "cost_usd": 0.1,
        "duration_s": 1.0,
        "num_turns": 1,
        "ok": True,
        "session_id": "s",
        "error_kind": None,
        "error_detail": None,
    }


def bench_args() -> argparse.Namespace:
    return argparse.Namespace(
        claude_bin="claude",
        timeout=5,
        model="pinned-model",
        effort="xhigh",
        base_url=None,
        auth_token=None,
    )


def sandbox(tmp_path: Path) -> SimpleNamespace:
    calls: list[dict[str, object]] = []
    private_root = tmp_path / "sandbox-private"
    private_root.mkdir(exist_ok=True)
    instance = SimpleNamespace(
        claude_bin="claude",
        clone=tmp_path,
        private_root=private_root,
        command_prefix=[],
        command_prefix_calls=calls,
        settings_json="{}",
        transcript_projects=tmp_path / "transcripts",
    )
    instance.command_prefix_for = lambda **kwargs: calls.append(dict(kwargs)) or []
    return instance


def test_capture_digest_binds_command_targets_and_raw_bytes(tmp_path: Path) -> None:
    root = tmp_path / "oracles"
    write_oracle(root)
    original = capture_task_oracle(oracle_task(), root=root)
    same = capture_task_oracle(oracle_task(), root=root)
    changed_command = capture_task_oracle(oracle_task(command="false"), root=root)
    (root / "oracle.test.ts").write_bytes(b"changed behavior")
    changed_bytes = capture_task_oracle(oracle_task(), root=root)

    assert same == original
    assert changed_command.digest != original.digest
    assert changed_command.command_digest != original.command_digest
    assert changed_bytes.digest != original.digest
    assert changed_bytes.manifest_digest != original.manifest_digest
    assert original.binding["oracle_files"] == [
        {
            "target": "nested/oracle.test.ts",
            "sha256": hashlib.sha256(b"hidden behavior").hexdigest(),
            "size": len(b"hidden behavior"),
        }
    ]


def test_clone_sanitization_prunes_harness_checkout_and_recoverable_history(tmp_path: Path) -> None:
    source = tmp_path / "source"
    source.mkdir()

    def git(repo: Path, *args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
        result = subprocess.run(
            ["git", "-C", str(repo), *args],
            check=False,
            capture_output=True,
            text=True,
        )
        if check and result.returncode != 0:
            pytest.fail(f"git {' '.join(args)} failed: {result.stderr}")
        return result

    git(source, "init", "--quiet", "--initial-branch=main")
    git(source, "config", "user.name", "Oracle Test")
    git(source, "config", "user.email", "oracle-test.invalid")
    (source / "visible.txt").write_text("model-visible source\n")
    hidden = source / "eval" / "workflow_bench" / "oracles"
    hidden.mkdir(parents=True)
    (hidden / "secret.oracle.test.ts").write_text("unique hidden behavioral assertion\n")
    (source / "eval" / "workflow_bench" / "tasks.scenarios.yaml").write_text("secret command\n")
    git(source, "add", "--all")
    git(source, "commit", "--quiet", "-m", "fixture with hidden oracle")
    git(source, "tag", "oracle-backup")

    clone = tmp_path / "clone"
    clone_result = subprocess.run(
        ["git", "clone", "--no-local", "--no-hardlinks", "--quiet", str(source), str(clone)],
        check=False,
        capture_output=True,
        text=True,
    )
    assert clone_result.returncode == 0, clone_result.stderr
    original_head = git(clone, "rev-parse", "HEAD").stdout.strip()
    hidden_tree = git(clone, "rev-parse", "HEAD:eval/workflow_bench").stdout.strip()

    sanitized_head = oracle_assets.sanitize_clone_for_hidden_oracles(clone)

    assert sanitized_head != original_head
    assert (clone / "visible.txt").read_text() == "model-visible source\n"
    assert not (clone / "eval" / "workflow_bench").exists()
    assert git(clone, "show", f"{original_head}:eval/workflow_bench/tasks.scenarios.yaml", check=False).returncode != 0
    assert git(clone, "cat-file", "-e", original_head, check=False).returncode != 0
    assert git(clone, "cat-file", "-e", hidden_tree, check=False).returncode != 0
    assert git(clone, "for-each-ref", "--format=%(refname)").stdout == ""
    assert git(clone, "show", "-s", "--format=%P", "HEAD").stdout.strip() == ""
    assert git(clone, "status", "--porcelain=v1", "--untracked-files=all").stdout == ""


def test_require_hidden_harness_absent_fails_closed_on_leftover_tree(tmp_path: Path) -> None:
    clone = tmp_path / "clone"
    hidden = clone / "eval" / "workflow_bench"
    hidden.mkdir(parents=True)
    (hidden / "review_cases").mkdir()

    with pytest.raises(ValueError, match="hidden harness visible"):
        require_hidden_harness_absent(clone)

    shutil.rmtree(hidden)
    require_hidden_harness_absent(clone)


def test_hidden_harness_apply_exclude_is_idempotent() -> None:
    raw = "git apply eval/workflow_bench/review_cases/pr.patch && rm -rf eval/workflow_bench"
    once = with_hidden_harness_apply_exclude(raw)
    assert once == review_case_setup_command("pr.patch")
    assert with_hidden_harness_apply_exclude(once) == once
    assert with_hidden_harness_apply_exclude("true") == "true"


def test_review_setup_skips_sanitized_harness_hunks(tmp_path: Path) -> None:
    repo = tmp_path / "repo"
    repo.mkdir()

    def git(*args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
        result = subprocess.run(
            ["git", "-C", str(repo), *args],
            check=False,
            capture_output=True,
            text=True,
        )
        if check and result.returncode != 0:
            pytest.fail(f"git {' '.join(args)} failed: {result.stderr}")
        return result

    git("init", "--quiet", "--initial-branch=main")
    git("config", "user.name", "Review Setup")
    git("config", "user.email", "review-setup.invalid")
    (repo / "visible.py").write_text("old\n")
    hidden = repo / "eval" / "workflow_bench"
    hidden.mkdir(parents=True)
    (hidden / "learnings.jsonl").write_text("{}\n")
    git("add", "--all")
    git("commit", "--quiet", "-m", "base with harness file")
    (repo / "visible.py").write_text("new\n")
    (hidden / "learnings.jsonl").write_text("{}\nextra\n")
    patch = git("diff").stdout
    git("checkout", "--", ".")
    shutil.rmtree(hidden)
    patch_path = hidden / "review_cases" / "case.patch"
    patch_path.parent.mkdir(parents=True)
    patch_path.write_text(patch)

    rejected = git("apply", "--check", str(patch_path.relative_to(repo)), check=False)
    assert rejected.returncode != 0
    assert "learnings.jsonl" in rejected.stderr

    setup = with_hidden_harness_apply_exclude(
        "git apply eval/workflow_bench/review_cases/case.patch && rm -rf eval/workflow_bench"
    )
    applied = subprocess.run(["/bin/sh", "-lc", setup], cwd=repo, check=False, capture_output=True, text=True)
    assert applied.returncode == 0, applied.stderr
    assert (repo / "visible.py").read_text() == "new\n"
    assert not hidden.exists()


def test_clone_sanitization_prunes_remote_history_when_head_never_had_harness(tmp_path: Path) -> None:
    source = tmp_path / "source"
    source.mkdir()

    def git(repo: Path, *args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
        result = subprocess.run(
            ["git", "-C", str(repo), *args],
            check=False,
            capture_output=True,
            text=True,
        )
        if check and result.returncode != 0:
            pytest.fail(f"git {' '.join(args)} failed: {result.stderr}")
        return result

    git(source, "init", "--quiet", "--initial-branch=main")
    git(source, "config", "user.name", "Oracle Test")
    git(source, "config", "user.email", "oracle-test.invalid")
    (source / "visible.txt").write_text("old task snapshot\n")
    git(source, "add", "--all")
    git(source, "commit", "--quiet", "-m", "old snapshot without harness")
    old_head = git(source, "rev-parse", "HEAD").stdout.strip()

    hidden = source / "eval" / "workflow_bench" / "oracles"
    hidden.mkdir(parents=True)
    secret = hidden / "future-secret.test.ts"
    secret.write_text("UNRECOVERABLE_REMOTE_ORACLE_BYTES\n")
    git(source, "add", "--all")
    git(source, "commit", "--quiet", "-m", "future remote-only oracle")
    future_head = git(source, "rev-parse", "HEAD").stdout.strip()

    sanitized_heads: list[str] = []
    for name in ("clone-one", "clone-two"):
        clone = tmp_path / name
        subprocess.run(
            ["git", "clone", "--no-local", "--no-hardlinks", "--quiet", str(source), str(clone)],
            check=True,
        )
        git(clone, "checkout", "--detach", "--quiet", old_head)
        assert git(clone, "show", f"{future_head}:eval/workflow_bench/oracles/future-secret.test.ts").stdout == (
            "UNRECOVERABLE_REMOTE_ORACLE_BYTES\n"
        )

        sanitized_heads.append(oracle_assets.sanitize_clone_for_hidden_oracles(clone))

        assert git(clone, "remote").stdout == ""
        assert git(clone, "for-each-ref", "--format=%(refname)").stdout == ""
        assert git(clone, "cat-file", "-e", future_head, check=False).returncode != 0
        assert (
            git(
                clone, "show", f"{future_head}:eval/workflow_bench/oracles/future-secret.test.ts", check=False
            ).returncode
            != 0
        )
        assert git(clone, "fsck", "--full", "--no-reflogs", "--unreachable").stdout == ""
        assert git(clone, "show", "-s", "--format=%P", "HEAD").stdout.strip() == ""

    assert sanitized_heads[0] == sanitized_heads[1]


@pytest.mark.skipif(os.name == "nt", reason="symlink contract is POSIX-specific")
def test_capture_rejects_symlinked_sources_and_parents(tmp_path: Path) -> None:
    outside = tmp_path / "outside"
    outside.mkdir()
    (outside / "oracle.test.ts").write_text("secret")

    source_link_root = tmp_path / "source-link-root"
    source_link_root.mkdir()
    (source_link_root / "oracle.test.ts").symlink_to(outside / "oracle.test.ts")
    with pytest.raises(ValueError, match="regular non-symlink"):
        capture_task_oracle(oracle_task(), root=source_link_root)

    parent_link_root = tmp_path / "parent-link-root"
    parent_link_root.mkdir()
    (parent_link_root / "linked").symlink_to(outside, target_is_directory=True)
    task = oracle_task(source="linked/oracle.test.ts")
    with pytest.raises(ValueError, match="parents must be real"):
        capture_task_oracle(task, root=parent_link_root)


@pytest.mark.parametrize(
    ("constant", "value", "expected"),
    [
        ("MAX_ORACLE_FILE_BYTES", 4, "bounded regular"),
        ("MAX_ORACLE_TOTAL_BYTES", 4, "total byte limit"),
        ("MAX_ORACLE_PATH_BYTES", 4, "bounded portable path"),
    ],
)
def test_capture_enforces_file_total_and_path_bounds(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
    constant: str,
    value: int,
    expected: str,
) -> None:
    root = tmp_path / "oracles"
    write_oracle(root, b"12345")
    monkeypatch.setattr(oracle_assets, constant, value)
    with pytest.raises(ValueError, match=expected):
        capture_task_oracle(oracle_task(), root=root)


def test_oracle_is_staged_privately_then_removed_and_mutation_is_rejected(tmp_path: Path) -> None:
    source = tmp_path / "oracles"
    write_oracle(source)
    snapshot = capture_task_oracle(oracle_task(), root=source)
    worktree = tmp_path / "worktree"
    worktree.mkdir()

    with staged_task_oracle(worktree, snapshot) as stage:
        assert stage.parent == worktree
        assert stage.name.startswith(".wfbench-oracle-")
        staged = stage / "nested" / "oracle.test.ts"
        assert staged.read_bytes() == b"hidden behavior"
        assert not any(path.name == "oracle.test.ts" for path in worktree.iterdir())
    assert not list(worktree.glob(".wfbench-oracle-*"))

    with pytest.raises(ValueError, match="changed during verification"):
        with staged_task_oracle(worktree, snapshot) as stage:
            staged = stage / "nested" / "oracle.test.ts"
            staged.chmod(0o600)
            staged.write_bytes(b"weakened")
    assert not list(worktree.glob(".wfbench-oracle-*"))


def test_vacuous_authored_test_cannot_self_certify_resolution(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    source = tmp_path / "oracles"
    write_oracle(source)
    snapshot = capture_task_oracle(oracle_task(), root=source)
    monkeypatch.setattr(runner, "run_claude", lambda *args, **kwargs: session_record())
    outcomes = iter([(True, "authored test passed"), (False, "hidden behavior failed")])
    monkeypatch.setattr(runner, "run_verify", lambda *args, **kwargs: next(outcomes))

    record = runner.run_arm(
        "baseline",
        {"prompt": "implement behavior", "verify": "true"},
        tmp_path,
        bench_args(),
        sandbox=sandbox(tmp_path),
        oracle_snapshot=snapshot,
    )

    assert record["authored_tests_passed"] is True
    assert record["oracle_passed"] is False
    assert record["resolved"] is False
    assert record["error_kind"] == "oracle-failed"


def test_host_unsafe_oracle_executes_beside_candidate_and_is_removed(tmp_path: Path) -> None:
    from workflow_bench.proposer_sandbox import prepare_sandbox

    source = tmp_path / "oracles"
    write_oracle(
        source,
        b"from pathlib import Path\nassert (Path(__file__).resolve().parents[2] / 'candidate.txt').read_text() == 'candidate'\n",
    )
    snapshot = capture_task_oracle(
        oracle_task(command='python3 "$GITNEXUS_BENCH_ORACLE_ROOT/nested/oracle.test.ts"'), root=source
    )
    clone = tmp_path / "clone"
    clone.mkdir()
    (clone / "candidate.txt").write_text("candidate")
    with prepare_sandbox(clone=clone, claude_bin=sys.executable, backend="host-unsafe") as session:
        passed, detail = runner._run_hidden_oracle(snapshot, clone, bench_args(), session)
        assert passed, detail
        assert not list(clone.glob(".wfbench-oracle-*"))


def test_oracle_path_and_bytes_appear_only_after_the_model_session(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    source = tmp_path / "oracles"
    write_oracle(source)
    snapshot = capture_task_oracle(oracle_task(), root=source)
    stages_seen: list[list[Path]] = []

    def fake_session(*args, **kwargs):
        assert not list(tmp_path.glob(".wfbench-oracle-*"))
        assert b"hidden behavior" not in b"".join(path.read_bytes() for path in tmp_path.glob("*.test.ts"))
        return session_record()

    def fake_verify(*args, **kwargs):
        stages_seen.append(list(tmp_path.glob(".wfbench-oracle-*")))
        return True, "ok"

    monkeypatch.setattr(runner, "run_claude", fake_session)
    monkeypatch.setattr(runner, "run_verify", fake_verify)
    record = runner.run_arm(
        "baseline",
        {"prompt": "implement behavior", "verify": "true"},
        tmp_path,
        bench_args(),
        sandbox=sandbox(tmp_path),
        oracle_snapshot=snapshot,
    )

    assert stages_seen[0] == []  # authored tests run before hidden files are staged
    assert len(stages_seen[1]) == 1
    assert record["resolved"] is True
    assert not list(tmp_path.glob(".wfbench-oracle-*"))


def test_hidden_oracle_uses_digest_bound_staged_config_not_candidate_config(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    source = tmp_path / "oracles"
    source.mkdir()
    hidden_config = b"export default { test: { passWithNoTests: false, setupFiles: [] } };\n"
    (source / "vitest.config.mts").write_bytes(hidden_config)
    (source / "oracle.test.ts").write_text("hidden test")
    task = oracle_task(
        command=(
            'npx vitest run --config "$GITNEXUS_BENCH_ORACLE_ROOT/vitest.config.mts" '
            '"$GITNEXUS_BENCH_ORACLE_ROOT/nested/oracle.test.ts"'
        )
    )
    task["oracle"]["files"].insert(  # type: ignore[index]
        0,
        {"source": "vitest.config.mts", "target": "vitest.config.mts"},
    )
    snapshot = capture_task_oracle(task, root=source)
    candidate_config = tmp_path / "vitest.config.ts"
    candidate_config.write_text("export default { test: { passWithNoTests: true } };\n")
    calls = 0

    sandbox_instance = sandbox(tmp_path)

    def fake_verify(command, *args, **kwargs):
        nonlocal calls
        calls += 1
        if calls == 1:
            return True, "authored"
        assert command == snapshot.command
        oracle_env_root = kwargs["env"][oracle_assets.ORACLE_ENV_VAR]
        assert oracle_env_root.startswith("/workspace/.wfbench-oracle-")
        assert Path(oracle_env_root).parent == Path("/workspace")
        # A hidden test's ../gitnexus import must resolve to the credited
        # candidate checkout, not to an unrelated /opt/gitnexus tree.
        assert Path(oracle_env_root).parent / "gitnexus" == Path("/workspace/gitnexus")
        prefix_options = sandbox_instance.command_prefix_calls[-1]
        assert prefix_options["read_only_workspace"] is True
        assert prefix_options["unshare_network"] is True
        oracle_mount = prefix_options["extra_read_only_mounts"][0]
        assert oracle_mount.target == oracle_env_root
        oracle_root = oracle_mount.source
        assert oracle_root.is_relative_to(sandbox_instance.private_root)
        assert (oracle_root / "vitest.config.mts").read_bytes() == hidden_config
        assert (oracle_root / "vitest.config.mts").read_bytes() != candidate_config.read_bytes()
        return True, "hidden"

    monkeypatch.setattr(runner, "run_claude", lambda *args, **kwargs: session_record())
    monkeypatch.setattr(runner, "run_verify", fake_verify)
    record = runner.run_arm(
        "baseline",
        {"prompt": "implement behavior", "verify": "true"},
        tmp_path,
        bench_args(),
        sandbox=sandbox_instance,
        oracle_snapshot=snapshot,
    )

    assert calls == 2
    assert record["resolved"] is True
    assert snapshot.command_digest == hashlib.sha256(snapshot.command.encode()).hexdigest()
    assert any(item.target == "vitest.config.mts" for item in snapshot.files)


@pytest.mark.skipif(os.name == "nt", reason="Vitest module-resolution fixture uses symlinks")
def test_hidden_vitest_config_executes_sibling_oracle_against_candidate_checkout(tmp_path: Path) -> None:
    """Exercise the shipped config with the same sibling layout used by bwrap."""

    repository_root = Path(__file__).resolve().parents[2]
    vitest = repository_root / "gitnexus" / "node_modules" / ".bin" / "vitest"
    if not vitest.is_file():
        pytest.skip("GitNexus Vitest dependencies are not installed")

    workspace = tmp_path / "workspace"
    candidate = workspace / "gitnexus"
    candidate.mkdir(parents=True)
    (candidate / "candidate.ts").write_text("export const candidateValue = 'candidate-workspace';\n")

    # The test file is a workspace sibling, so bare `vitest` imports resolve
    # through this harness dependency link while candidate-relative imports
    # resolve through ../gitnexus exactly as they do in the sandbox.
    (workspace / "node_modules").symlink_to(repository_root / "gitnexus" / "node_modules", target_is_directory=True)
    (candidate / "node_modules").symlink_to(repository_root / "gitnexus" / "node_modules", target_is_directory=True)

    hidden = workspace / ".wfbench-oracle-smoke"
    hidden.mkdir()
    shipped_config = repository_root / "eval" / "workflow_bench" / "oracles" / "vitest.config.mts"
    config = hidden / "vitest.config.mts"
    config.write_bytes(shipped_config.read_bytes())
    sentinel = tmp_path / "oracle-ran.txt"
    oracle = hidden / "candidate-import.oracle.test.ts"
    oracle.write_text(
        "import { writeFileSync } from 'node:fs';\n"
        "import { expect, test } from 'vitest';\n"
        "import { candidateValue } from '../gitnexus/candidate';\n"
        "test('uses the credited candidate checkout', () => {\n"
        "  expect(candidateValue).toBe('candidate-workspace');\n"
        "  writeFileSync(process.env.ORACLE_SENTINEL!, `ran:${candidateValue}`);\n"
        "});\n"
    )
    environment = os.environ.copy()
    environment["ORACLE_SENTINEL"] = str(sentinel)

    completed = subprocess.run(
        [str(vitest), "run", "--config", str(config), str(oracle)],
        cwd=candidate,
        env=environment,
        check=False,
        capture_output=True,
        text=True,
        timeout=30,
    )

    assert completed.returncode == 0, completed.stdout + completed.stderr
    assert sentinel.read_text() == "ran:candidate-workspace"


@pytest.fixture(scope="module")
def scenario_control_base(tmp_path_factory: pytest.TempPathFactory) -> tuple[Path, dict[str, dict[str, object]]]:
    """Use the immutable task source with the dependencies that grade it.

    Release evaluation grades tasks with a toolchain built at the pinned task
    commit. CI points GITNEXUS_ORACLE_TASK_DEPENDENCIES at that build; local runs
    fall back to the current checkout's dependencies.
    """

    repository = Path(__file__).resolve().parents[2]
    dependencies = Path(os.environ.get("GITNEXUS_ORACLE_TASK_DEPENDENCIES") or repository)
    if not (dependencies / "gitnexus" / "node_modules" / ".bin" / "vitest").is_file():
        pytest.skip("GitNexus Vitest dependencies are not installed")
    tasks = yaml.safe_load((repository / "eval/workflow_bench/tasks.scenarios.yaml").read_text())["tasks"]
    refs = {task["ref"] for task in tasks}
    assert refs == {"c4ecf398de637cf779ab2238e69958cfd5d601ce"}
    ref = next(iter(refs))
    environment = {**os.environ, "GIT_NO_LAZY_FETCH": "1"}
    revision = subprocess.run(
        ["git", "cat-file", "-e", f"{ref}^{{commit}}"],
        cwd=repository,
        env=environment,
        check=False,
        capture_output=True,
    )
    if revision.returncode:
        pytest.skip("pinned v1.6.12 task history is not available in this checkout")

    workspace = tmp_path_factory.mktemp("oracle-controls")
    archive = workspace / "task-base.tar"
    with archive.open("wb") as output:
        archived = subprocess.run(
            [
                "git",
                "archive",
                ref,
                "gitnexus/src",
                "gitnexus/vendor",
                "gitnexus/test",
                "gitnexus/tsconfig.json",
                "gitnexus/vitest.config.ts",
                "gitnexus/package.json",
                "gitnexus-shared/package.json",
            ],
            cwd=repository,
            env=environment,
            stdout=output,
            stderr=subprocess.PIPE,
            check=False,
            timeout=120,
        )
    if archived.returncode:
        pytest.skip("pinned task source blobs are unavailable; fetch v1.6.12 before running oracle controls")
    base = workspace / "base"
    base.mkdir()
    if not hasattr(tarfile, "data_filter"):
        pytest.skip("oracle controls need tarfile extraction filters (Python 3.11.4+)")
    with tarfile.open(archive) as source:
        source.extractall(base, filter="data")
    for relative in ("node_modules", "gitnexus/node_modules", "gitnexus-shared/node_modules", "gitnexus-shared/dist"):
        (base / relative).symlink_to(dependencies / relative, target_is_directory=True)
    subprocess.run(["git", "init", "--quiet", str(base)], check=True, capture_output=True)
    return base, {task["id"]: task for task in tasks}


def _oracle_control_candidate(base: Path, destination: Path) -> Path:
    shutil.copytree(base, destination, symlinks=True, ignore=shutil.ignore_patterns("vendor"))
    (destination / "gitnexus/vendor").symlink_to(base / "gitnexus/vendor", target_is_directory=True)
    return destination


def _verify_shipped_oracle(candidate: Path, task: dict[str, object]) -> subprocess.CompletedProcess[str]:
    snapshot = capture_task_oracle(task)
    with staged_task_oracle(candidate, snapshot) as staged:
        return subprocess.run(
            ["/bin/sh", "-c", snapshot.command],
            cwd=candidate,
            env={**os.environ, oracle_assets.ORACLE_ENV_VAR: str(staged)},
            capture_output=True,
            text=True,
            check=False,
            timeout=180,
        )


def _assert_behavioral_rejection(completed: subprocess.CompletedProcess[str], assertion: str = "") -> None:
    output = completed.stdout + completed.stderr
    assert completed.returncode == 1, output
    assert "AssertionError" in output, output
    assert assertion in output, output
    # An import/setup crash is not evidence that a behavioral control worked.
    assert "Failed Suites" not in output, output
    assert "Unhandled Errors" not in output, output


def _check_c_provenance_mutants(candidate: Path, base: Path, task: dict[str, object]) -> None:
    """Reject broken behavior, and distinguish module-load errors from evidence."""

    interpret = Path("gitnexus/src/core/ingestion/languages/c/interpret.ts")
    resolver = Path("gitnexus/src/core/ingestion/languages/c/scope-resolver.ts")
    valid_interpret = (candidate / interpret).read_text()
    valid_resolver = (candidate / resolver).read_text()
    system_guards = re.findall(
        r"\bconst\s+parsedImport\s*=\s*context\s*\?\.\s*parsedImport\s*;\s*"
        r"if\s*\(\s*parsedImport\s*\?\.\s*kind\s*===\s*'wildcard'\s*&&\s*"
        r"'isSystem'\s+in\s+parsedImport\s*&&\s*parsedImport\s*\.\s*isSystem\s*===\s*true\s*\)\s*"
        r"\{\s*return\s+null\s*;\s*\}",
        valid_resolver,
    )
    fallbacks = re.findall(
        r"\breturn\s+resolveCImportTarget\s*\(\s*targetRaw\s*,\s*fromFile\s*,\s*allFilePaths\s*,?\s*\)\s*;",
        valid_resolver,
    )
    assert len(system_guards) == 1, "expected exactly one C system-include guard"
    assert len(fallbacks) == 1, "expected exactly one C resolver fallback"
    system_guard, fallback = system_guards[0], fallbacks[0]
    late_guard_resolver = valid_resolver.replace(system_guard, "").replace(fallback, system_guard + "\n    " + fallback)
    assert valid_interpret.count("return parsedImport;") == 1
    assert valid_interpret.count("targetRaw: source,") == 1
    mutants = [
        (interpret, (base / interpret).read_text(), "literal include <stdio.h> must be retained"),
        (
            interpret,
            valid_interpret.replace("return parsedImport;", "return null;"),
            'literal include "stdio.h" must be retained',
        ),
        (interpret, valid_interpret.replace("targetRaw: source,", "targetRaw: '',"), "retained header identity"),
        (resolver, (base / resolver).read_text(), "include <stdio.h> resolution"),
        (resolver, late_guard_resolver, "include <stdio.h> resolution"),
    ]
    for relative, broken, assertion in mutants:
        path = candidate / relative
        original = path.read_text()
        assert broken != original
        path.write_text(broken)
        try:
            _assert_behavioral_rejection(_verify_shipped_oracle(candidate, task), assertion)
        finally:
            path.write_text(original)

    # A missing native-parser module must fail setup, never satisfy a negative
    # behavioral control just because Vitest exits nonzero.
    parser = candidate / "gitnexus/src/core/ingestion/languages/c/query.ts"
    original_parser = parser.read_bytes()
    parser.unlink()
    try:
        broken_setup = _verify_shipped_oracle(candidate, task)
        assert "Failed Suites" in broken_setup.stdout + broken_setup.stderr
        with pytest.raises(AssertionError):
            _assert_behavioral_rejection(broken_setup, "literal include <stdio.h> must be retained")
    finally:
        parser.write_bytes(original_parser)


@pytest.mark.skipif(os.name == "nt", reason="oracle controls use POSIX dependency symlinks and shell commands")
@pytest.mark.parametrize(
    ("task_id", "reference"),
    [
        ("trivial-status-json-alias", "trivial-status-json-alias"),
        ("inv-feature-c-include-provenance", "inv-feature-c-include-provenance-flag"),
        ("inv-feature-c-include-provenance", "inv-feature-c-include-provenance-delimiter"),
        ("inv-feature-list-repos-filter", "inv-feature-list-repos-filter"),
        ("cross-module-parse-retry", "cross-module-parse-retry"),
        ("cross-module-parse-retry", "cross-module-parse-retry-pool"),
    ],
)
def test_shipped_oracles_reject_task_base_and_accept_reference_fixes(
    tmp_path: Path,
    scenario_control_base: tuple[Path, dict[str, dict[str, object]]],
    task_id: str,
    reference: str,
) -> None:
    base, tasks = scenario_control_base
    candidate = _oracle_control_candidate(base, tmp_path / "candidate")
    negative = _verify_shipped_oracle(candidate, tasks[task_id])
    _assert_behavioral_rejection(
        negative,
        "literal include <stdio.h> must be retained" if task_id == "inv-feature-c-include-provenance" else "",
    )

    patches = Path(__file__).resolve().parents[1] / "workflow_bench/oracles/reference"
    subprocess.run(["git", "apply", str(patches / f"{reference}.patch")], cwd=candidate, check=True)
    positive = _verify_shipped_oracle(candidate, tasks[task_id])
    assert positive.returncode == 0, positive.stdout + positive.stderr
    if task_id == "inv-feature-c-include-provenance":
        # A reference passing transpile-only Vitest is not enough: use the same
        # typed, authored verification that an evaluated repair must satisfy.
        authored = subprocess.run(
            ["/bin/sh", "-c", str(tasks[task_id]["verify"])],
            cwd=candidate,
            capture_output=True,
            text=True,
            check=False,
            timeout=180,
        )
        assert authored.returncode == 0, authored.stdout + authored.stderr
        if reference == "inv-feature-c-include-provenance-flag":
            # This entry point is already selected by native CI; keep every C
            # validity control reachable without changing workflow selection.
            # Indent every existing code line differently without matching the
            # reference's formatting; the behavioral controls must still work.
            resolver = candidate / "gitnexus/src/core/ingestion/languages/c/scope-resolver.ts"
            resolver.write_text(re.sub(r"(?m)^([ \t]+)", r"\t\1", resolver.read_text()))
            positive = _verify_shipped_oracle(candidate, tasks[task_id])
            assert positive.returncode == 0, positive.stdout + positive.stderr
            _check_c_provenance_mutants(candidate, base, tasks[task_id])


@pytest.mark.skipif(os.name == "nt", reason="oracle controls use POSIX dependency symlinks and shell commands")
@pytest.mark.parametrize(
    "reference",
    ["cross-module-parse-retry-startup-only", "cross-module-parse-retry-pool-startup-only"],
)
def test_in_flight_retry_oracle_rejects_startup_only_repairs(
    tmp_path: Path,
    scenario_control_base: tuple[Path, dict[str, dict[str, object]]],
    reference: str,
) -> None:
    base, tasks = scenario_control_base
    candidate = _oracle_control_candidate(base, tmp_path / "candidate")
    patches = Path(__file__).resolve().parents[1] / "workflow_bench/oracles/reference"
    subprocess.run(["git", "apply", str(patches / f"{reference}.patch")], cwd=candidate, check=True)

    rejected = _verify_shipped_oracle(candidate, tasks["cross-module-parse-retry"])
    output = rejected.stdout + rejected.stderr
    assert rejected.returncode == 1, output
    assert "AssertionError" in output
    assert "recovers the same file after two transient" in output
    assert "stops after two retries" in output


def test_weakened_authored_tests_cannot_produce_a_promotion_decision() -> None:
    incumbent = runner.aggregate([{"class": "demo", "resolved": True, **_metrics()} for _ in range(3)])
    candidate = runner.aggregate(
        [
            {
                "class": "demo",
                "resolved": False,
                "authored_tests_passed": True,
                "oracle_passed": False,
                **_metrics(cost_usd=0.01),
            }
            for _ in range(3)
        ]
    )
    decision = evaluate_candidate(
        {"task": {"workflow": incumbent, "candidate_workflow": candidate}},
        incumbent_arm="workflow",
        candidate_arm="candidate_workflow",
        model="pinned-model",
        min_runs=3,
    )

    assert decision["decision"] == "keep_incumbent"
    assert any("resolution regressed" in reason for reason in decision["reasons"])


def _metrics(*, cost_usd: float = 1.0) -> dict[str, object]:
    return {
        "input_tokens": 10,
        "cache_creation_input_tokens": 0,
        "cache_read_input_tokens": 0,
        "output_tokens": 5,
        "cost_usd": cost_usd,
        "duration_s": 1.0,
        "num_turns": 1,
        "diff_files": 1,
        "diff_insertions": 1,
        "diff_deletions": 0,
    }