"""Regression tests for the correctness review of rounds five and six."""
import json
import os
import re
import shutil
import subprocess
import threading
import unittest
from pathlib import Path

os.environ["HARBOR_NO_TOASTS"] = "1"
from helpers import ROOT, HarborCase, make_home, run_module, temp_dir
from harbor import notify, projects
from harbor.core import Harbor, HarborError


class StopState(HarborCase):
    def test_02_trial_for_one_agent_while_everyone_is_stopped(self):
        self.h.set_flag("stop:all", "on", "User", "web")
        self.clock.t += 60
        self.h.set_flag("stop:Codex", "off", "User", "web", until="30m")
        self.assertEqual((self.h.stopped("Codex"), bool(self.h.trial("Codex"))), ([], True))
        self.assertTrue(self.h.stopped("Kimi"))
        self.clock.t += 60
        self.h.set_flag("stop:all", "on", "Claude", "cli")                     # an agent's re-stop changes nothing
        self.assertFalse(self.h.stopped("Codex"))
        self.h.set_flag("stop:all", "on", "User", "web")                       # the human pressing it again is a
        self.assertTrue(self.h.stopped("Codex"))                                # new decision: it covers everyone

    def test_03_coordinator_stop_and_lift_keep_the_trial_end(self):
        self.h.set_flag("stop:Codex", "on", "User", "web")
        self.h.set_flag("stop:Codex", "off", "User", "web", until="30m")
        self.h.set_flag("stop:Codex", "on", "Claude", "cli")
        self.h.set_flag("stop:Codex", "off", "Claude", "cli")
        self.assertIsNotNone(self.h.trial("Codex"))
        self.clock.t += 31 * 60
        self.assertTrue(self.h.stopped("Codex"))                               # the trial still ends

    def test_04_ended_trial_is_a_stop_everywhere(self):
        (self.dir / "x").mkdir()
        self.h.set_flag("stop:Codex", "on", "User", "web")
        self.h.set_flag("stop:Codex", "off", "User", "web", until="1m")
        self.clock.t += 120
        self.assertTrue(self.h.stopped("Codex"))
        self.assertIn("trial run ended", self.h.stopped("Codex")[0]["reason"])
        launch = {a["agent"]: a for a in self.h.launch_status()}
        self.assertEqual(launch["Codex"]["state"], "stopped")

    def test_07_launch_shows_team_trials(self):
        self.h.set_flag("stop:all", "on", "User", "web")
        self.clock.t += 60
        self.h.set_flag("stop:all", "off", "User", "web", until="15m")
        self.assertEqual({a["agent"]: a for a in self.h.launch_status()}["Kimi"]["state"], "trial")


class RollCalls(HarborCase):
    def test_06_second_roll_call_keeps_the_first(self):
        self.h.rollcall("User", "web")
        codex_q = next(q for q in self.h.questions(to="Codex"))
        self.h.answer("Codex", codex_q["id"], "Here")
        self.h.rollcall("User", "web", agents=["Kimi"])
        rows = {r["agent"]: r for r in self.h.rollcall_status()["agents"]}
        self.assertEqual(rows["Codex"]["answer"], "Here")
        self.assertEqual(len(rows), 5)
        self.assertEqual(len([q for q in self.h.questions(to="Kimi")]), 1)       # the older one is closed


class Notifications(HarborCase):
    def test_05_backlog_over_fifty(self):
        notify.set_enabled(self.h, True)
        for i in range(60):
            self.h.send("Codex", "All", f"broadcast {i}")
        self.h.send("Codex", ["User"], "the one that matters")
        pending, top = notify.pending_for_human(self.h)
        self.assertEqual([m["body"] for m in pending], ["the one that matters"])


class FlowFixes(unittest.TestCase):
    def setUp(self):
        self.dir = temp_dir()

    def tearDown(self):
        shutil.rmtree(self.dir, ignore_errors=True)

    def project(self, **extra):
        data = {"name": "f", "coordinators": ["Lead"], "database": str(self.dir / "f.db"),
                "agents": [{"name": "Lead"}, {"name": "Ann"}, {"name": "Bob"}], **extra}
        return projects.project_from_dict(data)

    def test_08_09_bulk_and_templates_by_a_worker(self):
        h = Harbor(self.project(templates={"t": {"steps": [{"name": "a", "title": "Fix {name} in {scope}",
                                                             "assignee": "Bob"}]}}))
        h.claim("Lead", "range:0x440000-0x47FFFF", kind="band", on_behalf_of="Bob")
        r = h.bulk_tasks("Ann", "0x445310 One\n0x445320 Two")
        self.assertEqual([t["assignee"] for t in r["created"]], [None, None])      # no routing to someone else
        with self.assertRaises(HarborError):
            h.run_template("Ann", "t", scope="fn:0x445400")
        self.assertEqual(len(h.tasks()), 2)                                          # nothing half-created
        task = h.run_template("Lead", "t", scope="fn:0x445400")[0]
        self.assertEqual(task["title"], "Fix {name} in fn:0x445400")
        h.close()

    def test_11_policy_checked_at_load(self):
        p = self.project(policy={"land_requires": "Ann"})
        self.assertEqual(p.policy["land_requires"], ["Ann"])
        for bad in ({"land_requires": ["Ghost"]}, {"all_per_hour": "lots"}):
            with self.assertRaises(projects.ProjectError):
                self.project(policy=bad)
        h = Harbor(self.project(policy={"land_requires": ["Ann"]}))
        item = h.submit("Ann", "x", commit="abc")
        h.transition("Lead", item["id"], "approved")
        h.transition("Lead", item["id"], "landed", commit="b4")                    # the author is not required
        h.close()

    def test_10_rebase_does_not_look_new(self):
        env = {**os.environ, "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@x", "GIT_COMMITTER_NAME": "t",
               "GIT_COMMITTER_EMAIL": "t@x", "GIT_AUTHOR_DATE": "2026-01-01T10:00:00", "GIT_COMMITTER_DATE": "2026-01-01T10:00:00"}
        repo = self.dir / "r"
        repo.mkdir()
        g = lambda *a, e=env: subprocess.run(["git", "-C", str(repo), *a], check=True, capture_output=True, env=e)
        g("init", "-q", "-b", "main")
        (repo / "a").write_text("a")
        g("add", "a")
        g("commit", "-q", "-m", "base")
        g("checkout", "-q", "-b", "work")
        (repo / "b").write_text("b")
        g("add", "b")
        g("commit", "-q", "-m", "draft")
        h = Harbor(projects.project_from_dict({"name": "g", "database": str(self.dir / "g.db"),
                                               "agents": [{"name": "Ann", "workdir": str(repo)}]}))
        h.clock = lambda: 1767268800.0 + 3600 * 5                                  # 2026-01-01 15:00 UTC
        sha = h.git_status("Ann", max_age=0)["last_commit"]["sha"]
        h.submit("Ann", "draft", commit=sha, reviewers=["User"])
        h.clock = lambda: 1767268800.0 + 3600 * 6
        g("checkout", "-q", "main")
        (repo / "c").write_text("c")
        g("add", "c")
        g("commit", "-q", "-m", "main moves", e={**env, "GIT_AUTHOR_DATE": "2026-01-01T16:00:00",
                                                  "GIT_COMMITTER_DATE": "2026-01-01T16:00:00"})
        g("checkout", "-q", "work")
        g("rebase", "-q", "main", e={**env, "GIT_COMMITTER_DATE": "2026-01-01T16:30:00"})
        self.assertEqual(h.git_status("Ann", max_age=0)["unsubmitted"], [])
        h.close()


class Pages(unittest.TestCase):
    def test_01_no_code_hidden_behind_a_line_comment(self):
        for page in ("app.html",):
            scripts = re.findall(r"<script>([\s\S]*?)</script>", (ROOT / "harbor" / "web" / page).read_text(encoding="utf-8"))
            self.assertTrue(scripts, page)
            for line in "\n".join(scripts).splitlines():
                code = re.sub(r"'(?:\\.|[^'\\])*'|\"(?:\\.|[^\"\\])*\"|`(?:\\.|[^`\\])*`|/(?:\\.|[^/\\\n])+/[gimsuy]*", "", line)
                if "//" in code:
                    after = code.split("//", 1)[1]
                    self.assertFalse(re.search(r"\b(const|let|var|function|if|return)\b", after), (page, line[:120]))


class Race(HarborCase):
    def test_answer_race_for_real(self):
        q = self.h.ask("Codex", "Ship?", ["Yes", "No"], fallback="Claude")
        self.h.set_presence("User", "web", "away")
        results, gate = [], threading.Barrier(2)

        def answer(who, value, origin):
            h = Harbor(self.project, clock=self.clock)
            try:
                gate.wait()
                results.append(h.answer(who, q["id"], value, origin=origin)["answer"])
            except HarborError as error:
                results.append(f"refused: {error}")
            finally:
                h.close()
        threads = [threading.Thread(target=answer, args=a) for a in (("Claude", "Yes", "cli"), ("User", "No", "web"))]
        [t.start() for t in threads]
        [t.join() for t in threads]
        self.assertEqual(sum(1 for r in results if not r.startswith("refused")), 1, results)
        self.assertEqual(len([m for m in self.h.inbox("Codex", unread_only=False) if m["kind"] == "answer"]), 1)


class LaunchFindings(unittest.TestCase):
    """What the first launch with the real agent apps showed."""

    def gate(self, command=None, tool="shell"):
        from harbor import hooks
        payload = {"tool_name": tool}
        if command is not None:
            payload["tool_input"] = {"command": command}
        return hooks._tool_is_harbor(payload)

    def test_messages_may_hold_shell_characters_inside_double_quotes(self):
        run = "py -3.12 -B R:/Workspace/Tools/Harbor/run.py --as Kimi "
        for c in [run + 'send Claude --thread launch "Kimi is online (through the command; MCP was blocked) & ready"',
                  'harbor ack 1708 --note "ok, I\'m stopped <as asked> {no work} @Claude ^"',
                  'harbor send Claude "it\u2019s fine"', "harbor send Claude 'plain words'"]:
            self.assertTrue(self.gate(c), c)
        for c in ['harbor send All "$(id)"', 'harbor send All "a" ; rm x', "harbor send All '\"'; rm x; '\"'",
                  'harbor send All "x\\" ; rm y ; \\"z"', 'harbor send All "100%"', 'harbor send All "line\nRemove-Item x"',
                  'harbor send All "a\u201d ; rm x ; \u201cb"', "harbor send All 'it (x)'", 'harbor send All "unterminated',
                  'harbor send All "`whoami`"', 'harbor send All \\"x\\"']:
            self.assertFalse(self.gate(c), c)

    def test_every_hosts_harbor_tool_names_and_tool_loaders(self):
        for name in ("mcp__plugin-harbor-open-seas_harbor__send", "mcp__harbor-open-seas__brief", "harbor-open-seas.inbox",
                     "mcp__harbor__wait_for_messages", "harbor_ack", "select_tools", "ToolSearch"):
            self.assertTrue(self.gate(tool=name), name)
        for name in ("mcp__plugin-harbor-open-seas_harbor__land", "mcp__plugin-harbor-open-seas_harbor__next_task",
                     "mcp__other__send", "my_harbor_tool", "Harbor", "mcp__plugin-harbor-open-seas_harbor__resume"):
            self.assertFalse(self.gate(tool=name), name)

    def test_the_stop_message_says_sending_is_allowed(self):
        from harbor import hooks

        class FakeHarbor:
            human = "User"
        text = hooks._stop_text(FakeHarbor(), "Kimi", [{"key": "stop:Kimi", "set_by": "User", "set_at": "now", "reason": None}])
        self.assertIn("send", text)
        self.assertIn("double quotes", text)


class AllocationWhileStopped(HarborCase):
    def test_coordinator_records_bands_for_stopped_agents_but_no_work(self):
        self.h.set_flag("stop:Codex", "on", "User", "web")
        self.h.claim("Claude", "range:0x401000-0x43FFFF", kind="band", on_behalf_of="Codex")
        self.h.claim("Claude", "fn:0x401200", kind="naming", on_behalf_of="Codex")
        with self.assertRaises(HarborError):
            self.h.claim("Claude", "fn:0x401300", kind="draft", on_behalf_of="Codex")
        with self.assertRaises(HarborError):
            self.h.claim("Codex", "fn:0x401400")


class BulkNotice(HarborCase):
    def test_bulk_sends_one_notice_per_assignee(self):
        self.h.claim("Claude", "range:0x401000-0x43FFFF", kind="band", on_behalf_of="Codex")
        before = len(self.h.inbox("Codex", unread_only=False, limit=500, mark_read=False))
        r = self.h.bulk_tasks("Claude", "0x401100 One\n0x401200 Two\n0x401300 Three")
        self.assertEqual([t["assignee"] for t in r["created"]], ["Codex"] * 3)
        new = self.h.inbox("Codex", unread_only=False, limit=500, mark_read=False)[before:]
        self.assertEqual(len(new), 1)
        self.assertIn("3 new tasks", new[0]["body"])


class Autopilot(HarborCase):
    def setUp(self):
        super().setUp()
        self.h.project.autopilot = True

    def test_running_agents_keep_going_and_stopped_or_stuck_ones_rest(self):
        self.h.claim("Claude", "range:0x401000-0x43FFFF", kind="band", on_behalf_of="Codex")
        t = self.h.add_task("Claude", "Make 0x401100 readable", scope="fn:0x401100", assignee="Codex")
        self.assertIn(f"#{t['id']}", self.h.continuation("Codex"))                 # offers the next task
        self.h.take_task("Codex", t["id"])
        self.assertIn(f"Continue task #{t['id']}", self.h.continuation("Codex"))
        self.assertIsNotNone(self.h.continuation("Codex"))
        self.assertIsNone(self.h.continuation("Codex"))                             # three idle turn ends: rest
        self.assertEqual([a["agent"] for a in self.h.attention() if a["kind"] == "resting"], ["Codex"])
        self.h.activity("Codex", "Edit", "codex", target="src/x.c")                 # real work resets the guard
        self.assertIsNotNone(self.h.continuation("Codex"))
        self.assertEqual([a for a in self.h.attention() if a["kind"] == "resting"], [])
        self.h.set_flag("stop:Codex", "on", "User", "web")
        self.assertIsNone(self.h.continuation("Codex"))                             # the human's stop rests it

    def test_band_fallback_and_reviewer_work(self):
        self.h.claim("Claude", "range:0x401000-0x43FFFF", kind="band", on_behalf_of="Codex")
        self.assertIn("own band", self.h.continuation("Codex"))
        self.assertIn("wait_for_messages", self.h.continuation("Kimi"))
        item = self.h.submit("Codex", "Readable 0x401100", commit="abc123")
        self.assertIn(f"#{item['id']}", self.h.continuation("Kimi"))


class LandMany(HarborCase):
    def test_batch_landing_sends_one_message_per_author_and_one_win(self):
        ids = [self.h.submit("Codex", f"Readable 0x40{i}000", commit=f"c{i}")["id"] for i in range(3)]
        before = self.h.conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0]
        landed = self.h.land_many("Claude", ids, "abc1234", note="readable check passed")
        self.assertEqual([i["state"] for i in landed], ["landed"] * 3)
        self.assertEqual(self.h.conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0] - before, 2)
        with self.assertRaises(HarborError):
            self.h.land_many("Codex", ids, "abc1234")


class AuthorRefreshesCommit(HarborCase):
    def test_author_hands_in_a_newer_commit_while_submitted_or_approved(self):
        item = self.h.submit("Codex", "Readable 0x401000", commit="c1")
        self.h.transition("Codex", item["id"], "submitted", note="newer", commit="c2")
        self.assertEqual(self.h.item(item["id"])[0]["commit_hash"], "c2")
        self.h.transition("Kimi", item["id"], "approved")
        self.h.transition("Codex", item["id"], "submitted", note="one more fix", commit="c3")
        self.assertEqual(self.h.item(item["id"])[0]["state"], "submitted")      # back for review with c3
        with self.assertRaises(HarborError):
            self.h.transition("Codex", item["id"], "submitted")                   # no commit: nothing to refresh
        with self.assertRaises(HarborError):
            self.h.transition("Kimi", item["id"], "submitted", commit="c4")       # only the author


class ProjectRefresh(unittest.TestCase):
    def test_edits_to_the_project_file_apply_without_a_restart(self):
        d = temp_dir()
        try:
            path = d / "p.toml"
            body = 'name = "p"\ndatabase = \'{db}\'\nreviewers = ["Ann"]\n[[agents]]\nname = "Ann"\n[[agents]]\nname = "Bob"\n'
            path.write_text(body.format(db=d / "p.db"), encoding="utf-8")
            h = Harbor(projects.load_file(path))
            self.assertFalse(h.refresh_project())
            path.write_text(body.format(db=d / "p.db").replace('reviewers = ["Ann"]', 'reviewers = ["Bob"]\nautopilot = true'),
                            encoding="utf-8")
            os.utime(path, (os.path.getmtime(path) + 5,) * 2)
            self.assertTrue(h.refresh_project())
            self.assertEqual((h.project.reviewers, h.project.autopilot), (("Bob",), True))
            path.write_text("this is not toml [", encoding="utf-8")
            os.utime(path, (os.path.getmtime(path) + 10,) * 2)
            self.assertFalse(h.refresh_project())                                   # keeps the last good one
            self.assertEqual(h.project.reviewers, ("Bob",))
            h.close()
        finally:
            shutil.rmtree(d, ignore_errors=True)


class SharedFiles(HarborCase):
    def test_shared_annotation_files_are_not_collisions(self):
        self.h.project.shared_files = ("annotations/*.csv", "annotations/prototypes.h")
        self.h.record_touches("Codex", ["annotations/names.csv", "annotations/prototypes.h", "src/a.c"])
        cols = self.h.record_touches("Kimi", ["annotations/names.csv", "annotations/prototypes.h", "src/a.c"])
        self.assertEqual([c["path"] for c in cols], ["src/a.c"])
        self.assertEqual([c["path"] for c in self.h.recent_collisions()], ["src/a.c"])


class WorkingWithoutHooks(HarborCase):
    def test_harbor_calls_and_file_scans_count_as_work_but_not_as_hooks(self):
        self.h.activity("DeepSeek", "harbor submit", "deepseek", channel="mcp")
        row = next(a for a in self.h.agents() if a["name"] == "DeepSeek")
        self.assertTrue(self.h.working(row))
        self.assertIsNone({r["agent"]: r for r in self.h.launch_status()}["DeepSeek"]["hooks"])
        self.h.activity("DeepSeek", "edit", "", target="src/x.c", channel="files")
        self.assertIn("files", {c["channel"] for c in self.h.connections()["DeepSeek"]})


class UnreadQuestions(HarborCase):
    def test_unread_questions_to_an_agent_need_attention_until_read(self):
        q = self.h.send("DeepSeek", ["Claude"], "May I take the 0x4C slice?", kind="question")
        self.h.send("DeepSeek", ["All"], "status for everyone", kind="request")
        kinds = lambda: [a for a in self.h.attention() if a["kind"] == "unread_questions"]
        self.assertEqual(kinds(), [])                                        # not yet 20 minutes old
        self.clock.t += 25 * 60
        found = kinds()
        self.assertEqual([(a["agent"], f"#{q['id']}" in a["text"]) for a in found], [("Claude", True)])
        self.h.inbox("Claude")                                               # reading it clears the item
        self.assertEqual(kinds(), [])


class AttentionWithoutFalseAlarms(HarborCase):
    def test_stopped_agents_owe_no_acks_and_file_scopes_carry_their_address(self):
        from harbor.util import parse_scope
        self.assertEqual(parse_scope("file:src/game/_unassigned/4C6540_UnkPediaTextHelper.c")[1:], (0x4C6540, 0x4C6540))
        self.assertEqual(parse_scope("file:annotations/names.csv")[1:], (None, None))
        m = self.h.send("Codex", ["Kimi"], "please confirm", ack_required=True)
        self.clock.t += 120 * 60
        self.assertTrue(any(a["kind"] == "ack_overdue" for a in self.h.attention()))
        self.h.set_flag("stop:Kimi", "on", "Claude", "cli")
        self.assertFalse(any(a["kind"] == "ack_overdue" and f"#{m['id']}" in a["text"] for a in self.h.attention()))


class KimiStopKeepGoing(unittest.TestCase):
    def test_kimi_gets_exit_2_and_stderr_others_get_json(self):
        import contextlib, io
        from harbor import hooks
        out, err = io.StringIO(), io.StringIO()
        with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
            code = hooks._keep_going("kimi", "keep going")
        self.assertEqual((code, out.getvalue(), err.getvalue().strip()), (2, "", "keep going"))
        out = io.StringIO()
        with contextlib.redirect_stdout(out):
            code = hooks._keep_going("codex", "keep going")
        self.assertEqual(code, 0)
        self.assertEqual(json.loads(out.getvalue())["decision"], "block")


class HooksVisible(HarborCase):
    def test_mcp_without_hooks_needs_attention(self):
        self.h.connect("Codex", "mcp", "codex")
        no_hooks = lambda: [a["agent"] for a in self.h.attention() if a["kind"] == "no_hooks"]
        launch = lambda: {r["agent"]: r for r in self.h.launch_status()}["Codex"]["hooks"]
        self.assertEqual((no_hooks(), launch()), (["Codex"], None))
        self.h.connect("Codex", "hook", "codex")
        self.assertEqual(no_hooks(), [])
        self.assertTrue(launch())


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