| 1 | """Tests for the task-loop lane's derived fields and lane registration."""
|
| 2 |
|
| 3 | from __future__ import annotations
|
| 4 |
|
| 5 | import sys
|
| 6 | import shutil
|
| 7 | import tempfile
|
| 8 | import unittest
|
| 9 | import http.client
|
| 10 | from pathlib import Path
|
| 11 | from unittest.mock import patch
|
| 12 |
|
| 13 | sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
|
| 14 |
|
| 15 | import task_loop as lane
|
| 16 | from oakbench.execution import TimedCapture
|
| 17 | from oakbench.rows import LANE_REQUIRED_KEYS
|
| 18 | from oakbench.subjects import Subject
|
| 19 | from oakbench.remotes import RemoteResolution
|
| 20 |
|
| 21 |
|
| 22 | class LoopSummaryFieldsTest(unittest.TestCase):
|
| 23 | def test_warm_delta_from_acquire_pair(self) -> None:
|
| 24 | rows = [
|
| 25 | {"operation": "task.acquire", "elapsed_ms": 1000.0},
|
| 26 | {"operation": "task.edit", "elapsed_ms": 5.0},
|
| 27 | {"operation": "followup.acquire", "elapsed_ms": 600.0},
|
| 28 | {"operation": "followup.edit", "elapsed_ms": 5.0},
|
| 29 | ]
|
| 30 | fields = lane.loop_summary_fields(rows)
|
| 31 | self.assertEqual(fields["followup_warm_delta_ms"], -400.0)
|
| 32 | self.assertEqual(fields["task_phase_ms"], 1005.0)
|
| 33 | self.assertEqual(fields["followup_phase_ms"], 605.0)
|
| 34 |
|
| 35 | def test_missing_acquire_yields_null_not_zero(self) -> None:
|
| 36 | fields = lane.loop_summary_fields([{"operation": "task.edit", "elapsed_ms": 5.0}])
|
| 37 | self.assertIsNone(fields["followup_warm_delta_ms"])
|
| 38 |
|
| 39 |
|
| 40 | class LaneRegistrationTest(unittest.TestCase):
|
| 41 | def test_task_loop_lane_registered(self) -> None:
|
| 42 | self.assertIn("task-loop", LANE_REQUIRED_KEYS)
|
| 43 | self.assertIn("command", LANE_REQUIRED_KEYS["task-loop"])
|
| 44 |
|
| 45 |
|
| 46 | class VirtualBranchReTest(unittest.TestCase):
|
| 47 | def test_parses_mount_status(self) -> None:
|
| 48 | text = "Mount: oak/bench-sync-tmp@abc (virtual branch tl-x-r0-task--9f00aa11)"
|
| 49 | match = lane.VIRTUAL_BRANCH_RE.search(text)
|
| 50 | self.assertEqual(match.group(1), "tl-x-r0-task--9f00aa11")
|
| 51 |
|
| 52 |
|
| 53 | class FailurePolicyTest(unittest.TestCase):
|
| 54 | def test_step_row_marks_unexpected_zero_exit_as_failure(self) -> None:
|
| 55 | subject = Subject("git", "git", "Git", Path("/usr/bin/git"))
|
| 56 | capture = TimedCapture(
|
| 57 | elapsed_ms=1.0,
|
| 58 | returncode=0,
|
| 59 | stdout_text="",
|
| 60 | stdout_bytes=0,
|
| 61 | stdout_truncated=False,
|
| 62 | stderr_text="",
|
| 63 | stderr_bytes=0,
|
| 64 | stderr_truncated=False,
|
| 65 | )
|
| 66 | with tempfile.TemporaryDirectory() as tmp, patch.object(lane, "run_timed", return_value=capture):
|
| 67 | row = lane.step_row(
|
| 68 | subject,
|
| 69 | "task_loop",
|
| 70 | "expect.fail",
|
| 71 | ["sh", "-c", "true"],
|
| 72 | Path(tmp),
|
| 73 | 0,
|
| 74 | {
|
| 75 | "bench_id": "test",
|
| 76 | "profile": "task-loop",
|
| 77 | "timestamp_utc": "test",
|
| 78 | "benchmark_track": "agent-default",
|
| 79 | },
|
| 80 | 100,
|
| 81 | expected_returncodes=(1,),
|
| 82 | )
|
| 83 |
|
| 84 | self.assertEqual(row["process_returncode"], 0)
|
| 85 | self.assertEqual(row["returncode"], 1)
|
| 86 |
|
| 87 | def test_oracle_false_marks_total_failed(self) -> None:
|
| 88 | total = {
|
| 89 | **lane.phase_total_row(
|
| 90 | [{"operation": "task.acquire", "elapsed_ms": 1.0, "returncode": 0, "command": []}],
|
| 91 | Subject("oak", "oak", "Oak", Path("/usr/bin/oak")),
|
| 92 | "task_loop",
|
| 93 | 0,
|
| 94 | {
|
| 95 | "bench_id": "test",
|
| 96 | "profile": "task-loop",
|
| 97 | "timestamp_utc": "test",
|
| 98 | "benchmark_track": "agent-default",
|
| 99 | },
|
| 100 | "task_loop.total",
|
| 101 | ),
|
| 102 | "oracle_payload_on_remote_branch": False,
|
| 103 | }
|
| 104 | lane.enforce_oracle_result(total, False)
|
| 105 | self.assertEqual(total["returncode"], 1)
|
| 106 | self.assertTrue(lane.has_lane_failures([total]))
|
| 107 |
|
| 108 | def test_lane_failures_ignore_skips_but_fail_on_real_errors(self) -> None:
|
| 109 | self.assertFalse(lane.has_lane_failures([{"returncode": lane.SKIP_RETURNCODE, "skipped": True}]))
|
| 110 | self.assertTrue(lane.has_lane_failures([{"returncode": 1}]))
|
| 111 |
|
| 112 |
|
| 113 | class LocalGitSeedTest(unittest.TestCase):
|
| 114 | def test_seeded_local_remote_advertises_main_as_head(self) -> None:
|
| 115 | subject = Subject("git", "git", "Git", Path("/usr/bin/git"))
|
| 116 | with tempfile.TemporaryDirectory() as tmp:
|
| 117 | root = Path(tmp)
|
| 118 | remote_path = root / "remote.git"
|
| 119 | init = lane.run_untimed(
|
| 120 | [str(subject.bin), "init", "--bare", "-q", str(remote_path)], root
|
| 121 | )
|
| 122 | self.assertEqual(init.returncode, 0, init.stderr)
|
| 123 | remote = RemoteResolution(
|
| 124 | repo=str(remote_path),
|
| 125 | transport="local_file",
|
| 126 | server="local-bare",
|
| 127 | env_var="generated",
|
| 128 | )
|
| 129 |
|
| 130 | error = lane.ensure_git_remote_seeded(subject, remote, root)
|
| 131 |
|
| 132 | self.assertIsNone(error)
|
| 133 | advertised = lane.run_untimed(
|
| 134 | [str(subject.bin), "symbolic-ref", "HEAD"], remote_path
|
| 135 | )
|
| 136 | self.assertEqual(advertised.returncode, 0, advertised.stderr)
|
| 137 | self.assertEqual(advertised.stdout.strip(), "refs/heads/main")
|
| 138 |
|
| 139 |
|
| 140 | class AgentTaskEvidenceTest(unittest.TestCase):
|
| 141 | def test_total_keeps_skips_unmeasured_and_aggregates_costs(self) -> None:
|
| 142 | subject = Subject("git", "git", "Git", Path("/usr/bin/git"))
|
| 143 | rows = [
|
| 144 | {
|
| 145 | "operation": "task.acquire",
|
| 146 | "elapsed_ms": 10.0,
|
| 147 | "returncode": 0,
|
| 148 | "command": ["git", "clone"],
|
| 149 | "tool_call_count": 1,
|
| 150 | "stdout_bytes": 7,
|
| 151 | "stderr_bytes": 0,
|
| 152 | "retry_count": 0,
|
| 153 | "human_action_count": 0,
|
| 154 | },
|
| 155 | {
|
| 156 | "operation": "task.ci.observe",
|
| 157 | "elapsed_ms": 0.0,
|
| 158 | "returncode": lane.SKIP_RETURNCODE,
|
| 159 | "command": [],
|
| 160 | "skipped": True,
|
| 161 | "skip_reason": "local transport has no hosted CI provider",
|
| 162 | "tool_call_count": 0,
|
| 163 | "stdout_bytes": None,
|
| 164 | "stderr_bytes": None,
|
| 165 | "retry_count": 0,
|
| 166 | "human_action_count": 0,
|
| 167 | },
|
| 168 | ]
|
| 169 |
|
| 170 | total = lane.phase_total_row(
|
| 171 | rows,
|
| 172 | subject,
|
| 173 | "agent_task_baseline_v1",
|
| 174 | 0,
|
| 175 | {
|
| 176 | "bench_id": "test",
|
| 177 | "profile": "task-loop",
|
| 178 | "timestamp_utc": "test",
|
| 179 | "benchmark_track": "agent-default",
|
| 180 | },
|
| 181 | "agent_task.total",
|
| 182 | )
|
| 183 |
|
| 184 | self.assertEqual(total["returncode"], 0)
|
| 185 | self.assertFalse(total["coverage_complete"])
|
| 186 | self.assertEqual(total["unmeasured_operations"], ["task.ci.observe"])
|
| 187 | self.assertEqual(total["raw_output_bytes"], 7)
|
| 188 | self.assertEqual(total["retry_count"], 0)
|
| 189 | self.assertEqual(total["human_action_count"], 0)
|
| 190 | self.assertEqual(total["phase_elapsed_ms"], {"task": 10.0})
|
| 191 |
|
| 192 | def test_interrupted_publication_summary_never_claims_retry(self) -> None:
|
| 193 | fields = lane.interrupted_publication_fields(
|
| 194 | request_count=1,
|
| 195 | request_bytes=4096,
|
| 196 | local_head_before="a" * 64,
|
| 197 | local_head_after="a" * 64,
|
| 198 | status_after={"needs_push": True, "dirty": False},
|
| 199 | request_complete_before_disconnect=True,
|
| 200 | )
|
| 201 |
|
| 202 | self.assertEqual(fields["publication_outcome"], "unknown")
|
| 203 | self.assertEqual(fields["mutation_requests_observed"], 1)
|
| 204 | self.assertEqual(fields["blind_replay_requests"], 0)
|
| 205 | self.assertTrue(fields["local_head_preserved"])
|
| 206 | self.assertTrue(fields["local_work_preserved"])
|
| 207 | self.assertEqual(fields["network_request_bytes_observed"], 4096)
|
| 208 |
|
| 209 | def test_interruption_server_reads_one_complete_push_then_drops_reply(self) -> None:
|
| 210 | server, thread = lane.start_interrupted_publication_server()
|
| 211 | host, port = server.server_address[:2]
|
| 212 | try:
|
| 213 | connection = http.client.HTTPConnection(host, port, timeout=2)
|
| 214 | connection.request("POST", "/api/repos", body=b"{}", headers={"Content-Type": "application/json"})
|
| 215 | response = connection.getresponse()
|
| 216 | self.assertEqual(response.status, 200)
|
| 217 | response.read()
|
| 218 | connection.close()
|
| 219 |
|
| 220 | connection = http.client.HTTPConnection(host, port, timeout=2)
|
| 221 | connection.request(
|
| 222 | "POST",
|
| 223 | "/api/oak/repo/push",
|
| 224 | body=b"fixture-payload",
|
| 225 | headers={"Content-Type": "application/json"},
|
| 226 | )
|
| 227 | with self.assertRaises(http.client.RemoteDisconnected):
|
| 228 | connection.getresponse()
|
| 229 | connection.close()
|
| 230 | finally:
|
| 231 | server.shutdown()
|
| 232 | server.server_close()
|
| 233 | thread.join(timeout=2)
|
| 234 |
|
| 235 | self.assertEqual(server.push_request_count, 1)
|
| 236 | self.assertEqual(server.push_request_bytes, len(b"fixture-payload"))
|
| 237 | self.assertTrue(server.request_complete_before_disconnect)
|
| 238 |
|
| 239 | @unittest.skipIf(shutil.which("git") is None, "git not installed")
|
| 240 | def test_local_git_agent_task_records_cold_warm_and_unmeasured_ci(self) -> None:
|
| 241 | subject = Subject("git", "git", "Git", Path(shutil.which("git") or "git"))
|
| 242 | with tempfile.TemporaryDirectory() as tmp:
|
| 243 | root = Path(tmp)
|
| 244 | remote = lane.oakbench_remotes.make_git_bare_remote(
|
| 245 | subject.bin, root, "agent-task.git"
|
| 246 | )
|
| 247 | rows = lane.run_git_agent_task_baseline(
|
| 248 | subject,
|
| 249 | remote,
|
| 250 | root,
|
| 251 | 0,
|
| 252 | {
|
| 253 | "bench_id": "test",
|
| 254 | "profile": "task-loop",
|
| 255 | "timestamp_utc": "test",
|
| 256 | "benchmark_track": "agent-default",
|
| 257 | },
|
| 258 | 20_000,
|
| 259 | )
|
| 260 |
|
| 261 | operations = [row["operation"] for row in rows]
|
| 262 | self.assertIn("cold.acquire", operations)
|
| 263 | self.assertIn("warm.acquire", operations)
|
| 264 | self.assertIn("cold.ci.observe", operations)
|
| 265 | total = next(row for row in rows if row["operation"] == "agent_task.total")
|
| 266 | self.assertEqual(total["returncode"], 0)
|
| 267 | self.assertFalse(total["coverage_complete"])
|
| 268 | self.assertEqual(total["oracle_payload_on_remote_branch"], True)
|
| 269 | self.assertEqual(total["acquisition_models"], {
|
| 270 | "cold": "fresh_full_clone",
|
| 271 | "warm": "shared_object_worktree",
|
| 272 | })
|
| 273 | ci = next(row for row in rows if row["operation"] == "cold.ci.observe")
|
| 274 | self.assertEqual(ci["returncode"], lane.SKIP_RETURNCODE)
|
| 275 | self.assertIn("no hosted CI", ci["skip_reason"])
|
| 276 |
|
| 277 |
|
| 278 | if __name__ == "__main__":
|
| 279 | unittest.main()
|