6krrt/tests/test_task_set.py

"""Validate the eval task set against reference solutions.

A check the task author cannot satisfy is a broken check, and it scores the
task set rather than the model. That has already happened here: a sequence
task asked for "the next number" and "the 11th term" in one breath, and every
model that read it correctly scored zero.

So every `code` task gets a reference implementation below, and every check
must pass against it. Every `exact` answer is recomputed rather than trusted.
These tests run offline and take milliseconds — they are the cheap guard
against spending an hour of API calls measuring a typo.
"""

from pathlib import Path

import pytest
import yaml

from eval_proficiency import score_code, score_exact

TASKS = yaml.safe_load((Path(__file__).resolve().parent.parent / "evals" / "tasks.yaml").read_text())["tasks"]
BY_ID = {t["id"]: t for t in TASKS}


# --- reference solutions --------------------------------------------------

REFERENCES = {
    "merge_intervals": '''
def merge_intervals(intervals):
    if not intervals:
        return []
    ordered = sorted(intervals, key=lambda iv: iv[0])
    out = [list(ordered[0])]
    for start, end in ordered[1:]:
        if start <= out[-1][1]:
            out[-1][1] = max(out[-1][1], end)
        else:
            out.append([start, end])
    return out
''',
    "parse_semver": r'''
import re
_SEMVER = re.compile(
    r"^(?P<major>0|[1-9]\d*)\.(?P<minor>0|[1-9]\d*)\.(?P<patch>0|[1-9]\d*)"
    r"(?:-(?P<prerelease>(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)"
    r"(?:\.(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?"
    r"(?:\+(?P<build>[0-9a-zA-Z-]+(?:\.[0-9a-zA-Z-]+)*))?$"
)

def parse_semver(version):
    m = _SEMVER.match(version or "")
    if not m:
        raise ValueError(version)
    d = m.groupdict()
    return {
        "major": int(d["major"]),
        "minor": int(d["minor"]),
        "patch": int(d["patch"]),
        "prerelease": d["prerelease"],
        "build": d["build"],
    }
''',
    "word_wrap": '''
def word_wrap(text, width):
    words = text.split()
    if not words:
        return []
    lines, current = [], words[0]
    for word in words[1:]:
        if len(current) + 1 + len(word) <= width:
            current += " " + word
        else:
            lines.append(current)
            current = word
    lines.append(current)
    return lines
''',
    "refactor_falsy_defaults": '''
_DEFAULTS = {"retries": 3, "timeout": 30, "verbose": False}

def apply_settings(overrides):
    return {k: overrides.get(k, v) for k, v in _DEFAULTS.items()}
''',
    "refactor_first_match": '''
def first_match(items, predicates):
    return next(
        (item for item in items if any(p(item) for p in predicates)), None
    )
''',
    "refactor_dispatch": '''
_TABLE = {200: "ok", 201: "created", 404: "not found", 500: "server error"}

def describe(code):
    return _TABLE.get(code, "unknown")
''',
    "debug_late_binding": '''
def make_multipliers(factors):
    return [lambda x, f=f: x * f for f in factors]
''',
    "debug_binary_search": '''
def bsearch(items, target):
    lo, hi = 0, len(items)
    while lo < hi:
        mid = (lo + hi) // 2
        if items[mid] == target:
            return mid
        elif items[mid] < target:
            lo = mid + 1
        else:
            hi = mid
    return -1
''',
    "debug_greedy_regex": '''
import re

def extract_tags(text):
    return re.findall(r"<([^<>]+)>", text)
''',
}


@pytest.mark.parametrize("task_id", sorted(REFERENCES))
def test_reference_solution_passes_every_check(task_id):
    task = BY_ID[task_id]
    score, detail = score_code(REFERENCES[task_id], task["checks"])
    assert score == 1.0, f"{task_id}: reference scored {score} ({detail})"


def test_every_code_task_has_a_reference():
    # A code task with no reference has never been validated, so its checks
    # could be wrong in exactly the way that costs an hour of API calls
    code_tasks = {t["id"] for t in TASKS if t["kind"] == "code"}
    assert code_tasks == set(REFERENCES), (
        f"unvalidated: {sorted(code_tasks - set(REFERENCES))}"
    )


# --- the buggy code really is buggy ---------------------------------------

BUGGY = {
    "refactor_first_match": '''
def first_match(items, predicates):
    found = None
    done = False
    for item in items:
        if done:
            break
        for p in predicates:
            if p(item):
                found = item
                done = True
                break
    return found
''',
    "debug_late_binding": '''
def make_multipliers(factors):
    out = []
    for f in factors:
        out.append(lambda x: x * f)
    return out
''',
    "debug_greedy_regex": '''
import re

def extract_tags(text):
    return re.findall(r"<(.+)>", text)
''',
}


def test_refactor_target_already_passes_its_own_checks():
    # A refactor task's ORIGINAL code must pass, or the task is secretly a
    # debugging task and "behaviour must not change" is a lie
    score, detail = score_code(BUGGY["refactor_first_match"], BY_ID["refactor_first_match"]["checks"])
    assert score == 1.0, f"refactor target fails its own checks: {detail}"


@pytest.mark.parametrize("task_id", ["debug_late_binding", "debug_greedy_regex"])
def test_debugging_tasks_start_broken(task_id):
    # The whole point is that the given code fails. If it passes, the task
    # measures nothing — a model could return the input unchanged.
    score, _ = score_code(BUGGY[task_id], BY_ID[task_id]["checks"])
    assert score < 1.0


# --- exact answers, recomputed --------------------------------------------

def test_percent_trap_answer():
    # 20% up then 20% down is a 4% net loss, not a wash
    original = 96 / (1.20 * 0.80)
    assert score_exact(str(original), BY_ID["math_percent_trap"]["answer"])[0] == 1.0


def test_rate_trap_answer():
    # 3 machines / 3 widgets / 3 min => one machine makes one widget in 3 min,
    # so 100 machines make 100 widgets in the same 3 minutes
    per_machine_minutes = 3
    assert score_exact(str(per_machine_minutes), BY_ID["math_rate_trap"]["answer"])[0] == 1.0


def test_counting_answer():
    # first digit 9 choices (not 0), then 9, 8, 7 for distinctness
    assert score_exact(str(9 * 9 * 8 * 7), BY_ID["math_counting"]["answer"])[0] == 1.0


def test_counting_answer_matches_brute_force():
    count = sum(
        1 for n in range(1000, 10000) if len(set(str(n))) == 4
    )
    assert str(count) == BY_ID["math_counting"]["answer"]


# --- task set hygiene -----------------------------------------------------

def test_every_task_has_the_fields_its_kind_needs():
    for task in TASKS:
        kind = task["kind"]
        assert task.get("prompt"), f"{task['id']} has no prompt"
        if kind == "code":
            assert task.get("checks"), f"{task['id']} has no checks"
        elif kind == "exact":
            assert task.get("answer") is not None, f"{task['id']} has no answer"
        elif kind == "judge":
            assert task.get("rubric"), f"{task['id']} has no rubric"
        elif kind == "tool":
            assert "expect_tool" in task, f"{task['id']} has no expect_tool"
            assert task.get("tools"), f"{task['id']} has no tools"


def test_task_ids_are_unique():
    ids = [t["id"] for t in TASKS]
    assert len(ids) == len(set(ids))