File size: 4,828 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 | from __future__ import annotations
from hashlib import sha256
import json
from pathlib import Path
from types import SimpleNamespace
import tempfile
import unittest
from agent_harness.components import Candidate
from agent_harness.live_agent_experiment import LiveToolHarness
from agent_harness.protocol_experiment import ProtocolWorkspace
from agent_harness.specs import (
load_embeddings,
load_harnesses,
load_models,
load_task_split,
load_tasks,
)
ROOT = Path(__file__).resolve().parents[1]
class Study5ExperimentTests(unittest.TestCase):
def test_frozen_manifest_counts_and_hashes(self) -> None:
expected = {"E13": 1440, "E14": 540, "E15": 540, "E16": 306}
for experiment_id, count in expected.items():
path = ROOT / "configs" / "study5" / f"{experiment_id}_cells.json"
value = json.loads(path.read_text(encoding="utf-8"))
digest = value.pop("design_sha256")
observed = sha256(
json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")
).hexdigest()
self.assertEqual(digest, observed)
self.assertEqual(len(value["cells"]), count)
identities = {
(item["task_id"], item["harness_id"], item["interface_id"], item["model_id"])
for item in value["cells"]
}
self.assertEqual(len(identities), count)
def test_e16_selection_is_frozen_before_outcomes(self) -> None:
selection = json.loads(
(ROOT / "configs" / "study5" / "E16_selection.json").read_text(encoding="utf-8")
)
digest = selection.pop("selection_sha256")
observed = sha256(
json.dumps(selection, sort_keys=True, separators=(",", ":")).encode("utf-8")
).hexdigest()
self.assertEqual(digest, observed)
self.assertEqual(
selection["selected_harnesses"],
["H000", "H001", "H002", "H007", "H008", "H014"],
)
self.assertTrue(selection["selection_is_outcome_blind_to_e16"])
def test_fresh_validation_split_is_unique_and_disjoint(self) -> None:
tasks = load_tasks(ROOT)
split = load_task_split(ROOT / "tasks" / "splits" / "study5_fresh.txt")
self.assertEqual(len(split), 17)
commits = {tasks[item].gold_commit for item in split}
prior = {task.gold_commit for task_id, task in tasks.items() if task_id not in split}
self.assertEqual(len(commits), 17)
self.assertFalse(commits & prior)
def _tools(self, harness_id: str, tree: Path) -> LiveToolHarness:
task = load_tasks(ROOT)["TASK_CR_001"]
workspace = ProtocolWorkspace(tree, ("example.go",), task, 2)
return LiveToolHarness(
load_harnesses(ROOT)[harness_id],
SimpleNamespace(),
workspace,
SimpleNamespace(),
load_models(ROOT)["M002"],
load_embeddings(ROOT)["EMB002"],
lambda transition: None,
)
def test_one_shot_and_iterative_search_policies_are_operational(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
tree = Path(temporary)
(tree / "example.go").write_text("package example\nfunc Work() {}\n", encoding="utf-8")
one_shot = self._tools("H008", tree)
one_shot._begin_search()
with self.assertRaisesRegex(ValueError, "exactly one"):
one_shot._begin_search()
iterative = self._tools("H010", tree)
iterative._begin_search()
iterative._begin_search()
self.assertEqual(iterative.search_call_count, 2)
def test_search_observation_uses_declared_packing(self) -> None:
candidate = Candidate(
path="example.go",
line_start=1,
line_end=2,
text="package example\nfunc Work() {}",
source="test",
score=1.0,
)
with tempfile.TemporaryDirectory() as temporary:
tree = Path(temporary)
(tree / "example.go").write_text("package example\nfunc Work() {}\n", encoding="utf-8")
skeleton = self._tools("H013", tree)._packed_records((candidate,))[0]
summary = self._tools("H015", tree)._packed_records((candidate,))[0]
whole = self._tools("H014", tree)._packed_records((candidate,))[0]
self.assertEqual(skeleton["packing"], "skeletons")
self.assertIn("Work", skeleton["snippet"])
self.assertEqual(summary["packing"], "role_summaries")
self.assertIn("Work", summary["snippet"])
self.assertEqual(whole["packing"], "whole_files")
self.assertIn("package example", whole["snippet"])
if __name__ == "__main__":
unittest.main()
|