from __future__ import annotations

import json

import shutil

import subprocess

import sys

import tempfile

import threading

import time

import unittest

import urllib.request

from pathlib import Path

from unittest import mock

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

import stale_land  # noqa: E402

from oakbench import ci_stub  # noqa: E402

from oakbench.rows import validate_rows  # noqa: E402

def tree(files: dict[str, str]) -> Path:

    root = Path(tempfile.mkdtemp(prefix="sl-test-"))

    stale_land.write_files(root, files)

    return root

class FixtureOracleTests(unittest.TestCase):

    """The semantic-conflict oracle must be real: green alone, red combined."""

    def check(self, *patches: str) -> bool:

        files = dict(stale_land.base_files())

        for name in patches:

            files.update(stale_land.agent_patch(name))

        root = tree(files)

        try:

            return stale_land.run_check(root)[0]

        finally:

            shutil.rmtree(root)

    def test_base_is_green(self) -> None:

        self.assertTrue(self.check())

    def test_pair_members_are_green_alone(self) -> None:

        self.assertTrue(self.check("pair_rename"))

        self.assertTrue(self.check("pair_consumer"))

    def test_pair_is_red_combined(self) -> None:

        self.assertFalse(self.check("pair_rename", "pair_consumer"))

    def test_ordinary_agents_are_green_together_with_either_pair_member(self) -> None:

        self.assertTrue(self.check("agent00", "agent01", "agent02", "pair_rename"))

        self.assertTrue(self.check("agent00", "agent01", "agent02", "pair_consumer"))

    def test_patches_touch_disjoint_paths(self) -> None:

        seen: dict[str, str] = {}

        for name in stale_land.agent_names(8):

            for path in stale_land.agent_patch(name):

                self.assertNotIn(path, seen, f"{name} overlaps {seen.get(path)} on {path}")

                seen[path] = name

        movers = {stale_land.mover_file(j)[0] for j in range(50)}

        self.assertFalse(movers & set(seen))

    def test_check_writes_no_bytecode_into_tree(self) -> None:

        root = tree(stale_land.base_files())

        try:

            stale_land.run_check(root)

            self.assertEqual([], [p for p in root.rglob("__pycache__")])

        finally:

            shutil.rmtree(root)

class FinalStateClassificationTests(unittest.TestCase):

    def test_exact_with_one_pair_member(self) -> None:

        files = stale_land.expected_final_files(4, 3, "pair_rename")

        result = stale_land.classify_final(files, 4, 3)

        self.assertTrue(result["intended_changes_exact"])

        self.assertEqual(["pair_rename"], result["pair_landed"])

        self.assertEqual([], result["missing_agents"])

    def test_both_pair_members_is_not_exact(self) -> None:

        files = stale_land.expected_final_files(1, 1, "pair_rename")

        files.update(stale_land.agent_patch("pair_consumer"))

        result = stale_land.classify_final(files, 1, 1)

        self.assertFalse(result["intended_changes_exact"])

        self.assertTrue(result["both_pair_landed"])

    def test_missing_agent_and_extra_file_are_detected(self) -> None:

        files = stale_land.expected_final_files(2, 1, "pair_consumer")

        for path in stale_land.agent_patch("agent01"):

            files.pop(path)

        result = stale_land.classify_final(files, 2, 1)

        self.assertEqual(["agent01"], result["missing_agents"])

        self.assertFalse(result["intended_changes_exact"])

        extra = stale_land.expected_final_files(2, 1, "pair_consumer")

        extra["stray.txt"] = "x"

        self.assertFalse(stale_land.classify_final(extra, 2, 1)["intended_changes_exact"])

    def test_read_tree_ignores_vcs_metadata_but_keeps_workflows(self) -> None:

        root = tree(stale_land.base_files())

        try:

            (root / ".oak" / "oak.db").write_text("db")

            (root / ".git").mkdir()

            (root / ".git" / "HEAD").write_text("ref")

            self.assertEqual(stale_land.base_files(), stale_land.read_tree(root))

            self.assertEqual(ci_stub.tree_digest(root), stale_land.files_digest(stale_land.base_files()))

        finally:

            shutil.rmtree(root)

class HelperTests(unittest.TestCase):

    def test_last_json_skips_human_preamble(self) -> None:

        text = "Updating 'main'…\nMerged.\n{\"merged\": true, \"commit_hash\": \"abc\"}\n"

        self.assertEqual({"merged": True, "commit_hash": "abc"}, stale_land.last_json(text))

        self.assertIsNone(stale_land.last_json("no json here"))

    def test_retry_after_parses_429_hint_only(self) -> None:

        hit = subprocess.CompletedProcess([], 6, "", "Server error: HTTP 429 Too Many Requests: an integrity proof for this repository is already running; retry after 2 seconds")

        self.assertEqual(2.0, stale_land.retry_after(hit))

        self.assertIsNone(stale_land.retry_after(subprocess.CompletedProcess([], 1, "", "HTTP 409 Conflict")))

    def test_agent_counts_tool_calls_and_retries_429(self) -> None:

        agent = stale_land.AgentRun(name="agent00", arm="oak_safe")

        responses = [

            subprocess.CompletedProcess([], 6, "", "HTTP 429 Too Many Requests: busy; retry after 0 seconds"),

            subprocess.CompletedProcess([], 0, "{}", ""),

        ]

        with mock.patch.object(stale_land.subprocess, "run", side_effect=responses):

            proc = agent.call_retrying(["oak", "clone"], Path("."), {})

        self.assertEqual(0, proc.returncode)

        self.assertEqual(2, agent.tool_calls)

        self.assertEqual(1, agent.rate_limited)

    def test_retryable_conflict_requires_explicit_retry_signal(self) -> None:

        push409 = subprocess.CompletedProcess([], 6, '{"error":{"message":"Server error: HTTP 409 Conflict: a verified blob mapping changed before branch publication; retry"}}', "")

        merge409 = subprocess.CompletedProcess([], 6, "", 'HTTP 409 Conflict {"code":"main_moved","retryable":true}')

        conflict = subprocess.CompletedProcess([], 6, "", "HTTP 409 Conflict: merge conflict in pkg/core.py")

        self.assertTrue(stale_land.retryable_conflict(push409))

        self.assertTrue(stale_land.retryable_conflict(merge409))

        self.assertFalse(stale_land.retryable_conflict(conflict))

        self.assertFalse(stale_land.retryable_conflict(subprocess.CompletedProcess([], 0, "; retry 409", "")))

    def test_conflict_retries_are_bounded_and_opt_in(self) -> None:

        busy = subprocess.CompletedProcess([], 6, "", "HTTP 409 Conflict: mapping changed; retry")

        agent = stale_land.AgentRun(name="a", arm="oak_safe")

        with mock.patch.object(stale_land.subprocess, "run", return_value=busy), mock.patch.object(stale_land.time, "sleep"):

            self.assertEqual(6, agent.call_retrying(["oak", "push"], Path("."), {}).returncode)

            self.assertEqual(1, agent.tool_calls)

            agent.call_retrying(["oak", "push"], Path("."), {}, retry_conflicts=2)

        self.assertEqual(1 + 3, agent.tool_calls)

        self.assertEqual(2, agent.conflict_retries)

    def test_schedule_expands_oak_arms_per_client_and_git_once(self) -> None:

        args = stale_land.parse_args(

            ["--arms", "oak_safe,git_competent", "--agents", "1,4", "--behind", "1", "--runs", "2",

             "--oak-bin", "a=/x/oak", "--oak-bin", "b=/y/oak"]

        )

        plan = stale_land.schedule(args)

        self.assertEqual(2 * 1 * 2 * (2 + 1), len(plan))

        self.assertEqual({"a", "b", "git"}, {item[1][0] for item in plan})

        with self.assertRaises(SystemExit):

            stale_land.schedule(stale_land.parse_args(["--arms", "bogus"]))

class GitArmEndToEndTests(unittest.TestCase):

    """Git arms run fully locally, so they exercise the whole driver and oracles."""

    @unittest.skipUnless(shutil.which("git"), "git not installed")

    def test_git_arms_land_everything_and_refuse_the_red_pair_member(self) -> None:

        scratch = Path(tempfile.mkdtemp(prefix="sl-e2e-"))

        try:

            for arm in ("git_competent", "git_fifo"):

                args = stale_land.parse_args(["--scratch", str(scratch), "--ci-added-seconds", "0"])

                with mock.patch.dict("os.environ", {"OAK_BENCH_LOCK_PATH": str(scratch / "lock")}):

                    rows = stale_land.run_trial(args, arm, 2, 3, 1, None, {"bench_id": "t"}, ("git", "git", None))

                self.assertEqual([], validate_rows(rows, "contention"))

                trial = rows[-1]

                self.assertEqual("land.trial", trial["operation"])

                self.assertEqual(0, trial["red_main_events"], arm)

                self.assertTrue(trial["intended_changes_exact"], arm)

                self.assertTrue(trial["final_main_green"], arm)

                self.assertEqual(1.0, trial["combined_tested_fraction"], arm)

                outcomes = {r["agent"]: r["outcome"] for r in rows if r["operation"] == "land.agent"}

                self.assertEqual("landed", outcomes["agent00"])

                self.assertEqual("landed", outcomes["agent01"])

                self.assertEqual(

                    ["landed", "refused_red_combined"],

                    sorted([outcomes["pair_rename"], outcomes["pair_consumer"]]),

                )

                if arm == "git_fifo":

                    self.assertEqual(0, trial["retries_total"])

        finally:

            shutil.rmtree(scratch, ignore_errors=True)

class BuildRowsNullHonestyTests(unittest.TestCase):

    def test_unlanded_agent_has_null_tested_flag_and_null_time(self) -> None:

        agent = stale_land.AgentRun(name="pair_consumer", arm="oak_safe", outcome="refused_red_combined", finished_at=2.0, attempts=1)

        oracle = {"heads": [], "red": 0, "digests": {}, "final_files": stale_land.expected_final_files(0, 0, "pair_rename"), "ci_trial": [], "ci_setup": []}

        rows = stale_land.build_rows(meta={"bench_id": "t"}, arm="oak_safe", n=0, k=0, run_index=1, agents=[agent], start=1.0, end=2.0,

                                     setup={"setup_ms": 1.0, "main_commits_before": 1}, oracle=oracle, subject="oak_installed",

                                     subject_kind="oak", binary_hash=None)

        agent_row = rows[0]

        self.assertIsNone(agent_row["combined_tree_tested_before_land"])

        self.assertIsNone(agent_row["time_to_landed_ms"])

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

        self.assertIsNone(rows[-1]["combined_tested_fraction"])

        self.assertEqual([], validate_rows(rows, "contention"))

class CiStubProtocolTests(unittest.TestCase):

    """The stub must speak oakspace's signed dispatch/callback protocol."""

    def setUp(self) -> None:

        self.tmp = Path(tempfile.mkdtemp(prefix="ci-stub-test-"))

        self.posts: list[tuple[str, dict]] = []

    def tearDown(self) -> None:

        shutil.rmtree(self.tmp, ignore_errors=True)

    def make(self, **kwargs) -> ci_stub.CiStub:  # type: ignore[no-untyped-def]

        return ci_stub.CiStub(api_url="http://127.0.0.1:9", secret="s3cret", ledger=self.tmp / "ledger.jsonl",

                              workdir=self.tmp / "work", post=lambda path, body: self.posts.append((path, body)) or True, **kwargs)

    def test_refuses_non_loopback_bind(self) -> None:

        with self.assertRaises(ValueError):

            self.make(host="0.0.0.0")

    def test_rejects_unsigned_jobs_and_accepts_signed(self) -> None:

        stub = self.make().start()

        try:

            url = f"http://127.0.0.1:{stub.port}/jobs"

            body = json.dumps({"run_id": 7, "repo_id": 1, "commit": "c", "fetch_token": "", "steps": []}).encode()

            bad = urllib.request.Request(url, data=body, method="POST", headers={"x-oak-signature-256": "sha256=00"})

            with self.assertRaises(urllib.error.HTTPError) as ctx:

                urllib.request.urlopen(bad, timeout=5)

            self.assertEqual(401, ctx.exception.code)

            ctx.exception.close()

            good = urllib.request.Request(url, data=body, method="POST", headers={"x-oak-signature-256": "sha256=" + ci_stub.sign("s3cret", body)})

            with urllib.request.urlopen(good, timeout=5) as response:

                self.assertEqual(202, response.status)

            stub.join_runs(10)

            self.assertEqual(("/api/ci/runs/7/conclusion", {"conclusion": "success", "error": None}), self.posts[-1])

        finally:

            stub.stop()

    def test_failing_step_reports_step_and_run_failure_then_skips_rest(self) -> None:

        stub = self.make()

        spec = {"run_id": 3, "repo_id": 1, "commit": "c", "fetch_token": "", "env": {"X": "1"},

                "steps": [{"step_id": 1, "name": "a", "run": "test \"$X\" = 1 && test \"$OAK_CI_RUN_ID\" = 3"},

                          {"step_id": 2, "name": "b", "run": "exit 4"},

                          {"step_id": 3, "name": "c", "run": "true"}]}

        record = stub.execute(spec)

        self.assertEqual("failure", record["conclusion"])

        self.assertEqual(

            [("/api/ci/runs/3/steps", {"step_id": 1, "conclusion": "success", "exit_code": 0}),

             ("/api/ci/runs/3/steps", {"step_id": 2, "conclusion": "failure", "exit_code": 4}),

             ("/api/ci/runs/3/steps", {"step_id": 3, "conclusion": "skipped", "exit_code": None}),

             ("/api/ci/runs/3/conclusion", {"conclusion": "failure", "error": "step 'b' exited 4"})],

            self.posts,

        )

        ledger = ci_stub.read_ledger(self.tmp / "ledger.jsonl")

        self.assertEqual(1, len(ledger))

        self.assertIsNotNone(ledger[0]["tree_digest"])

    def test_added_seconds_delays_conclusion(self) -> None:

        stub = self.make(added_seconds=0.3)

        start = time.perf_counter()

        stub.execute({"run_id": 1, "repo_id": 1, "commit": "c", "fetch_token": "", "steps": []})

        self.assertGreaterEqual(time.perf_counter() - start, 0.3)

    def test_fetch_failure_concludes_failure_not_wedged(self) -> None:

        stub = self.make()

        with mock.patch.object(stub, "fetch", return_value=subprocess.CompletedProcess([], 128, "", "fatal: nope")):

            record = stub.execute({"run_id": 9, "repo_id": 1, "commit": "c", "fetch_token": "tok", "steps": [{"step_id": 1, "name": "a", "run": "true"}]})

        self.assertEqual("failure", record["conclusion"])

        self.assertIn("source fetch failed", record["error"])

        self.assertEqual("/api/ci/runs/9/conclusion", self.posts[-1][0])

if __name__ == "__main__":

    unittest.main()