diff options
Diffstat (limited to 'subagents/jobs.lua')
| -rw-r--r-- | subagents/jobs.lua | 69 |
1 files changed, 45 insertions, 24 deletions
diff --git a/subagents/jobs.lua b/subagents/jobs.lua index bab8f30..3f9a8af 100644 --- a/subagents/jobs.lua +++ b/subagents/jobs.lua @@ -14,10 +14,15 @@ -- Waiting parks the CALLING coroutine — the tool handler's — and a poll -- callback resumes exactly that coroutine. subagents/workflow.lua's rule that -- a workflow callback must never run on a nested coroutine follows from this: --- the parked thread is the one the resume goes to. Where there is no coroutine --- to park (a plain script) or a job with no wake pipe, the identical drain --- runs in a loop instead — same gate, same settle bookkeeping, only the wait --- differs. +-- the parked thread is the one the resume goes to. +-- +-- The wake pipe is therefore mandatory wherever luv is: a started job whose +-- pipe or poll could not be armed is a start failure, not a degraded job. +-- The drain-and-sleep fallback below only ever serves a luv-less host or a +-- caller with no coroutine to park (a plain script), because it cannot serve a +-- child that dispatches tools: those tool batches are posted to this very +-- thread, so a loop that never returns to uv would wait for work only it can +-- do. A visible "could not be armed" beats an unrecoverable hang. -- -- A settled result is read and cached the moment it appears, because -- `job:close()` frees it. Jobs are otherwise left open until close_all() ends @@ -127,7 +132,9 @@ local function settle(handle, raw) end end handle.settled = result - if handle.state == "running" then + -- A closing job still holds its slot: close_all could not join its pump, so + -- the child is still up. It gives the slot back here, once it really exits. + if handle.state == "running" or handle.state == "closing" then running = running - 1 end handle.state = "settled" @@ -142,6 +149,21 @@ end local function launch(handle) handle.fds = open_pipe() + + -- Arm the wake before the child exists, so a host that cannot give us one + -- never leaves a pump running with nothing to drain it. + if ok_uv then + if handle.fds == nil then + return false, "the subagent wake pipe could not be armed" + end + local armed, poll = pcall(uv.new_poll, handle.fds.read) + if not armed or not poll then + close_pipe(handle) + return false, "the subagent wake pipe could not be armed" + end + handle.poll = poll + end + local ok, job, err = pcall(handle.spec.build, handle.fds and handle.fds.write or nil) if not ok then close_pipe(handle) @@ -156,15 +178,11 @@ local function launch(handle) handle.state = "running" running = running + 1 - if handle.fds and ok_uv then - local armed, poll = pcall(uv.new_poll, handle.fds.read) - if armed and poll then - handle.poll = poll - poll:start("r", function() - drain(handle) - wake() - end) - end + if handle.poll then + handle.poll:start("r", function() + drain(handle) + wake() + end) end return true end @@ -265,7 +283,7 @@ end -- -- startspec = { -- build = function(wake_fd) -> job | nil, err -- calls agent:run_async --- label = string?, id = string?, one_shot = boolean? +-- id = string? -- on_event = function(event)? -- one call per drained run_async event -- shape = function(raw) -> result? -- maps the settled run_async result -- onto the caller's result table @@ -279,9 +297,7 @@ function M.start(spec) local handle = setmetatable({ spec = spec, - label = spec.label, id = spec.id, - one_shot = spec.one_shot == true, state = "queued", }, handle_mt) live[#live + 1] = handle @@ -306,15 +322,16 @@ function M.start(spec) return handle end --- True while a child with this id has a turn in flight. A child cancelled by --- close_all is not one: its turn belongs to the turn that ended, and the next --- one must not be refused because that pump has not finished exiting yet. +-- True while a child with this id has a turn in flight, including one that +-- close_all cancelled but whose pump has not exited yet: it still owns that +-- child's session file, and a second writer over the same file loses data. The +-- refusal is transient — the pump's settle clears it on the next drain. function M.active(id) if id == nil then return false end for _, handle in ipairs(live) do - if handle.id == id and handle.settled == nil and handle.state ~= "closing" then + if handle.id == id and handle.settled == nil then return true end end @@ -439,8 +456,10 @@ function M.close_all() end end - -- Whatever is still unsettled stays live so its poll (and any fallback - -- drain) still reaches it; the next close_all sweeps up what settled since. + -- Whatever is still unsettled stays live with its wake poll still armed — + -- start() refuses a job that has none — so the byte its pump writes on the + -- way out still drives the settle that closes it and frees its pipe. The + -- next close_all sweeps up whatever settled since. local closing = {} for _, handle in ipairs(live) do if handle.job and handle.settled == nil then @@ -453,7 +472,9 @@ function M.close_all() end end live, queued, waiters = closing, {}, {} - running = 0 + -- Those children are still running against the same bound; each releases + -- its slot in settle() when its pump finally exits. + running = #closing end return M |