path: root/spec/test_jobs.lua

-- 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. Real wake pipes are armed (a job that cannot get one is refused), but
-- nothing writes to them and no loop runs: a plain script cannot park, so
-- awaiting drains its jobs in place and `settle = N` means "settles on the Nth
-- poll", which is how the ordering cases stay deterministic without a 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({
            id = spec.id,
            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 five 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", "g" }) do
                handles[#handles + 1] = assert(start(name))
            end
            assert(#built == 5, "the gate holds at five running, saw " .. #built)
            assert(handles[6]:result() == nil, "a queued child has not settled")

            local results = jobs.await(handles, "all")
            assert(#built == 7, "the queue drains as slots free up, saw " .. #built)
            assert(#results == 7, "every child reports")
            assert(results[6].text == "f" and results[7].text == "g", "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", "f" }) do
                handles[#handles + 1] = assert(start(name))
            end
            handles[6]:cancel()

            local result = handles[6]: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 == 5, "the queued child was never built")
            assert(not contains(built, "f"), "the queued child was never built")

            jobs.await({ handles[1], handles[2], handles[3], handles[4], handles[5] }, "all")
            assert(#built == 5, "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", "f" }) do
                handles[#handles + 1] = assert(start(name, { settle = 5 }))
            end
            jobs.cancel_all()

            for _, name in ipairs({ "a", "b", "c", "d", "e" }) do
                assert(made[name]._cancel_requested, "running child " .. name .. " was not cancelled")
            end
            assert(#built == 5, "cancel_all must not start the queued child")
            assert(handles[6]: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 },

    -- Under luv the wake pipe is not optional: without it nothing would ever
    -- return the owner thread to the loop the child's own tool batches need.
    { "a child whose wake pipe cannot be armed is refused instead of started", function()
        local ok, uv = pcall(require, "luv")
        if not ok or type(uv) ~= "table" then
            return "skip", "luv is not installed"
        end
        with_jobs(function()
            local start, built = starter()
            uv.pipe = function()
                return nil, "EMFILE"
            end
            local handle, err = start("alpha")
            uv.pipe = nil -- back to the real luv through the harness metatable
            assert(handle == nil, "a child with no wake pipe must not start")
            assert(tostring(err):find("the subagent wake pipe could not be armed", 1, true), tostring(err))
            assert(#built == 0, "and its turn is never built")
        end)
    end },

    -- close_all cannot join a pump that is still up, so that child is still
    -- running against the same bound and still owns its session file.
    { "a closing child holds its slot until it settles", function()
        with_jobs(function()
            local start, built = starter()
            local closing = assert(start("alpha", { settle = 99 }))
            jobs.close_all()

            local queued = assert(start("beta"))
            assert(#built == 1, "the closing child still holds the only slot, saw " .. #built)

            jobs.await({ closing }, "all")
            assert(#built == 2, "the slot comes back when the closing child settles")

            jobs.await({ queued }, "all")
            local last = assert(start("gamma"))
            assert(#built == 3, "and the bound is not leaked once everything has settled")
            jobs.await({ last }, "all")
        end, 1)
    end },

    { "a closing child still reports its id as active", function()
        with_jobs(function()
            local start = starter()
            local handle = assert(start("alpha", { id = "child-1", settle = 99 }))
            assert(jobs.active("child-1"), "a running child is active")

            jobs.close_all()
            assert(jobs.active("child-1"), "a cancelled child still owns its session file")

            jobs.await({ handle }, "all")
            assert(not jobs.active("child-1"), "the settle releases the id")
        end)
    end },
}