1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
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"))
|
