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

from __future__ import annotations

import sys

import os

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):

    @unittest.skipUnless(

        os.environ.get("OAK_BENCH_TEST_OAK_BIN"),

        "set OAK_BENCH_TEST_OAK_BIN to exercise the real local Oak server",

    )

    def test_local_oak_agent_task_uses_fresh_checkouts_and_remote_oracles(self) -> None:

        oak_bin = Path(os.environ["OAK_BENCH_TEST_OAK_BIN"])

        subject = Subject("oak_local", "oak", "Oak local", oak_bin)

        with tempfile.TemporaryDirectory() as tmp:

            rows = lane.run_oak_local_agent_task_baseline(

                subject,

                Path(tmp),

                0,

                {

                    "bench_id": "test",

                    "profile": "task-loop",

                    "timestamp_utc": "test",

                    "benchmark_track": "agent-default",

                },

                20_000,

            )

        total = next(row for row in rows if row["operation"] == "agent_task_oak_local.total")

        self.assertEqual(total["returncode"], 0)

        self.assertFalse(total["coverage_complete"])

        self.assertTrue(total["oracle_payload_on_remote_branch"])

        self.assertTrue(total["oracle_branch_head_matches"])

        self.assertTrue(total["server_process_stopped"])

        self.assertEqual(

            total["acquisition_models"],

            {"cold": "fresh_full_clone", "followup": "second_fresh_full_clone"},

        )

        self.assertEqual(total["repository_cache_state"]["followup"], "no_shared_object_store")

        self.assertIn("cold.ci.observe", total["unmeasured_operations"])

        self.assertIn("followup.review_handoff", total["unmeasured_operations"])

    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",

            b"Content-Length: +3\r\n",

            b"Content-Length: 3\r\nContent-Length: 3\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_accepts_http_ows_around_content_length(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:\t3 \t\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_declared_request_bytes, 3)

        self.assertTrue(server.push_request_framing_valid)

        self.assertTrue(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()