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
|
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_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_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())
|
