jobs: only settle may close; e2e proof harness

close_all no longer joins pump threads from the owner thread: job:close()
joins the pump, and a pump parked in an owner-posted tool batch needs the
owner back in the loop, so a synchronous close at turn_end could deadlock.
close_all now cancels everything, closes only already-settled jobs, and
defers the rest to their own settle; jobs.active() ignores closing children
so the next turn can resume the id.

e2e/ reconstructs the end-to-end proof against the real binary in print
mode: a scripted fixture protocol plays primary and children, and the suite
asserts the product surface from process output and JSONL bytes on disk —
parallel batch with unknown-agent error, lane identity, tool inheritance
minus subagents.*, resume with manifest re-identification, workflow diamond,
sandbox fan-out with an ephemeral structured worker, catalog round-trip,
child-store isolation from the session picker. mise run e2e; check stays
independent.

6 files changed, 1241 insertions, 12 deletions

diff --git a/e2e/fixture.lua b/e2e/fixture.lua
new file mode 100644
index 0000000..a2926ff
--- /dev/null
+++ b/e2e/fixture.lua
@@ -0,0 +1,363 @@
+-- The e2e fixture extension: one scripted provider protocol plus the handful of
+-- tools a child may call. Loaded by e2e/run.lua into a throwaway config layer,
+-- never by a real session.
+--
+-- The protocol interprets a prompt as a tiny script, so one fixture plays every
+-- role the scenarios need — primary and child alike. The first LINE of the most
+-- recent plain user prompt is the script; everything after it is data (a
+-- workflow step's prompt arrives with its dependency sections appended, and
+-- those must not be parsed). The script is split on ";;" into one step per
+-- round, a round being one assistant message already in the conversation, and a
+-- step may hold several "&&"-joined commands.
+--
+--   say <text>            answer with <text>
+--   tool <name> <json>    call a tool (repeat with && for one batch)
+--   results               answer with every tool result in the last user message
+--   find <needle>         "found <needle>" / "missing <needle>" over the history
+--   tag <label> <needle>  "<label>:found" / "<label>:missing"
+--   sys <needle>          the same over request.system_prompt
+--   tools                 answer with the offered tool names, comma-joined
+--   body                  answer with the whole prompt, first line included
+--   emit <name> <json>    a forced output-tool call (one-shot structured worker)
+--   err <message>         fail the stream mid-turn
+--
+-- CONSTRAINT: a child's protocol calls run on the primary's non-yielding path,
+-- so nothing in open/next/close may yield. Everything that has to park lives in
+-- a TOOL handler instead — those run as coroutines, which is what `pair` below
+-- uses to prove two children are genuinely in flight at once.
+
+local panto = require("panto")
+local uv = require("luv")
+
+-- The rock's own dependencies (lyaml, toml2lua, dkjson) are required lazily
+-- during a turn, and panto's luarocks tree has no reason to hold them; the
+-- driver points this at the repo's ./.rocks tree instead. APPENDING keeps
+-- panto's own luv — the one built against the host libuv — ahead of the copy
+-- living in that tree.
+local rocks = os.getenv("PANTO_E2E_ROCKS")
+if rocks ~= nil and rocks ~= "" then
+    package.path = table.concat({
+        package.path,
+        rocks .. "/share/lua/5.4/?.lua",
+        rocks .. "/share/lua/5.4/?/init.lua",
+    }, ";")
+    package.cpath = package.cpath .. ";" .. rocks .. "/lib/lua/5.4/?.so"
+end
+
+-- Marks are appended to a per-scenario file so the driver can assert on
+-- interleaving after the process is gone. Absent variable = tracing off.
+local trace_path = os.getenv("PANTO_E2E_TRACE")
+
+local function mark(text)
+    if trace_path == nil or trace_path == "" then
+        return
+    end
+    local fh = io.open(trace_path, "a")
+    if fh == nil then
+        return
+    end
+    fh:write(text, "\n")
+    fh:close()
+end
+
+-- Park this coroutine on a real libuv timer. Tool handlers only.
+local function sleep_ms(ms)
+    local co = coroutine.running()
+    local timer = uv.new_timer()
+    uv.timer_start(timer, ms, 0, function()
+        uv.timer_stop(timer)
+        uv.close(timer)
+        coroutine.resume(co)
+    end)
+    coroutine.yield()
+end
+
+-- ---------------------------------------------------------------------------
+-- Script parsing
+-- ---------------------------------------------------------------------------
+
+local function split(text, sep)
+    local out, start = {}, 1
+    while true do
+        local from, to = text:find(sep, start, true)
+        if from == nil then
+            out[#out + 1] = text:sub(start)
+            return out
+        end
+        out[#out + 1] = text:sub(start, from - 1)
+        start = to + 1
+    end
+end
+
+local function trim(text)
+    return (text:gsub("^%s+", ""):gsub("%s+$", ""))
+end
+
+-- Where the current turn starts: the most recent PLAIN user message. On a tool
+-- round trip the last user message is the tool result, which carries no text
+-- block, so this reaches past it; on a resumed conversation everything before
+-- it belongs to an earlier turn and must not be counted as this one's.
+local function prompt_index(request)
+    for i = #(request.messages or {}), 1, -1 do
+        local message = request.messages[i]
+        if message.role == "user" then
+            for _, block in ipairs(message.blocks or {}) do
+                if block.type == "text" then
+                    return i
+                end
+            end
+        end
+    end
+    return 0
+end
+
+local function prompt_of(request)
+    local index = prompt_index(request)
+    for _, block in ipairs(index > 0 and request.messages[index].blocks or {}) do
+        if block.type == "text" then
+            return block.text
+        end
+    end
+    return ""
+end
+
+-- Rounds are scoped to the current turn: one per assistant message the model
+-- has already produced since this turn's prompt.
+local function round_of(request)
+    local seen = 0
+    for i = prompt_index(request) + 1, #(request.messages or {}) do
+        if request.messages[i].role == "assistant" then
+            seen = seen + 1
+        end
+    end
+    return seen + 1
+end
+
+-- Every tool result text in this turn's last user message that has any,
+-- newline-joined. Earlier turns' results stay out of it.
+local function tool_results_text(request)
+    for i = #(request.messages or {}), prompt_index(request) + 1, -1 do
+        local found = {}
+        for _, block in ipairs(request.messages[i].blocks or {}) do
+            if block.type == "tool_result" then
+                for _, part in ipairs(block.content or {}) do
+                    if part.type == "text" then
+                        found[#found + 1] = part.text
+                    end
+                end
+            end
+        end
+        if #found > 0 then
+            return table.concat(found, "\n")
+        end
+    end
+    return ""
+end
+
+-- Everything the turn can see except the system prompt: the transcript panto
+-- serialized for us plus every block text in the active window.
+local function haystack(request)
+    local parts = { request.history_transcript or "" }
+    for _, message in ipairs(request.messages or {}) do
+        for _, block in ipairs(message.blocks or {}) do
+            if type(block.text) == "string" then
+                parts[#parts + 1] = block.text
+            end
+            for _, part in ipairs(block.content or {}) do
+                if type(part.text) == "string" then
+                    parts[#parts + 1] = part.text
+                end
+            end
+        end
+    end
+    return table.concat(parts, "\n")
+end
+
+-- ---------------------------------------------------------------------------
+-- Streams
+-- ---------------------------------------------------------------------------
+
+local function stream_of(steps)
+    local index = 0
+    return {
+        next = function()
+            index = index + 1
+            local step = steps[index]
+            if step == nil then
+                return { type = "done" }
+            end
+            return step()
+        end,
+    }
+end
+
+local function text_stream(text)
+    return stream_of({ function()
+        return { type = "text_delta", text = text }
+    end })
+end
+
+local function tool_call_steps(commands)
+    local steps = {}
+    for index, command in ipairs(commands) do
+        local verb, rest = command:match("^(%S+)%s*(.*)$")
+        if verb ~= "tool" and verb ~= "emit" then
+            return nil, "step " .. index .. " is not a tool call: " .. command
+        end
+        local name, input_json = rest:match("^(%S+)%s+(.*)$")
+        if name == nil then
+            return nil, "step " .. index .. " has no tool input: " .. command
+        end
+        steps[index] = function()
+            return {
+                type = "tool_call",
+                id = "fx" .. index,
+                name = name,
+                input_json = input_json,
+            }
+        end
+    end
+    return steps
+end
+
+local function build(request, step)
+    local commands = {}
+    for _, part in ipairs(split(step, "&&")) do
+        local command = trim(part)
+        if command ~= "" then
+            commands[#commands + 1] = command
+        end
+    end
+    if #commands == 0 then
+        return text_stream("")
+    end
+
+    local first = commands[1]
+    local verb, rest = first:match("^(%S+)%s*(.*)$")
+    verb = verb or ""
+
+    if verb == "tool" or verb == "emit" then
+        local steps, err = tool_call_steps(commands)
+        if steps == nil then
+            return stream_of({ function()
+                return { type = "error", kind = "invalid_request", message = err }
+            end })
+        end
+        return stream_of(steps)
+    end
+
+    if verb == "say" then
+        return text_stream(rest)
+    end
+    if verb == "results" then
+        return text_stream(tool_results_text(request))
+    end
+    if verb == "body" then
+        return text_stream(prompt_of(request))
+    end
+    if verb == "tools" then
+        local names = {}
+        for _, tool in ipairs(request.tools or {}) do
+            names[#names + 1] = tool.name
+        end
+        table.sort(names)
+        return text_stream(table.concat(names, ","))
+    end
+    if verb == "find" then
+        local hit = haystack(request):find(rest, 1, true) ~= nil
+        return text_stream((hit and "found " or "missing ") .. rest)
+    end
+    if verb == "tag" then
+        local label, needle = rest:match("^(%S+)%s+(.*)$")
+        local hit = label ~= nil and haystack(request):find(needle, 1, true) ~= nil
+        return text_stream(tostring(label) .. (hit and ":found" or ":missing"))
+    end
+    if verb == "sys" then
+        local hit = (request.system_prompt or ""):find(rest, 1, true) ~= nil
+        return text_stream((hit and "sys-found " or "sys-missing ") .. rest)
+    end
+    if verb == "model" then
+        return text_stream(string.format(
+            "model=%s effort=%s", tostring(request.model), tostring(request.effort or "")))
+    end
+    if verb == "err" then
+        return stream_of({ function()
+            return { type = "error", kind = "terminal", message = rest }
+        end })
+    end
+
+    return stream_of({ function()
+        return { type = "error", kind = "invalid_request", message = "unknown verb " .. verb }
+    end })
+end
+
+-- ---------------------------------------------------------------------------
+-- Registration
+-- ---------------------------------------------------------------------------
+--
+-- Every candidate source is evaluated before the allow/deny policy runs, so
+-- registration belongs in activate(), never at file scope.
+
+local function activate()
+    panto.ext.register_protocol {
+        name = "fixture",
+        effort_levels = {
+            { label = "tiny", detail = "quick" },
+            { label = "deep", detail = "thorough" },
+        },
+        open = function(request)
+            local script = prompt_of(request):match("^([^\n]*)") or ""
+            local rounds = split(script, ";;")
+            local round = round_of(request)
+            -- Past the end of the script the turn reports its tool results and
+            -- stops. That keeps a tool round trip to one scripted step, and it
+            -- is what makes an unexpected extra round show up as a wrong answer
+            -- instead of an endless loop.
+            local step = trim(rounds[round] or "results")
+            mark(string.format("open %s round=%d", tostring(request.model), round))
+            return build(request, step)
+        end,
+    }
+
+    panto.ext.register_tool {
+        name = "echo",
+        description = "echoes its input",
+        schema = { type = "object", properties = { text = { type = "string" } } },
+        handler = function(input)
+            return "echo:" .. tostring(input and input.text)
+        end,
+    }
+
+    panto.ext.register_tool {
+        name = "whoami",
+        description = "reports the calling agent's session id",
+        schema = { type = "object" },
+        handler = function()
+            return panto.ext.agent:session_id()
+        end,
+    }
+
+    panto.ext.register_tool {
+        name = "pair",
+        description = "rendezvous with a sibling child",
+        schema = { type = "object", properties = { text = { type = "string" } } },
+        handler = function(input)
+            -- Mark the window this handler spends parked. Two windows can only
+            -- interleave if two children are genuinely in flight together; a
+            -- serialized pair would mark enter/leave/enter/leave instead.
+            local tag = tostring(input and input.text)
+            _G.e2e_paired = (_G.e2e_paired or 0) + 1
+            mark("enter " .. tag)
+            local spins = 0
+            while (_G.e2e_paired or 0) < 2 and spins < 400 do
+                sleep_ms(5)
+                spins = spins + 1
+            end
+            sleep_ms(20)
+            mark("leave " .. tag)
+            return "pair:" .. tag
+        end,
+    }
+end
+
+return { name = "fixture", activate = activate }
diff --git a/e2e/run.lua b/e2e/run.lua
new file mode 100644
index 0000000..c240813
--- /dev/null
+++ b/e2e/run.lua
@@ -0,0 +1,789 @@
+-- End-to-end harness: the real `panto` binary, the real rock, a scripted
+-- provider.
+--
+-- Every scenario builds a throwaway config layer under a fresh temp root, drops
+-- e2e/fixture.lua and a symlink to this repo into an `extensions.paths`
+-- directory, and runs `panto -p <script>` in a throwaway working directory. The
+-- fixture protocol turns the prompt into a scripted turn (see e2e/fixture.lua),
+-- so the primary emits exactly the tool batch a scenario needs and its children
+-- answer deterministically. Nothing below the extension API is stubbed: real
+-- agents, real luv parking, real JSONL on disk.
+--
+-- Why print mode rather than a PTY: `panto -p` builds the identical
+-- SessionWorld the TUI does (agent, Lua interpreter, ExtHost, extension
+-- discovery and activation, protocol router) and runs the turn on a pump thread
+-- while the main thread owns the libuv loop — which is precisely the seam the
+-- children's wake pipes and tool batches ride on. The only things it does not
+-- provide are the TUI-only surfaces: `turn_start`/`turn_end`/`turn_interrupt`
+-- never fire (no components, no Escape), so the progress cards and the
+-- interrupt path degrade to no-ops here and stay covered by the rock specs.
+--
+-- Run: `mise run e2e` (or `lua e2e/run.lua [--only <name>] [--keep]`).
+
+local function sh(text)
+    return "'" .. tostring(text):gsub("'", "'\\''") .. "'"
+end
+
+local function popen_line(command)
+    local pipe = io.popen(command, "r")
+    if pipe == nil then
+        return nil
+    end
+    local text = pipe:read("a")
+    pipe:close()
+    return (text or ""):gsub("%s+$", "")
+end
+
+-- Absolute: every scenario runs with its cwd inside a temp directory, so a
+-- path relative to the invocation would stop resolving the moment a turn runs.
+local REPO = popen_line("cd \"$(dirname " .. sh(arg[0]) .. ")/..\" && pwd")
+assert(REPO ~= nil and REPO ~= "", "could not resolve the repository root")
+
+local PANTO = os.getenv("PANTO_BIN")
+if PANTO == nil or PANTO == "" then
+    PANTO = REPO .. "/../pantograph/zig-out/bin/panto"
+end
+
+-- The luarocks tree panto bootstraps for itself, in the developer's real data
+-- home. Every scenario gets a throwaway data home (so the auto-generated base
+-- config/models layer cannot drift a scenario's assertions from one machine to
+-- the next) with this tree symlinked in, which is the one thing there too
+-- expensive to rebuild per run.
+local REAL_ROCKS = (os.getenv("XDG_DATA_HOME") or (os.getenv("HOME") or "") .. "/.local/share")
+    .. "/panto/rocks"
+
+-- A wedged child is the failure this harness exists to catch; without a cap it
+-- would present as a hung suite instead of a diff.
+local TIMEOUT = popen_line("command -v timeout || command -v gtimeout || true")
+local TIMEOUT_SECONDS = tonumber(os.getenv("PANTO_E2E_TIMEOUT") or "") or 120
+
+local only = nil
+local keep = false
+for index = 1, #arg do
+    if arg[index] == "--only" then
+        only = arg[index + 1]
+    elseif arg[index] == "--keep" then
+        keep = true
+    end
+end
+
+-- ---------------------------------------------------------------------------
+-- Shell and filesystem helpers
+-- ---------------------------------------------------------------------------
+
+local function run_shell(command)
+    local ok, how, code = os.execute(command)
+    if ok == true then
+        return 0
+    end
+    if how == "exit" then
+        return code or 1
+    end
+    return 128 + (code or 0)
+end
+
+local function mkdirp(path)
+    assert(run_shell("mkdir -p " .. sh(path)) == 0, "mkdir failed: " .. path)
+end
+
+local function write_file(path, text)
+    local fh = assert(io.open(path, "w"))
+    fh:write(text)
+    fh:close()
+end
+
+local function read_file(path)
+    local fh = io.open(path, "r")
+    if fh == nil then
+        return nil
+    end
+    local text = fh:read("a")
+    fh:close()
+    return text
+end
+
+local function list_dir(path)
+    local out = popen_line("ls -1 " .. sh(path) .. " 2>/dev/null")
+    local names = {}
+    for name in (out or ""):gmatch("[^\n]+") do
+        names[#names + 1] = name
+    end
+    table.sort(names)
+    return names
+end
+
+local function lines_of(text)
+    local out = {}
+    for line in (text or ""):gmatch("[^\n]+") do
+        out[#out + 1] = line
+    end
+    return out
+end
+
+-- ---------------------------------------------------------------------------
+-- Assertions
+-- ---------------------------------------------------------------------------
+
+local failures = {}
+local checks = 0
+local scenario_name = "?"
+local scenario_context = {}
+
+local function fail(what, detail)
+    failures[#failures + 1] = { scenario = scenario_name, what = what, detail = detail }
+    io.write(string.format("  FAIL  %s\n", what))
+    if detail ~= nil and detail ~= "" then
+        for _, line in ipairs(lines_of(detail)) do
+            io.write("        " .. line .. "\n")
+        end
+    end
+end
+
+local function pass(what)
+    checks = checks + 1
+    io.write(string.format("  ok    %s\n", what))
+end
+
+-- Truncated context for a failing assertion: the haystack is often a whole
+-- turn's stdout or a JSONL file, and dumping all of it buries the diff.
+local function excerpt(text, limit)
+    text = tostring(text)
+    limit = limit or 1600
+    if #text <= limit then
+        return text
+    end
+    return text:sub(1, limit) .. "\n... (" .. (#text - limit) .. " more bytes)"
+end
+
+local function check(what, ok, detail)
+    if ok then
+        pass(what)
+    else
+        fail(what, detail)
+    end
+    return ok
+end
+
+local function check_contains(what, haystack, needle)
+    return check(what, (haystack or ""):find(needle, 1, true) ~= nil,
+        "expected to find:\n" .. needle .. "\nin:\n" .. excerpt(haystack))
+end
+
+local function check_absent(what, haystack, needle)
+    return check(what, (haystack or ""):find(needle, 1, true) == nil,
+        "expected NOT to find:\n" .. needle .. "\nin:\n" .. excerpt(haystack))
+end
+
+local function check_equal(what, got, want)
+    return check(what, got == want,
+        "got:  " .. tostring(got) .. "\nwant: " .. tostring(want))
+end
+
+local function count_occurrences(haystack, needle)
+    local seen, start = 0, 1
+    while true do
+        local from, to = (haystack or ""):find(needle, start, true)
+        if from == nil then
+            return seen
+        end
+        seen = seen + 1
+        start = to + 1
+    end
+end
+
+-- ---------------------------------------------------------------------------
+-- The throwaway panto installation
+-- ---------------------------------------------------------------------------
+
+local tmp_root = popen_line("mktemp -d " .. sh((os.getenv("TMPDIR") or "/tmp"):gsub("/$", "") .. "/panto-subagents-e2e-XXXXXX"))
+assert(tmp_root ~= nil and tmp_root ~= "", "could not create a temp root")
+
+local CONFIG = [[
+[api]
+timeout = 60
+retries = 0
+
+# The children park the primary's tool batch for as long as they take; the
+# watchdog would otherwise fail a batch this harness deliberately holds open.
+[tools]
+batch_warn_seconds = 0
+batch_kill_seconds = 0
+
+[providers.fixture-provider]
+protocol = "fixture"
+
+[defaults]
+model = "fixture-provider:mini"
+reasoning = "tiny"
+
+# The second path is this repository, added exactly the way a local checkout is
+# meant to be: its init.lua is a single-file source whose package root is the
+# checkout, so `require("subagents.jobs")` resolves to subagents/jobs.lua just
+# as it does from an installed rock.
+#
+# A pure whitelist: the real base layer's shipped extensions are still
+# discovered and evaluated, but only these two activate, so a child's inherited
+# tool set is exactly what this fixture registers.
+[extensions]
+paths = ["%s", "%s"]
+deny = ["**"]
+allow = ["subagents", "fixture"]
+]]
+
+local MODELS = [[
+[fixture-provider.mini]
+model = "mini-wire"
+
+[fixture-provider.beta]
+model = "beta-wire"
+]]
+
+local PROFILE_ALPHA = [[
+---
+name: alpha
+description: the alpha profile
+---
+
+ALPHA-PROFILE-BODY
+]]
+
+local PROFILE_BETA = [[
+---
+name: beta
+description: the beta profile
+model: fixture-provider:beta
+reasoning: deep
+---
+
+BETA-PROFILE-BODY
+]]
+
+local WORKFLOW_DIAMOND = [==[
+name = "diamond"
+description = "one root, two independent branches, one sink"
+
+[[steps]]
+id = "root"
+agent = "alpha"
+prompt = "say ROOT-OUT"
+
+[[steps]]
+id = "left"
+agent = "alpha"
+prompt = "tag LEFT ROOT-OUT"
+needs = ["root"]
+
+[[steps]]
+id = "right"
+agent = "beta"
+prompt = "tag RIGHT ROOT-OUT"
+needs = ["root"]
+
+[[steps]]
+id = "sink"
+agent = "alpha"
+prompt = "body"
+needs = ["left", "right"]
+]==]
+
+-- One installation per scenario: its own config layer, working directory,
+-- sessions base and trace file, so a scenario can assert on "every session in
+-- this directory" without seeing another's.
+local function make_install(name)
+    local root = tmp_root .. "/" .. name
+    local ext = root .. "/ext"
+    local work = root .. "/work"
+    mkdirp(root .. "/config/panto")
+    mkdirp(root .. "/data/panto")
+    mkdirp(ext)
+    mkdirp(work .. "/.panto/agents")
+    mkdirp(work .. "/.panto/workflows")
+    mkdirp(root .. "/sessions")
+    assert(run_shell("ln -sfn " .. sh(REAL_ROCKS) .. " " .. sh(root .. "/data/panto/rocks")) == 0)
+
+    write_file(root .. "/config/panto/config.toml", CONFIG:format(ext, REPO))
+    write_file(root .. "/config/panto/models.toml", MODELS)
+    write_file(work .. "/.panto/agents/alpha.md", PROFILE_ALPHA)
+    write_file(work .. "/.panto/agents/beta.md", PROFILE_BETA)
+    write_file(work .. "/.panto/workflows/diamond.toml", WORKFLOW_DIAMOND)
+    -- The project layer replaces the built-in system prompt outright, so the
+    -- context a child inherits is exactly this marker.
+    write_file(work .. "/.panto/SYSTEM.md", "PRIMARY-CONTEXT-MARK\n")
+
+    assert(run_shell("cp " .. sh(REPO .. "/e2e/fixture.lua") .. " " .. sh(ext .. "/fixture.lua")) == 0)
+
+    return {
+        name = name,
+        root = root,
+        work = work,
+        sessions = root .. "/sessions",
+        trace = root .. "/trace.txt",
+        config_home = root .. "/config",
+        data_home = root .. "/data",
+        runs = 0,
+    }
+end
+
+-- panto's per-cwd session directory below the PANTO_SESSION_DIR base.
+local function session_dir(install)
+    local entries = list_dir(install.sessions)
+    if #entries ~= 1 then
+        return nil, string.format("expected one per-cwd session directory under %s, saw %d",
+            install.sessions, #entries)
+    end
+    return install.sessions .. "/" .. entries[1]
+end
+
+-- The binary, in this installation's environment. Every layer panto reads is
+-- redirected into the temp root — base included, so the developer's own
+-- providers, model aliases and extension paths cannot reach a scenario; only
+-- the bootstrapped luarocks tree is borrowed, by symlink.
+local function invocation(install)
+    local parts = {
+        "cd", sh(install.work), "&&",
+        "XDG_CONFIG_HOME=" .. sh(install.config_home),
+        "XDG_DATA_HOME=" .. sh(install.data_home),
+        "PANTO_SESSION_DIR=" .. sh(install.sessions),
+        "PANTO_E2E_TRACE=" .. sh(install.trace),
+        "PANTO_E2E_ROCKS=" .. sh(REPO .. "/.rocks"),
+    }
+    if TIMEOUT ~= nil and TIMEOUT ~= "" then
+        parts[#parts + 1] = sh(TIMEOUT)
+        parts[#parts + 1] = tostring(TIMEOUT_SECONDS)
+    end
+    parts[#parts + 1] = sh(PANTO)
+    return parts
+end
+
+-- Run one turn. `resume` is a primary session id (or nil for a new session).
+local function turn(install, script, resume)
+    install.runs = install.runs + 1
+    local log = string.format("%s/run-%d.stderr", install.root, install.runs)
+    local parts = invocation(install)
+    if resume ~= nil then
+        parts[#parts + 1] = "--resume"
+        parts[#parts + 1] = sh(resume)
+    end
+    parts[#parts + 1] = "-p"
+    parts[#parts + 1] = sh(script)
+    parts[#parts + 1] = "2>" .. sh(log)
+
+    local command = table.concat(parts, " ")
+    local pipe = assert(io.popen(command, "r"))
+    local out = pipe:read("a") or ""
+    local ok, how, code = pipe:close()
+
+    local status = (ok == true) and 0 or (how == "exit" and (code or 1) or 128 + (code or 0))
+    scenario_context = {
+        command = command,
+        stdout = out,
+        stderr = read_file(log) or "",
+        status = status,
+    }
+    -- A turn that did not finish makes every assertion after it noise, so this
+    -- aborts the scenario with the one failure that matters.
+    if status == 124 then
+        error(string.format("panto timed out after %ds\nstdout so far:\n%s",
+            TIMEOUT_SECONDS, excerpt(out)), 0)
+    elseif status ~= 0 then
+        error(string.format("panto exited %d\nstdout:\n%s", status, excerpt(out)), 0)
+    end
+    io.write(string.format("  ..    turn %d (%s): %d bytes of output\n",
+        install.runs, resume and ("resume " .. resume:sub(1, 8)) or "new", #out))
+    if os.getenv("PANTO_E2E_VERBOSE") then
+        for _, line in ipairs(lines_of(out)) do
+            io.write("        | " .. line .. "\n")
+        end
+    end
+    return out, status
+end
+
+-- ---------------------------------------------------------------------------
+-- Reading the store back
+-- ---------------------------------------------------------------------------
+
+-- Per-message metadata is stored as a JSON *string* inside the JSONL record, so
+-- its quotes are escaped one level deeper than the record's own fields. These
+-- assertions read the bytes on disk, not a decoded view, so they escape too.
+local function meta(needle)
+    return (needle:gsub('"', '\\"'))
+end
+
+local function jsonl_ids(dir)
+    local ids = {}
+    for _, name in ipairs(list_dir(dir)) do
+        local stem = name:match("^(.+)%.jsonl$")
+        if stem then
+            ids[#ids + 1] = stem
+        end
+    end
+    return ids
+end
+
+-- What `panto sessions` shows for this working directory: the user-facing view
+-- that must never mention a child.
+local function sessions_listing(install)
+    local parts = invocation(install)
+    parts[#parts + 1] = "sessions"
+    parts[#parts + 1] = "2>&1"
+    return popen_line(table.concat(parts, " ")) or ""
+end
+
+local function child_dir(install, primary_id)
+    local dir, err = session_dir(install)
+    if dir == nil then
+        return nil, err
+    end
+    return dir .. "/subagents/" .. primary_id
+end
+
+-- Every `subagents.run` result block, keyed in stdout order.
+local function result_blocks(out)
+    local blocks = {}
+    for block in ("\n" .. out):gmatch("\nid: [^\n]*\nagent: [^\n]*\nstatus: [^\n]*\nresumable: [^\n]*\n%-%-%- output %-%-%-\n") do
+        blocks[#blocks + 1] = block
+    end
+    return blocks
+end
+
+-- ---------------------------------------------------------------------------
+-- Scenarios
+-- ---------------------------------------------------------------------------
+
+local scenarios = {}
+
+local function scenario(name, body)
+    scenarios[#scenarios + 1] = { name = name, body = body }
+end
+
+-- 1. One tool batch, seven calls: two children that must overlap, three that
+--    report what they were handed, one that fails mid-stream, and one unknown
+--    profile that never becomes a child. Six jobs against a bound of four also
+--    means two of them start only when a slot frees.
+scenario("parallel-batch", function()
+    local install = make_install("parallel-batch")
+    local script = table.concat({
+        [[tool subagents.run {"agent":"alpha","prompt":"tool pair {\"text\":\"A\"}"}]],
+        [[tool subagents.run {"agent":"beta","prompt":"tool pair {\"text\":\"B\"}"}]],
+        [[tool subagents.run {"agent":"alpha","prompt":"tool whoami {}"}]],
+        [[tool subagents.run {"agent":"beta","prompt":"tools"}]],
+        [[tool subagents.run {"agent":"alpha","prompt":"sys PRIMARY-CONTEXT-MARK"}]],
+        [[tool subagents.run {"agent":"alpha","prompt":"err BOOM-CHILD"}]],
+        [[tool subagents.run {"agent":"ghost","prompt":"say never"}]],
+    }, " && ")
+    local out = turn(install, script)
+
+    check_equal("seven results came back",
+        #result_blocks(out) + count_occurrences(out, "Error: unknown agent"), 7)
+    check_contains("the unknown profile failed by name", out,
+        "Error: unknown agent 'ghost'; known: alpha, beta")
+    check_contains("child A reported its rendezvous", out, "pair:A")
+    check_contains("child B reported its rendezvous", out, "pair:B")
+    check_contains("the primary's system context reached a child", out, "sys-found PRIMARY-CONTEXT-MARK")
+
+    -- One child's stream failed; its five siblings settled on their own terms.
+    check_equal("five children completed", count_occurrences(out, "status: completed"), 5)
+    check_equal("exactly one child failed", count_occurrences(out, "status: failed"), 1)
+    check_contains("the failure carries the provider's message", out, "BOOM-CHILD")
+    -- It died before its first assistant message, so the store flushed nothing
+    -- and the result must not promise a continuation.
+    check_contains("a child with no durable file is not resumable", out,
+        "status: failed\nresumable: false")
+
+    -- Lane identity: `panto.ext.agent` inside a child's tool handler is that
+    -- child, so the id it printed must be the id the result block reports.
+    local block_id, whoami_id = out:match("id: ([%x%-]+)\nagent: alpha\nstatus: completed\nresumable: true\n%-%-%- output %-%-%-\n([%x%-]+)")
+    check("a child's whoami is its own session id",
+        block_id ~= nil and block_id == whoami_id,
+        "block id: " .. tostring(block_id) .. "\nwhoami:   " .. tostring(whoami_id))
+
+    -- Tool inheritance: everything the primary has except the delegation tools.
+    local offered = out:match("\n%-%-%- output %-%-%-\n(echo[^\n]*)")
+    check_equal("a child is offered exactly the primary's non-delegation tools", offered, "echo,pair,whoami")
+
+    -- Overlap: both children entered `pair` before either left it. A serialized
+    -- pair would read enter/leave/enter/leave.
+    --
+    -- This is also the proof that jobs.lua really parked. `pair` runs on the
+    -- owner thread, so the children's tool batches are only served while the
+    -- primary's own handler is suspended and the loop is running. Had the wake
+    -- pipe or its uv poll failed to arm, `jobs.await` would have fallen back to
+    -- its drain-and-sleep loop, never returned to the loop, and these two
+    -- handlers would never have run at all.
+    local trace = lines_of(read_file(install.trace) or "")
+    local order = {}
+    for _, line in ipairs(trace) do
+        if line:match("^enter ") or line:match("^leave ") then
+            order[#order + 1] = line
+        end
+    end
+    check("two children were in flight at once",
+        #order == 4 and order[1]:match("^enter") and order[2]:match("^enter")
+            and order[3]:match("^leave") and order[4]:match("^leave"),
+        "rendezvous trace:\n" .. table.concat(order, "\n"))
+
+    -- Bytes on disk.
+    local dir = assert(session_dir(install))
+    local top = jsonl_ids(dir)
+    check_equal("the primary store holds exactly one top-level session", #top, 1)
+    local primary_id = top[1]
+
+    local children = assert(child_dir(install, primary_id))
+    local child_ids = jsonl_ids(children)
+    -- Six children started; the one whose stream failed left no file behind,
+    -- which is the same fact `resumable: false` reported above.
+    check_equal("the five children that answered persisted under subagents/<primary>",
+        #child_ids, 5)
+    local top_text = table.concat(top, "\n")
+    local top_level = {}
+    for _, id in ipairs(child_ids) do
+        if top_text:find(id, 1, true) then
+            top_level[#top_level + 1] = id
+        end
+    end
+    check("no child id appears as a top-level session", #top_level == 0,
+        "found at the top level:\n" .. table.concat(top_level, "\n"))
+
+    -- The picker's own view, not just the directory: `list`/`resolve` look at
+    -- direct .jsonl children only, so a child is invisible to the user.
+    -- Row-counting rather than id-matching: `sessions` prints short ids, and
+    -- every UUIDv7 minted in the same millisecond shares its first eight
+    -- characters, so a substring test would pass for the wrong reason.
+    local listing = sessions_listing(install)
+    local rows = {}
+    for _, line in ipairs(lines_of(listing)) do
+        if line:match("^%x%x%x%x%x%x%x%x%s") then
+            rows[#rows + 1] = line
+        end
+    end
+    check_equal("panto sessions lists one session, not seven", #rows, 1)
+    check("panto sessions lists the primary",
+        rows[1] ~= nil and rows[1]:sub(1, 8) == primary_id:sub(1, 8),
+        "listing:\n" .. excerpt(listing))
+
+    -- The manifest rides on the profile system message; the per-turn model and
+    -- reasoning ride on the user message.
+    local seen_manifest, seen_turn, seen_beta = 0, 0, 0
+    for index, id in ipairs(child_ids) do
+        local who = "child " .. index
+        local text = read_file(children .. "/" .. id .. ".jsonl") or ""
+        for _, line in ipairs(lines_of(text)) do
+            if line:find(meta('"subagents"'), 1, true) and line:find(meta('"owner":"' .. primary_id .. '"'), 1, true) then
+                seen_manifest = seen_manifest + 1
+                check(who .. ": the manifest is on a system message",
+                    line:find('"role":"system"', 1, true) ~= nil, excerpt(line, 400))
+                check(who .. ": the manifest names the profile",
+                    line:find(meta('"agent":"alpha"'), 1, true) ~= nil
+                        or line:find(meta('"agent":"beta"'), 1, true) ~= nil, excerpt(line, 400))
+            elseif line:find(meta('"subagents"'), 1, true) and line:find(meta('"model":'), 1, true) then
+                seen_turn = seen_turn + 1
+                check(who .. ": per-turn metadata is on a user message",
+                    line:find('"role":"user"', 1, true) ~= nil, excerpt(line, 400))
+                if line:find(meta('"model":"fixture-provider:beta"'), 1, true) then
+                    seen_beta = seen_beta + 1
+                    check(who .. ": the beta profile's reasoning came with its model",
+                        line:find(meta('"reasoning":"deep"'), 1, true) ~= nil, excerpt(line, 400))
+                else
+                    check(who .. ": an alpha child inherited the primary's model and reasoning",
+                        line:find(meta('"model":"fixture-provider:mini"'), 1, true) ~= nil
+                            and line:find(meta('"reasoning":"tiny"'), 1, true) ~= nil,
+                        excerpt(line, 400))
+                end
+            end
+        end
+        check_contains(who .. ": the child-role instruction was seeded", text,
+            "You are a subagent working inside another agent's session")
+        check_contains(who .. ": the primary's system context was seeded", text,
+            "PRIMARY-CONTEXT-MARK")
+    end
+    check_equal("every child carries one manifest", seen_manifest, 5)
+    check_equal("every child recorded one turn's model/reasoning", seen_turn, 5)
+    check_equal("the two beta children ran on the profile's model", seen_beta, 2)
+end)
+
+-- 2. A child outlives its process: a second `panto --resume` continues it, and a
+--    third, unrelated primary cannot.
+scenario("resume", function()
+    local install = make_install("resume")
+    local first = turn(install,
+        [[tool subagents.run {"agent":"alpha","model":"fixture-provider:beta","reasoning":"deep","prompt":"say HELLO-ONE"}]])
+    check_contains("the first turn completed", first, "status: completed")
+    check_contains("the first turn is resumable", first, "resumable: true")
+    check_contains("the child answered", first, "HELLO-ONE")
+
+    local child_id = first:match("id: ([%x%-]+)")
+    check("the first turn reported a child id", child_id ~= nil, excerpt(first))
+    if child_id == nil then
+        return
+    end
+
+    local dir = assert(session_dir(install))
+    local primary_id = jsonl_ids(dir)[1]
+    local children = assert(child_dir(install, primary_id))
+
+    local second = turn(install,
+        string.format([[tool subagents.run {"id":"%s","prompt":"find HELLO-ONE"}]], child_id),
+        primary_id)
+    check_contains("the resumed child saw its own history", second, "found HELLO-ONE")
+    check_contains("the resumed child completed", second, "status: completed")
+    check_contains("the resumed child re-identified itself from the manifest", second, "agent: alpha")
+    check_contains("the resumed child kept its id", second, "id: " .. child_id)
+
+    local text = read_file(children .. "/" .. child_id .. ".jsonl") or ""
+    local turns = 0
+    for _, line in ipairs(lines_of(text)) do
+        if line:find(meta('"subagents"'), 1, true) and line:find(meta('"model":'), 1, true) then
+            turns = turns + 1
+            -- The second turn passed no overrides: these values came back out
+            -- of the stored conversation, not out of the primary's defaults.
+            check("turn " .. turns .. " recorded the call's model override",
+                line:find(meta('"model":"fixture-provider:beta"'), 1, true) ~= nil
+                    and line:find(meta('"reasoning":"deep"'), 1, true) ~= nil,
+                excerpt(line, 400))
+        end
+    end
+    check_equal("the child's file records two turns", turns, 2)
+    check_equal("one conversation, one file", #jsonl_ids(children), 1)
+
+    -- A different primary session shares neither the catalog directory nor the
+    -- id space, so the same id is simply not there.
+    local third = turn(install, string.format([[tool subagents.run {"id":"%s","prompt":"say nope"}]], child_id))
+    check_contains("another primary cannot resolve the child", third,
+        string.format("Error: unknown subagent id '%s' for this session", child_id))
+end)
+
+-- 3. A discovered TOML workflow: a diamond, so both branches run concurrently
+--    off one root and the sink sees both labelled outputs.
+scenario("workflow-diamond", function()
+    local install = make_install("workflow-diamond")
+    local out = turn(install, [[tool subagents.workflow {"name":"diamond","prompt":"DIAMOND-INPUT"}]])
+
+    check_equal("only the sink is terminal", count_occurrences("\n" .. out, "\nstep: "), 1)
+    check_contains("the sink is the terminal node", out, "step: sink")
+    check_contains("the sink completed", out, "step: sink\nstatus: completed")
+    check_contains("every step received the workflow input", out, "## Workflow input\n\nDIAMOND-INPUT")
+    check_contains("the left branch saw the root's output", out, "## Output of left\n\nLEFT:found")
+    check_contains("the right branch saw the root's output", out, "## Output of right\n\nRIGHT:found")
+
+    local dir = assert(session_dir(install))
+    local primary_id = jsonl_ids(dir)[1]
+    check_equal("four steps became four children", #jsonl_ids(assert(child_dir(install, primary_id))), 4)
+end)
+
+-- 4. The restricted `subagents.lua` sandbox: a one-shot structured worker whose
+--    decoded output drives the fan-out.
+scenario("sandbox-fanout", function()
+    local install = make_install("sandbox-fanout")
+    local source = table.concat({
+        "return subagents.workflow(function(ctx, input)",
+        "  local split = ctx:agent({",
+        "    agent = 'alpha',",
+        [[    prompt = 'emit emit_result {\"items\":[\"X\",\"Y\"]}',]],
+        "    output = { description = 'the work items', schema = { type = 'object',",
+        "      required = { 'items' },",
+        "      properties = { items = { type = 'array', items = { type = 'string' } } } } },",
+        "  }):await()",
+        "  local handles = {}",
+        "  for _, item in ipairs(split.output.items) do",
+        "    handles[#handles + 1] = ctx:agent({ agent = 'beta', prompt = 'say ITEM-' .. item })",
+        "  end",
+        "  return ctx:await(handles, 'all')",
+        "end)",
+    }, "\\n")
+    local out = turn(install, string.format(
+        [[tool subagents.lua {"prompt":"FANOUT-INPUT","source":"%s"}]], source))
+
+    check_contains("the first fan-out branch ran", out, "ITEM-X")
+    check_contains("the second fan-out branch ran", out, "ITEM-Y")
+    check_equal("the fan-out produced two results", count_occurrences(out, "status: completed"), 2)
+
+    -- The one-shot worker is ephemeral: only the two conversational children
+    -- have a durable file.
+    local dir = assert(session_dir(install))
+    local primary_id = jsonl_ids(dir)[1]
+    check_equal("the structured worker left no session behind",
+        #jsonl_ids(assert(child_dir(install, primary_id))), 2)
+end)
+
+-- 5. The catalog tool, in all four shapes, against a provider whose reasoning
+--    levels come from the protocol rather than models.toml.
+scenario("catalog", function()
+    local install = make_install("catalog")
+    local out = turn(install, table.concat({
+        [[tool subagents.models {}]],
+        [[tool subagents.models {"model":"fixture-provider:beta"}]],
+        [[tool subagents.models {"agent":"alpha"}]],
+        [[tool subagents.models {"agent":"beta"}]],
+        [[tool subagents.models {"query":"beta"}]],
+        [[tool subagents.models {"model":"fixture-provider:nope"}]],
+    }, " && "))
+
+    check_contains("the overview reports the inherited model", out, "inherited model: fixture-provider:mini")
+    check_contains("the overview reports the inherited reasoning", out, "inherited reasoning: tiny")
+    check_contains("the overview counts the fixture provider's models", out, "fixture-provider (")
+    check_contains("the exact lookup resolves the wire name", out, "wire model: beta-wire")
+    check_contains("the exact lookup reports the protocol's effort levels", out, "reasoning levels: tiny, deep")
+    check_contains("a profile without a model says it inherits", out, "model: inherits the primary model")
+    check_contains("a profile with a model resolves it", out, "model: fixture-provider:beta")
+    check_contains("a profile's reasoning is reported", out, "profile reasoning: deep")
+    check_contains("the search found the model", out, "1 match(es):")
+    check_contains("an unknown model is reported, not resolved", out,
+        "No configured model matches 'fixture-provider:nope'")
+end)
+
+-- ---------------------------------------------------------------------------
+-- Main
+-- ---------------------------------------------------------------------------
+
+io.write("panto:  " .. PANTO .. "\n")
+io.write("root:   " .. tmp_root .. "\n")
+if run_shell("test -x " .. sh(PANTO)) ~= 0 then
+    io.write("\nerror: no panto binary at " .. PANTO ..
+        "\n       build it: cd ../pantograph && mise exec -- zig build\n")
+    os.exit(1)
+end
+-- Each scenario's data home is a fresh directory, so a missing tree here is not
+-- a slow first run — it is every scenario bootstrapping luarocks from scratch.
+if run_shell("test -d " .. sh(REAL_ROCKS)) ~= 0 then
+    io.write("\nerror: no bootstrapped luarocks tree at " .. REAL_ROCKS ..
+        "\n       create it once: " .. PANTO .. " bootstrap\n")
+    os.exit(1)
+end
+-- Without lyaml the profiles parse as body-only, every model/reasoning
+-- assertion drifts, and the diff blames the wrong thing. Say so instead.
+if run_shell("test -d " .. sh(REPO .. "/.rocks/share/lua/5.4/lyaml")) ~= 0 then
+    io.write("\nerror: no ./.rocks tree with lyaml (the rock reads profile frontmatter with it)" ..
+        "\n       install it: mise run deps\n")
+    os.exit(1)
+end
+
+local started_all = os.time()
+for _, entry in ipairs(scenarios) do
+    if only == nil or only == entry.name then
+        scenario_name = entry.name
+        io.write("\n== " .. entry.name .. "\n")
+        local ok, err = pcall(entry.body)
+        if not ok then
+            fail("the scenario raised", tostring(err) .. "\nlast command:\n" ..
+                tostring(scenario_context.command) .. "\nstderr:\n" ..
+                excerpt(scenario_context.stderr))
+        end
+    end
+end
+
+io.write(string.format("\n%d checks, %d failures, %ds wall\n", checks, #failures, os.time() - started_all))
+if #failures > 0 then
+    io.write("\nfailed:\n")
+    for _, entry in ipairs(failures) do
+        io.write(string.format("  %s: %s\n", entry.scenario, entry.what))
+    end
+    io.write("\nartifacts kept at " .. tmp_root .. "\n")
+    os.exit(1)
+end
+
+if keep then
+    io.write("artifacts kept at " .. tmp_root .. "\n")
+else
+    run_shell("rm -rf " .. sh(tmp_root))
+end
+os.exit(0)
diff --git a/mise.toml b/mise.toml
index 429d8dd..97d78c5 100644
--- a/mise.toml
+++ b/mise.toml
@@ -25,6 +25,27 @@ luarocks --lua-version 5.4 --tree .rocks install luv
 luarocks --lua-version 5.4 --tree .rocks install dkjson
 """
 
+# `mise run e2e` drives the REAL panto binary against the REAL rock through a
+# scripted provider protocol (see e2e/run.lua). It is deliberately not a
+# dependency of `check`: it needs the sibling pantograph checkout built, which a
+# rock-only contributor has no reason to have. Point PANTO_REPO elsewhere, or
+# PANTO_BIN straight at a binary, to skip the build.
+[tasks.e2e]
+description = "End-to-end: the real panto binary, the real rock, a scripted provider"
+dir = "{{config_root}}"
+run = """
+set -e
+PANTO_REPO="${PANTO_REPO:-$(cd .. && pwd)/pantograph}"
+if [ -z "${PANTO_BIN:-}" ] && [ -d "$PANTO_REPO" ]; then
+    (cd "$PANTO_REPO" && mise exec -- zig build)
+    PANTO_BIN="$PANTO_REPO/zig-out/bin/panto"
+fi
+# The rock's lazily required dependencies (lyaml, toml2lua, dkjson) are handed
+# to the embedded interpreter from this tree; see e2e/fixture.lua.
+[ -d .rocks ] || mise run deps
+PANTO_BIN="$PANTO_BIN" lua e2e/run.lua
+"""
+
 [tasks.check]
 description = "Run the specs against ./.rocks"
 dir = "{{config_root}}"
diff --git a/spec/test_init.lua b/spec/test_init.lua
index 800dcaa..338b7a7 100644
--- a/spec/test_init.lua
+++ b/spec/test_init.lua
@@ -115,7 +115,11 @@ return {
                 handle.emit("turn_interrupt", { phase = "interrupt" })
                 assert(job._cancel_requested, "an interrupted turn asks its children to stop")
                 handle.emit("turn_end", { phase = "end", reason = "interrupted" })
-                assert(job._closed, "the end of the turn joins every child")
+                assert(not job._closed, "the end of the turn never joins a pump from the owner thread")
+                -- The pump exits (the fake settles cancelled on its next poll)
+                -- and the drain that notices it does the close.
+                jobs.await({ started }, "all")
+                assert(job._closed, "a child closes as soon as its pump exits")
             end)
             jobs.close_all()
             assert(checked, failure)
diff --git a/spec/test_jobs.lua b/spec/test_jobs.lua
index e6ce097..ce0525c 100644
--- a/spec/test_jobs.lua
+++ b/spec/test_jobs.lua
@@ -233,6 +233,26 @@ return {
         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()
diff --git a/subagents/jobs.lua b/subagents/jobs.lua
index a8cde6d..bab8f30 100644
--- a/subagents/jobs.lua
+++ b/subagents/jobs.lua
@@ -22,6 +22,14 @@
 -- 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.
+--
+-- `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")
 
@@ -75,6 +83,16 @@ local function close_pipe(handle)
     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
@@ -114,6 +132,11 @@ local function settle(handle, raw)
     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
 
@@ -283,13 +306,15 @@ function M.start(spec)
     return handle
 end
 
--- True while a child with this id has a turn in flight.
+-- 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.
 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
+        if handle.id == id and handle.settled == nil and handle.state ~= "closing" then
             return true
         end
     end
@@ -353,7 +378,7 @@ local function can_park(handles)
         return false
     end
     for _, handle in ipairs(handles) do
-        if handle.state == "running" and handle.poll == nil then
+        if handle.job ~= nil and handle.settled == nil and handle.poll == nil then
             return false
         end
     end
@@ -404,23 +429,30 @@ function M.cancel_all()
     end
 end
 
--- The turn is over: join every pump and drop the state. Requests go out first
--- so the joins overlap instead of running one child's teardown at a time.
+-- 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 so its poll (and any fallback
+    -- drain) still reaches it; the next close_all sweeps up what settled since.
+    local closing = {}
     for _, handle in ipairs(live) do
-        if handle.job then
-            pcall(handle.job.close, handle.job)
-            handle.job = nil
+        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
-        close_pipe(handle)
-        handle.state = "closed"
     end
-    live, queued, waiters = {}, {}, {}
+    live, queued, waiters = closing, {}, {}
     running = 0
 end