Log in
benchmarksPublic

Viewing files at commit 290f11640be1Commit detailsView latest

Browse files
tests/test_task_loop.py 705 lines · 28.2 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(
241
            all(row["materialization_timing_controls_met"] is False for row in totals)
242
        )
243
244
    def test_legacy_selected_receipt_requires_exact_two_phase_route(self) -> None:
245
        receipt = {
246
            "result": {"snapshot_bound": False, "head": "a" * 64, "manifest": "b" * 64},
247
            "observed": {
248
                "pull_phases": [
249
                    {"name": "initial", "materialization_passes": 0},
250
                    {"name": "legacy_selected_branch", "materialization_passes": 1},
251
                ],
252
                "materialization": {"passes_observed": 2, "written_paths": 194},
253
            },
254
        }
255
256
        evidence = lane.legacy_selected_receipt_evidence(receipt)
257
258
        self.assertTrue(evidence["legacy_route_observed"])
259
        self.assertEqual(evidence["materialization_passes_observed"], 2)
260
        self.assertEqual(evidence["head"], "a" * 64)
261
        self.assertEqual(evidence["manifest"], "b" * 64)
262
263
    def test_materialization_fixture_environment_drops_inherited_oak_configuration(self) -> None:
264
        with tempfile.TemporaryDirectory() as tmp, patch.dict(
265
            os.environ,
266
            {
267
                "HOME": "/original/home",
268
                "OAK_API_KEY": "secret",
269
                "OAK_REMOTE": "https://production.invalid",
270
                "GITHUB_TOKEN": "github-secret",
271
                "AWS_SECRET_ACCESS_KEY": "aws-secret",
272
            },
273
        ):
274
            env = lane.isolated_oak_fixture_env(Path(tmp) / "config")
275
276
        self.assertEqual(env["HOME"], "/original/home")
277
        self.assertNotIn("OAK_API_KEY", env)
278
        self.assertNotIn("OAK_REMOTE", env)
279
        self.assertNotIn("GITHUB_TOKEN", env)
280
        self.assertNotIn("AWS_SECRET_ACCESS_KEY", env)
281
        self.assertEqual(env["OAK_AUTHOR"], "oak-materialization-bench")
282
        self.assertEqual(env["OAK_NO_UPDATE_CHECK"], "1")
283
284
    def test_materialization_timing_controls_require_explicit_harness_contract(self) -> None:
285
        held = {"measurement_lock": "held"}
286
        self.assertTrue(lane.materialization_timing_controls_met(held, 10, True, True))
287
        self.assertFalse(lane.materialization_timing_controls_met(held, 9, True, True))
288
        self.assertFalse(lane.materialization_timing_controls_met(held, 10, False, True))
289
        self.assertFalse(
290
            lane.materialization_timing_controls_met(
291
                {"measurement_lock": "disabled"}, 10, True, True
292
            )
293
        )
294
        self.assertFalse(lane.materialization_timing_controls_met(held, 10, True, False))
295
296
    def test_materialization_campaign_refuses_relative_or_nonexecutable_serve(self) -> None:
297
        subject = Subject("control", "oak", "Control", Path(sys.executable))
298
        metadata = {"measurement_lock": "held"}
299
        with tempfile.TemporaryDirectory() as tmp:
300
            root = Path(tmp)
301
            relative = lane.run_legacy_selected_materialization_campaign(
302
                [subject],
303
                Path("oak"),
304
                None,
305
                "a" * 64,
306
                "control",
307
                root,
308
                1,
309
                metadata,
310
                100,
311
                False,
312
                1,
313
                False,
314
            )
315
            nonexecutable = root / "serve"
316
            nonexecutable.write_text("not executable")
317
            refused = lane.run_legacy_selected_materialization_campaign(
318
                [subject],
319
                nonexecutable,
320
                None,
321
                "a" * 64,
322
                "control",
323
                root,
324
                1,
325
                metadata,
326
                100,
327
                False,
328
                1,
329
                False,
330
            )
331
332
        self.assertEqual(relative[0]["returncode"], lane.SKIP_RETURNCODE)
333
        self.assertEqual(refused[0]["returncode"], lane.SKIP_RETURNCODE)
334
335
    def test_materialization_campaign_requires_named_control(self) -> None:
336
        subject = Subject("candidate", "oak", "Candidate", Path(sys.executable))
337
        with tempfile.TemporaryDirectory() as tmp:
338
            rows = lane.run_legacy_selected_materialization_campaign(
339
                [subject],
340
                Path(sys.executable),
341
                None,
342
                "a" * 64,
343
                "missing",
344
                Path(tmp),
345
                1,
346
                {"measurement_lock": "held"},
347
                100,
348
                False,
349
                1,
350
                False,
351
            )
352
353
        self.assertEqual(rows[0]["returncode"], lane.SKIP_RETURNCODE)
354
        self.assertIn("baseline-subject", rows[0]["skip_reason"])
355
356
    def test_normalized_worktree_snapshot_records_bytes_modes_and_ignores_vcs_state(self) -> None:
357
        with tempfile.TemporaryDirectory() as tmp:
358
            root = Path(tmp)
359
            (root / ".oak").mkdir()
360
            (root / ".oak/oak.db").write_bytes(b"private")
361
            (root / "nested").mkdir()
362
            file_path = root / "nested/tool.sh"
363
            file_path.write_bytes(b"#!/bin/sh\n")
364
            file_path.chmod(0o755)
365
366
            rows = lane.normalized_worktree_snapshot(root)
367
368
        self.assertEqual(len(rows), 1)
369
        self.assertEqual(rows[0]["path"], "nested/tool.sh")
370
        self.assertEqual(rows[0]["mode"], 0o755)
371
        self.assertEqual(rows[0]["kind"], "file")
372
        self.assertEqual(rows[0]["sha256"], "a8076d3d28d21e02012b20eaf7dbf75409a6277134439025f282e368e3305abf")
373
374
    @unittest.skipUnless(
375
        os.environ.get("OAK_BENCH_TEST_OAK_BIN"),
376
        "set OAK_BENCH_TEST_OAK_BIN to exercise the real local Oak server",
377
    )
378
    def test_local_oak_agent_task_uses_fresh_checkouts_and_remote_oracles(self) -> None:
379
        oak_bin = Path(os.environ["OAK_BENCH_TEST_OAK_BIN"])
380
        subject = Subject("oak_local", "oak", "Oak local", oak_bin)
381
        with tempfile.TemporaryDirectory() as tmp:
382
            rows = lane.run_oak_local_agent_task_baseline(
383
                subject,
384
                Path(tmp),
385
                0,
386
                {
387
                    "bench_id": "test",
388
                    "profile": "task-loop",
389
                    "timestamp_utc": "test",
390
                    "benchmark_track": "agent-default",
391
                },
392
                20_000,
393
            )
394
395
        total = next(row for row in rows if row["operation"] == "agent_task_oak_local.total")
396
        self.assertEqual(total["returncode"], 0)
397
        self.assertFalse(total["coverage_complete"])
398
        self.assertTrue(total["oracle_payload_on_remote_branch"])
399
        self.assertTrue(total["oracle_branch_head_matches"])
400
        self.assertTrue(total["server_process_stopped"])
401
        self.assertEqual(
402
            total["acquisition_models"],
403
            {"cold": "fresh_full_clone", "followup": "second_fresh_full_clone"},
404
        )
405
        self.assertEqual(total["repository_cache_state"]["followup"], "no_shared_object_store")
406
        self.assertIn("cold.ci.observe", total["unmeasured_operations"])
407
        self.assertIn("followup.review_handoff", total["unmeasured_operations"])
408
409
    def test_total_keeps_skips_unmeasured_and_aggregates_costs(self) -> None:
410
        subject = Subject("git", "git", "Git", Path("/usr/bin/git"))
411
        rows = [
412
            {
413
                "operation": "task.acquire",
414
                "elapsed_ms": 10.0,
415
                "returncode": 0,
416
                "command": ["git", "clone"],
417
                "tool_call_count": 1,
418
                "stdout_bytes": 7,
419
                "stderr_bytes": 0,
420
                "retry_count": 0,
421
                "human_action_count": 0,
422
            },
423
            {
424
                "operation": "task.ci.observe",
425
                "elapsed_ms": 0.0,
426
                "returncode": lane.SKIP_RETURNCODE,
427
                "command": [],
428
                "skipped": True,
429
                "skip_reason": "local transport has no hosted CI provider",
430
                "tool_call_count": 0,
431
                "stdout_bytes": None,
432
                "stderr_bytes": None,
433
                "retry_count": 0,
434
                "human_action_count": 0,
435
            },
436
        ]
437
438
        total = lane.phase_total_row(
439
            rows,
440
            subject,
441
            "agent_task_baseline_v1",
442
            0,
443
            {
444
                "bench_id": "test",
445
                "profile": "task-loop",
446
                "timestamp_utc": "test",
447
                "benchmark_track": "agent-default",
448
            },
449
            "agent_task.total",
450
        )
451
452
        self.assertEqual(total["returncode"], 0)
453
        self.assertFalse(total["coverage_complete"])
454
        self.assertEqual(total["unmeasured_operations"], ["task.ci.observe"])
455
        self.assertEqual(total["raw_output_bytes"], 7)
456
        self.assertEqual(total["retry_count"], 0)
457
        self.assertEqual(total["human_action_count"], 0)
458
        self.assertEqual(total["phase_elapsed_ms"], {"task": 10.0})
459
460
    def test_interrupted_publication_summary_never_claims_retry(self) -> None:
461
        fields = lane.interrupted_publication_fields(
462
            request_count=1,
463
            request_bytes=4096,
464
            declared_request_bytes=4096,
465
            request_framing_valid=True,
466
            request_body_timed_out=False,
467
            local_head_before="a" * 64,
468
            local_head_after="a" * 64,
469
            status_after={"needs_push": True, "dirty": False},
470
            worktree_payload_before="payload-hash",
471
            worktree_payload_after="payload-hash",
472
            cli_observation={"reported_outcome": "unknown", "observation": "json_error_unknown"},
473
        )
474
475
        self.assertEqual(fields["publication_outcome"], "unknown")
476
        self.assertEqual(fields["mutation_requests_observed"], 1)
477
        self.assertEqual(fields["blind_replay_requests"], 0)
478
        self.assertTrue(fields["local_head_preserved"])
479
        self.assertTrue(fields["local_work_preserved"])
480
        self.assertTrue(fields["worktree_payload_preserved"])
481
        self.assertEqual(fields["network_request_bytes_observed"], 4096)
482
483
    def test_interruption_separates_cli_report_from_fixture_delivery_knowledge(self) -> None:
484
        generic_error = {
485
            "stdout": '{"schema_version":1,"error":{"code":"http","message":"connection closed"}}',
486
            "stderr": "",
487
            "process_returncode": 6,
488
            "timed_out": False,
489
        }
490
        cli = lane.cli_publication_observation(generic_error, json_requested=True)
491
        self.assertEqual(cli["reported_outcome"], "unreported")
492
        self.assertEqual(cli["observation"], "json_error_without_publication_outcome")
493
494
        fields = lane.interrupted_publication_fields(
495
            request_count=1,
496
            request_bytes=3,
497
            declared_request_bytes=100,
498
            request_framing_valid=True,
499
            request_body_timed_out=False,
500
            local_head_before="a" * 64,
501
            local_head_after="a" * 64,
502
            status_after={"needs_push": True, "dirty": False},
503
            worktree_payload_before="payload",
504
            worktree_payload_after="payload",
505
            cli_observation=cli,
506
        )
507
508
        self.assertEqual(fields["publication_outcome"], "unreported")
509
        self.assertEqual(fields["cli_observation"], "json_error_without_publication_outcome")
510
        self.assertEqual(fields["fixture_delivery_state"], "partial_request_observed")
511
        self.assertEqual(
512
            fields["fixture_outcome_knowledge"],
513
            "indeterminate_after_partial_request_disconnect",
514
        )
515
516
    def test_interruption_requires_exact_payload_and_unpublished_state(self) -> None:
517
        fields = lane.interrupted_publication_fields(
518
            request_count=1,
519
            request_bytes=100,
520
            declared_request_bytes=100,
521
            request_framing_valid=True,
522
            request_body_timed_out=False,
523
            local_head_before="a" * 64,
524
            local_head_after="a" * 64,
525
            status_after={"needs_push": True, "dirty": False},
526
            worktree_payload_before="before-hash",
527
            worktree_payload_after="after-hash",
528
            cli_observation={"reported_outcome": "unknown", "observation": "json_error_unknown"},
529
        )
530
531
        self.assertTrue(fields["unpublished_state_preserved"])
532
        self.assertFalse(fields["worktree_payload_preserved"])
533
        self.assertFalse(fields["local_work_preserved"])
534
535
    def test_interruption_server_reads_one_complete_push_then_drops_reply(self) -> None:
536
        server, thread = lane.start_interrupted_publication_server()
537
        host, port = server.server_address[:2]
538
        try:
539
            connection = http.client.HTTPConnection(host, port, timeout=2)
540
            connection.request("POST", "/api/repos", body=b"{}", headers={"Content-Type": "application/json"})
541
            response = connection.getresponse()
542
            self.assertEqual(response.status, 200)
543
            response.read()
544
            connection.close()
545
546
            connection = http.client.HTTPConnection(host, port, timeout=2)
547
            connection.request(
548
                "POST",
549
                "/api/oak/repo/push",
550
                body=b"fixture-payload",
551
                headers={"Content-Type": "application/json"},
552
            )
553
            with self.assertRaises(http.client.RemoteDisconnected):
554
                connection.getresponse()
555
            connection.close()
556
        finally:
557
            server.shutdown()
558
            server.server_close()
559
            thread.join(timeout=2)
560
561
        self.assertEqual(server.push_request_count, 1)
562
        self.assertEqual(server.push_request_bytes, len(b"fixture-payload"))
563
        self.assertTrue(server.request_complete_before_disconnect)
564
565
    def test_interruption_server_never_certifies_a_truncated_request(self) -> None:
566
        server, thread = lane.start_interrupted_publication_server()
567
        host, port = server.server_address[:2]
568
        try:
569
            with socket.create_connection((host, port), timeout=2) as connection:
570
                connection.sendall(
571
                    b"POST /api/oak/repo/push HTTP/1.1\r\n"
572
                    b"Host: local\r\n"
573
                    b"Content-Length: 100\r\n"
574
                    b"Content-Type: application/json\r\n\r\n"
575
                    b"abc"
576
                )
577
                connection.shutdown(socket.SHUT_WR)
578
                connection.recv(1024)
579
        finally:
580
            server.shutdown()
581
            server.server_close()
582
            thread.join(timeout=2)
583
584
        self.assertEqual(server.push_request_count, 1)
585
        self.assertEqual(server.push_request_bytes, 3)
586
        self.assertEqual(server.push_declared_request_bytes, 100)
587
        self.assertTrue(server.push_request_framing_valid)
588
        self.assertFalse(server.request_complete_before_disconnect)
589
590
    def test_interruption_server_rejects_absent_and_invalid_framing(self) -> None:
591
        for framing in (
592
            b"",
593
            b"Content-Length: nope\r\n",
594
            b"Content-Length: +3\r\n",
595
            b"Content-Length: 3\r\nContent-Length: 3\r\n",
596
            f"Content-Length: {lane.INTERRUPTED_PUBLICATION_MAX_REQUEST_BYTES + 1}\r\n".encode(),
597
        ):
598
            with self.subTest(framing=framing):
599
                server, thread = lane.start_interrupted_publication_server()
600
                host, port = server.server_address[:2]
601
                try:
602
                    with socket.create_connection((host, port), timeout=2) as connection:
603
                        connection.sendall(
604
                            b"POST /api/oak/repo/push HTTP/1.1\r\n"
605
                            b"Host: local\r\n"
606
                            + framing
607
                            + b"Content-Type: application/json\r\n\r\nabc"
608
                        )
609
                        connection.shutdown(socket.SHUT_WR)
610
                        connection.recv(1024)
611
                finally:
612
                    server.shutdown()
613
                    server.server_close()
614
                    thread.join(timeout=2)
615
616
                self.assertEqual(server.push_request_count, 1)
617
                self.assertIsNone(server.push_declared_request_bytes)
618
                self.assertFalse(server.push_request_framing_valid)
619
                self.assertFalse(server.request_complete_before_disconnect)
620
621
    def test_interruption_server_accepts_http_ows_around_content_length(self) -> None:
622
        server, thread = lane.start_interrupted_publication_server()
623
        host, port = server.server_address[:2]
624
        try:
625
            with socket.create_connection((host, port), timeout=2) as connection:
626
                connection.sendall(
627
                    b"POST /api/oak/repo/push HTTP/1.1\r\n"
628
                    b"Host: local\r\n"
629
                    b"Content-Length:\t3 \t\r\n\r\nabc"
630
                )
631
                connection.shutdown(socket.SHUT_WR)
632
                connection.recv(1024)
633
        finally:
634
            server.shutdown()
635
            server.server_close()
636
            thread.join(timeout=2)
637
638
        self.assertEqual(server.push_declared_request_bytes, 3)
639
        self.assertTrue(server.push_request_framing_valid)
640
        self.assertTrue(server.request_complete_before_disconnect)
641
642
    def test_interruption_server_bounds_a_partial_request_while_sender_stays_open(self) -> None:
643
        server, thread = lane.start_interrupted_publication_server()
644
        host, port = server.server_address[:2]
645
        started = time.monotonic()
646
        try:
647
            with socket.create_connection((host, port), timeout=4) as connection:
648
                connection.sendall(
649
                    b"POST /api/oak/repo/push HTTP/1.1\r\n"
650
                    b"Host: local\r\n"
651
                    b"Content-Length: 100\r\n\r\nabc"
652
                )
653
                connection.recv(1024)
654
        finally:
655
            server.shutdown()
656
            server.server_close()
657
            thread.join(timeout=2)
658
659
        self.assertLess(time.monotonic() - started, 4)
660
        self.assertEqual(server.push_request_bytes, 3)
661
        self.assertTrue(server.push_request_framing_valid)
662
        self.assertTrue(server.push_request_body_timed_out)
663
        self.assertFalse(server.request_complete_before_disconnect)
664
665
    @unittest.skipIf(shutil.which("git") is None, "git not installed")
666
    def test_local_git_agent_task_records_cold_warm_and_unmeasured_ci(self) -> None:
667
        subject = Subject("git", "git", "Git", Path(shutil.which("git") or "git"))
668
        with tempfile.TemporaryDirectory() as tmp:
669
            root = Path(tmp)
670
            remote = lane.oakbench_remotes.make_git_bare_remote(
671
                subject.bin, root, "agent-task.git"
672
            )
673
            rows = lane.run_git_agent_task_baseline(
674
                subject,
675
                remote,
676
                root,
677
                0,
678
                {
679
                    "bench_id": "test",
680
                    "profile": "task-loop",
681
                    "timestamp_utc": "test",
682
                    "benchmark_track": "agent-default",
683
                },
684
                20_000,
685
            )
686
687
        operations = [row["operation"] for row in rows]
688
        self.assertIn("cold.acquire", operations)
689
        self.assertIn("warm.acquire", operations)
690
        self.assertIn("cold.ci.observe", operations)
691
        total = next(row for row in rows if row["operation"] == "agent_task.total")
692
        self.assertEqual(total["returncode"], 0)
693
        self.assertFalse(total["coverage_complete"])
694
        self.assertEqual(total["oracle_payload_on_remote_branch"], True)
695
        self.assertEqual(total["acquisition_models"], {
696
            "cold": "fresh_full_clone",
697
            "warm": "shared_object_worktree",
698
        })
699
        ci = next(row for row in rows if row["operation"] == "cold.ci.observe")
700
        self.assertEqual(ci["returncode"], lane.SKIP_RETURNCODE)
701
        self.assertIn("no hosted CI", ci["skip_reason"])
702
703
704
if __name__ == "__main__":
705
    unittest.main()