import json
import os
import subprocess
import sys
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from wflib import tasks as T
from wflib import lanes as L
from test_cli import Cli
from test_model import LANES

FOUR = LANES.replace("## Needs human", "- **t-four** [P3] (1h): Four.\n\n## Needs human")


class HeldPickTest(unittest.TestCase):
    def test_pick_skips_held(self):
        item, skipped = L.pick(T.parse(FOUR), set(), L.DEFAULT_LANES, "1h", None, "opus",
                               held={"t-one": "opus session uds:/a"})
        self.assertEqual((item.id, [(i.id, why) for i, why in skipped]),
                         ("t-two", [("t-one", "in progress by opus session uds:/a")]))


class ClaimCliTest(Cli):
    tasks_text = FOUR

    def env(self, name, pid):
        sock = self.root / f"{name}.sock"
        sock.write_text("")
        return {"CLAUDE_CODE_MESSAGING_SOCKET": str(sock), "CLAUDE_PID": str(pid), "CLAUDE_CODE_SESSION_ID": name}

    def dead_pid(self):
        p = subprocess.Popen(["true"])
        p.wait()
        return p.pid

    def claim(self, id):
        return json.loads((self.root / ".wf" / "claims" / f"{id}.json").read_text())

    def test_progress_claims_and_other_session_skips(self):
        a, b = self.env("a", os.getpid()), self.env("b", os.getppid())
        self.ok("next", "--lane", "fast", "--as", "opus", "--brief", env=a)
        self.ok("status", "t-three", "progress", "x", env=a)
        c = self.claim("t-three")
        self.assertEqual({k: c[k] for k in ("pid", "socket", "lane", "model")},
                         {"pid": os.getpid(), "socket": str(self.root / "a.sock"), "lane": "fast", "model": "opus"})
        self.assertEqual(self.ok("next", "--lane", "fast", "--as", "opus", "--brief", env=a).splitlines()[0],
                         "- **t-three** [P2] (<1h) (in progress: x): Three.")
        code, out, err = self.wf("next", "--lane", "fast", "--as", "opus", env=b)
        self.assertEqual((code, err), (1, "wf: nothing pickable for fast (opus) in Pending\n"))
        self.assertIn(f"===== Skipped =====\n- t-three: in progress by opus session uds:{self.root / 'a.sock'}\n", out)
        out = self.ok("next", "--lane", "slow", "--as", "opus", env=b)
        self.assertIn("===== Next task =====\n- **t-one**", out)

    def test_dead_claim_ignored(self):
        self.ok("status", "t-three", "progress", "x", env=self.env("a", self.dead_pid()))
        self.assertEqual(self.ok("next", "--lane", "fast", "--as", "opus", "--brief", env=self.env("b", os.getpid())).splitlines()[0],
                         "- **t-three** [P2] (<1h) (in progress: x): Three.")

    def test_claim_without_in_progress_status_ignored(self):
        self.ok("status", "t-three", "progress", "x", env=self.env("a", os.getppid()))
        self.ok("status", "t-three", "clear", env=self.env("b", os.getpid()))
        self.assertFalse((self.root / ".wf" / "claims" / "t-three.json").exists())
        self.assertEqual(self.ok("next", "--lane", "fast", "--as", "opus", "--brief", env=self.env("b", os.getpid())).splitlines()[0],
                         "- **t-three** [P2] (<1h): Three.")

    def test_done_and_blocked_release(self):
        a = self.env("a", os.getpid())
        self.ok("status", "t-three", "progress", "x", env=a)
        self.ok("done", "t-three", "-m", "ok", env=a)
        self.assertFalse((self.root / ".wf" / "claims" / "t-three.json").exists())
        self.ok("status", "t-four", "progress", "x", env=a)
        self.ok("add", "-s", "awaiting", "Q?", env=a)
        self.ok("status", "t-four", "blocked", "a-q", env=a)
        self.assertFalse((self.root / ".wf" / "claims" / "t-four.json").exists())

    def test_no_env_no_claim(self):
        self.ok("status", "t-three", "progress", "x", env={"CLAUDE_PID": "", "CLAUDE_CODE_MESSAGING_SOCKET": ""})
        self.assertFalse((self.root / ".wf" / "claims").exists())
        self.assertFalse((self.root / ".wf" / "sessions").exists())

    def test_same_lane_live_session_warns_and_keeps_registry(self):
        a, b = self.env("a", os.getppid()), self.env("b", os.getpid())
        self.ok("next", "--lane", "fast", "--as", "opus", "--brief", env=a)
        out = self.ok("next", "--lane", "fast", "--as", "opus", "--brief", env=b)
        self.assertEqual(out.splitlines()[0],
                         f"another live fast session holds this lane: uds:{self.root / 'a.sock'} "
                         "(claims keep tasks apart; tell the owner if unintended)")
        reg = json.loads((self.root / ".wf" / "sessions" / "fast.json").read_text())
        self.assertEqual(reg["socket"], str(self.root / "a.sock"))
        self.ok("next", "--lane", "fast", "--as", "opus", "--brief", env=a)  # own re-register: no warning
        self.assertNotIn("another live", self.ok("next", "--lane", "fast", "--as", "opus", "--brief", env=a))

    def test_lanes_unregister_drops_own_record(self):
        a, b = self.env("a", os.getpid()), self.env("b", os.getppid())
        self.ok("next", "--lane", "fast", "--as", "opus", "--brief", env=a)
        self.ok("next", "--lane", "slow", "--as", "sonnet", "--brief", env=b)
        out = self.ok("lanes", "--unregister", env=a)
        self.assertFalse((self.root / ".wf" / "sessions" / "fast.json").exists())
        self.assertTrue((self.root / ".wf" / "sessions" / "slow.json").exists())
        self.assertNotIn(str(self.root / "a.sock"), out)


def git(cwd, *args):
    subprocess.run(["git", "-C", str(cwd), *args], check=True, capture_output=True,
                   env={**os.environ, "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@t", "GIT_COMMITTER_NAME": "t",
                        "GIT_COMMITTER_EMAIL": "t@t"})


class WorktreeTest(Cli):
    tasks_text = FOUR

    def setUp(self):
        super().setUp()
        git(self.root, "init", "-q", "-b", "master")
        (self.root / ".gitignore").write_text(".worktrees/\n")
        git(self.root, "add", "-A")
        git(self.root, "commit", "-qm", "init")
        self.wt = self.root / ".worktrees" / "fast"
        git(self.root, "worktree", "add", "-q", str(self.wt), "-b", "fast/t-three")

    def test_worktree_writes_main_tree(self):
        none = {"CLAUDE_CODE_MESSAGING_SOCKET": ""}
        self.wf("status", "t-three", "progress", "x", project=False, cwd=self.wt, env=none)
        self.assertIn("(in progress: x)", (self.root / "TASKS.md").read_text())
        self.assertNotIn("in progress", (self.wt / "TASKS.md").read_text())
        code, out, err = self.wf("done", "t-four", "-m", "ok", project=False, cwd=self.wt / "docs", env=none)
        self.assertEqual(code, 0, err)
        self.assertIn("**t-four**", self.archive())
        self.assertNotIn("t-four", (self.wt / "tasks" / "archive.md").read_text())

    def test_done_in_worktree_prints_merge_steps(self):
        code, out, err = self.wf("done", "t-four", "-m", "ok", project=False, cwd=self.wt,
                                 env={"CLAUDE_CODE_MESSAGING_SOCKET": ""})
        self.assertEqual(code, 0, err)
        self.assertTrue(out.endswith(
            "worktree mode (branch fast/t-three), after verify: commit your code here (explicit paths), then:\n"
            "  wf merge   (rebase, ff-merge into master, commit TASKS.md tasks/archive.md, push home; "
            "conflict → git rebase master, resolve, verify, wf merge again)\n"), out)
        self.assertNotIn("worktree mode", self.ok("done", "t-three", "-m", "ok"))

    def env(self, name, pid):
        sock = self.root / f"{name}.sock"
        sock.write_text("")
        return {"CLAUDE_CODE_MESSAGING_SOCKET": str(sock), "CLAUDE_PID": str(pid), "CLAUDE_CODE_SESSION_ID": name}

    def test_next_says_worktree_mode_when_two_sessions_live(self):
        son, me = self.env("s", os.getppid()), self.env("o", os.getpid())
        self.assertNotIn("Multi-session", self.ok("next", "--lane", "fast", "--as", "opus", env=me))
        self.ok("next", "--lane", "slow", "--as", "sonnet", "--brief", env=son)
        out = self.ok("next", "--lane", "fast", "--as", "opus", env=me)
        self.assertIn("===== Multi-session =====\n"
                      "2 live sessions here: work in your lane's worktree, never on master:\n"
                      "  cd .worktrees/fast && git switch -c fast/<task> master  (wf there writes this TASKS.md)\n\n", out)
        self.ok("next", "--lane", "slow", "--as", "sonnet", "--brief", env=son)
        out = self.ok("next", "--lane", "slow", "--as", "sonnet", env=son)
        self.assertIn("  git worktree add .worktrees/slow -b slow/<task> master  (wf there writes this TASKS.md)\n", out)
        code, out, err = self.wf("next", "--lane", "fast", "--as", "opus", project=False, cwd=self.wt, env=me)
        self.assertIn("===== Multi-session =====\n2 live sessions here: you are in worktree .worktrees/fast "
                      "(branch fast/t-three); wf writes the main tree's TASKS.md\n", out)


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


class StaleTest(ClaimCliTest):
    def test_stale_list_and_clear(self):
        self.ok("status", "t-three", "progress", "x", env=self.env("a", self.dead_pid()))
        self.ok("status", "t-four", "progress", "y", env=self.env("b", os.getpid()))
        self.assertEqual(self.ok("list", "--stale").splitlines()[0].split()[0], "t-three")
        self.assertNotIn("t-four", self.ok("list", "--stale"))
        self.assertIn("cleared t-three", self.ok("status", "--clear-stale"))
        self.assertEqual(self.ok("list", "--progress").count("prog"), 1)
        self.assertIn("t-four", self.ok("list", "--progress"))
        self.assertFalse((self.root / ".wf" / "claims" / "t-three.json").exists())

    def test_no_claim_listed(self):
        self.ok("status", "t-three", "progress", "x")
        (self.root / ".wf" / "claims" / "t-three.json").unlink(missing_ok=True)
        self.assertIn("t-three", self.ok("list", "--stale"))