tests/test_task_loop.py · oak/benchmarks

Viewing files at commit 54b9920be8dfCommit details View latest

Browse files

tests/test_task_loop.py 449 lines · 17.9 KB · python

"""Tests for the task-loop lane's derived fields and lane registration."""

from __future__ import annotations

import sys
import shutil
import socket
import tempfile
import time
import unittest
import http.client
from pathlib import Path
from unittest.mock import patch

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))

import task_loop as lane
from oakbench.execution import TimedCapture
from oakbench.rows import LANE_REQUIRED_KEYS
from oakbench.subjects import Subject
from oakbench.remotes import RemoteResolution


class LoopSummaryFieldsTest(unittest.TestCase):
    def test_warm_delta_from_acquire_pair(self) -> None:
        rows = [
            {"operation": "task.acquire", "elapsed_ms": 1000.0},
            {"operation": "task.edit", "elapsed_ms": 5.0},
            {"operation": "followup.acquire", "elapsed_ms": 600.0},
            {"operation": "followup.edit", "elapsed_ms": 5.0},
        ]
        fields = lane.loop_summary_fields(rows)
        self.assertEqual(fields["followup_warm_delta_ms"], -400.0)
        self.assertEqual(fields["task_phase_ms"], 1005.0)
        self.assertEqual(fields["followup_phase_ms"], 605.0)

    def test_missing_acquire_yields_null_not_zero(self) -> None:
        fields = lane.loop_summary_fields([{"operation": "task.edit", "elapsed_ms": 5.0}])
        self.assertIsNone(fields["followup_warm_delta_ms"])


class LaneRegistrationTest(unittest.TestCase):
    def test_task_loop_lane_registered(self) -> None:
        self.assertIn("task-loop", LANE_REQUIRED_KEYS)
        self.assertIn("command", LANE_REQUIRED_KEYS["task-loop"])


class VirtualBranchReTest(unittest.TestCase):
    def test_parses_mount_status(self) -> None:
        text = "Mount: oak/bench-sync-tmp@abc (virtual branch tl-x-r0-task--9f00aa11)"
        match = lane.VIRTUAL_BRANCH_RE.search(text)
        self.assertEqual(match.group(1), "tl-x-r0-task--9f00aa11")


class FailurePolicyTest(unittest.TestCase):
    def test_step_row_never_treats_a_process_timeout_as_an_expected_error(self) -> None:
        subject = Subject("oak", "oak", "Oak", Path("/usr/bin/oak"))
        capture = TimedCapture(
            elapsed_ms=100.0,
            returncode=124,
            stdout_text="",
            stdout_bytes=0,
            stdout_truncated=False,
            stderr_text="",
            stderr_bytes=0,
            stderr_truncated=False,
            timed_out=True,
        )
        with tempfile.TemporaryDirectory() as tmp, patch.object(
            lane, "run_timed", return_value=capture
        ) as timed:
            row = lane.step_row(
                subject,
                "interrupted_publication_after_request_v1",
                "publication.attempt",
                ["oak", "push"],
                Path(tmp),
                0,
                {"bench_id": "test"},
                100,
                expected_returncodes=tuple(range(1, 128)),
                timeout_seconds=0.1,
            )

        self.assertEqual(timed.call_args.kwargs["timeout_seconds"], 0.1)
        self.assertTrue(row["timed_out"])
        self.assertEqual(row["returncode"], 124)

    def test_step_row_marks_unexpected_zero_exit_as_failure(self) -> None:
        subject = Subject("git", "git", "Git", Path("/usr/bin/git"))
        capture = TimedCapture(
            elapsed_ms=1.0,
            returncode=0,
            stdout_text="",
            stdout_bytes=0,
            stdout_truncated=False,
            stderr_text="",
            stderr_bytes=0,
            stderr_truncated=False,
        )
        with tempfile.TemporaryDirectory() as tmp, patch.object(lane, "run_timed", return_value=capture):
            row = lane.step_row(
                subject,
                "task_loop",
                "expect.fail",
                ["sh", "-c", "true"],
                Path(tmp),
                0,
                {
                    "bench_id": "test",
                    "profile": "task-loop",
                    "timestamp_utc": "test",
                    "benchmark_track": "agent-default",
                },
                100,
                expected_returncodes=(1,),
            )

        self.assertEqual(row["process_returncode"], 0)
        self.assertEqual(row["returncode"], 1)

    def test_oracle_false_marks_total_failed(self) -> None:
        total = {
            **lane.phase_total_row(
                [{"operation": "task.acquire", "elapsed_ms": 1.0, "returncode": 0, "command": []}],
                Subject("oak", "oak", "Oak", Path("/usr/bin/oak")),
                "task_loop",
                0,
                {
                    "bench_id": "test",
                    "profile": "task-loop",
                    "timestamp_utc": "test",
                    "benchmark_track": "agent-default",
                },
                "task_loop.total",
            ),
            "oracle_payload_on_remote_branch": False,
        }
        lane.enforce_oracle_result(total, False)
        self.assertEqual(total["returncode"], 1)
        self.assertTrue(lane.has_lane_failures([total]))

    def test_lane_failures_ignore_skips_but_fail_on_real_errors(self) -> None:
        self.assertFalse(lane.has_lane_failures([{"returncode": lane.SKIP_RETURNCODE, "skipped": True}]))
        self.assertTrue(lane.has_lane_failures([{"returncode": 1}]))


class LocalGitSeedTest(unittest.TestCase):
    def test_seeded_local_remote_advertises_main_as_head(self) -> None:
        subject = Subject("git", "git", "Git", Path("/usr/bin/git"))
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            remote_path = root / "remote.git"
            init = lane.run_untimed(
                [str(subject.bin), "init", "--bare", "-q", str(remote_path)], root
            )
            self.assertEqual(init.returncode, 0, init.stderr)
            remote = RemoteResolution(
                repo=str(remote_path),
                transport="local_file",
                server="local-bare",
                env_var="generated",
            )

            error = lane.ensure_git_remote_seeded(subject, remote, root)

            self.assertIsNone(error)
            advertised = lane.run_untimed(
                [str(subject.bin), "symbolic-ref", "HEAD"], remote_path
            )
            self.assertEqual(advertised.returncode, 0, advertised.stderr)
            self.assertEqual(advertised.stdout.strip(), "refs/heads/main")


class AgentTaskEvidenceTest(unittest.TestCase):
    def test_total_keeps_skips_unmeasured_and_aggregates_costs(self) -> None:
        subject = Subject("git", "git", "Git", Path("/usr/bin/git"))
        rows = [
            {
                "operation": "task.acquire",
                "elapsed_ms": 10.0,
                "returncode": 0,
                "command": ["git", "clone"],
                "tool_call_count": 1,
                "stdout_bytes": 7,
                "stderr_bytes": 0,
                "retry_count": 0,
                "human_action_count": 0,
            },
            {
                "operation": "task.ci.observe",
                "elapsed_ms": 0.0,
                "returncode": lane.SKIP_RETURNCODE,
                "command": [],
                "skipped": True,
                "skip_reason": "local transport has no hosted CI provider",
                "tool_call_count": 0,
                "stdout_bytes": None,
                "stderr_bytes": None,
                "retry_count": 0,
                "human_action_count": 0,
            },
        ]

        total = lane.phase_total_row(
            rows,
            subject,
            "agent_task_baseline_v1",
            0,
            {
                "bench_id": "test",
                "profile": "task-loop",
                "timestamp_utc": "test",
                "benchmark_track": "agent-default",
            },
            "agent_task.total",
        )

        self.assertEqual(total["returncode"], 0)
        self.assertFalse(total["coverage_complete"])
        self.assertEqual(total["unmeasured_operations"], ["task.ci.observe"])
        self.assertEqual(total["raw_output_bytes"], 7)
        self.assertEqual(total["retry_count"], 0)
        self.assertEqual(total["human_action_count"], 0)
        self.assertEqual(total["phase_elapsed_ms"], {"task": 10.0})

    def test_interrupted_publication_summary_never_claims_retry(self) -> None:
        fields = lane.interrupted_publication_fields(
            request_count=1,
            request_bytes=4096,
            declared_request_bytes=4096,
            request_framing_valid=True,
            request_body_timed_out=False,
            local_head_before="a" * 64,
            local_head_after="a" * 64,
            status_after={"needs_push": True, "dirty": False},
            worktree_payload_before="payload-hash",
            worktree_payload_after="payload-hash",
            cli_observation={"reported_outcome": "unknown", "observation": "json_error_unknown"},
        )

        self.assertEqual(fields["publication_outcome"], "unknown")
        self.assertEqual(fields["mutation_requests_observed"], 1)
        self.assertEqual(fields["blind_replay_requests"], 0)
        self.assertTrue(fields["local_head_preserved"])
        self.assertTrue(fields["local_work_preserved"])
        self.assertTrue(fields["worktree_payload_preserved"])
        self.assertEqual(fields["network_request_bytes_observed"], 4096)

    def test_interruption_separates_cli_report_from_fixture_delivery_knowledge(self) -> None:
        generic_error = {
            "stdout": '{"schema_version":1,"error":{"code":"http","message":"connection closed"}}',
            "stderr": "",
            "process_returncode": 6,
            "timed_out": False,
        }
        cli = lane.cli_publication_observation(generic_error, json_requested=True)
        self.assertEqual(cli["reported_outcome"], "unreported")
        self.assertEqual(cli["observation"], "json_error_without_publication_outcome")

        fields = lane.interrupted_publication_fields(
            request_count=1,
            request_bytes=3,
            declared_request_bytes=100,
            request_framing_valid=True,
            request_body_timed_out=False,
            local_head_before="a" * 64,
            local_head_after="a" * 64,
            status_after={"needs_push": True, "dirty": False},
            worktree_payload_before="payload",
            worktree_payload_after="payload",
            cli_observation=cli,
        )

        self.assertEqual(fields["publication_outcome"], "unreported")
        self.assertEqual(fields["cli_observation"], "json_error_without_publication_outcome")
        self.assertEqual(fields["fixture_delivery_state"], "partial_request_observed")
        self.assertEqual(
            fields["fixture_outcome_knowledge"],
            "indeterminate_after_partial_request_disconnect",
        )

    def test_interruption_requires_exact_payload_and_unpublished_state(self) -> None:
        fields = lane.interrupted_publication_fields(
            request_count=1,
            request_bytes=100,
            declared_request_bytes=100,
            request_framing_valid=True,
            request_body_timed_out=False,
            local_head_before="a" * 64,
            local_head_after="a" * 64,
            status_after={"needs_push": True, "dirty": False},
            worktree_payload_before="before-hash",
            worktree_payload_after="after-hash",
            cli_observation={"reported_outcome": "unknown", "observation": "json_error_unknown"},
        )

        self.assertTrue(fields["unpublished_state_preserved"])
        self.assertFalse(fields["worktree_payload_preserved"])
        self.assertFalse(fields["local_work_preserved"])

    def test_interruption_server_reads_one_complete_push_then_drops_reply(self) -> None:
        server, thread = lane.start_interrupted_publication_server()
        host, port = server.server_address[:2]
        try:
            connection = http.client.HTTPConnection(host, port, timeout=2)
            connection.request("POST", "/api/repos", body=b"{}", headers={"Content-Type": "application/json"})
            response = connection.getresponse()
            self.assertEqual(response.status, 200)
            response.read()
            connection.close()

            connection = http.client.HTTPConnection(host, port, timeout=2)
            connection.request(
                "POST",
                "/api/oak/repo/push",
                body=b"fixture-payload",
                headers={"Content-Type": "application/json"},
            )
            with self.assertRaises(http.client.RemoteDisconnected):
                connection.getresponse()
            connection.close()
        finally:
            server.shutdown()
            server.server_close()
            thread.join(timeout=2)

        self.assertEqual(server.push_request_count, 1)
        self.assertEqual(server.push_request_bytes, len(b"fixture-payload"))
        self.assertTrue(server.request_complete_before_disconnect)

    def test_interruption_server_never_certifies_a_truncated_request(self) -> None:
        server, thread = lane.start_interrupted_publication_server()
        host, port = server.server_address[:2]
        try:
            with socket.create_connection((host, port), timeout=2) as connection:
                connection.sendall(
                    b"POST /api/oak/repo/push HTTP/1.1\r\n"
                    b"Host: local\r\n"
                    b"Content-Length: 100\r\n"
                    b"Content-Type: application/json\r\n\r\n"
                    b"abc"
                )
                connection.shutdown(socket.SHUT_WR)
                connection.recv(1024)
        finally:
            server.shutdown()
            server.server_close()
            thread.join(timeout=2)

        self.assertEqual(server.push_request_count, 1)
        self.assertEqual(server.push_request_bytes, 3)
        self.assertEqual(server.push_declared_request_bytes, 100)
        self.assertTrue(server.push_request_framing_valid)
        self.assertFalse(server.request_complete_before_disconnect)

    def test_interruption_server_rejects_absent_and_invalid_framing(self) -> None:
        for framing in (
            b"",
            b"Content-Length: nope\r\n",
            f"Content-Length: {lane.INTERRUPTED_PUBLICATION_MAX_REQUEST_BYTES + 1}\r\n".encode(),
        ):
            with self.subTest(framing=framing):
                server, thread = lane.start_interrupted_publication_server()
                host, port = server.server_address[:2]
                try:
                    with socket.create_connection((host, port), timeout=2) as connection:
                        connection.sendall(
                            b"POST /api/oak/repo/push HTTP/1.1\r\n"
                            b"Host: local\r\n"
                            + framing
                            + b"Content-Type: application/json\r\n\r\nabc"
                        )
                        connection.shutdown(socket.SHUT_WR)
                        connection.recv(1024)
                finally:
                    server.shutdown()
                    server.server_close()
                    thread.join(timeout=2)

                self.assertEqual(server.push_request_count, 1)
                self.assertIsNone(server.push_declared_request_bytes)
                self.assertFalse(server.push_request_framing_valid)
                self.assertFalse(server.request_complete_before_disconnect)

    def test_interruption_server_bounds_a_partial_request_while_sender_stays_open(self) -> None:
        server, thread = lane.start_interrupted_publication_server()
        host, port = server.server_address[:2]
        started = time.monotonic()
        try:
            with socket.create_connection((host, port), timeout=4) as connection:
                connection.sendall(
                    b"POST /api/oak/repo/push HTTP/1.1\r\n"
                    b"Host: local\r\n"
                    b"Content-Length: 100\r\n\r\nabc"
                )
                connection.recv(1024)
        finally:
            server.shutdown()
            server.server_close()
            thread.join(timeout=2)

        self.assertLess(time.monotonic() - started, 4)
        self.assertEqual(server.push_request_bytes, 3)
        self.assertTrue(server.push_request_framing_valid)
        self.assertTrue(server.push_request_body_timed_out)
        self.assertFalse(server.request_complete_before_disconnect)

    @unittest.skipIf(shutil.which("git") is None, "git not installed")
    def test_local_git_agent_task_records_cold_warm_and_unmeasured_ci(self) -> None:
        subject = Subject("git", "git", "Git", Path(shutil.which("git") or "git"))
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            remote = lane.oakbench_remotes.make_git_bare_remote(
                subject.bin, root, "agent-task.git"
            )
            rows = lane.run_git_agent_task_baseline(
                subject,
                remote,
                root,
                0,
                {
                    "bench_id": "test",
                    "profile": "task-loop",
                    "timestamp_utc": "test",
                    "benchmark_track": "agent-default",
                },
                20_000,
            )

        operations = [row["operation"] for row in rows]
        self.assertIn("cold.acquire", operations)
        self.assertIn("warm.acquire", operations)
        self.assertIn("cold.ci.observe", operations)
        total = next(row for row in rows if row["operation"] == "agent_task.total")
        self.assertEqual(total["returncode"], 0)
        self.assertFalse(total["coverage_complete"])
        self.assertEqual(total["oracle_payload_on_remote_branch"], True)
        self.assertEqual(total["acquisition_models"], {
            "cold": "fresh_full_clone",
            "warm": "shared_object_worktree",
        })
        ci = next(row for row in rows if row["operation"] == "cold.ci.observe")
        self.assertEqual(ci["returncode"], lane.SKIP_RETURNCODE)
        self.assertIn("no hosted CI", ci["skip_reason"])


if __name__ == "__main__":
    unittest.main()

About

Reversible clean-root candidate rebuilt from oak/benchmarks main@009dead264ca; original repository retained.

Download

RunnersDetails

No runs on main yet.

Open branchesHistory

mrmrs-516758agent_loop_hosted_v1: hosted agent-loop gap map lane (Oak vs GitHub vs ideal)checking…6 behind1mo ago

mrmrs-b26e28Add stale_branch_land_v1: baseline for landing branches that are behind mainchecking…6 behind1mo ago

mrmrs-fad256task_loop: shallow_worker_branch_refresh_v1 (fresh shallow worker -> task branch -> first parent refresh, Oak vs Git, HTTP request counts)checking…6 behind1mo ago

oakmark-throughput-benchmark-transplantAdd oakmark: per-repo push/clone throughput benchmark for oak (ForgeMark analogue)checking…11 behind3mo ago

All branches →

Activity12 commits to main

May 2026now

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 socket`
8	`import tempfile`
9	`import time`
10	`import unittest`
11	`import http.client`
12	`from pathlib import Path`
13	`from unittest.mock import patch`
14
15	`sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))`
16
17	`import task_loop as lane`
18	`from oakbench.execution import TimedCapture`
19	`from oakbench.rows import LANE_REQUIRED_KEYS`
20	`from oakbench.subjects import Subject`
21	`from oakbench.remotes import RemoteResolution`
22
23
24	`class LoopSummaryFieldsTest(unittest.TestCase):`
25	`def test_warm_delta_from_acquire_pair(self) -> None:`
26	`rows = [`
27	`{"operation": "task.acquire", "elapsed_ms": 1000.0},`
28	`{"operation": "task.edit", "elapsed_ms": 5.0},`
29	`{"operation": "followup.acquire", "elapsed_ms": 600.0},`
30	`{"operation": "followup.edit", "elapsed_ms": 5.0},`
31	`]`
32	`fields = lane.loop_summary_fields(rows)`
33	`self.assertEqual(fields["followup_warm_delta_ms"], -400.0)`
34	`self.assertEqual(fields["task_phase_ms"], 1005.0)`
35	`self.assertEqual(fields["followup_phase_ms"], 605.0)`
36
37	`def test_missing_acquire_yields_null_not_zero(self) -> None:`
38	`fields = lane.loop_summary_fields([{"operation": "task.edit", "elapsed_ms": 5.0}])`
39	`self.assertIsNone(fields["followup_warm_delta_ms"])`
40
41
42	`class LaneRegistrationTest(unittest.TestCase):`
43	`def test_task_loop_lane_registered(self) -> None:`
44	`self.assertIn("task-loop", LANE_REQUIRED_KEYS)`
45	`self.assertIn("command", LANE_REQUIRED_KEYS["task-loop"])`
46
47
48	`class VirtualBranchReTest(unittest.TestCase):`
49	`def test_parses_mount_status(self) -> None:`
50	`text = "Mount: oak/bench-sync-tmp@abc (virtual branch tl-x-r0-task--9f00aa11)"`
51	`match = lane.VIRTUAL_BRANCH_RE.search(text)`
52	`self.assertEqual(match.group(1), "tl-x-r0-task--9f00aa11")`
53
54
55	`class FailurePolicyTest(unittest.TestCase):`
56	`def test_step_row_never_treats_a_process_timeout_as_an_expected_error(self) -> None:`
57	`subject = Subject("oak", "oak", "Oak", Path("/usr/bin/oak"))`
58	`capture = TimedCapture(`
59	`elapsed_ms=100.0,`
60	`returncode=124,`
61	`stdout_text="",`
62	`stdout_bytes=0,`
63	`stdout_truncated=False,`
64	`stderr_text="",`
65	`stderr_bytes=0,`
66	`stderr_truncated=False,`
67	`timed_out=True,`
68	`)`
69	`with tempfile.TemporaryDirectory() as tmp, patch.object(`
70	`lane, "run_timed", return_value=capture`
71	`) as timed:`
72	`row = lane.step_row(`
73	`subject,`
74	`"interrupted_publication_after_request_v1",`
75	`"publication.attempt",`
76	`["oak", "push"],`
77	`Path(tmp),`
78	`0,`
79	`{"bench_id": "test"},`
80	`100,`
81	`expected_returncodes=tuple(range(1, 128)),`
82	`timeout_seconds=0.1,`
83	`)`
84
85	`self.assertEqual(timed.call_args.kwargs["timeout_seconds"], 0.1)`
86	`self.assertTrue(row["timed_out"])`
87	`self.assertEqual(row["returncode"], 124)`
88
89	`def test_step_row_marks_unexpected_zero_exit_as_failure(self) -> None:`
90	`subject = Subject("git", "git", "Git", Path("/usr/bin/git"))`
91	`capture = TimedCapture(`
92	`elapsed_ms=1.0,`
93	`returncode=0,`
94	`stdout_text="",`
95	`stdout_bytes=0,`
96	`stdout_truncated=False,`
97	`stderr_text="",`
98	`stderr_bytes=0,`
99	`stderr_truncated=False,`
100	`)`
101	`with tempfile.TemporaryDirectory() as tmp, patch.object(lane, "run_timed", return_value=capture):`
102	`row = lane.step_row(`
103	`subject,`
104	`"task_loop",`
105	`"expect.fail",`
106	`["sh", "-c", "true"],`
107	`Path(tmp),`
108	`0,`
109	`{`
110	`"bench_id": "test",`
111	`"profile": "task-loop",`
112	`"timestamp_utc": "test",`
113	`"benchmark_track": "agent-default",`
114	`},`
115	`100,`
116	`expected_returncodes=(1,),`
117	`)`
118
119	`self.assertEqual(row["process_returncode"], 0)`
120	`self.assertEqual(row["returncode"], 1)`
121
122	`def test_oracle_false_marks_total_failed(self) -> None:`
123	`total = {`
124	`**lane.phase_total_row(`
125	`[{"operation": "task.acquire", "elapsed_ms": 1.0, "returncode": 0, "command": []}],`
126	`Subject("oak", "oak", "Oak", Path("/usr/bin/oak")),`
127	`"task_loop",`
128	`0,`
129	`{`
130	`"bench_id": "test",`
131	`"profile": "task-loop",`
132	`"timestamp_utc": "test",`
133	`"benchmark_track": "agent-default",`
134	`},`
135	`"task_loop.total",`
136	`),`
137	`"oracle_payload_on_remote_branch": False,`
138	`}`
139	`lane.enforce_oracle_result(total, False)`
140	`self.assertEqual(total["returncode"], 1)`
141	`self.assertTrue(lane.has_lane_failures([total]))`
142
143	`def test_lane_failures_ignore_skips_but_fail_on_real_errors(self) -> None:`
144	`self.assertFalse(lane.has_lane_failures([{"returncode": lane.SKIP_RETURNCODE, "skipped": True}]))`
145	`self.assertTrue(lane.has_lane_failures([{"returncode": 1}]))`
146
147
148	`class LocalGitSeedTest(unittest.TestCase):`
149	`def test_seeded_local_remote_advertises_main_as_head(self) -> None:`
150	`subject = Subject("git", "git", "Git", Path("/usr/bin/git"))`
151	`with tempfile.TemporaryDirectory() as tmp:`
152	`root = Path(tmp)`
153	`remote_path = root / "remote.git"`
154	`init = lane.run_untimed(`
155	`[str(subject.bin), "init", "--bare", "-q", str(remote_path)], root`
156	`)`
157	`self.assertEqual(init.returncode, 0, init.stderr)`
158	`remote = RemoteResolution(`
159	`repo=str(remote_path),`
160	`transport="local_file",`
161	`server="local-bare",`
162	`env_var="generated",`
163	`)`
164
165	`error = lane.ensure_git_remote_seeded(subject, remote, root)`
166
167	`self.assertIsNone(error)`
168	`advertised = lane.run_untimed(`
169	`[str(subject.bin), "symbolic-ref", "HEAD"], remote_path`
170	`)`
171	`self.assertEqual(advertised.returncode, 0, advertised.stderr)`
172	`self.assertEqual(advertised.stdout.strip(), "refs/heads/main")`
173
174
175	`class AgentTaskEvidenceTest(unittest.TestCase):`
176	`def test_total_keeps_skips_unmeasured_and_aggregates_costs(self) -> None:`
177	`subject = Subject("git", "git", "Git", Path("/usr/bin/git"))`
178	`rows = [`
179	`{`
180	`"operation": "task.acquire",`
181	`"elapsed_ms": 10.0,`
182	`"returncode": 0,`
183	`"command": ["git", "clone"],`
184	`"tool_call_count": 1,`
185	`"stdout_bytes": 7,`
186	`"stderr_bytes": 0,`
187	`"retry_count": 0,`
188	`"human_action_count": 0,`
189	`},`
190	`{`
191	`"operation": "task.ci.observe",`
192	`"elapsed_ms": 0.0,`
193	`"returncode": lane.SKIP_RETURNCODE,`
194	`"command": [],`
195	`"skipped": True,`
196	`"skip_reason": "local transport has no hosted CI provider",`
197	`"tool_call_count": 0,`
198	`"stdout_bytes": None,`
199	`"stderr_bytes": None,`
200	`"retry_count": 0,`
201	`"human_action_count": 0,`
202	`},`
203	`]`
204
205	`total = lane.phase_total_row(`
206	`rows,`
207	`subject,`
208	`"agent_task_baseline_v1",`
209	`0,`
210	`{`
211	`"bench_id": "test",`
212	`"profile": "task-loop",`
213	`"timestamp_utc": "test",`
214	`"benchmark_track": "agent-default",`
215	`},`
216	`"agent_task.total",`
217	`)`
218
219	`self.assertEqual(total["returncode"], 0)`
220	`self.assertFalse(total["coverage_complete"])`
221	`self.assertEqual(total["unmeasured_operations"], ["task.ci.observe"])`
222	`self.assertEqual(total["raw_output_bytes"], 7)`
223	`self.assertEqual(total["retry_count"], 0)`
224	`self.assertEqual(total["human_action_count"], 0)`
225	`self.assertEqual(total["phase_elapsed_ms"], {"task": 10.0})`
226
227	`def test_interrupted_publication_summary_never_claims_retry(self) -> None:`
228	`fields = lane.interrupted_publication_fields(`
229	`request_count=1,`
230	`request_bytes=4096,`
231	`declared_request_bytes=4096,`
232	`request_framing_valid=True,`
233	`request_body_timed_out=False,`
234	`local_head_before="a" * 64,`
235	`local_head_after="a" * 64,`
236	`status_after={"needs_push": True, "dirty": False},`
237	`worktree_payload_before="payload-hash",`
238	`worktree_payload_after="payload-hash",`
239	`cli_observation={"reported_outcome": "unknown", "observation": "json_error_unknown"},`
240	`)`
241
242	`self.assertEqual(fields["publication_outcome"], "unknown")`
243	`self.assertEqual(fields["mutation_requests_observed"], 1)`
244	`self.assertEqual(fields["blind_replay_requests"], 0)`
245	`self.assertTrue(fields["local_head_preserved"])`
246	`self.assertTrue(fields["local_work_preserved"])`
247	`self.assertTrue(fields["worktree_payload_preserved"])`
248	`self.assertEqual(fields["network_request_bytes_observed"], 4096)`
249
250	`def test_interruption_separates_cli_report_from_fixture_delivery_knowledge(self) -> None:`
251	`generic_error = {`
252	`"stdout": '{"schema_version":1,"error":{"code":"http","message":"connection closed"}}',`
253	`"stderr": "",`
254	`"process_returncode": 6,`
255	`"timed_out": False,`
256	`}`
257	`cli = lane.cli_publication_observation(generic_error, json_requested=True)`
258	`self.assertEqual(cli["reported_outcome"], "unreported")`
259	`self.assertEqual(cli["observation"], "json_error_without_publication_outcome")`
260
261	`fields = lane.interrupted_publication_fields(`
262	`request_count=1,`
263	`request_bytes=3,`
264	`declared_request_bytes=100,`
265	`request_framing_valid=True,`
266	`request_body_timed_out=False,`
267	`local_head_before="a" * 64,`
268	`local_head_after="a" * 64,`
269	`status_after={"needs_push": True, "dirty": False},`
270	`worktree_payload_before="payload",`
271	`worktree_payload_after="payload",`
272	`cli_observation=cli,`
273	`)`
274
275	`self.assertEqual(fields["publication_outcome"], "unreported")`
276	`self.assertEqual(fields["cli_observation"], "json_error_without_publication_outcome")`
277	`self.assertEqual(fields["fixture_delivery_state"], "partial_request_observed")`
278	`self.assertEqual(`
279	`fields["fixture_outcome_knowledge"],`
280	`"indeterminate_after_partial_request_disconnect",`
281	`)`
282
283	`def test_interruption_requires_exact_payload_and_unpublished_state(self) -> None:`
284	`fields = lane.interrupted_publication_fields(`
285	`request_count=1,`
286	`request_bytes=100,`
287	`declared_request_bytes=100,`
288	`request_framing_valid=True,`
289	`request_body_timed_out=False,`
290	`local_head_before="a" * 64,`
291	`local_head_after="a" * 64,`
292	`status_after={"needs_push": True, "dirty": False},`
293	`worktree_payload_before="before-hash",`
294	`worktree_payload_after="after-hash",`
295	`cli_observation={"reported_outcome": "unknown", "observation": "json_error_unknown"},`
296	`)`
297
298	`self.assertTrue(fields["unpublished_state_preserved"])`
299	`self.assertFalse(fields["worktree_payload_preserved"])`
300	`self.assertFalse(fields["local_work_preserved"])`
301
302	`def test_interruption_server_reads_one_complete_push_then_drops_reply(self) -> None:`
303	`server, thread = lane.start_interrupted_publication_server()`
304	`host, port = server.server_address[:2]`
305	`try:`
306	`connection = http.client.HTTPConnection(host, port, timeout=2)`
307	`connection.request("POST", "/api/repos", body=b"{}", headers={"Content-Type": "application/json"})`
308	`response = connection.getresponse()`
309	`self.assertEqual(response.status, 200)`
310	`response.read()`
311	`connection.close()`
312
313	`connection = http.client.HTTPConnection(host, port, timeout=2)`
314	`connection.request(`
315	`"POST",`
316	`"/api/oak/repo/push",`
317	`body=b"fixture-payload",`
318	`headers={"Content-Type": "application/json"},`
319	`)`
320	`with self.assertRaises(http.client.RemoteDisconnected):`
321	`connection.getresponse()`
322	`connection.close()`
323	`finally:`
324	`server.shutdown()`
325	`server.server_close()`
326	`thread.join(timeout=2)`
327
328	`self.assertEqual(server.push_request_count, 1)`
329	`self.assertEqual(server.push_request_bytes, len(b"fixture-payload"))`
330	`self.assertTrue(server.request_complete_before_disconnect)`
331
332	`def test_interruption_server_never_certifies_a_truncated_request(self) -> None:`
333	`server, thread = lane.start_interrupted_publication_server()`
334	`host, port = server.server_address[:2]`
335	`try:`
336	`with socket.create_connection((host, port), timeout=2) as connection:`
337	`connection.sendall(`
338	`b"POST /api/oak/repo/push HTTP/1.1\r\n"`
339	`b"Host: local\r\n"`
340	`b"Content-Length: 100\r\n"`
341	`b"Content-Type: application/json\r\n\r\n"`
342	`b"abc"`
343	`)`
344	`connection.shutdown(socket.SHUT_WR)`
345	`connection.recv(1024)`
346	`finally:`
347	`server.shutdown()`
348	`server.server_close()`
349	`thread.join(timeout=2)`
350
351	`self.assertEqual(server.push_request_count, 1)`
352	`self.assertEqual(server.push_request_bytes, 3)`
353	`self.assertEqual(server.push_declared_request_bytes, 100)`
354	`self.assertTrue(server.push_request_framing_valid)`
355	`self.assertFalse(server.request_complete_before_disconnect)`
356
357	`def test_interruption_server_rejects_absent_and_invalid_framing(self) -> None:`
358	`for framing in (`
359	`b"",`
360	`b"Content-Length: nope\r\n",`
361	`f"Content-Length: {lane.INTERRUPTED_PUBLICATION_MAX_REQUEST_BYTES + 1}\r\n".encode(),`
362	`):`
363	`with self.subTest(framing=framing):`
364	`server, thread = lane.start_interrupted_publication_server()`
365	`host, port = server.server_address[:2]`
366	`try:`
367	`with socket.create_connection((host, port), timeout=2) as connection:`
368	`connection.sendall(`
369	`b"POST /api/oak/repo/push HTTP/1.1\r\n"`
370	`b"Host: local\r\n"`
371	`+ framing`
372	`+ b"Content-Type: application/json\r\n\r\nabc"`
373	`)`
374	`connection.shutdown(socket.SHUT_WR)`
375	`connection.recv(1024)`
376	`finally:`
377	`server.shutdown()`
378	`server.server_close()`
379	`thread.join(timeout=2)`
380
381	`self.assertEqual(server.push_request_count, 1)`
382	`self.assertIsNone(server.push_declared_request_bytes)`
383	`self.assertFalse(server.push_request_framing_valid)`
384	`self.assertFalse(server.request_complete_before_disconnect)`
385
386	`def test_interruption_server_bounds_a_partial_request_while_sender_stays_open(self) -> None:`
387	`server, thread = lane.start_interrupted_publication_server()`
388	`host, port = server.server_address[:2]`
389	`started = time.monotonic()`
390	`try:`
391	`with socket.create_connection((host, port), timeout=4) as connection:`
392	`connection.sendall(`
393	`b"POST /api/oak/repo/push HTTP/1.1\r\n"`
394	`b"Host: local\r\n"`
395	`b"Content-Length: 100\r\n\r\nabc"`
396	`)`
397	`connection.recv(1024)`
398	`finally:`
399	`server.shutdown()`
400	`server.server_close()`
401	`thread.join(timeout=2)`
402
403	`self.assertLess(time.monotonic() - started, 4)`
404	`self.assertEqual(server.push_request_bytes, 3)`
405	`self.assertTrue(server.push_request_framing_valid)`
406	`self.assertTrue(server.push_request_body_timed_out)`
407	`self.assertFalse(server.request_complete_before_disconnect)`
408
409	`@unittest.skipIf(shutil.which("git") is None, "git not installed")`
410	`def test_local_git_agent_task_records_cold_warm_and_unmeasured_ci(self) -> None:`
411	`subject = Subject("git", "git", "Git", Path(shutil.which("git") or "git"))`
412	`with tempfile.TemporaryDirectory() as tmp:`
413	`root = Path(tmp)`
414	`remote = lane.oakbench_remotes.make_git_bare_remote(`
415	`subject.bin, root, "agent-task.git"`
416	`)`
417	`rows = lane.run_git_agent_task_baseline(`
418	`subject,`
419	`remote,`
420	`root,`
421	`0,`
422	`{`
423	`"bench_id": "test",`
424	`"profile": "task-loop",`
425	`"timestamp_utc": "test",`
426	`"benchmark_track": "agent-default",`
427	`},`
428	`20_000,`
429	`)`
430
431	`operations = [row["operation"] for row in rows]`
432	`self.assertIn("cold.acquire", operations)`
433	`self.assertIn("warm.acquire", operations)`
434	`self.assertIn("cold.ci.observe", operations)`
435	`total = next(row for row in rows if row["operation"] == "agent_task.total")`
436	`self.assertEqual(total["returncode"], 0)`
437	`self.assertFalse(total["coverage_complete"])`
438	`self.assertEqual(total["oracle_payload_on_remote_branch"], True)`
439	`self.assertEqual(total["acquisition_models"], {`
440	`"cold": "fresh_full_clone",`
441	`"warm": "shared_object_worktree",`
442	`})`
443	`ci = next(row for row in rows if row["operation"] == "cold.ci.observe")`
444	`self.assertEqual(ci["returncode"], lane.SKIP_RETURNCODE)`
445	`self.assertIn("no hosted CI", ci["skip_reason"])`
446
447
448	`if __name__ == "__main__":`
449	`unittest.main()`