File size: 6,301 Bytes
d61821a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 | from __future__ import annotations
from pathlib import Path
import tempfile
import unittest
from agent_harness.protocol_experiment import (
ProtocolToolHarness,
ProtocolWorkspace,
_identity,
protocol_tool_definitions,
)
from agent_harness.repair_experiment import PatchOutputError
from agent_harness.specs import (
load_edit_interfaces,
load_experiments,
load_harnesses,
load_models,
load_task_split,
load_tasks,
)
from agent_harness.study2_experiment import tokenizer_for
ROOT = Path(__file__).resolve().parents[1]
class ProtocolExperimentTests(unittest.TestCase):
def setUp(self) -> None:
self.task = load_tasks(ROOT)["TASK_CR_001"]
self.interfaces = load_edit_interfaces(ROOT)
def test_frozen_matrix_has_540_cells(self) -> None:
experiment = load_experiments(ROOT)["E09"]
split = load_task_split(ROOT / "tasks" / "splits" / "study3_protocol.txt")
self.assertEqual(len(split), 60)
self.assertEqual(experiment.cells_per_task(), 9)
self.assertEqual(experiment.cells_per_task() * len(split), 540)
self.assertEqual(experiment.edit_interface_ids, ("P001", "P002", "P003"))
self.assertEqual(experiment.model_ids, ("M002", "M003", "M004"))
def test_identity_records_ancillary_seed_and_context(self) -> None:
identity = _identity(
load_experiments(ROOT)["E12"],
load_tasks(ROOT)["TASK_S4_R001_003"],
self.interfaces["P002"],
load_models(ROOT)["M002"],
"a" * 40,
seed=2,
context_budget=16384,
retrieval_harness=load_harnesses(ROOT)["H007"],
)
self.assertEqual(identity.seed, 2)
self.assertEqual(identity.context_budget, 16384)
def test_every_protocol_model_has_a_frozen_tokenizer(self) -> None:
models = load_models(ROOT)
for model_id in ("M002", "M003", "M004"):
tokenizer = tokenizer_for(models[model_id])
self.assertTrue(tokenizer.path.is_file())
self.assertGreater(tokenizer.count("protocol compatibility"), 0)
def test_structured_edits_create_canonical_final_diffs(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
tree = Path(temporary)
target = tree / "example.go"
target.write_text("package example\n\nconst value = 1\n", encoding="utf-8")
workspace = ProtocolWorkspace(tree, ("example.go",), self.task, 2)
replaced = workspace.replace_text("example.go", "value = 1", "value = 2")
self.assertTrue(replaced["accepted"])
self.assertIn("-const value = 1", workspace.final_patch())
self.assertIn("+const value = 2", workspace.final_patch())
written = workspace.write_file(
"example.go", "package example\n\nconst value = 3\n"
)
self.assertTrue(written["accepted"])
self.assertIn("+const value = 3", workspace.final_patch())
def test_exact_replace_rejects_ambiguous_source(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
tree = Path(temporary)
(tree / "example.go").write_text("x := 1\nx := 1\n", encoding="utf-8")
workspace = ProtocolWorkspace(tree, ("example.go",), self.task, 2)
with self.assertRaisesRegex(ValueError, "exactly once"):
workspace.replace_text("example.go", "x := 1", "x := 2")
def test_each_arm_exposes_only_its_assigned_edit_tool(self) -> None:
for interface in self.interfaces.values():
names = [
item["function"]["name"]
for item in protocol_tool_definitions(interface, self.task)
]
self.assertEqual(
set(names), {"read_file", interface.edit_tool, "run_tests", "finish"}
)
def test_retrieval_protocol_preserves_one_tool_signature(self) -> None:
harnesses = load_harnesses(ROOT)
for harness_id in ("H000", "H007"):
names = [
item["function"]["name"]
for item in protocol_tool_definitions(
self.interfaces["P003"], self.task, harnesses[harness_id]
)
]
self.assertEqual(
names, ["search_code", "read_file", "write_file", "run_tests", "finish"]
)
oracle_names = [
item["function"]["name"]
for item in protocol_tool_definitions(
self.interfaces["P002"], self.task, harnesses["H018"]
)
]
self.assertNotIn("search_code", oracle_names)
def test_specialized_protocol_exposes_separate_search_actions(self) -> None:
names = [
item["function"]["name"]
for item in protocol_tool_definitions(
self.interfaces["P002"], self.task, load_harnesses(ROOT)["H011"]
)
]
self.assertEqual(
names[:5],
["search_exact", "search_lexical", "search_syntax", "search_dense", "search_graph"],
)
def test_calling_an_unassigned_edit_tool_is_rejected(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
tree = Path(temporary)
(tree / "example.go").write_text("package example\n", encoding="utf-8")
workspace = ProtocolWorkspace(tree, ("example.go",), self.task, 2)
tools = ProtocolToolHarness(self.interfaces["P002"], workspace)
with self.assertRaisesRegex(ValueError, "unavailable"):
tools.execute("apply_patch", {"patch": "not used"})
def test_malformed_raw_diff_uses_the_scored_tool_error_domain(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
tree = Path(temporary)
(tree / "example.go").write_text("package example\n", encoding="utf-8")
workspace = ProtocolWorkspace(tree, ("example.go",), self.task, 2)
tools = ProtocolToolHarness(self.interfaces["P001"], workspace)
with self.assertRaisesRegex(PatchOutputError, "no modified repository path"):
tools.execute("apply_patch", {"patch": "not a unified diff"})
if __name__ == "__main__":
unittest.main()
|