-- Start child agent jobs, bound how many run at once, and drain their events.
--
-- One module owns the session-wide concurrency bound and the event pump, so
-- the policy that decides *what* to start (subagents/spawn.lua) never touches
-- luv, and the callers that wait for a result (subagents/run.lua, the workflow
-- API) never touch a job.
--
-- Threading and ownership: everything here runs on panto's Lua owner thread.
-- A job's pump runs on its own thread inside the binding; it buffers events on
-- the job and writes one byte to the wake pipe handed to it here. That byte is
-- an edge trigger, never a count, so every wake drains the pipe, then drains
-- `job:next_event()` to exhaustion, then checks `job:result()`.
--
-- Waiting parks the CALLING coroutine and a poll callback resumes that exact
-- coroutine. Foreground tools use their handler coroutine; background
-- workflows provide a dedicated coroutine anchored in their registry record.
--
-- 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
-- the turn, so a result can still be read after its coroutine resumed.
--
-- A settled consumer may close a job immediately after caching its result;
-- otherwise close_all() handles it. `job:close()` also JOINS the pump, and a pump parked in a tool batch is
-- waiting on the owner thread to come back to the uv loop — the very thread
-- every entry point here runs on. Closing an unsettled job would therefore
-- deadlock, so nothing does: only settle() closes a job, once its pump has
-- exited. close_all() asks what is still running to cancel, marks it
-- close-on-settle, and lets the wake byte that carries the settle drive the
-- close.

local ok_uv, uv = pcall(require, "luv")

local READ_CHUNK = 4096

local M = {}

-- The activation-time default is five; init.lua may replace it from the
-- layered `[subagents] max_concurrent` setting before any child can start.
M.MAX_CONCURRENT = 5

local handle_mt = {}
handle_mt.__index = handle_mt
handle_mt.__name = "subagents.job"

-- Every handle that has not been closed, in start order; the queue is the
-- subset waiting for a slot; `running` counts the started-but-unsettled ones.
local live = {}
local queued = {}
local waiters = {}
local running = 0
local pumping = false
local cancelling = false

-- ---------------------------------------------------------------------------
-- Wake pipes
-- ---------------------------------------------------------------------------

local function open_pipe()
    if not ok_uv or type(uv.pipe) ~= "function" then
        return nil
    end
    local ok, pair = pcall(uv.pipe, { nonblock = true }, { nonblock = true })
    if not ok or type(pair) ~= "table" then
        return nil
    end
    return pair
end

-- Close the poll before the fds: the pump has already exited by the time a job
-- settles, so nothing can be mid-write when the read end goes away.
local function close_pipe(handle)
    if handle.poll then
        pcall(handle.poll.stop, handle.poll)
        pcall(handle.poll.close, handle.poll)
        handle.poll = nil
    end
    if handle.fds then
        pcall(uv.fs_close, handle.fds.read)
        pcall(uv.fs_close, handle.fds.write)
        handle.fds = nil
    end
end

-- Free the binding job. Safe only after the result is cached (close() frees it)
-- and the pump has exited (close() joins it), which together mean: after settle.
local function close_job(handle)
    if handle.job then
        pcall(handle.job.close, handle.job)
        handle.job = nil
    end
    handle.state = "closed"
end

local function drain_pipe(handle)
    if not handle.fds then
        return
    end
    while true do
        local data = uv.fs_read(handle.fds.read, READ_CHUNK, -1)
        if type(data) ~= "string" or #data < READ_CHUNK then
            return
        end
    end
end

-- ---------------------------------------------------------------------------
-- Gate, drain, settle
-- ---------------------------------------------------------------------------

local drain
local pump_queue
local wake

local function settle(handle, raw)
    if handle.settled ~= nil then
        return
    end
    local result = raw
    if handle.spec.shape then
        local ok, shaped = pcall(handle.spec.shape, raw)
        if ok and type(shaped) == "table" then
            result = shaped
        elseif not ok then
            result = { status = "failed", error = tostring(shaped), resumable = false }
        end
    end
    handle.settled = result
    -- 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"
    close_pipe(handle)
    -- The turn ended while this one was still running: close_all() could not
    -- join the pump then, but the pump is gone now, so the join is free.
    if handle.close_on_settle then
        close_job(handle)
    end
    pump_queue()
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)
        return false, tostring(job)
    end
    if not job then
        close_pipe(handle)
        return false, err and tostring(err) or "the child could not be started"
    end

    handle.job = job
    handle.state = "running"
    running = running + 1

    if handle.poll then
        handle.poll:start("r", function()
            drain(handle)
            wake()
        end)
    end
    return true
end

-- Start queued jobs while the gate has room. Reentrant: a job that settles the
-- instant it starts calls back in here, and the outer loop keeps going.
function pump_queue()
    if pumping or cancelling then
        return
    end
    pumping = true
    while running < M.MAX_CONCURRENT do
        local handle = table.remove(queued, 1)
        if handle == nil then
            break
        end
        if handle.state == "queued" then
            local ok, err = launch(handle)
            if not ok then
                settle(handle, { status = "failed", error = tostring(err) })
            end
        end
    end
    pumping = false
end

-- Drain one job: the wake pipe, then every buffered event, then the result.
-- Returns true when anything moved, which is what the fallback wait loop uses
-- to decide whether it is spinning.
function drain(handle)
    local job = handle.job
    if job == nil or handle.settled ~= nil then
        return false
    end
    drain_pipe(handle)

    local moved = false
    local on_event = handle.spec.on_event
    local event = job:next_event()
    while event ~= nil do
        moved = true
        if on_event then
            pcall(on_event, event)
        end
        event = job:next_event()
    end

    local raw = job:result()
    if raw ~= nil then
        settle(handle, raw)
        return true
    end
    return moved
end

local function drain_all()
    local moved = false
    local index = 1
    while index <= #live do
        if drain(live[index]) then
            moved = true
        end
        index = index + 1
    end
    return moved
end

-- ---------------------------------------------------------------------------
-- Handles
-- ---------------------------------------------------------------------------

function handle_mt:result()
    return self.settled
end

-- A queued child settles without its build ever running: no user message is
-- appended, so nothing on disk claims a turn that never happened.
function handle_mt:cancel()
    if self.settled ~= nil then
        return
    end
    if self.state == "queued" then
        for index, other in ipairs(queued) do
            if other == self then
                table.remove(queued, index)
                break
            end
        end
        settle(self, { status = "cancelled", error = "cancelled before the child started" })
        return
    end
    if self.job then
        pcall(self.job.request_cancel, self.job)
    end
end

-- start(startspec) -> handle | nil, err
--
-- startspec = {
--   build     = function(wake_fd) -> job | nil, err   -- calls agent:run_async
--   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
-- }
--
-- Over the bound the job is queued and `build` is not called yet.
function M.start(spec)
    if type(spec) ~= "table" or type(spec.build) ~= "function" then
        return nil, "jobs.start needs a build function"
    end

    local handle = setmetatable({
        spec = spec,
        id = spec.id,
        state = "queued",
    }, handle_mt)
    live[#live + 1] = handle

    if running >= M.MAX_CONCURRENT then
        queued[#queued + 1] = handle
        return handle
    end

    local ok, err = launch(handle)
    if not ok then
        for index, other in ipairs(live) do
            if other == handle then
                table.remove(live, index)
                break
            end
        end
        return nil, err
    end
    -- Deliberately not drained here: starting a child must not settle it, and
    -- the first wake byte is already in the pipe by the time anyone waits.
    return handle
end

-- 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 then
            return true
        end
    end
    return false
end

-- ---------------------------------------------------------------------------
-- Awaiting
-- ---------------------------------------------------------------------------

-- "all" is satisfied only when every handle has settled; "first" as soon as one
-- has. Both answer in input order, and "first" hands the rest back by identity
-- so a caller can await them again.
local function collect(handles, mode)
    if mode == "first" then
        local results, remaining = {}, {}
        for _, handle in ipairs(handles) do
            if handle.settled ~= nil then
                results[#results + 1] = handle.settled
            else
                remaining[#remaining + 1] = handle
            end
        end
        if #results == 0 and #remaining > 0 then
            return nil
        end
        return results, remaining
    end

    local results = {}
    for index, handle in ipairs(handles) do
        if handle.settled == nil then
            return nil
        end
        results[index] = handle.settled
    end
    return results, {}
end

-- Resume every parked await whose condition now holds. Never resumes the
-- running coroutine: a coroutine that is executing cannot also be parked.
function wake()
    local self_co = coroutine.running()
    local index = 1
    while index <= #waiters do
        local waiter = waiters[index]
        local results, remaining = collect(waiter.handles, waiter.mode)
        if results ~= nil and waiter.co ~= self_co and coroutine.status(waiter.co) == "suspended" then
            table.remove(waiters, index)
            coroutine.resume(waiter.co, results, remaining)
        else
            index = index + 1
        end
    end
end

-- Parking is only safe when something will wake us: a started job with no wake
-- pipe is drained by asking it directly instead.
local function can_park(handles)
    if not coroutine.isyieldable() then
        return false
    end
    for _, handle in ipairs(handles) do
        if handle.job ~= nil and handle.settled == nil and handle.poll == nil then
            return false
        end
    end
    return true
end

-- await(handles, mode) -> results, remaining
function M.await(handles, mode)
    mode = mode or "all"
    if type(handles) ~= "table" then
        error("jobs.await expects an array of job handles", 2)
    end
    if mode ~= "all" and mode ~= "first" then
        error("jobs.await mode must be \"all\" or \"first\"", 2)
    end

    while true do
        local moved = drain_all()
        local results, remaining = collect(handles, mode)
        if results ~= nil then
            return results, remaining
        end
        wake()

        if can_park(handles) then
            waiters[#waiters + 1] = { co = coroutine.running(), handles = handles, mode = mode }
            local woken, rest = coroutine.yield()
            if woken ~= nil then
                return woken, rest
            end
        elseif not moved and ok_uv then
            -- Nothing to drain and nothing to park on: yield the core rather
            -- than burn it while the pump threads work.
            uv.sleep(1)
        end
    end
end

-- ---------------------------------------------------------------------------
-- Turn lifecycle
-- ---------------------------------------------------------------------------

-- The turn was interrupted: ask every child to stop. Cancellation is a request,
-- not a settle — each job still reports its own cancelled result.
function M.cancel_all()
    cancelling = true
    for index = #live, 1, -1 do
        live[index]:cancel()
    end
    cancelling = false
    pump_queue()
    wake()
end

-- Close settled jobs without disturbing queued or running background work.
-- Called at ordinary turn boundaries; session teardown still uses close_all.
function M.reap()
    local kept = {}
    for _, handle in ipairs(live) do
        if handle.settled ~= nil then
            close_pipe(handle)
            close_job(handle)
        else
            kept[#kept + 1] = handle
        end
    end
    live = kept
end

-- The turn is over: cancel every child and drop the state. Never blocks — a
-- child whose pump is still up is left open and closed by its own settle, so
-- the owner thread stays free for the tool batches those pumps are waiting on.
function M.close_all()
    for _, handle in ipairs(live) do
        if handle.job and handle.settled == nil then
            pcall(handle.job.request_cancel, handle.job)
        end
    end

    -- 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
            handle.close_on_settle = true
            handle.state = "closing"
            closing[#closing + 1] = handle
        else
            close_pipe(handle)
            close_job(handle)
        end
    end
    live, queued, waiters = closing, {}, {}
    -- 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