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
|
import json
import os
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_check import Base, CLEAN
SESS = """\
# Tasks — demo
## Awaiting your decision
## Pending
- **t-solo** [P1] (1h): Solo.
- Sessions: solo — lib bump
Model: sonnet
- **t-own** [P1] (1h, interactive): Own.
- **t-owner** [P2] (1h): Owner.
Sessions: owner
- **t-par** [P2] (<1h): Par.
Sessions: parallel
- **t-plain** [P3] (<1h): Plain.
## Needs human
## Deferred
"""
class SessionsLineTest(unittest.TestCase):
def setUp(self):
self.doc = T.parse(SESS)
def test_values(self):
self.assertEqual([(i.id, i.sessions) for i in self.doc.section("pending").items],
[("t-solo", "solo"), ("t-own", "owner"), ("t-owner", "owner"),
("t-par", "parallel"), ("t-plain", "parallel")])
def test_set_replaces_in_place_and_keeps_tail_order(self):
T.set_fields(self.doc, "t-solo", sessions="owner")
self.assertEqual(self.doc.item("t-solo").body, [" Sessions: owner", " Model: sonnet"])
def test_set_adds_before_model(self):
T.set_fields(self.doc, "t-plain", sessions="solo")
self.assertEqual(self.doc.item("t-plain").body, [" Sessions: solo"])
def test_set_empty_removes(self):
T.set_fields(self.doc, "t-owner", sessions="")
self.assertEqual(self.doc.item("t-owner").body, [])
def test_set_drops_interactive_flag(self):
T.set_fields(self.doc, "t-own", sessions="owner")
item = self.doc.item("t-own")
self.assertEqual(item.lines(), ["- **t-own** [P1] (1h): Own.", " Sessions: owner"])
def test_set_bad_value(self):
with self.assertRaises(T.TaskError):
T.set_fields(self.doc, "t-plain", sessions="many")
def test_note_goes_before_sessions_line(self):
T.add_note(self.doc, "t-owner", "x")
self.assertEqual(self.doc.item("t-owner").body, [" - x", " Sessions: owner"])
class SessionsPickTest(unittest.TestCase):
def setUp(self):
self.doc = T.parse(SESS)
def pick(self, lane, model, **kw):
return L.pick(self.doc, set(), L.DEFAULT_LANES, "1h", lane, model, **kw)
def test_alone_solo_picked_owner_skipped(self):
item, skipped = self.pick(None, "sonnet")
self.assertEqual(item.id, "t-solo")
item, skipped = self.pick("slow", "opus", others=1)
self.assertEqual((item.id, [(i.id, why) for i, why in skipped]),
("t-par", [("t-solo", "solo: 1 other live session"),
("t-own", "owner: needs the owner (wf next --owner)"),
("t-owner", "owner: needs the owner (wf next --owner)")]))
def test_owner_present(self):
item, _ = self.pick("slow", "opus", others=1, owner=True)
self.assertEqual(item.id, "t-own")
def test_solo_skipped_with_other_live_sessions(self):
item, skipped = self.pick(None, "sonnet", others=1)
self.assertEqual((item, [(i.id, why) for i, why in skipped]),
(None, [("t-solo", "solo: 1 other live session")]))
_, skipped = self.pick(None, "sonnet", others=2)
self.assertEqual(skipped[0][1], "solo: 2 other live sessions")
def test_solo_running(self):
self.assertEqual(T.solo_running(self.doc, {"t-solo": "sonnet session uds:/a"}),
("t-solo", "sonnet session uds:/a"))
self.assertIsNone(T.solo_running(self.doc, {"t-par": "x"}))
self.assertIsNone(T.solo_running(self.doc, {}))
def test_solo_done_block(self):
sessions = {"opus": {"socket": "/o", "alive": True, "pid": 1},
"haiku": {"socket": "/h", "alive": False, "pid": 2},
"sonnet": {"socket": "/s", "alive": True, "pid": 3}}
self.assertEqual(L.solo_done_block(["t-solo"], sessions, "3"),
["notify opus uds:/o: solo t-solo done, run wf next"])
self.assertEqual(L.solo_done_block([], sessions, "3"), [])
class SessionsCheckTest(Base):
def test_bad_word_error_interactive_warning(self):
text = CLEAN.replace("- **t-one** [P1] (1h)", "- **t-one** [P1] (1h, interactive)").replace(
"## Needs human", "- **t-x** [P3] (1h): X.\n Sessions: lots\n\n## Needs human")
self.tasks(text)
errors, warnings = self.run_check()
self.assertTrue(any(e.endswith("t-x: Sessions 'lots' (want parallel, solo, owner)") for e in errors), errors)
self.assertTrue(any(w.endswith("t-one: 'interactive' flag: write 'Sessions: owner' "
"(wf set t-one --sessions owner)") for w in warnings), warnings)
class SessionsCliTest(Cli):
tasks_text = SESS
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 register(self, lane, model, sock, pid):
d = self.root / ".wf" / "sessions"
d.mkdir(parents=True, exist_ok=True)
(d / f"{lane}.json").write_text(json.dumps({"lane": lane, "model": model, "socket": str(sock), "pid": pid,
"session": "s", "at": "2026-10-04T10:00"}))
def test_list_marks(self):
self.assertEqual(self.ok("list"),
"t-solo P1 1h - sonnet slow [solo] Solo\n"
"t-own P1 1h - opus slow [owner] Own\n"
"t-owner P2 1h - opus slow [owner] Owner\n"
"t-par P2 <1h - opus fast Par\n"
"t-plain P3 <1h - opus fast Plain\n"
"pending 5 · human 0 · awaiting 0 · deferred 0\n")
def test_add_and_set(self):
self.ok("add", "Four.", "-p", "3", "-e", "1h", "--sessions", "solo", "--model", "haiku")
self.assertEqual(self.item("t-four"), "- **t-four** [P3] (1h): Four.\n Sessions: solo\n Model: haiku\n")
self.ok("set", "t-four", "--sessions", "")
self.assertEqual(self.item("t-four"), "- **t-four** [P3] (1h): Four.\n Model: haiku\n")
self.fails("set", "t-four", "--sessions", "many", code=2)
self.fails("add", "Five.", "-p", "3", "-e", "1h", "--sessions", "many", code=2)
def test_interactive_option_is_old_spelling(self):
code, out, err = self.wf("add", "Four.", "-p", "3", "-e", "1h", "--interactive")
self.assertEqual((code, err), (0, "wf: --interactive is now --sessions owner\n"))
self.assertEqual(self.item("t-four"), "- **t-four** [P3] (1h): Four.\n Sessions: owner\n")
code, out, err = self.wf("set", "t-plain", "--interactive", "yes")
self.assertEqual((code, err), (0, "wf: --interactive is now --sessions owner\n"))
self.assertEqual(self.item("t-plain"), "- **t-plain** [P3] (<1h): Plain.\n Sessions: owner\n")
def test_next_owner(self):
self.ok("done", "t-solo", "-m", "ok")
out = self.ok("next", "--lane", "slow", "--as", "opus", env={"CLAUDE_PID": "", "CLAUDE_CODE_MESSAGING_SOCKET": ""})
self.assertIn("- t-own: owner: needs the owner (wf next --owner)\n", out)
self.assertIn("===== Next task =====\n- **t-par**", out) # slow empty → fallback fast
self.assertTrue(self.ok("next", "--lane", "slow", "--as", "opus", "--owner", "--brief").startswith("- **t-own**"))
def test_next_solo_with_other_live_session(self):
sock = self.root / "o.sock"
sock.write_text("")
self.register("fast", "opus", sock, os.getppid())
err = self.fails("next", "--as", "sonnet", env=self.env("s", os.getpid()))
self.assertEqual(err, "wf: nothing pickable for all lanes (sonnet) in Pending\n")
out = self.wf("next", "--as", "sonnet", env=self.env("s", os.getpid()))[1]
self.assertIn("- t-solo: solo: 1 other live session\n", out)
def test_solo_in_progress_blocks_others_and_done_notifies(self):
s, o = self.env("s", os.getpid()), self.env("o", os.getppid())
self.ok("next", "--lane", "slow", "--as", "sonnet", "--brief", env=s)
self.ok("status", "t-solo", "progress", "x", env=s)
code, out, err = self.wf("next", "--lane", "fast", "--as", "opus", env=o)
self.assertEqual((code, err), (1, f"wf: solo t-solo in progress by sonnet session uds:{self.root / 's.sock'}: "
"wait (its wf done notifies you)\n"))
self.assertNotIn("Next task", out)
out = self.ok("done", "t-solo", "-m", "ok", env=s)
self.assertIn(f"notify fast uds:{self.root / 'o.sock'}: solo t-solo done, run wf next\n", out)
self.assertTrue(self.ok("next", "--lane", "fast", "--as", "opus", "--brief", env=o).startswith("- **t-par**"))
if __name__ == "__main__":
unittest.main()
|
