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
|
-- subagents/jobs.lua: the concurrency gate, the queue, cancellation, and the
-- await contract every caller (run.lua, workflow.lua) is written against.
--
-- These cases drive the job machinery directly with hand-made fake jobs rather
-- than through a child, so a failure here points at the gate and not at spawn
-- policy. The fake jobs have no wake pipe (the harness has none), so awaiting
-- drains them in place; `settle = N` means "settles on the Nth poll", which is
-- how the ordering cases stay deterministic without an event loop.
local fake = require("spec.fake_ext")
local jobs = require("subagents.jobs")
-- The host may report one result or an array of them; both are normalized here
-- exactly as workflow.lua normalizes them.
local function as_array(value)
if type(value) ~= "table" then
return {}
end
if value.status ~= nil then
return { value }
end
return value
end
-- Every case leaves the module clean: close_all drops whatever is still live.
local function with_jobs(fn, max_concurrent)
local original = jobs.MAX_CONCURRENT
if max_concurrent then
jobs.MAX_CONCURRENT = max_concurrent
end
local ok, err = pcall(fn)
pcall(jobs.close_all)
jobs.MAX_CONCURRENT = original
if not ok then
error(err, 0)
end
end
-- A starter that records which builds actually ran, and the jobs they made.
local function starter()
local built, made = {}, {}
local function start(name, spec)
spec = spec or {}
return jobs.start({
label = name,
id = spec.id,
one_shot = spec.one_shot,
on_event = spec.on_event,
build = function()
built[#built + 1] = name
if spec.build_error then
return nil, spec.build_error
end
local job = fake.job({
settle = spec.settle,
events = spec.events,
result = spec.result or { status = "completed", text = name },
})
made[name] = job
return job
end,
})
end
return start, built, made
end
local function contains(list, value)
for _, entry in ipairs(list) do
if entry == value then
return true
end
end
return false
end
return {
{ "a started job settles through await", function()
with_jobs(function()
local start = starter()
local handle = assert(start("alpha"))
assert(handle:result() == nil, "a job that has not settled has no result")
local results = jobs.await({ handle }, "all")
assert(#results == 1, "one handle, one result")
assert(results[1].status == "completed", tostring(results[1].status))
assert(results[1].text == "alpha", tostring(results[1].text))
assert(handle:result().text == "alpha", "the settled result stays on the handle")
end)
end },
{ "await all returns results in input order, not settle order", function()
with_jobs(function()
local start = starter()
local handles = {
assert(start("alpha", { settle = 3 })),
assert(start("beta", { settle = 1 })),
assert(start("gamma", { settle = 2 })),
}
local results = jobs.await(handles, "all")
assert(#results == 3, "expected three results")
assert(results[1].text == "alpha", tostring(results[1].text))
assert(results[2].text == "beta", tostring(results[2].text))
assert(results[3].text == "gamma", tostring(results[3].text))
end)
end },
{ "await first returns the earliest settler and the remaining handles", function()
with_jobs(function()
local start = starter()
local handles = {
assert(start("alpha", { settle = 3 })),
assert(start("beta", { settle = 1 })),
assert(start("gamma", { settle = 2 })),
}
local seen = {}
while #handles > 0 do
local results, remaining = jobs.await(handles, "first")
results = as_array(results)
assert(#results >= 1, "an await that returns must settle something")
for _, result in ipairs(results) do
seen[#seen + 1] = result.text
end
assert(type(remaining) == "table", "first mode reports what is still running")
assert(#remaining < #handles, "every await makes progress")
handles = remaining
end
assert(seen[1] == "beta", "the lowest settle key comes back first: " .. table.concat(seen, ","))
assert(contains(seen, "gamma") and contains(seen, "alpha"), table.concat(seen, ","))
assert(#seen == 3, table.concat(seen, ","))
end)
end },
{ "the gate runs four at a time and queues the rest", function()
with_jobs(function()
local start, built = starter()
local handles = {}
for _, name in ipairs({ "a", "b", "c", "d", "e", "f" }) do
handles[#handles + 1] = assert(start(name))
end
assert(#built == 4, "the gate holds at four running, saw " .. #built)
assert(handles[5]:result() == nil, "a queued child has not settled")
local results = jobs.await(handles, "all")
assert(#built == 6, "the queue drains as slots free up, saw " .. #built)
assert(#results == 6, "every child reports")
assert(results[5].text == "e" and results[6].text == "f", "queued children keep their place")
end)
end },
{ "a child cancelled while queued never starts", function()
with_jobs(function()
local start, built = starter()
local handles = {}
for _, name in ipairs({ "a", "b", "c", "d", "e" }) do
handles[#handles + 1] = assert(start(name))
end
handles[5]:cancel()
local result = handles[5]:result()
assert(type(result) == "table", "a cancelled queued child settles immediately")
assert(result.status == "cancelled", tostring(result.status))
assert(result.error == "cancelled before the child started", tostring(result.error))
assert(#built == 4, "the queued child was never built")
assert(not contains(built, "e"), "the queued child was never built")
jobs.await({ handles[1], handles[2], handles[3], handles[4] }, "all")
assert(#built == 4, "a cancelled child does not start when a slot frees up")
end)
end },
{ "cancelling a running child asks its job to cancel", function()
with_jobs(function()
local start, _, made = starter()
local handle = assert(start("alpha", { settle = 5 }))
handle:cancel()
assert(made.alpha._cancel_requested, "the job was asked to cancel")
local results = jobs.await({ handle }, "all")
assert(results[1].status == "cancelled", tostring(results[1].status))
end)
end },
{ "events reach on_event before the job settles", function()
with_jobs(function()
local seen = {}
local start = starter()
local handle = assert(start("alpha", {
settle = 2,
events = {
{ type = "content_delta", text = "half " },
{ type = "content_delta", text = "a thought" },
},
on_event = function(event)
seen[#seen + 1] = event.text
end,
}))
jobs.await({ handle }, "all")
assert(table.concat(seen) == "half a thought", "events arrive in order: " .. table.concat(seen, "|"))
end)
end },
{ "cancel_all stops the running children and drops the queued ones", function()
with_jobs(function()
local start, built, made = starter()
local handles = {}
for _, name in ipairs({ "a", "b", "c", "d", "e" }) do
handles[#handles + 1] = assert(start(name, { settle = 5 }))
end
jobs.cancel_all()
for _, name in ipairs({ "a", "b", "c", "d" }) do
assert(made[name]._cancel_requested, "running child " .. name .. " was not cancelled")
end
assert(#built == 4, "cancel_all must not start the queued child")
assert(handles[5]:result().status == "cancelled", "the queued child settles cancelled")
local results = jobs.await(handles, "all")
for index, result in ipairs(results) do
assert(result.status == "cancelled", "child " .. index .. " is " .. tostring(result.status))
end
end)
end },
{ "close_all closes every job it started", function()
with_jobs(function()
local start, _, made = starter()
local handle = assert(start("alpha"))
jobs.await({ handle }, "all")
assert(not made.alpha._closed, "awaiting does not close a job")
jobs.close_all()
assert(made.alpha._closed, "turn end closes the job")
end)
end },
-- close() joins the pump, and a pump parked in a tool batch needs the owner
-- thread — the thread close_all itself runs on. So an unsettled job is never
-- closed here; it is cancelled, and the drain that sees it settle closes it.
{ "close_all leaves an unsettled job to close on its own settle", function()
with_jobs(function()
local start, _, made = starter()
local handle = assert(start("alpha", { settle = 99 }))
jobs.close_all()
assert(made.alpha._cancel_requested, "close_all asks a running child to stop")
assert(not made.alpha._closed, "close_all must not join a pump that is still up")
assert(handle:result() == nil, "close_all does not settle the child itself")
local results = jobs.await({ handle }, "all")
assert(results[1].status == "cancelled", tostring(results[1].status))
assert(made.alpha._closed, "the job closes as soon as it settles")
assert(handle:result().status == "cancelled", "the result is cached before the close")
end)
end },
{ "a build failure is a nil return, never an exception", function()
with_jobs(function()
local start = starter()
local handle, err = start("alpha", { build_error = "the host refused" })
assert(handle == nil, "a failed build starts nothing")
assert(tostring(err):find("the host refused", 1, true), tostring(err))
end)
end },
}
|
