Log in
benchmarksPublic

Viewing files at commit b4b51d4daa88Commit detailsView latest

Browse files
tests/test_task_loop.py 703 lines · 28.1 KB · python
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 os
7
import shutil
8
import socket
9
import tempfile
10
import time
11
import unittest
12
import http.client
13
from pathlib import Path
14
from unittest.mock import patch
15
16
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
17
18
import task_loop as lane
19
from oakbench.execution import TimedCapture
20
from oakbench.rows import LANE_REQUIRED_KEYS
21
from oakbench.subjects import Subject
22
from oakbench.remotes import RemoteResolution
23
24
25
class LoopSummaryFieldsTest(unittest.TestCase):
26
    def test_warm_delta_from_acquire_pair(self) -> None:
27
        rows = [
28
            {"operation": "task.acquire", "elapsed_ms": 1000.0},
29
            {"operation": "task.edit", "elapsed_ms": 5.0},
30
            {"operation": "followup.acquire", "elapsed_ms": 600.0},
31
            {"operation": "followup.edit", "elapsed_ms": 5.0},
32
        ]
33
        fields = lane.loop_summary_fields(rows)
34
        self.assertEqual(fields["followup_warm_delta_ms"], -400.0)
35
        self.assertEqual(fields["task_phase_ms"], 1005.0)
36
        self.assertEqual(fields["followup_phase_ms"], 605.0)
37
38
    def test_missing_acquire_yields_null_not_zero(self) -> None:
39
        fields = lane.loop_summary_fields([{"operation": "task.edit", "elapsed_ms": 5.0}])
40
        self.assertIsNone(fields["followup_warm_delta_ms"])
41
42
43
class LaneRegistrationTest(unittest.TestCase):
44
    def test_task_loop_lane_registered(self) -> None:
45
        self.assertIn("task-loop", LANE_REQUIRED_KEYS)
46
        self.assertIn("command", LANE_REQUIRED_KEYS["task-loop"])
47
48
49
class VirtualBranchReTest(unittest.TestCase):
50
    def test_parses_mount_status(self) -> None:
51
        text = "Mount: oak/bench-sync-tmp@abc (virtual branch tl-x-r0-task--9f00aa11)"
52
        match = lane.VIRTUAL_BRANCH_RE.search(text)
53
        self.assertEqual(match.group(1), "tl-x-r0-task--9f00aa11")
54
55
56
class FailurePolicyTest(unittest.TestCase):
57
    def test_step_row_never_treats_a_process_timeout_as_an_expected_error(self) -> None:
58
        subject = Subject("oak", "oak", "Oak", Path("/usr/bin/oak"))
59
        capture = TimedCapture(
60
            elapsed_ms=100.0,
61
            returncode=124,
62
            stdout_text="",
63
            stdout_bytes=0,
64
            stdout_truncated=False,
65
            stderr_text="",
66
            stderr_bytes=0,
67
            stderr_truncated=False,
68
            timed_out=True,
69
        )
70
        with tempfile.TemporaryDirectory() as tmp, patch.object(
71
            lane, "run_timed", return_value=capture
72
        ) as timed:
73
            row = lane.step_row(
74
                subject,
75
                "interrupted_publication_after_request_v1",
76
                "publication.attempt",
77
                ["oak", "push"],
78
                Path(tmp),
79
                0,
80
                {"bench_id": "test"},
81
                100,
82
                expected_returncodes=tuple(range(1, 128)),
83
                timeout_seconds=0.1,
84
            )
85
86
        self.assertEqual(timed.call_args.kwargs["timeout_seconds"], 0.1)
87
        self.assertTrue(row["timed_out"])
88
        self.assertEqual(row["returncode"], 124)
89
90
    def test_step_row_marks_unexpected_zero_exit_as_failure(self) -> None:
91
        subject = Subject("git", "git", "Git", Path("/usr/bin/git"))
92
        capture = TimedCapture(
93
            elapsed_ms=1.0,
94
            returncode=0,
95
            stdout_text="",
96
            stdout_bytes=0,
97
            stdout_truncated=False,
98
            stderr_text="",
99
            stderr_bytes=0,
100
            stderr_truncated=False,
101
        )
102
        with tempfile.TemporaryDirectory() as tmp, patch.object(lane, "run_timed", return_value=capture):
103
            row = lane.step_row(
104
                subject,
105
                "task_loop",
106
                "expect.fail",
107
                ["sh", "-c", "true"],
108
                Path(tmp),
109
                0,
110
                {
111
                    "bench_id": "test",
112
                    "profile": "task-loop",
113
                    "timestamp_utc": "test",
114
                    "benchmark_track": "agent-default",
115
                },
116
                100,
117
                expected_returncodes=(1,),
118
            )
119
120
        self.assertEqual(row["process_returncode"], 0)
121
        self.assertEqual(row["returncode"], 1)
122
123
    def test_oracle_false_marks_total_failed(self) -> None:
124
        total = {
125
            **lane.phase_total_row(
126
                [{"operation": "task.acquire", "elapsed_ms": 1.0, "returncode": 0, "command": []}],
127
                Subject("oak", "oak", "Oak", Path("/usr/bin/oak")),
128
                "task_loop",
129
                0,
130
                {
131
                    "bench_id": "test",
132
                    "profile": "task-loop",
133
                    "timestamp_utc": "test",
134
                    "benchmark_track": "agent-default",
135
                },
136
                "task_loop.total",
137
            ),
138
            "oracle_payload_on_remote_branch": False,
139
        }
140
        lane.enforce_oracle_result(total, False)
141
        self.assertEqual(total["returncode"], 1)
142
        self.assertTrue(lane.has_lane_failures([total]))
143
144
    def test_lane_failures_ignore_skips_but_fail_on_real_errors(self) -> None:
145
        self.assertFalse(lane.has_lane_failures([{"returncode": lane.SKIP_RETURNCODE, "skipped": True}]))
146
        self.assertTrue(lane.has_lane_failures([{"returncode": 1}]))
147
148
149
class LocalGitSeedTest(unittest.TestCase):
150
    def test_seeded_local_remote_advertises_main_as_head(self) -> None:
151
        subject = Subject("git", "git", "Git", Path("/usr/bin/git"))
152
        with tempfile.TemporaryDirectory() as tmp:
153
            root = Path(tmp)
154
            remote_path = root / "remote.git"
155
            init = lane.run_untimed(
156
                [str(subject.bin), "init", "--bare", "-q", str(remote_path)], root
157
            )
158
            self.assertEqual(init.returncode, 0, init.stderr)
159
            remote = RemoteResolution(
160
                repo=str(remote_path),
161
                transport="local_file",
162
                server="local-bare",
163
                env_var="generated",
164
            )
165
166
            error = lane.ensure_git_remote_seeded(subject, remote, root)
167
168
            self.assertIsNone(error)
169
            advertised = lane.run_untimed(
170
                [str(subject.bin), "symbolic-ref", "HEAD"], remote_path
171
            )
172
            self.assertEqual(advertised.returncode, 0, advertised.stderr)
173
            self.assertEqual(advertised.stdout.strip(), "refs/heads/main")
174
175
176
class AgentTaskEvidenceTest(unittest.TestCase):
177
    @unittest.skipUnless(
178
        all(
179
            os.environ.get(key)
180
            for key in (
181
                "OAK_BENCH_TEST_MATERIALIZATION_BASELINE",
182
                "OAK_BENCH_TEST_MATERIALIZATION_CANDIDATE",
183
                "OAK_BENCH_TEST_MATERIALIZATION_SERVE",
184
                "OAK_BENCH_TEST_MATERIALIZATION_SERVE_HEAD",
185
            )
186
        ),
187
        "set all OAK_BENCH_TEST_MATERIALIZATION_* paths for the actual Serve tracer",
188
    )
189
    def test_legacy_selected_materialization_actual_serve_tracer(self) -> None:
190
        baseline = Subject(
191
            "oak_baseline",
192
            "oak",
193
            "Oak baseline",
194
            Path(os.environ["OAK_BENCH_TEST_MATERIALIZATION_BASELINE"]),
195
        )
196
        candidate = Subject(
197
            "oak_candidate",
198
            "oak",
199
            "Oak candidate",
200
            Path(os.environ["OAK_BENCH_TEST_MATERIALIZATION_CANDIDATE"]),
201
        )
202
        with tempfile.TemporaryDirectory() as tmp:
203
            rows = lane.run_legacy_selected_materialization_campaign(
204
                [baseline, candidate],
205
                Path(os.environ["OAK_BENCH_TEST_MATERIALIZATION_SERVE"]),
206
                (
207
                    Path(os.environ["OAK_BENCH_TEST_MATERIALIZATION_SERVE_SOURCE"])
208
                    if os.environ.get("OAK_BENCH_TEST_MATERIALIZATION_SERVE_SOURCE")
209
                    else None
210
                ),
211
                os.environ["OAK_BENCH_TEST_MATERIALIZATION_SERVE_HEAD"],
212
                "oak_baseline",
213
                Path(tmp),
214
                1,
215
                {
216
                    "bench_id": "test",
217
                    "profile": "task-loop",
218
                    "timestamp_utc": "test",
219
                    "benchmark_track": "oak-local-loopback-materialization",
220
                },
221
                20_000,
222
                False,
223
                468,
224
                False,
225
            )
226
227
        totals = [
228
            row
229
            for row in rows
230
            if row.get("operation") == "legacy_selected_clone_materialization.total"
231
        ]
232
        self.assertEqual([row["subject"] for row in totals], ["oak_baseline", "oak_candidate"])
233
        self.assertEqual([row["materialization_passes_observed"] for row in totals], [2, 1])
234
        self.assertTrue(all(row["oracle_passed"] for row in totals))
235
        self.assertTrue(all(row["oracle_network_work_matches_baseline"] for row in totals))
236
        self.assertEqual(
237
            [row["stat_cache_metadata_valid_before_status"] for row in totals],
238
            [False, True],
239
        )
240
        self.assertTrue(all(row["materialization_timing_claimable"] is False for row in totals))
241
242
    def test_legacy_selected_receipt_requires_exact_two_phase_route(self) -> None:
243
        receipt = {
244
            "result": {"snapshot_bound": False, "head": "a" * 64, "manifest": "b" * 64},
245
            "observed": {
246
                "pull_phases": [
247
                    {"name": "initial", "materialization_passes": 0},
248
                    {"name": "legacy_selected_branch", "materialization_passes": 1},
249
                ],
250
                "materialization": {"passes_observed": 2, "written_paths": 194},
251
            },
252
        }
253
254
        evidence = lane.legacy_selected_receipt_evidence(receipt)
255
256
        self.assertTrue(evidence["legacy_route_observed"])
257
        self.assertEqual(evidence["materialization_passes_observed"], 2)
258
        self.assertEqual(evidence["head"], "a" * 64)
259
        self.assertEqual(evidence["manifest"], "b" * 64)
260
261
    def test_materialization_fixture_environment_drops_inherited_oak_configuration(self) -> None:
262
        with tempfile.TemporaryDirectory() as tmp, patch.dict(
263
            os.environ,
264
            {
265
                "HOME": "/original/home",
266
                "OAK_API_KEY": "secret",
267
                "OAK_REMOTE": "https://production.invalid",
268
                "GITHUB_TOKEN": "github-secret",
269
                "AWS_SECRET_ACCESS_KEY": "aws-secret",
270
            },
271
        ):
272
            env = lane.isolated_oak_fixture_env(Path(tmp) / "config")
273
274
        self.assertEqual(env["HOME"], "/original/home")
275
        self.assertNotIn("OAK_API_KEY", env)
276
        self.assertNotIn("OAK_REMOTE", env)
277
        self.assertNotIn("GITHUB_TOKEN", env)
278
        self.assertNotIn("AWS_SECRET_ACCESS_KEY", env)
279
        self.assertEqual(env["OAK_AUTHOR"], "oak-materialization-bench")
280
        self.assertEqual(env["OAK_NO_UPDATE_CHECK"], "1")
281
282
    def test_materialization_timing_requires_explicit_full_control_contract(self) -> None:
283
        held = {"measurement_lock": "held"}
284
        self.assertTrue(lane.materialization_timing_is_claimable(held, 10, True, True))
285
        self.assertFalse(lane.materialization_timing_is_claimable(held, 9, True, True))
286
        self.assertFalse(lane.materialization_timing_is_claimable(held, 10, False, True))
287
        self.assertFalse(
288
            lane.materialization_timing_is_claimable(
289
                {"measurement_lock": "disabled"}, 10, True, True
290
            )
291
        )
292
        self.assertFalse(lane.materialization_timing_is_claimable(held, 10, True, False))
293
294
    def test_materialization_campaign_refuses_relative_or_nonexecutable_serve(self) -> None:
295
        subject = Subject("control", "oak", "Control", Path(sys.executable))
296
        metadata = {"measurement_lock": "held"}
297
        with tempfile.TemporaryDirectory() as tmp:
298
            root = Path(tmp)
299
            relative = lane.run_legacy_selected_materialization_campaign(
300
                [subject],
301
                Path("oak"),
302
                None,
303
                "a" * 64,
304
                "control",
305
                root,
306
                1,
307
                metadata,
308
                100,
309
                False,
310
                1,
311
                False,
312
            )
313
            nonexecutable = root / "serve"
314
            nonexecutable.write_text("not executable")
315
            refused = lane.run_legacy_selected_materialization_campaign(
316
                [subject],
317
                nonexecutable,
318
                None,
319
                "a" * 64,
320
                "control",
321
                root,
322
                1,
323
                metadata,
324
                100,
325
                False,
326
                1,
327
                False,
328
            )
329
330
        self.assertEqual(relative[0]["returncode"], lane.SKIP_RETURNCODE)
331
        self.assertEqual(refused[0]["returncode"], lane.SKIP_RETURNCODE)
332
333
    def test_materialization_campaign_requires_named_control(self) -> None:
334
        subject = Subject("candidate", "oak", "Candidate", Path(sys.executable))
335
        with tempfile.TemporaryDirectory() as tmp:
336
            rows = lane.run_legacy_selected_materialization_campaign(
337
                [subject],
338
                Path(sys.executable),
339
                None,
340
                "a" * 64,
341
                "missing",
342
                Path(tmp),
343
                1,
344
                {"measurement_lock": "held"},
345
                100,
346
                False,
347
                1,
348
                False,
349
            )
350
351
        self.assertEqual(rows[0]["returncode"], lane.SKIP_RETURNCODE)
352
        self.assertIn("baseline-subject", rows[0]["skip_reason"])
353
354
    def test_normalized_worktree_snapshot_records_bytes_modes_and_ignores_vcs_state(self) -> None:
355
        with tempfile.TemporaryDirectory() as tmp:
356
            root = Path(tmp)
357
            (root / ".oak").mkdir()
358
            (root / ".oak/oak.db").write_bytes(b"private")
359
            (root / "nested").mkdir()
360
            file_path = root / "nested/tool.sh"
361
            file_path.write_bytes(b"#!/bin/sh\n")
362
            file_path.chmod(0o755)
363
364
            rows = lane.normalized_worktree_snapshot(root)
365
366
        self.assertEqual(len(rows), 1)
367
        self.assertEqual(rows[0]["path"], "nested/tool.sh")
368
        self.assertEqual(rows[0]["mode"], 0o755)
369
        self.assertEqual(rows[0]["kind"], "file")
370
        self.assertEqual(rows[0]["sha256"], "a8076d3d28d21e02012b20eaf7dbf75409a6277134439025f282e368e3305abf")
371
372
    @unittest.skipUnless(
373
        os.environ.get("OAK_BENCH_TEST_OAK_BIN"),
374
        "set OAK_BENCH_TEST_OAK_BIN to exercise the real local Oak server",
375
    )
376
    def test_local_oak_agent_task_uses_fresh_checkouts_and_remote_oracles(self) -> None:
377
        oak_bin = Path(os.environ["OAK_BENCH_TEST_OAK_BIN"])
378
        subject = Subject("oak_local", "oak", "Oak local", oak_bin)
379
        with tempfile.TemporaryDirectory() as tmp:
380
            rows = lane.run_oak_local_agent_task_baseline(
381
                subject,
382
                Path(tmp),
383
                0,
384
                {
385
                    "bench_id": "test",
386
                    "profile": "task-loop",
387
                    "timestamp_utc": "test",
388
                    "benchmark_track": "agent-default",
389
                },
390
                20_000,
391
            )
392
393
        total = next(row for row in rows if row["operation"] == "agent_task_oak_local.total")
394
        self.assertEqual(total["returncode"], 0)
395
        self.assertFalse(total["coverage_complete"])
396
        self.assertTrue(total["oracle_payload_on_remote_branch"])
397
        self.assertTrue(total["oracle_branch_head_matches"])
398
        self.assertTrue(total["server_process_stopped"])
399
        self.assertEqual(
400
            total["acquisition_models"],
401
            {"cold": "fresh_full_clone", "followup": "second_fresh_full_clone"},
402
        )
403
        self.assertEqual(total["repository_cache_state"]["followup"], "no_shared_object_store")
404
        self.assertIn("cold.ci.observe", total["unmeasured_operations"])
405
        self.assertIn("followup.review_handoff", total["unmeasured_operations"])
406
407
    def test_total_keeps_skips_unmeasured_and_aggregates_costs(self) -> None:
408
        subject = Subject("git", "git", "Git", Path("/usr/bin/git"))
409
        rows = [
410
            {
411
                "operation": "task.acquire",
412
                "elapsed_ms": 10.0,
413
                "returncode": 0,
414
                "command": ["git", "clone"],
415
                "tool_call_count": 1,
416
                "stdout_bytes": 7,
417
                "stderr_bytes": 0,
418
                "retry_count": 0,
419
                "human_action_count": 0,
420
            },
421
            {
422
                "operation": "task.ci.observe",
423
                "elapsed_ms": 0.0,
424
                "returncode": lane.SKIP_RETURNCODE,
425
                "command": [],
426
                "skipped": True,
427
                "skip_reason": "local transport has no hosted CI provider",
428
                "tool_call_count": 0,
429
                "stdout_bytes": None,
430
                "stderr_bytes": None,
431
                "retry_count": 0,
432
                "human_action_count": 0,
433
            },
434
        ]
435
436
        total = lane.phase_total_row(
437
            rows,
438
            subject,
439
            "agent_task_baseline_v1",
440
            0,
441
            {
442
                "bench_id": "test",
443
                "profile": "task-loop",
444
                "timestamp_utc": "test",
445
                "benchmark_track": "agent-default",
446
            },
447
            "agent_task.total",
448
        )
449
450
        self.assertEqual(total["returncode"], 0)
451
        self.assertFalse(total["coverage_complete"])
452
        self.assertEqual(total["unmeasured_operations"], ["task.ci.observe"])
453
        self.assertEqual(total["raw_output_bytes"], 7)
454
        self.assertEqual(total["retry_count"], 0)
455
        self.assertEqual(total["human_action_count"], 0)
456
        self.assertEqual(total["phase_elapsed_ms"], {"task": 10.0})
457
458
    def test_interrupted_publication_summary_never_claims_retry(self) -> None:
459
        fields = lane.interrupted_publication_fields(
460
            request_count=1,
461
            request_bytes=4096,
462
            declared_request_bytes=4096,
463
            request_framing_valid=True,
464
            request_body_timed_out=False,
465
            local_head_before="a" * 64,
466
            local_head_after="a" * 64,
467
            status_after={"needs_push": True, "dirty": False},
468
            worktree_payload_before="payload-hash",
469
            worktree_payload_after="payload-hash",
470
            cli_observation={"reported_outcome": "unknown", "observation": "json_error_unknown"},
471
        )
472
473
        self.assertEqual(fields["publication_outcome"], "unknown")
474
        self.assertEqual(fields["mutation_requests_observed"], 1)
475
        self.assertEqual(fields["blind_replay_requests"], 0)
476
        self.assertTrue(fields["local_head_preserved"])
477
        self.assertTrue(fields["local_work_preserved"])
478
        self.assertTrue(fields["worktree_payload_preserved"])
479
        self.assertEqual(fields["network_request_bytes_observed"], 4096)
480
481
    def test_interruption_separates_cli_report_from_fixture_delivery_knowledge(self) -> None:
482
        generic_error = {
483
            "stdout": '{"schema_version":1,"error":{"code":"http","message":"connection closed"}}',
484
            "stderr": "",
485
            "process_returncode": 6,
486
            "timed_out": False,
487
        }
488
        cli = lane.cli_publication_observation(generic_error, json_requested=True)
489
        self.assertEqual(cli["reported_outcome"], "unreported")
490
        self.assertEqual(cli["observation"], "json_error_without_publication_outcome")
491
492
        fields = lane.interrupted_publication_fields(
493
            request_count=1,
494
            request_bytes=3,
495
            declared_request_bytes=100,
496
            request_framing_valid=True,
497
            request_body_timed_out=False,
498
            local_head_before="a" * 64,
499
            local_head_after="a" * 64,
500
            status_after={"needs_push": True, "dirty": False},
501
            worktree_payload_before="payload",
502
            worktree_payload_after="payload",
503
            cli_observation=cli,
504
        )
505
506
        self.assertEqual(fields["publication_outcome"], "unreported")
507
        self.assertEqual(fields["cli_observation"], "json_error_without_publication_outcome")
508
        self.assertEqual(fields["fixture_delivery_state"], "partial_request_observed")
509
        self.assertEqual(
510
            fields["fixture_outcome_knowledge"],
511
            "indeterminate_after_partial_request_disconnect",
512
        )
513
514
    def test_interruption_requires_exact_payload_and_unpublished_state(self) -> None:
515
        fields = lane.interrupted_publication_fields(
516
            request_count=1,
517
            request_bytes=100,
518
            declared_request_bytes=100,
519
            request_framing_valid=True,
520
            request_body_timed_out=False,
521
            local_head_before="a" * 64,
522
            local_head_after="a" * 64,
523
            status_after={"needs_push": True, "dirty": False},
524
            worktree_payload_before="before-hash",
525
            worktree_payload_after="after-hash",
526
            cli_observation={"reported_outcome": "unknown", "observation": "json_error_unknown"},
527
        )
528
529
        self.assertTrue(fields["unpublished_state_preserved"])
530
        self.assertFalse(fields["worktree_payload_preserved"])
531
        self.assertFalse(fields["local_work_preserved"])
532
533
    def test_interruption_server_reads_one_complete_push_then_drops_reply(self) -> None:
534
        server, thread = lane.start_interrupted_publication_server()
535
        host, port = server.server_address[:2]
536
        try:
537
            connection = http.client.HTTPConnection(host, port, timeout=2)
538
            connection.request("POST", "/api/repos", body=b"{}", headers={"Content-Type": "application/json"})
539
            response = connection.getresponse()
540
            self.assertEqual(response.status, 200)
541
            response.read()
542
            connection.close()
543
544
            connection = http.client.HTTPConnection(host, port, timeout=2)
545
            connection.request(
546
                "POST",
547
                "/api/oak/repo/push",
548
                body=b"fixture-payload",
549
                headers={"Content-Type": "application/json"},
550
            )
551
            with self.assertRaises(http.client.RemoteDisconnected):
552
                connection.getresponse()
553
            connection.close()
554
        finally:
555
            server.shutdown()
556
            server.server_close()
557
            thread.join(timeout=2)
558
559
        self.assertEqual(server.push_request_count, 1)
560
        self.assertEqual(server.push_request_bytes, len(b"fixture-payload"))
561
        self.assertTrue(server.request_complete_before_disconnect)
562
563
    def test_interruption_server_never_certifies_a_truncated_request(self) -> None:
564
        server, thread = lane.start_interrupted_publication_server()
565
        host, port = server.server_address[:2]
566
        try:
567
            with socket.create_connection((host, port), timeout=2) as connection:
568
                connection.sendall(
569
                    b"POST /api/oak/repo/push HTTP/1.1\r\n"
570
                    b"Host: local\r\n"
571
                    b"Content-Length: 100\r\n"
572
                    b"Content-Type: application/json\r\n\r\n"
573
                    b"abc"
574
                )
575
                connection.shutdown(socket.SHUT_WR)
576
                connection.recv(1024)
577
        finally:
578
            server.shutdown()
579
            server.server_close()
580
            thread.join(timeout=2)
581
582
        self.assertEqual(server.push_request_count, 1)
583
        self.assertEqual(server.push_request_bytes, 3)
584
        self.assertEqual(server.push_declared_request_bytes, 100)
585
        self.assertTrue(server.push_request_framing_valid)
586
        self.assertFalse(server.request_complete_before_disconnect)
587
588
    def test_interruption_server_rejects_absent_and_invalid_framing(self) -> None:
589
        for framing in (
590
            b"",
591
            b"Content-Length: nope\r\n",
592
            b"Content-Length: +3\r\n",
593
            b"Content-Length: 3\r\nContent-Length: 3\r\n",
594
            f"Content-Length: {lane.INTERRUPTED_PUBLICATION_MAX_REQUEST_BYTES + 1}\r\n".encode(),
595
        ):
596
            with self.subTest(framing=framing):
597
                server, thread = lane.start_interrupted_publication_server()
598
                host, port = server.server_address[:2]
599
                try:
600
                    with socket.create_connection((host, port), timeout=2) as connection:
601
                        connection.sendall(
602
                            b"POST /api/oak/repo/push HTTP/1.1\r\n"
603
                            b"Host: local\r\n"
604
                            + framing
605
                            + b"Content-Type: application/json\r\n\r\nabc"
606
                        )
607
                        connection.shutdown(socket.SHUT_WR)
608
                        connection.recv(1024)
609
                finally:
610
                    server.shutdown()
611
                    server.server_close()
612
                    thread.join(timeout=2)
613
614
                self.assertEqual(server.push_request_count, 1)
615
                self.assertIsNone(server.push_declared_request_bytes)
616
                self.assertFalse(server.push_request_framing_valid)
617
                self.assertFalse(server.request_complete_before_disconnect)
618
619
    def test_interruption_server_accepts_http_ows_around_content_length(self) -> None:
620
        server, thread = lane.start_interrupted_publication_server()
621
        host, port = server.server_address[:2]
622
        try:
623
            with socket.create_connection((host, port), timeout=2) as connection:
624
                connection.sendall(
625
                    b"POST /api/oak/repo/push HTTP/1.1\r\n"
626
                    b"Host: local\r\n"
627
                    b"Content-Length:\t3 \t\r\n\r\nabc"
628
                )
629
                connection.shutdown(socket.SHUT_WR)
630
                connection.recv(1024)
631
        finally:
632
            server.shutdown()
633
            server.server_close()
634
            thread.join(timeout=2)
635
636
        self.assertEqual(server.push_declared_request_bytes, 3)
637
        self.assertTrue(server.push_request_framing_valid)
638
        self.assertTrue(server.request_complete_before_disconnect)
639
640
    def test_interruption_server_bounds_a_partial_request_while_sender_stays_open(self) -> None:
641
        server, thread = lane.start_interrupted_publication_server()
642
        host, port = server.server_address[:2]
643
        started = time.monotonic()
644
        try:
645
            with socket.create_connection((host, port), timeout=4) as connection:
646
                connection.sendall(
647
                    b"POST /api/oak/repo/push HTTP/1.1\r\n"
648
                    b"Host: local\r\n"
649
                    b"Content-Length: 100\r\n\r\nabc"
650
                )
651
                connection.recv(1024)
652
        finally:
653
            server.shutdown()
654
            server.server_close()
655
            thread.join(timeout=2)
656
657
        self.assertLess(time.monotonic() - started, 4)
658
        self.assertEqual(server.push_request_bytes, 3)
659
        self.assertTrue(server.push_request_framing_valid)
660
        self.assertTrue(server.push_request_body_timed_out)
661
        self.assertFalse(server.request_complete_before_disconnect)
662
663
    @unittest.skipIf(shutil.which("git") is None, "git not installed")
664
    def test_local_git_agent_task_records_cold_warm_and_unmeasured_ci(self) -> None:
665
        subject = Subject("git", "git", "Git", Path(shutil.which("git") or "git"))
666
        with tempfile.TemporaryDirectory() as tmp:
667
            root = Path(tmp)
668
            remote = lane.oakbench_remotes.make_git_bare_remote(
669
                subject.bin, root, "agent-task.git"
670
            )
671
            rows = lane.run_git_agent_task_baseline(
672
                subject,
673
                remote,
674
                root,
675
                0,
676
                {
677
                    "bench_id": "test",
678
                    "profile": "task-loop",
679
                    "timestamp_utc": "test",
680
                    "benchmark_track": "agent-default",
681
                },
682
                20_000,
683
            )
684
685
        operations = [row["operation"] for row in rows]
686
        self.assertIn("cold.acquire", operations)
687
        self.assertIn("warm.acquire", operations)
688
        self.assertIn("cold.ci.observe", operations)
689
        total = next(row for row in rows if row["operation"] == "agent_task.total")
690
        self.assertEqual(total["returncode"], 0)
691
        self.assertFalse(total["coverage_complete"])
692
        self.assertEqual(total["oracle_payload_on_remote_branch"], True)
693
        self.assertEqual(total["acquisition_models"], {
694
            "cold": "fresh_full_clone",
695
            "warm": "shared_object_worktree",
696
        })
697
        ci = next(row for row in rows if row["operation"] == "cold.ci.observe")
698
        self.assertEqual(ci["returncode"], lane.SKIP_RETURNCODE)
699
        self.assertIn("no hosted CI", ci["skip_reason"])
700
701
702
if __name__ == "__main__":
703
    unittest.main()