-- 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 },
}