GitNexus/eval/tests/test_evolve.py

"""Unit tests for the pure evidence/apply/driver helpers of workflow_bench.evolve."""

import hashlib
import json
import os
import subprocess
import sys
import time
from datetime import UTC, datetime, timedelta

import pytest

from workflow_bench import evolve
from workflow_bench.runner_sessions import PARENT_EVENT_STREAM_SOURCE
from workflow_bench.evolve import (
    build_parser,
    build_proposer_prompt,
    generation_timeout_seconds,
    load_jsonl,
    proposer_evidence_entries,
    read_learnings,
    resolve_incumbent_arms,
    runner_argv,
    select_evidence,
    summarize_gate,
    validate_promotion_for_apply,
)
from workflow_bench.process_control import run_managed
from workflow_bench.proposer_sandbox import pid_namespace_command, preflight_bubblewrap


def row(**overrides):
    base = {
        "task": "demo-task",
        "class": "trivial",
        "arm": "workflow",
        "run": 0,
        "resolved": True,
        "error_kind": None,
        "cost_usd": 1.0,
        "num_turns": 10,
        "output_tokens": 400,
        "session_ids": ["sess-1"],
        "verify_output": "ok",
    }
    base.update(overrides)
    return base


def test_select_evidence_puts_unresolved_before_expensive_resolved():
    rows = [
        row(task="cheap", resolved=True, cost_usd=0.5),
        row(task="fail", resolved=False, error_kind="verify-failed"),
        row(task="pricey", resolved=True, cost_usd=9.0),
    ]
    picked = select_evidence(rows)
    assert [r["task"] for r in picked] == ["fail", "pricey", "cheap"]


def test_select_evidence_excludes_infra_error_rows_and_caps():
    rows = [
        row(task="harness-died", resolved=False, error_kind="infra-error"),
        row(task="session-died", resolved=False, error_kind="session-error"),
        row(
            task="missing-transcript",
            resolved=False,
            error_kind="evidence-unverified",
        ),
    ]
    rows += [row(task=f"t{i}", cost_usd=float(i)) for i in range(20)]
    picked = select_evidence(rows, max_rows=5)
    assert len(picked) == 5
    assert all(r["error_kind"] != "infra-error" for r in picked)
    assert [r["task"] for r in picked] == ["t19", "t18", "t17", "t16", "t15"]


def test_select_evidence_tolerates_an_explicit_null_cost():
    # A foreign --seed-results row (e.g. hand-edited or from another tool)
    # can carry an explicit JSON null rather than omitting the key; .get's
    # default only covers the missing-key case, so this must not raise.
    rows = [row(task="no-cost", resolved=True, cost_usd=None), row(task="priced", cost_usd=5.0)]
    picked = select_evidence(rows)
    assert [r["task"] for r in picked] == ["priced", "no-cost"]


def test_load_jsonl_skips_blank_and_malformed_lines(tmp_path):
    path = tmp_path / "learnings.jsonl"
    path.write_text('{"skill": "gitnexus-plan"}\n\nnot json\n[1, 2]\n{"skill": "gitnexus-work"}\n')
    assert load_jsonl(path) == [{"skill": "gitnexus-plan"}, {"skill": "gitnexus-work"}]


def test_load_jsonl_missing_file_is_empty(tmp_path):
    assert load_jsonl(tmp_path / "absent.jsonl") == []


def test_read_learnings_keeps_the_most_recent_entries(tmp_path):
    path = tmp_path / "learnings.jsonl"
    rows = [{"skill": "gitnexus-work", "n": i} for i in range(10)] + [
        {"skill": "gitnexus-review", "n": 10},
        {"skill": "gitnexus-lfg", "n": 11},
    ]
    path.write_text("\n".join(json.dumps(row) for row in rows) + "\n")
    assert read_learnings(path, cap=3) == [
        {"skill": "gitnexus-work", "n": 7},
        {"skill": "gitnexus-work", "n": 8},
        {"skill": "gitnexus-work", "n": 9},
    ]


def test_summarize_gate_one_line_per_decision():
    promotion = {
        "decisions": [
            {
                "candidate_arm": "candidate_workflow",
                "decision": "keep_incumbent",
                "reasons": ["a", "b", "c", "d"],
            }
        ]
    }
    lines = summarize_gate(promotion)
    assert lines == ["candidate_workflow: keep_incumbent — a; b; c"]


def test_build_proposer_prompt_carries_evidence_constraints_and_paths(tmp_path):
    prompt = build_proposer_prompt(
        results_dir=tmp_path / "bench",
        evidence=[row(task="fail", resolved=False, error_kind="verify-failed")],
        learnings=[{"skill": "gitnexus-work", "friction": "budget blown on reruns"}],
        gate_summary=["candidate_workflow: keep_incumbent — quality regressed"],
        overlay_dir=tmp_path / "overlay",
        proposal_path=tmp_path / "proposal.md",
        incumbent_arms=["workflow"],
    )
    assert str(tmp_path / "overlay") in prompt
    assert str(tmp_path / "proposal.md") in prompt
    assert "gitnexus-plan, gitnexus-work" in prompt
    assert "node .gitnexus/run.cjs analyze" in prompt
    assert "1 row(s) in /evidence/learnings.json" in prompt
    assert "1 selected row(s) in /evidence/selected-rows.json" in prompt
    assert "1 decision(s) in /evidence/gate-summary.json" in prompt
    assert "budget blown on reruns" not in prompt
    assert "verify-failed" not in prompt
    assert "~/.claude/projects" not in prompt


def test_build_proposer_prompt_first_generation_has_no_results_dir(tmp_path):
    prompt = build_proposer_prompt(
        results_dir=None,
        evidence=[],
        learnings=[],
        gate_summary=[],
        overlay_dir=tmp_path / "overlay",
        proposal_path=tmp_path / "proposal.md",
        incumbent_arms=["workflow_direct"],
    )
    assert "none (first generation)" in prompt
    assert "none yet — use the incumbent skills and staged learning queue" in prompt


def test_proposer_reads_only_digest_bound_transcripts_below_results(tmp_path, monkeypatch):
    results = tmp_path / "results"
    transcripts = results / "transcripts"
    transcripts.mkdir(parents=True, mode=0o700)
    transcripts.chmod(0o700)
    payload = b'{"message":{"content":[{"type":"text","text":"bound transcript"}]}}\n'
    artifact = transcripts / "task-workflow-run0-session.jsonl"
    artifact.write_bytes(payload)
    artifact.chmod(0o600)
    metadata = {
        "path": "transcripts/task-workflow-run0-session.jsonl",
        "sha256": hashlib.sha256(payload).hexdigest(),
        "bytes": len(payload),
        "source": PARENT_EVENT_STREAM_SOURCE,
    }
    foreign_home = tmp_path / "foreign-home"
    foreign = foreign_home / ".claude" / "projects" / "other" / "private.jsonl"
    foreign.parent.mkdir(parents=True)
    foreign.write_text("foreign host transcript")
    monkeypatch.setenv("HOME", str(foreign_home))

    entries = proposer_evidence_entries(
        results_dir=results,
        evidence=[row(session_ids=["**/*"], transcript_artifacts=[metadata])],
        learnings=[],
        gate_summary=[],
    )

    assert entries["transcript-0-0.jsonl"] == payload.decode()
    assert "foreign host transcript" not in json.dumps(entries)

    bad_digest = {**metadata, "sha256": "0" * 64}
    with pytest.raises(evolve.SandboxError, match="digest does not match"):
        proposer_evidence_entries(
            results_dir=results,
            evidence=[row(transcript_artifacts=[bad_digest])],
            learnings=[],
            gate_summary=[],
        )


@pytest.mark.skipif(os.name == "nt", reason="transcript symlink containment is POSIX-only")
def test_proposer_rejects_symlink_and_foreign_transcript_artifacts(tmp_path):
    results = tmp_path / "results"
    transcripts = results / "transcripts"
    transcripts.mkdir(parents=True, mode=0o700)
    transcripts.chmod(0o700)
    outside = tmp_path / "outside.jsonl"
    outside.write_text("outside")
    linked = transcripts / "linked.jsonl"
    linked.symlink_to(outside)
    link_metadata = {
        "path": "transcripts/linked.jsonl",
        "sha256": hashlib.sha256(outside.read_bytes()).hexdigest(),
        "bytes": outside.stat().st_size,
        "source": PARENT_EVENT_STREAM_SOURCE,
    }

    with pytest.raises(evolve.SandboxError, match="regular non-symlink"):
        proposer_evidence_entries(
            results_dir=results,
            evidence=[row(transcript_artifacts=[link_metadata])],
            learnings=[],
            gate_summary=[],
        )
    with pytest.raises(evolve.SandboxError, match="unsafe results artifact path"):
        proposer_evidence_entries(
            results_dir=results,
            evidence=[
                row(
                    transcript_artifacts=[
                        {
                            "path": "../outside.jsonl",
                            "sha256": "0" * 64,
                            "bytes": 0,
                            "source": PARENT_EVENT_STREAM_SOURCE,
                        }
                    ]
                )
            ],
            learnings=[],
            gate_summary=[],
        )


def test_proposer_rejects_duplicate_transcript_metadata_before_materializing():
    metadata = {
        "path": "transcripts/repeated.jsonl",
        "sha256": "0" * 64,
        "bytes": 0,
        "source": PARENT_EVENT_STREAM_SOURCE,
    }

    with pytest.raises(evolve.SandboxError, match="duplicate transcript artifact path"):
        proposer_evidence_entries(
            results_dir=None,
            evidence=[
                row(
                    transcript_artifacts=[
                        metadata,
                        {**metadata, "path": "transcripts//repeated.jsonl"},
                    ]
                )
            ],
            learnings=[],
            gate_summary=[],
        )


def test_proposer_bounds_transcript_metadata_per_row_and_globally_before_materializing():
    def metadata(index):
        return {
            "path": f"transcripts/session-{index}.jsonl",
            "sha256": "0" * 64,
            "bytes": 0,
            "source": PARENT_EVENT_STREAM_SOURCE,
        }

    with pytest.raises(evolve.SandboxError, match="per-row session limit"):
        proposer_evidence_entries(
            results_dir=None,
            evidence=[row(transcript_artifacts=[metadata(index) for index in range(3)])],
            learnings=[],
            gate_summary=[],
        )

    rows = [
        row(
            run=index,
            transcript_artifacts=[metadata(2 * index), metadata(2 * index + 1)],
        )
        for index in range(evolve.MAX_EVIDENCE_ROWS + 1)
    ]
    with pytest.raises(evolve.SandboxError, match="global evidence limit"):
        proposer_evidence_entries(
            results_dir=None,
            evidence=rows,
            learnings=[],
            gate_summary=[],
        )


def test_parser_defaults_match_the_gate_minimums():
    args = build_parser().parse_args(["--tasks", "t.yaml", "--model", "pinned"])
    assert args.runs == 3
    assert args.generations == 1
    assert args.arms is None
    assert args.apply is False
    assert args.learnings.name == "learnings.jsonl"


@pytest.mark.parametrize(
    "arguments",
    [
        ["--model", "Auto"],
        ["--model", "provider/latest"],
        ["--model", "pinned-model", "--proposer-model", "vendor@LATEST"],
    ],
)
def test_evolve_rejects_mutable_model_aliases(monkeypatch, tmp_path, capsys, arguments):
    monkeypatch.setattr(
        sys,
        "argv",
        ["workflow_bench.evolve", "--tasks", str(tmp_path / "missing.yaml"), *arguments],
    )
    with pytest.raises(SystemExit):
        evolve.main()
    assert "mutable auto/latest" in capsys.readouterr().err


def test_evolve_proposer_failure_returns_nonzero(monkeypatch, tmp_path):
    tasks = tmp_path / "tasks.yaml"
    tasks.write_text(
        """tasks:
  - id: demo
    class: test
    repo: .
    prompt: implement
    verify: "true"
    oracle:
      command: "true"
      files:
        - source: hidden.test.ts
          target: hidden.test.ts
"""
    )
    monkeypatch.setattr(
        sys,
        "argv",
        [
            "workflow_bench.evolve",
            "--tasks",
            str(tasks),
            "--model",
            "pinned-model",
            "--out-root",
            str(tmp_path / "out"),
        ],
    )
    monkeypatch.setattr(evolve.runner, "selected_task_bindings", lambda _tasks: [{"id": "demo"}])
    monkeypatch.setattr(evolve, "preflight_bubblewrap", lambda: tmp_path / "bwrap")
    monkeypatch.setattr(evolve, "require_claude_sandbox_helpers", lambda: None)
    monkeypatch.setattr(
        evolve,
        "run_proposer",
        lambda *args, **kwargs: {"ok": False, "error_detail": "proposer failed"},
    )

    assert evolve.main() == 1


def test_proposer_session_record_is_redacted_before_upload(monkeypatch, tmp_path):
    tasks = tmp_path / "tasks.yaml"
    tasks.write_text(
        """tasks:
  - id: demo
    class: test
    repo: .
    prompt: implement
    verify: "true"
    oracle:
      command: "true"
      files:
        - source: hidden.test.ts
          target: hidden.test.ts
"""
    )
    literal_token = "secret-LITERAL-XYZ"
    pattern_token = "sk-ant-FAKEEXAMPLE0000"
    monkeypatch.setattr(
        sys,
        "argv",
        [
            "workflow_bench.evolve",
            "--tasks",
            str(tasks),
            "--model",
            "pinned-model",
            "--out-root",
            str(tmp_path / "out"),
            "--auth-token",
            literal_token,
        ],
    )
    monkeypatch.setattr(evolve.runner, "selected_task_bindings", lambda _tasks: [{"id": "demo"}])
    monkeypatch.setattr(evolve, "preflight_bubblewrap", lambda: tmp_path / "bwrap")
    monkeypatch.setattr(evolve, "require_claude_sandbox_helpers", lambda: None)
    # A session error whose stderr echoed both the literal API key and an
    # sk-ant-shaped token into the record that gets written to the artifact.
    monkeypatch.setattr(
        evolve,
        "run_proposer",
        lambda *args, **kwargs: {
            "ok": False,
            "error_detail": {"stderr_tail": f"boom {literal_token} {pattern_token}"},
        },
    )

    assert evolve.main() == 1

    written = (tmp_path / "out" / "gen-0" / "proposer-session.json").read_text()
    assert literal_token not in written
    assert pattern_token not in written
    assert "[REDACTED]" in written


def test_runner_argv_pairs_each_incumbent_with_its_candidate(tmp_path):
    args = build_parser().parse_args(
        [
            "--tasks",
            "t.yaml",
            "--model",
            "pinned",
            "--arms",
            "workflow",
            "--include-expensive",
        ]
    )
    overlay = tmp_path / "overlay"
    skill = overlay / ".claude" / "skills" / "gitnexus-plan" / "SKILL.md"
    skill.parent.mkdir(parents=True)
    skill.write_text("candidate")
    task_bindings = [{"id": "task", "resolved_sha": "a" * 40}]
    target_bases = {".claude/skills/gitnexus-plan/SKILL.md": "b" * 64}
    argv = runner_argv(
        args,
        tmp_path / "bench",
        overlay,
        task_bindings=task_bindings,
        target_base_digests=target_bases,
        proposer_model="pinned",
    )
    arms = argv[argv.index("--arms") + 1 : argv.index("--promotion-metric")]
    assert arms == ["workflow", "candidate_workflow"]
    assert str(overlay) in argv
    assert str(tmp_path / "bench") in argv
    assert "pinned" in argv
    assert argv[argv.index("--proposer-model") + 1] == "pinned"
    assert "--include-expensive" in argv
    assert json.loads(argv[argv.index("--task-bindings-json") + 1]) == task_bindings
    assert json.loads(argv[argv.index("--promotion-target-bases-json") + 1]) == target_bases


def test_runner_argv_omits_proposer_for_manual_overlay(tmp_path):
    args = build_parser().parse_args(["--tasks", "t.yaml", "--model", "pinned"])
    overlay = tmp_path / "overlay"
    skill = overlay / ".claude" / "skills" / "gitnexus-plan" / "SKILL.md"
    skill.parent.mkdir(parents=True)
    skill.write_text("candidate")

    argv = runner_argv(
        args,
        tmp_path / "bench",
        overlay,
        task_bindings=[{"id": "task"}],
        target_base_digests={},
        proposer_model=None,
    )

    assert "--proposer-model" not in argv


def test_runner_argv_keeps_task_commit_pinned_when_ref_moves(tmp_path):
    repo = tmp_path / "task-repo"
    repo.mkdir()

    def git(*arguments):
        return subprocess.run(
            ["git", "-C", str(repo), *arguments],
            check=True,
            capture_output=True,
            text=True,
        ).stdout.strip()

    git("init", "-b", "main")
    git("config", "user.name", "Workflow Bench Test")
    git("config", "user.email", "workflow-bench@example.invalid")
    tracked = repo / "tracked.txt"
    tracked.write_text("one")
    git("add", "tracked.txt")
    git("commit", "-m", "first")
    first_sha = git("rev-parse", "HEAD")
    task = {
        "id": "moving-ref",
        "class": "test",
        "repo": str(repo),
        "ref": "main",
        "prompt": "test prompt",
        "verify": "true",
        "oracle": {
            "command": "true",
            "files": [
                {
                    "source": "trivial-version-alias.oracle.test.ts",
                    "target": "oracle.test.ts",
                }
            ],
        },
    }
    bindings = evolve.runner.selected_task_bindings([task])

    tracked.write_text("two")
    git("commit", "-am", "second")
    assert git("rev-parse", "main") != first_sha

    args = build_parser().parse_args(["--tasks", "t.yaml", "--model", "pinned"])
    overlay = tmp_path / "overlay"
    skill = overlay / ".claude" / "skills" / "gitnexus-plan" / "SKILL.md"
    skill.parent.mkdir(parents=True)
    skill.write_text("candidate")
    argv = runner_argv(
        args,
        tmp_path / "bench",
        overlay,
        task_bindings=bindings,
        target_base_digests={},
    )
    forwarded = json.loads(argv[argv.index("--task-bindings-json") + 1])

    assert forwarded[0]["resolved_sha"] == first_sha
    assert evolve.runner.resolve_task_bindings([task], forwarded)[0]["resolved_sha"] == first_sha


def test_generation_timeout_budgets_three_task_workflow_pair():
    timeout = generation_timeout_seconds(
        task_count=3,
        runs=3,
        session_timeout=3600,
        incumbent_arms=["workflow"],
    )

    per_task_preparation = (
        evolve.TASK_BINDING_GIT_PHASES * evolve.GIT_COMMAND_TIMEOUT_SECONDS
        + 2 * evolve.TASK_SNAPSHOT_TIMEOUT_SECONDS
        + evolve.WORKTREE_PREPARATION_TIMEOUT_SECONDS
        + evolve.GRAPH_SOURCE_PREPARATION_TIMEOUT_SECONDS
        + evolve.GRAPH_BUILD_TIMEOUT_SECONDS
        + 2 * evolve.GRAPH_QUERY_TIMEOUT_SECONDS
        + evolve.CLEANUP_TIMEOUT_SECONDS
    )
    paired_arm_cells = 2
    session_slots = 4
    workspace_snapshot_slots = 4
    per_task_run = session_slots * (3600 + evolve.SESSION_FINALIZATION_TIMEOUT_SECONDS) + paired_arm_cells * (
        evolve.WORKTREE_PREPARATION_TIMEOUT_SECONDS
        + evolve.ARM_ASSET_MATERIALIZATION_PHASES * evolve.TASK_SNAPSHOT_TIMEOUT_SECONDS
        + evolve.SETUP_TIMEOUT_SECONDS
        + 2 * 3600
        + evolve.ARM_EVIDENCE_GIT_PHASES * evolve.GIT_COMMAND_TIMEOUT_SECONDS
        + evolve.CLEANUP_TIMEOUT_SECONDS
    )
    per_task_run += workspace_snapshot_slots * evolve.TASK_SNAPSHOT_TIMEOUT_SECONDS
    per_task_run += evolve.CANDIDATE_OVERLAY_GIT_PHASES * evolve.GIT_COMMAND_TIMEOUT_SECONDS

    assert timeout == (
        evolve.PROMOTION_BASE_TIMEOUT_SECONDS
        + 3 * (per_task_preparation + 3 * per_task_run)
        + evolve.DRIVER_OVERHEAD_SECONDS
    )
    # The old deadline omitted clone sanitization entirely. Every graph seed
    # and every paired arm cell must now receive the full bounded envelope.
    assert timeout >= 3 * (1 + 3 * paired_arm_cells) * evolve.WORKTREE_PREPARATION_TIMEOUT_SECONDS


@pytest.mark.skipif(sys.platform != "linux", reason="Bubblewrap PID namespaces require Linux")
def test_outer_runner_pid_namespace_kills_setsid_descendant(tmp_path):
    try:
        bwrap = preflight_bubblewrap()
    except evolve.SandboxError as exc:
        pytest.skip(str(exc))
        raise AssertionError("pytest.skip() returned unexpectedly")
    sentinel = tmp_path / "escaped"
    child = (
        "import os,subprocess,sys,time; "
        f"subprocess.Popen([sys.executable,'-c',\"import time,pathlib;time.sleep(1);pathlib.Path({str(sentinel)!r}).touch()\"],preexec_fn=os.setsid); "
        "time.sleep(10)"
    )
    result = run_managed(
        pid_namespace_command([sys.executable, "-c", child], bwrap_bin=bwrap),
        timeout=0.15,
        terminate_grace=0.1,
        require_pid_namespace=True,
    )
    time.sleep(1.1)

    assert not result.ok
    assert result.state in {"timeout", "forced-kill"}
    assert not sentinel.exists()


def test_resolve_incumbent_arms_rejects_incomplete_and_extra_explicit_sets(tmp_path):
    plan = tmp_path / "plan"
    plan_skill = plan / ".claude" / "skills" / "gitnexus-plan" / "SKILL.md"
    plan_skill.parent.mkdir(parents=True)
    plan_skill.write_text("plan")
    assert resolve_incumbent_arms(plan, None) == ["workflow"]
    with pytest.raises(ValueError, match="exactly"):
        resolve_incumbent_arms(plan, ["workflow", "workflow_direct"])

    work = tmp_path / "work"
    work_skill = work / ".claude" / "skills" / "gitnexus-work" / "SKILL.md"
    work_skill.parent.mkdir(parents=True)
    work_skill.write_text("work")
    assert resolve_incumbent_arms(work, None) == ["workflow", "workflow_direct"]
    with pytest.raises(ValueError, match="exactly"):
        resolve_incumbent_arms(work, ["workflow"])


def bound_task_fixture():
    return {
        "id": "task",
        "prompt_digest": "prompt",
        "oracle_digest": "a" * 64,
        "oracle_command_digest": "b" * 64,
        "oracle_manifest_digest": "c" * 64,
        "sandbox_dependency_content_digest": "e" * 64,
        "sandbox_dependency_manifest_digest": "f" * 64,
        "oracle_files": [{"target": "oracle.test.ts", "sha256": "d" * 64, "size": 10}],
    }


def promotion_fixture(*, decisions=None, expires_delta=timedelta(days=1)):
    now = datetime.now(UTC)
    return {
        "schema_version": 3,
        "generated_at": now.isoformat(),
        "evidence_expires_at": (now + expires_delta).isoformat(),
        "benchmark_model": "bench-model",
        "proposer_model": "proposer-model",
        "candidate_origin": "model-proposer",
        "candidate_overlay_digest": "digest",
        "target_base_digests": {"path": "base"},
        "required_candidate_arms": ["candidate_workflow"],
        "selected_tasks": [bound_task_fixture()],
        "policy": {
            "metric": "cost_usd",
            "min_runs": 3,
            "min_improvement_pct": 5.0,
            "max_task_regression_pct": 20.0,
        },
        "decisions": (
            decisions
            if decisions is not None
            else [
                {
                    "incumbent_arm": "workflow",
                    "candidate_arm": "candidate_workflow",
                    "decision": "promote",
                    "metric": "cost_usd",
                }
            ]
        ),
    }


def validate_fixture(promotion):
    return validate_promotion_for_apply(
        promotion,
        overlay_digest="digest",
        benchmark_model="bench-model",
        proposer_model="proposer-model",
        selected_tasks=[bound_task_fixture()],
        target_base_digests={"path": "base"},
        required_candidate_arms=["candidate_workflow"],
        policy={
            "metric": "cost_usd",
            "min_runs": 3,
            "min_improvement_pct": 5.0,
            "max_task_regression_pct": 20.0,
        },
    )


def test_promotion_apply_requires_one_promote_for_every_bound_arm():
    assert [d["candidate_arm"] for d in validate_fixture(promotion_fixture())] == ["candidate_workflow"]

    for decisions in (
        [],
        [
            {
                "incumbent_arm": "workflow",
                "candidate_arm": "candidate_workflow",
                "decision": "keep_incumbent",
                "metric": "cost_usd",
            }
        ],
        [
            {
                "incumbent_arm": "workflow",
                "candidate_arm": "candidate_workflow",
                "decision": "promote",
                "metric": "cost_usd",
            },
            {
                "incumbent_arm": "workflow",
                "candidate_arm": "candidate_workflow",
                "decision": "promote",
                "metric": "cost_usd",
            },
        ],
        [
            {
                "incumbent_arm": "workflow_direct",
                "candidate_arm": "candidate_workflow_direct",
                "decision": "promote",
                "metric": "cost_usd",
            }
        ],
    ):
        with pytest.raises(ValueError):
            validate_fixture(promotion_fixture(decisions=decisions))


def test_manual_initial_overlay_has_no_fictitious_proposer_model():
    promotion = promotion_fixture()
    promotion["proposer_model"] = None
    promotion["candidate_origin"] = "manual-initial-overlay"

    decisions = validate_promotion_for_apply(
        promotion,
        overlay_digest="digest",
        benchmark_model="bench-model",
        proposer_model=None,
        selected_tasks=[bound_task_fixture()],
        target_base_digests={"path": "base"},
        required_candidate_arms=["candidate_workflow"],
        policy=promotion["policy"],
    )

    assert decisions[0]["decision"] == "promote"


def test_promotion_apply_rejects_pre_oracle_schema_and_missing_oracle_bindings():
    legacy = promotion_fixture()
    legacy["schema_version"] = 2
    with pytest.raises(ValueError, match="unsupported schema"):
        validate_fixture(legacy)

    weak_task = {"id": "task", "prompt_digest": "prompt"}
    weak = promotion_fixture()
    weak["selected_tasks"] = [weak_task]
    with pytest.raises(ValueError, match="hidden-oracle or dependency digests"):
        validate_promotion_for_apply(
            weak,
            overlay_digest="digest",
            benchmark_model="bench-model",
            proposer_model="proposer-model",
            selected_tasks=[weak_task],
            target_base_digests={"path": "base"},
            required_candidate_arms=["candidate_workflow"],
            policy={
                "metric": "cost_usd",
                "min_runs": 3,
                "min_improvement_pct": 5.0,
                "max_task_regression_pct": 20.0,
            },
        )


@pytest.mark.parametrize(
    ("field", "value"),
    [
        ("benchmark_model", "other"),
        ("proposer_model", "other"),
        ("candidate_overlay_digest", "other"),
        ("target_base_digests", {"path": "other"}),
        ("required_candidate_arms", ["candidate_workflow_direct"]),
        ("selected_tasks", [{"id": "other", "prompt_digest": "prompt"}]),
        (
            "policy",
            {
                "metric": "cost_usd",
                "min_runs": 4,
                "min_improvement_pct": 5.0,
                "max_task_regression_pct": 20.0,
            },
        ),
    ],
)
def test_promotion_apply_rejects_mismatched_evidence_bindings(field, value):
    promotion = promotion_fixture()
    promotion[field] = value
    with pytest.raises(ValueError, match="binding"):
        validate_fixture(promotion)


def test_promotion_apply_rejects_expired_evidence():
    with pytest.raises(ValueError, match="expired"):
        validate_fixture(promotion_fixture(expires_delta=timedelta(seconds=-1)))


def test_promotion_apply_rejects_extended_or_future_dated_evidence():
    with pytest.raises(ValueError, match="expired"):
        validate_fixture(promotion_fixture(expires_delta=timedelta(days=91)))

    promotion = promotion_fixture()
    future = datetime.now(UTC) + timedelta(days=1)
    promotion["generated_at"] = future.isoformat()
    promotion["evidence_expires_at"] = (future + timedelta(days=1)).isoformat()
    with pytest.raises(ValueError, match="future"):
        validate_fixture(promotion)


def test_promotion_apply_rejects_decision_metric_mismatch():
    promotion = promotion_fixture()
    promotion["decisions"][0]["metric"] = "output_tokens"
    with pytest.raises(ValueError, match="metric mismatch"):
        validate_fixture(promotion)