workflow: initial public history

1 files changed, 182 insertions, 0 deletions

diff --git a/tests/test_claims.py b/tests/test_claims.py
new file mode 100644
index 0000000..e6b5c5d
--- /dev/null
+++ b/tests/test_claims.py
@@ -0,0 +1,182 @@
+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"))