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
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
|
import os
import subprocess
import unittest
from pathlib import Path
from test_cli import TOML, Cli
from test_claims import git
from test_merge import IDENT
class SplitCli(Cli):
"""Private project (self.root, git, workflow.toml with code_root) + sibling code repo (self.code)."""
toml = TOML + 'code_root = "../code"\n'
def setUp(self):
super().setUp()
git(self.root, "init", "-q", "-b", "master")
(self.root / ".gitignore").write_text(".wf/\nout/\n")
git(self.root, "add", "-A")
git(self.root, "commit", "-qm", "init")
self.code = self.root.parent / "code"
self.code.mkdir()
git(self.code, "init", "-q", "-b", "master")
(self.code / ".gitignore").write_text(".worktrees/\n.wf-home\n")
(self.code / "lib.txt").write_text("a\n")
git(self.code, "add", "-A")
git(self.code, "commit", "-qm", "init")
def add_wt(self, lane="sonnet", id="t-three"):
"""Code worktree on branch <lane>/<id> with .wf-home (what wf start makes)."""
wt = self.code / ".worktrees" / lane
git(self.code, "worktree", "add", "-q", str(wt), "-b", f"{lane}/{id}")
(wt / ".wf-home").write_text(f"{self.root}\n")
return wt
def log(self, repo, rev="master"):
return subprocess.run(["git", "log", "--format=%s", rev], cwd=repo, capture_output=True, text=True).stdout
def sha(self, repo, rev="master"):
return subprocess.run(["git", "rev-parse", "--short", rev], cwd=repo, capture_output=True, text=True).stdout.strip()
class DiscoveryTest(SplitCli):
def test_wf_in_code_worktree_finds_private_project(self):
wt = self.add_wt()
code, out, err = self.wf("show", "t-three", project=False, cwd=wt)
self.assertEqual(code, 0, err)
self.assertIn("**t-three**", out)
def test_dangling_wf_home_is_config_error(self):
wt = self.add_wt()
(wt / ".wf-home").write_text(f"{self.root.parent / 'gone'}\n")
code, out, err = self.wf("show", "t-three", project=False, cwd=wt)
self.assertNotEqual(code, 0)
self.assertIn(".wf-home points to", err)
self.assertNotIn("Traceback", err)
def test_split_and_code_main(self):
import sys
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from wflib import config
cfg = config.load(self.root)
self.assertTrue(cfg.split)
self.assertEqual(config.code_main(cfg), self.code)
self.assertEqual(cfg.push, [])
def test_code_root_inside_same_repo_not_split(self):
import sys
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from wflib import config
(self.root / "workflow.toml").write_text(TOML + 'code_root = "."\npush = ["true"]\n')
cfg = config.load(self.root)
self.assertFalse(cfg.split)
self.assertIsNone(config.code_main(cfg))
self.assertEqual(cfg.push, ["true"])
class StartSplitTest(SplitCli):
toml = SplitCli.toml + 'worktree_setup = ["pwd > where.txt"]\n'
def test_start_makes_code_worktree_with_wf_home(self):
wt = self.code / ".worktrees" / "slow"
code, out, err = self.wf("start", "t-three", "--worktree", str(wt), "--branch", "slow/t-three",
project=False, env=IDENT)
self.assertEqual(code, 0, out + err)
self.assertEqual((wt / ".wf-home").read_text(), f"{self.root}\n")
self.assertEqual(subprocess.run(["git", "branch", "--show-current"], cwd=wt, capture_output=True,
text=True).stdout.strip(), "slow/t-three")
self.assertEqual((wt / "where.txt").read_text().strip(), str(wt)) # setup cwd = code worktree
self.assertIn("(in progress: slow/t-three)", (self.root / "TASKS.md").read_text())
self.assertIn(f"cd {wt} && (make test) && ", out)
self.assertFalse((self.root / ".worktrees").exists()) # nothing in the private repo
def test_start_refuses_private_repo_worktree(self):
wt = self.root / ".worktrees" / "slow"
git(self.root, "worktree", "add", "-q", str(wt), "-b", "slow/old")
code, out, err = self.wf("start", "t-three", "--worktree", str(wt), "--branch", "slow/t-three",
project=False, env=IDENT)
self.assertNotEqual(code, 0)
self.assertIn("is not a worktree of", err)
self.assertFalse((wt / ".wf-home").exists())
self.assertEqual(subprocess.run(["git", "branch", "--show-current"], cwd=wt, capture_output=True,
text=True).stdout.strip(), "slow/old") # not switched
self.assertIn("- **t-three** [P2] (<1h): Third", (self.root / "TASKS.md").read_text()) # no progress
def test_start_reuses_existing_code_worktree(self):
wt = self.code / ".worktrees" / "slow"
git(self.code, "worktree", "add", "-q", str(wt), "-b", "slow/old")
code, out, err = self.wf("start", "t-three", "--worktree", str(wt), "--branch", "slow/t-three",
project=False, env=IDENT)
self.assertEqual(code, 0, out + err)
self.assertEqual((wt / ".wf-home").read_text(), f"{self.root}\n")
def test_setup_in_code_worktree(self):
wt = self.add_wt()
code, out, err = self.wf("setup", project=False, cwd=wt)
self.assertEqual(code, 0, err)
self.assertEqual((wt / "where.txt").read_text().strip(), str(wt))
def env(self, name, pid): # copy of tests/test_claims.py ClaimTest.env
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_advice_names_wf_start_in_code_repo(self):
# pattern: tests/test_claims.py::test_next_says_worktree_mode_when_two_sessions_live
son, me = self.env("s", os.getppid()), self.env("o", os.getpid())
self.wf("next", "--lane", "slow", "--as", "sonnet", "--brief", env=son)
code, out, err = self.wf("next", "--lane", "fast", "--as", "opus", env=me)
self.assertIn(f" wf start <task> --worktree {self.code / '.worktrees' / 'fast'} --branch fast/<task> "
f"(run in {self.root}; wf there writes this TASKS.md)\n", out)
class FinishSplitTest(SplitCli):
def setUp(self):
super().setUp()
self.wt = self.add_wt()
def finish(self, *args):
return self.wf("finish", *args, "--no-push", project=False, cwd=self.wt, env=IDENT)
def test_code_only(self):
(self.wt / "code.txt").write_text("x\n")
code, out, err = self.finish("t-three", "-m", "ok", "--commit", "impl", "code.txt")
self.assertEqual(code, 0, out + err)
self.assertEqual(self.log(self.code), "impl\ninit\n")
csha = self.sha(self.code)
self.assertEqual(self.log(self.root), f"t-three done (code {csha})\ninit\n")
self.assertTrue(out.endswith(f"report: commit {csha} books {self.sha(self.root)}\n"), out)
self.assertEqual(subprocess.run(["git", "branch", "--list", "sonnet/t-three"], cwd=self.code,
capture_output=True, text=True).stdout, "")
self.assertIn("t-three", (self.root / "tasks" / "archive.md").read_text())
def test_code_and_private_path(self):
(self.wt / "code.txt").write_text("x\n")
(self.root / "docs" / "notes.md").write_text("n\n")
code, out, err = self.finish("t-three", "-m", "ok", "--commit", "impl", "code.txt",
str(self.root / "docs" / "notes.md"))
self.assertEqual(code, 0, out + err)
self.assertEqual(self.log(self.code), "impl\ninit\n")
self.assertEqual(self.log(self.root), f"impl (code {self.sha(self.code)})\ninit\n")
self.assertIn("docs/notes.md", subprocess.run(["git", "show", "--stat", "HEAD"], cwd=self.root,
capture_output=True, text=True).stdout)
def test_private_only(self):
(self.root / "docs" / "notes.md").write_text("n\n")
code, out, err = self.finish("t-three", "-m", "ok", "--commit", "areas", str(self.root / "docs" / "notes.md"))
self.assertEqual(code, 0, out + err)
self.assertEqual(self.log(self.code), "init\n") # no empty code commit
self.assertEqual(self.log(self.root), "areas\ninit\n")
self.assertTrue(out.endswith(f"report: commit - books {self.sha(self.root)}\n"), out)
self.assertEqual(subprocess.run(["git", "branch", "--list", "sonnet/t-three"], cwd=self.code,
capture_output=True, text=True).stdout, "")
def test_unignored_wf_home_does_not_block(self):
(self.code / ".gitignore").write_text(".worktrees/\n")
git(self.code, "commit", "-qam", "no wf-home ignore")
git(self.wt, "rebase", "-q", "master")
(self.wt / "code.txt").write_text("x\n")
code, out, err = self.finish("t-three", "-m", "ok", "--commit", "impl", "code.txt")
self.assertEqual(code, 0, out + err)
self.assertNotIn(".wf-home", subprocess.run(["git", "show", "--stat", "master"], cwd=self.code,
capture_output=True, text=True).stdout)
def test_stray_code_file_refused_task_open(self):
(self.wt / "code.txt").write_text("x\n")
(self.wt / "stray.txt").write_text("s\n")
code, out, err = self.finish("t-three", "-m", "ok", "--commit", "impl", "code.txt")
self.assertNotEqual(code, 0)
self.assertIn("stray.txt", err)
self.assertIn("- **t-three**", (self.root / "TASKS.md").read_text())
def test_path_in_neither_repo_refused(self):
other = self.root.parent / "elsewhere.txt"
other.write_text("e\n")
code, out, err = self.finish("t-three", "-m", "ok", "--commit", "impl", str(other))
self.assertNotEqual(code, 0)
self.assertIn("in neither repo", err)
def test_rebase_conflict_commits_nothing_private(self):
(self.wt / "lib.txt").write_text("branch\n")
(self.code / "lib.txt").write_text("master\n")
git(self.code, "commit", "-qam", "moved")
code, out, err = self.finish("t-three", "-m", "ok", "--commit", "impl", "lib.txt")
self.assertNotEqual(code, 0)
self.assertIn("conflicts", err)
self.assertEqual(self.log(self.root), "init\n") # books not committed yet
(self.wt / "lib.txt").write_text("both\n") # resolve by hand as the worker would
git(self.wt, "reset", "-q", "--soft", "master")
git(self.wt, "commit", "-qam", "impl")
code, out, err = self.wf("merge", "--no-push", project=False, cwd=self.wt, env=IDENT)
self.assertEqual(code, 0, out + err)
self.assertTrue(self.log(self.root).startswith("t-three done (code "), self.log(self.root))
def test_wip_commits_both(self):
(self.wt / "code.txt").write_text("x\n")
(self.root / "docs" / "notes.md").write_text("n\n")
code, out, err = self.wf("wip", "t-three", "-m", "half", "--commit", "wip", "code.txt",
str(self.root / "docs" / "notes.md"), project=False, cwd=self.wt, env=IDENT)
self.assertEqual(code, 0, out + err)
self.assertEqual(self.log(self.code, "sonnet/t-three"), "wip\ninit\n")
self.assertEqual(self.log(self.root), "t-three wip: half\ninit\n")
def test_done_lines_name_private_books(self):
code, out, err = self.wf("done", "t-three", "-m", "ok", project=False, cwd=self.wt, env=IDENT)
self.assertEqual(code, 0, out + err)
self.assertIn("commit TASKS.md tasks/archive.md, push home", out)
class PushSplitTest(SplitCli):
toml = SplitCli.toml + 'push = ["echo pushing", "echo boom; exit 3"]\n'
def setUp(self):
super().setUp()
self.wt = self.add_wt()
(self.wt / "code.txt").write_text("x\n")
def test_push_failure_keeps_merge_marks_and_reports(self):
code, out, err = self.wf("finish", "t-three", "-m", "ok", "--commit", "impl", "code.txt",
project=False, cwd=self.wt, env=IDENT)
self.assertEqual(code, 0, out + err)
self.assertIn("not pushed (exit 3): echo boom; exit 3\n", out)
self.assertIn(" boom\n", out)
self.assertIn("rerun: wf push", out)
self.assertRegex(out, r"report: commit [0-9a-f]+ books [0-9a-f]+ push-failed 3\n$")
self.assertEqual(self.log(self.code), "impl\ninit\n")
self.assertTrue((self.root / ".wf" / "push-failed").is_file())
code, out, err = self.wf("next", "--brief", project=False, cwd=self.root)
self.assertIn("push failed", out)
def test_wf_push_success_clears_marker(self):
self.wf("finish", "t-three", "-m", "ok", "--commit", "impl", "code.txt", project=False, cwd=self.wt, env=IDENT)
(self.root / "workflow.toml").write_text(SplitCli.toml + 'push = ["true"]\n')
code, out, err = self.wf("push", project=False, cwd=self.root)
self.assertEqual(code, 0, out + err)
self.assertFalse((self.root / ".wf" / "push-failed").exists())
def test_push_runs_in_private_root(self):
(self.root / "workflow.toml").write_text(SplitCli.toml + 'push = ["pwd > pushed-from.txt"]\n')
code, out, err = self.wf("finish", "t-three", "-m", "ok", "--commit", "impl", "code.txt",
project=False, cwd=self.wt, env=IDENT)
self.assertEqual(code, 0, out + err)
self.assertEqual((self.root / "pushed-from.txt").read_text().strip(), str(self.root))
class OrchSplitTest(SplitCli):
def setUp(self):
super().setUp()
self.wf("set", "t-three", "--model", "sonnet", "--done", "x")
def test_pick_worktree_in_code_repo(self):
code, out, err = self.wf("orch", "pick", "fast", "--id", "t-three", project=False, env=IDENT)
self.assertEqual(code, 0, out + err)
self.assertIn(f"Main tree: {self.root} Worktree: {self.code / '.worktrees' / 'fast'} Branch: fast/t-three", out)
def test_post_done_checks_code_branch_and_stops_on_push_failed(self):
self.wf("orch", "pick", "fast", "--id", "t-three", project=False, env=IDENT)
wt = self.code / ".worktrees" / "fast"
self.wf("start", "t-three", "--worktree", str(wt), "--branch", "fast/t-three", project=False, env=IDENT)
(wt / "code.txt").write_text("x\n")
(self.root / "workflow.toml").write_text(SplitCli.toml + 'push = ["exit 1"]\n')
code, out, err = self.wf("finish", "t-three", "-m", "ok", "--commit", "impl", "code.txt",
project=False, cwd=wt, env=IDENT)
self.assertEqual(code, 0, out + err)
code, out, err = self.wf("orch", "post", "t-three", "fast", "--result", "done", project=False, env=IDENT)
self.assertEqual(code, 0, out + err)
self.assertIn("post: t-three push-failed", out)
self.assertIn("stop lane fast: push-failed", out)
self.assertNotIn("branch fast/t-three still there", out)
self.assertIn(f" {self.sha(self.code)} ", (self.root / "out" / "wf-orch.log").read_text())
def _finish_then_dirty_books(self):
self.wf("orch", "pick", "fast", "--id", "t-three", project=False, env=IDENT)
wt = self.code / ".worktrees" / "fast"
self.wf("start", "t-three", "--worktree", str(wt), "--branch", "fast/t-three", project=False, env=IDENT)
(wt / "code.txt").write_text("x\n")
code, out, err = self.wf("finish", "t-three", "-m", "ok", "--commit", "impl", "code.txt",
project=False, cwd=wt, env=IDENT)
self.assertEqual(code, 0, out + err)
self.wf("note", "t-one", "later edit", project=False, env=IDENT) # books dirty in the private tree
self.assertTrue(subprocess.run(["git", "status", "--porcelain", "TASKS.md"], cwd=self.root,
capture_output=True, text=True).stdout.strip())
def test_post_done_commits_leftover_books_with_code_sha(self):
self._finish_then_dirty_books()
csha = self.sha(self.code)
code, out, err = self.wf("orch", "post", "t-three", "fast", "--result", "done", "--commit", csha,
"--no-pick", project=False, env=IDENT)
self.assertEqual(code, 0, out + err)
self.assertEqual(self.log(self.root).splitlines()[:2], [f"t-three done (code {csha})"] * 2)
self.assertFalse(subprocess.run(["git", "status", "--porcelain", "TASKS.md"], cwd=self.root,
capture_output=True, text=True).stdout.strip())
self.assertIn(f" t-three done {csha} ", (self.root / "out" / "wf-orch.log").read_text())
def test_post_stale_private_sha_replaced_by_code_sha(self):
self._finish_then_dirty_books()
stale = self.sha(self.root) # worker reported the private repo's HEAD
code, out, err = self.wf("orch", "post", "t-three", "fast", "--result", "done", "--commit", stale,
"--no-pick", project=False, env=IDENT)
self.assertEqual(code, 0, out + err)
csha = self.sha(self.code)
self.assertIn(f"commit {stale} not in code repo", out)
self.assertIn(f" t-three done {csha} ", (self.root / "out" / "wf-orch.log").read_text())
self.assertEqual(self.log(self.root).splitlines()[:2], [f"t-three done (code {csha})"] * 2)
self.assertFalse(subprocess.run(["git", "status", "--porcelain", "TASKS.md"], cwd=self.root,
capture_output=True, text=True).stdout.strip())
class CheckSplitTest(SplitCli):
def test_cloud_error_push_and_ignore_warnings(self):
(self.root / "workflow.toml").write_text(SplitCli.toml + "cloud = true\n")
(self.code / ".gitignore").write_text(".worktrees/\n")
git(self.code, "commit", "-qam", "no wf-home ignore")
code, out, err = self.wf("check")
self.assertEqual(code, 1)
self.assertIn("cloud lane needs a single repo", out)
self.assertIn("split project without push", out)
self.assertIn(".wf-home not git-ignored", out)
def test_clean_split_project(self):
(self.root / "workflow.toml").write_text(SplitCli.toml + 'push = ["true"]\n')
code, out, err = self.wf("check")
self.assertEqual(code, 0, out)
self.assertNotIn("split", out)
|
