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
|
import json
import os
import subprocess
import unittest
from test_cli import TOML, Cli
from test_claims import git
from test_merge import IDENT
TASKS = """\
# Tasks — demo
## Awaiting your decision
## Pending
- **t-one** [P1] (<1h): One.
- Done: one works
- Model: sonnet
- **t-two** [P2] (<1h): Two.
- Done: two works
- **t-raw** [P2] (<1h): Raw, no Done line.
- **t-big** [P2] (1h): Big.
- Done: big works
## Needs human
## Deferred
"""
class OrchTest(Cli):
tasks_text = TASKS
toml = TOML.replace('verify = ["make test"]\n', "")
def setUp(self):
super().setUp()
git(self.root, "init", "-q", "-b", "master")
(self.root / ".gitignore").write_text(".worktrees/\n.wf/\nout/\n")
git(self.root, "add", "-A")
git(self.root, "commit", "-qm", "init")
def orch(self, *args, code=0):
got, out, err = self.wf("orch", *args, env=IDENT)
self.assertEqual(got, code, out + err)
self.assertEqual(err, "")
return out
def run_wf(self, cwd, *args):
got, out, err = self.wf(*args, project=False, cwd=cwd, env=IDENT)
self.assertEqual(got, 0, out + err)
return out
def record(self, id):
return json.loads((self.root / ".wf" / "orch" / f"{id}.json").read_text())
def worker_done(self, id="t-one", lane="fast", merge=True):
"""What a worker does: wf start, code, wf done/finish (finish merges, done alone leaves the branch)."""
wt = self.root / ".worktrees" / lane
self.run_wf(self.root, "start", id, "--worktree", str(wt), "--branch", f"{lane}/{id}")
(wt / "code.txt").write_text("x\n")
if merge:
self.run_wf(wt, "finish", id, "-m", "ok", "--commit", "impl", "code.txt", "--no-push")
else:
git(wt, "add", "code.txt")
git(wt, "commit", "-qm", "impl")
self.run_wf(wt, "done", id, "-m", "ok")
return wt
def log(self):
return (self.root / "out" / "wf-orch.log").read_text()
def test_pick_claims_and_prints_prompt(self):
out = self.orch("pick", "fast")
wt = self.root / ".worktrees" / "fast"
self.assertEqual(out, "pick: t-one (lane fast, model sonnet, effort <1h) · claimed (in progress: worker)\n"
"agent: subagent_type wf-worker, model sonnet, no isolation; prompt:\n"
"Task: t-one Lane: fast Model: sonnet\n"
f"Main tree: {self.root} Worktree: {wt} Branch: fast/t-one\n"
"Final message: the 4 report lines only.\n")
self.assertIn("- **t-one** [P1] (<1h) (in progress: worker): One.", self.tasks())
self.assertEqual(self.record("t-one")["worktree"], str(wt))
def test_second_pick_skips_in_progress_and_busy_worktree(self):
self.orch("pick", "fast")
out = self.orch("pick", "fast")
self.assertIn("pick: t-two (lane fast, model opus", out)
self.assertIn(f"Worktree: {self.root / '.worktrees' / 'fast-2'} Branch: fast/t-two\n", out)
def test_worktree_on_other_branch_not_reused(self):
git(self.root, "worktree", "add", "-q", str(self.root / ".worktrees" / "fast"), "-b", "fast/t-old")
self.assertIn(".worktrees/fast-2 Branch: fast/t-one", self.orch("pick", "fast"))
def test_clean_detached_worktree_reused(self):
git(self.root, "worktree", "add", "-q", "--detach", str(self.root / ".worktrees" / "fast"), "master")
self.assertIn(".worktrees/fast Branch: fast/t-one", self.orch("pick", "fast"))
def test_worktree_with_live_session_not_reused(self):
wt = self.root / ".worktrees" / "fast"
git(self.root, "worktree", "add", "-q", "--detach", str(wt), "master")
folder = self.root.parent / "no-claude" / "sessions"
folder.mkdir(parents=True)
(folder / "1.json").write_text(json.dumps({"pid": os.getpid(), "cwd": str(wt / "docs")}))
self.assertIn(".worktrees/fast-2 Branch: fast/t-one", self.orch("pick", "fast"))
def test_stop_file_spawns_nothing(self):
(self.root / "out").mkdir()
(self.root / "out" / "wf-batch.stop").write_text("")
self.assertEqual(self.orch("pick", "fast"), "stop: out/wf-batch.stop exists: spawn nothing (let running workers finish)\n")
self.assertNotIn("in progress", self.tasks())
def test_none_pickable(self):
self.orch("pick", "fast")
self.orch("pick", "fast")
self.assertTrue(self.orch("pick", "fast").startswith("none: lane fast has no runner-ready task"))
def test_explicit_id_and_recovery_line(self):
out = self.orch("pick", "slow", "--id", "t-big", "--recovery", "crashed")
self.assertIn("Branch: slow/t-big\nRecovery: crashed\nFinal message", out)
def test_post_done_logs_and_picks_next(self):
self.orch("pick", "fast")
self.worker_done()
out = self.orch("post", "t-one", "fast", "--result", "done", "--commit", "abc1234", "--duration", "75")
self.assertTrue(out.startswith("post: t-one done\n\npick: t-two"), out)
self.assertRegex(self.log(), r"^\S+ fast sonnet t-one done abc1234 1m15s\n$")
self.assertFalse((self.root / ".wf" / "orch" / "t-one.json").exists())
def test_post_no_next_claims_nothing(self):
self.orch("pick", "fast")
self.worker_done()
out = self.orch("post", "t-one", "fast", "--result", "done", "--no-next", "--duration", "5")
self.assertTrue(out.startswith("post: t-one done"), out)
self.assertNotIn("pick:", out)
self.assertEqual([f.name for f in (self.root / ".wf" / "orch").glob("*.json")], [])
def test_post_done_without_record_takes_model_from_history(self):
self.orch("pick", "fast")
self.worker_done()
(self.root / ".wf" / "orch" / "t-one.json").unlink()
self.orch("post", "t-one", "fast", "--result", "done", "--no-pick", "--duration", "5")
self.assertRegex(self.log(), r"^\S+ fast sonnet t-one done ")
def test_post_done_merges_unmerged_branch(self):
self.orch("pick", "fast")
wt = self.worker_done(merge=False)
out = self.orch("post", "t-one", "fast", "--result", "done", "--no-pick", "--no-push")
self.assertEqual(out, "post: t-one done\n merged worktree HEAD: merged fast/t-one into master\n")
self.assertEqual(subprocess.run(["git", "-C", str(self.root), "show", "master:code.txt"],
capture_output=True, text=True).stdout, "x\n")
self.assertEqual(subprocess.run(["git", "-C", str(wt), "branch", "--show-current"],
capture_output=True, text=True).stdout, "")
def test_post_done_without_archive_is_red(self):
self.orch("pick", "fast")
out = self.orch("post", "t-one", "fast", "--result", "done")
self.assertEqual(out, "post: t-one post-check-red\n no archive line for t-one\n"
"stop lane fast: post-check-red → tell the owner\n")
self.assertIn(" post-check-red ", self.log())
def backdate_pick(self, id="t-one", secs=125):
f = self.root / ".wf" / "orch" / f"{id}.json"
rec = json.loads(f.read_text())
rec["at"] -= secs
f.write_text(json.dumps(rec) + "\n")
def test_post_zero_duration_falls_back_to_pick_time(self):
self.orch("pick", "fast")
self.backdate_pick()
self.worker_done()
self.orch("post", "t-one", "fast", "--result", "done", "--no-pick", "--duration", "0")
self.assertRegex(self.log(), r" t-one done \S+ 2m0[5-9]s\n$")
def test_post_red_keeps_pick_time_for_the_repost(self):
self.orch("pick", "fast")
self.backdate_pick()
self.orch("post", "t-one", "fast", "--result", "done")
self.assertTrue((self.root / ".wf" / "orch" / "t-one.json").exists())
self.worker_done()
self.orch("post", "t-one", "fast", "--result", "done", "--no-pick")
self.assertRegex(self.log().splitlines()[-1], r" t-one done \S+ 2m0[5-9]s$")
self.assertFalse((self.root / ".wf" / "orch" / "t-one.json").exists())
def test_post_handback_commits_leftovers_and_stops(self):
self.orch("pick", "fast")
out = self.orch("post", "t-one", "fast", "--result", "handback unclear spec")
self.assertEqual(out, "post: t-one handback\n committed leftover TASKS.md tasks/archive.md\n"
"stop lane fast: handback → tell the owner\n")
self.assertEqual(subprocess.run(["git", "-C", str(self.root), "log", "-1", "--format=%s"],
capture_output=True, text=True).stdout, "t-one handback (orchestrator)\n")
def test_post_done_on_slice_job_counts_as_sliced(self):
self.orch("pick", "slow", "--id", "t-big")
self.run_wf(self.root, "add", "--parent", "t-big", "-e", "<1h", "--done", "x", "Slice one")
out = self.orch("post", "t-big", "slow", "--result", "done", "--no-next", "--duration", "5")
self.assertTrue(out.startswith("post: t-big sliced"), out)
self.assertNotIn("post-check-red", self.log())
self.assertIn(" t-big sliced ", self.log())
def test_post_model_raised_continues(self):
self.orch("pick", "fast")
self.wf("set", "t-one", "--model", "opus")
self.wf("status", "t-one", "clear")
out = self.orch("post", "t-one", "fast", "--result", "handback beyond sonnet")
self.assertTrue(out.startswith("post: t-one model-raised\n"), out)
self.assertIn("pick: t-one (lane fast, model opus", out)
def test_post_no_report_suggests_recovery(self):
self.orch("pick", "fast")
self.assertIn('wf orch pick fast --id t-one --recovery "<why>"', self.orch("post", "t-one", "fast"))
def test_post_agent_without_transcript_still_logs(self):
self.orch("pick", "fast")
out = self.orch("post", "t-one", "fast", "--result", "wip", "--agent", "abc")
self.assertIn(" cost: not logged (no transcript for agent abc)\n", out)
def test_refused_in_linked_worktree(self):
git(self.root, "worktree", "add", "-q", "--detach", str(self.root / ".worktrees" / "x"), "master")
got, _, err = self.wf("orch", "pick", "fast", project=False, cwd=self.root / ".worktrees" / "x")
self.assertEqual((got, err), (1, "wf: orch runs in the main tree (the orchestrator's)\n"))
if __name__ == "__main__":
unittest.main()
|
