tests/test_specs.py

from __future__ import annotations

from pathlib import Path
import unittest

from agent_harness.specs import (
    load_agent_systems,
    load_edit_interfaces,
    load_experiments,
    load_harnesses,
    load_models,
    load_repositories,
    load_task_split,
    load_tasks,
    validate_configuration_tree,
)


ROOT = Path(__file__).resolve().parents[1]


class HarnessCatalogTests(unittest.TestCase):
    def test_catalog_is_complete_and_hashes_are_unique(self) -> None:
        harnesses = load_harnesses(ROOT)
        self.assertEqual(list(harnesses), [f"H{number:03d}" for number in range(21)])
        self.assertEqual(len({item.config_hash for item in harnesses.values()}), 21)
        self.assertEqual(len({item.treatment_hash for item in harnesses.values()}), 21)

    def test_core_retrieval_block_is_full_factorial(self) -> None:
        harnesses = load_harnesses(ROOT)
        observed = {
            (
                harnesses[f"H{number:03d}"].lexical,
                harnesses[f"H{number:03d}"].syntax == "tree_sitter",
                harnesses[f"H{number:03d}"].dense,
            )
            for number in range(8)
        }
        expected = {
            (lexical, syntax, dense)
            for lexical in (False, True)
            for syntax in (False, True)
            for dense in (False, True)
        }
        self.assertEqual(observed, expected)

    def test_control_harnesses_have_no_retrieval_capabilities(self) -> None:
        harnesses = load_harnesses(ROOT)
        for identifier in ("H016", "H017", "H018", "H019"):
            harness = harnesses[identifier]
            self.assertFalse(harness.exact_search)
            self.assertFalse(harness.lexical)
            self.assertFalse(harness.dense)
            self.assertEqual(harness.syntax, "raw")
            self.assertEqual(harness.graph_hops, 0)

    def test_configuration_tree_is_valid_and_embedding_is_ready(self) -> None:
        errors, warnings = validate_configuration_tree(ROOT)
        self.assertEqual(errors, [])
        self.assertFalse(any("dense retrieval" in warning for warning in warnings))
        self.assertEqual(warnings, [])

    def test_fixed_model_and_end_to_end_plan(self) -> None:
        models = load_models(ROOT)
        experiments = load_experiments(ROOT)
        self.assertEqual(models["M001"].canonical_name, "Qwen3.6-35B-A3B")
        self.assertEqual(experiments["E03"].cells_per_task(), 10)
        self.assertEqual(experiments["E03"].model_ids, ("M001",))
        self.assertEqual(models["M002"].context_length, 65536)
        self.assertEqual(experiments["E07"].cells_per_task(), 7)
        self.assertEqual(experiments["E07"].model_ids, ("M002",))

    def test_task_template_has_a_valid_schema_but_is_not_in_a_split(self) -> None:
        tasks = load_tasks(ROOT)
        self.assertEqual(len(tasks), 108)
        self.assertEqual(
            {item for item in tasks if item.startswith("TASK_CR_")},
            {f"TASK_CR_{number:03d}" for number in range(1, 16)},
        )
        self.assertEqual(tasks["TASK_EXAMPLE"].base_commit, "0" * 40)
        for task_id in ("TASK_GR_001", "TASK_GR_002", "TASK_GR_003", "TASK_GR_004", "TASK_GR_005"):
            self.assertEqual(tasks[task_id].validation_status, "retrieval_ready")
            self.assertEqual(tasks[task_id].repository_url, "https://gitlab.com/gitlab-org/gitlab-runner.git")

        end_to_end = {"TASK_CR_001", "TASK_CR_002", "TASK_CR_003", "TASK_CR_005", "TASK_CR_006", "TASK_CR_007", "TASK_CR_008", "TASK_CR_009", "TASK_CR_012", "TASK_CR_013"}
        for task_id in end_to_end:
            self.assertEqual(tasks[task_id].validation_status, "end_to_end_ready")
            self.assertTrue(tasks[task_id].gold_patch)
            self.assertTrue(tasks[task_id].test_patch)

    def test_study2_catalog_and_powered_matrix_are_frozen(self) -> None:
        systems = load_agent_systems(ROOT)
        repositories = load_repositories(ROOT)
        experiments = load_experiments(ROOT)
        split = load_task_split(ROOT / "tasks" / "splits" / "study2_confirmatory.txt")
        self.assertEqual(set(systems), {"A001", "A002"})
        self.assertEqual(set(repositories), {"R001", "R002", "R003"})
        self.assertEqual(len(split), 60)
        self.assertEqual(experiments["E08"].cells_per_task(), 14)
        self.assertEqual(experiments["E08"].cells_per_task() * len(split), 840)

    def test_study3_edit_interfaces_are_separate_and_frozen(self) -> None:
        interfaces = load_edit_interfaces(ROOT)
        self.assertEqual(set(interfaces), {"P001", "P002", "P003"})
        self.assertEqual(
            {item.edit_tool for item in interfaces.values()},
            {"apply_patch", "replace_text", "write_file"},
        )
        self.assertEqual(len({item.treatment_hash for item in interfaces.values()}), 3)

    def test_study4_fresh_task_matrix_is_frozen(self) -> None:
        tasks = load_tasks(ROOT)
        experiments = load_experiments(ROOT)
        split = load_task_split(ROOT / "tasks" / "splits" / "study4_fresh.txt")
        self.assertEqual(len(split), 20)
        self.assertEqual(
            {task_id.split("_")[2] for task_id in split}, {"R001", "R002", "R003"}
        )
        self.assertEqual(
            [sum(task_id.startswith(f"TASK_S4_{repo}_") for task_id in split) for repo in ("R001", "R002", "R003")],
            [3, 7, 10],
        )
        self.assertTrue(all(tasks[task_id].validation_status == "end_to_end_ready" for task_id in split))
        experiment = experiments["E10"]
        self.assertEqual(experiment.harness_ids, ("H000", "H007", "H018"))
        self.assertEqual(experiment.model_ids, ("M002", "M003", "M004"))


if __name__ == "__main__":
    unittest.main()