Add background Lua workflows, inline child prompts, and layered config

subagents.lua now starts a workflow on its own coroutine and returns a
session-scoped id immediately, so fan-out continues while the primary keeps
working; completion wakes the primary, and later calls read immutable records
from subagents.workflows. Every ctx:agent takes a workflow-unique name so those
records are addressable. A session_start guidance message tells the primary when
to reach for run vs lua.

Children no longer inherit the primary's system context: a child starts from the
fixed child-role instruction plus its profile, and subagents.run/ctx:agent
accept an inline system_prompt instead of a profile. The now-redundant `agent`
form of subagents.models is gone.

Concurrency defaults to five and is configurable through [subagents]
max_concurrent in any layered config.toml; turn boundaries reap only settled
jobs so background workflows survive, while interrupt and session end cancel.

Config roots come from panto.ext.dirs.layers rather than a hand-rolled XDG
lookup, which picks up the base and git-ignored local layers for both agents/
and workflows/.

TOML workflows tighten up: the subagents.workflow tool takes a discovered name
only (inline `steps` duplicated subagents.lua at less power), an optional
top-level `output` array chooses the reported steps and their order instead of
the terminal set, a step with no workflow input gets no empty input heading, and
a workflow naming an undiscovered agent is rejected at discovery rather than
part-way through a run.

1 files changed, 156 insertions, 174 deletions

diff --git a/spec/test_luatool.lua b/spec/test_luatool.lua
index 65723db..d002bca 100644
--- a/spec/test_luatool.lua
+++ b/spec/test_luatool.lua
@@ -1,17 +1,11 @@
--- subagents/luatool.lua: the restricted environment, the instruction budget,
--- and one real fan-out through the fake host.
---
--- The sandbox cases check the environment the guest actually gets rather than
--- only the errors an escape attempt produces, because a missing global is the
--- whole mechanism. One documented gap is asserted as a gap: the real string
--- metatable is reachable from any literal, so `("").dump` exists. Without
--- `load` there is no way to run bytecode, so it stays noise rather than an
--- escape — the assertion is here so a future change to that reasoning is
--- deliberate.
+-- subagents/luatool.lua: sandboxing plus the session-scoped asynchronous
+-- workflow API exposed to model-authored Lua.
 
 local fake = require("spec.fake_ext")
+local jobs = require("subagents.jobs")
 local luatool = require("subagents.luatool")
-local progress = require("subagents.progress")
+local workflow = require("subagents.workflow")
+local uv = require("luv")
 
 local function has(text, needle)
     assert(type(text) == "string", "expected a string, got " .. type(text))
@@ -28,13 +22,22 @@ local function profile_set()
     return set
 end
 
+local function pump(id)
+    for _ = 1, 2000 do
+        uv.run("nowait")
+        local record = workflow.workflows[id]
+        if record and record.status ~= "running" then return record end
+        uv.sleep(1)
+    end
+    error("workflow did not settle: " .. tostring(id), 0)
+end
+
 local function with_host(fn)
     local handle = fake.install()
     local ok, err = pcall(fn, handle, profile_set())
+    pcall(jobs.close_all)
     handle.restore()
-    if not ok then
-        error(err, 0)
-    end
+    if not ok then error(err, 0) end
 end
 
 return {
@@ -47,211 +50,190 @@ return {
         }) do
             assert(env[name] == nil, "the guest can reach " .. name)
         end
-        assert(env._G == env, "_G must point at the restricted table")
-        assert(type(env.subagents.workflow) == "function", "the workflow constructor is the whole API")
-        assert(env.string.dump == nil, "string.dump is removed from the guest copy")
-        assert(env.string ~= string, "the guest gets a copy it may safely mutate")
-        assert(env.print() == nil, "print is a no-op")
+        assert(env._G == env)
+        assert(type(env.subagents.workflow) == "function")
+        assert(type(env.subagents.workflows) == "table")
+        assert(env.string.dump == nil and env.string ~= string)
+        assert(env.print() == nil)
     end },
 
-    { "the string metatable stays reachable, and stays harmless", function()
+    { "the string metatable stays reachable and harmless", function()
         local env = luatool.build_env()
-        -- Documented gap: ("").dump resolves through the real string metatable.
-        assert(type(("").dump) == "function", "the gap this note describes has moved")
-        assert(env.load == nil and env.loadstring == nil,
-            "bytecode is only dangerous with a loader, and there is none")
+        assert(type(("").dump) == "function")
+        assert(env.load == nil and env.loadstring == nil)
     end },
 
-    { "an escape attempt inside the guest fails at the call", function()
-        with_host(function(handle, profiles)
-            local source = [[
-                return subagents.workflow(function(ctx, input)
-                    return { status = "completed", output = require("os").time() }
-                end)
-            ]]
-            local text = luatool.handle({ prompt = "x", source = source }, profiles)
-            has(text, "Error:")
-            has(text, "nil value")
-            assert(#handle.spawns == 0, "the guest started no children")
+    { "source is required and ordinary source values return directly", function()
+        with_host(function(_, profiles)
+            assert(luatool.handle({ source = "return 42" }, profiles) == "42")
+            has(luatool.handle({}, profiles), "Error: source is required")
+            has(luatool.handle({ source = "return (" }, profiles), "Error: source did not compile")
+            has(luatool.handle({ source = "error('nope')" }, profiles), "Error: source failed to run")
         end)
     end },
 
-    { "source that does not return a workflow is refused", function()
+    { "a workflow returns an id immediately then records named results", function()
         with_host(function(handle, profiles)
-            has(luatool.handle({ prompt = "x", source = "return 42" }, profiles),
-                "Error: source must return subagents.workflow(function(ctx, input) ... end)")
-            has(luatool.handle({ prompt = "x", source = "return (" }, profiles),
-                "Error: source did not compile")
-            has(luatool.handle({ prompt = "x", source = "error('nope')" }, profiles),
-                "Error: source failed to run")
-        end)
-    end },
+            handle.queue_for("alpha", { id = "0198-a", output = "alpha output" })
+            handle.queue_for("beta", { id = "0198-b", output = "beta output" })
+            local id = luatool.handle({ source = [[
+                return subagents.workflow(function(ctx)
+                    local a = ctx:agent{name="research", agent="alpha", prompt="research"}
+                    local b = ctx:agent{name="review", agent="beta", prompt="review"}
+                    local results = ctx:await({a, b}, "all")
+                    return results[1].output .. " + " .. results[2].output
+                end)
+            ]] }, profiles, { tool_call_id = "lua-call" })
 
-    { "prompt and source are both required", function()
-        with_host(function(handle, profiles)
-            has(luatool.handle({ source = "return 1" }, profiles), "Error: prompt is required")
-            has(luatool.handle({ prompt = "x" }, profiles), "Error: source is required")
-            has(luatool.handle({ prompt = "", source = "return 1" }, profiles), "Error: prompt is required")
+            has(id, "workflow-")
+            assert(workflow.workflows[id].status == "running")
+            assert(#handle.spawns == 0, "the callback starts after the tool returns")
+
+            local record = pump(id)
+            assert(record.status == "completed", tostring(record.error))
+            assert(record.result == "alpha output + beta output", tostring(record.result))
+            assert(#record.agents == 2)
+            assert(record.agents.research.output == "alpha output")
+            assert(record.agents[2].name == "review")
+            assert(record.agents.review.status == "completed")
+            assert(#handle.submissions == 1, "completion wakes the primary once")
+            has(handle.submissions[1], id)
+            assert(handle.emitted[1] == "agent_submission", "the host pipeline is explicitly woken")
         end)
     end },
 
-    { "a runaway guest is stopped by the instruction budget", function()
-        with_host(function(handle, profiles)
-            local source = [[
-                return subagents.workflow(function(ctx, input)
-                    local n = 0
-                    while true do n = n + 1 end
-                end)
-            ]]
-            local text = luatool.handle({ prompt = "x", source = source }, profiles)
-            has(text, "Error:")
-            has(text, "instruction budget exceeded")
+    { "a later Lua call inspects workflows and records are read-only", function()
+        with_host(function(_, profiles)
+            local id = luatool.handle({ source = [[
+                return subagents.workflow(function(ctx) return "done" end)
+            ]] }, profiles)
+            pump(id)
+            local query = string.format(
+                "local w=subagents.workflows[%q]; return w.status .. '|' .. w.result", id)
+            assert(luatool.handle({ source = query }, profiles) == "completed|done")
+
+            local mutation = string.format(
+                "subagents.workflows[%q].status='forged'; return 'bad'", id)
+            has(luatool.handle({ source = mutation }, profiles), "read-only")
+            assert(workflow.workflows[id].status == "completed")
         end)
     end },
 
-    { "the guest cannot catch the budget error and spin again", function()
+    { "completed agent output is inspectable while its workflow still runs", function()
         with_host(function(handle, profiles)
-            -- Bounded so a regression fails this case instead of hanging it: with
-            -- pcall back in the environment the guest would burn three budgets and
-            -- then report success.
-            local source = [[
-                return subagents.workflow(function(ctx, input)
-                    for _ = 1, 3 do
-                        pcall(function() while true do end end)
-                    end
-                    return { status = "completed", output = "outlived the budget" }
+            handle.queue_for("alpha", { output = "early" })
+            handle.queue_for("beta", { output = "late", settle = 100000 })
+            local id = luatool.handle({ source = [[
+                return subagents.workflow(function(ctx)
+                    local first = ctx:agent{name="first", agent="alpha", prompt="first"}:await()
+                    local second = ctx:agent{name="second", agent="beta", prompt="second"}:await()
+                    return first.output .. second.output
                 end)
-            ]]
-            local text = luatool.handle({ prompt = "x", source = source }, profiles)
-            has(text, "Error:")
-            has(text, "pcall")
+            ]] }, profiles)
+            for _ = 1, 100 do
+                uv.run("nowait")
+                local w = workflow.workflows[id]
+                if w.agents.first and w.agents.first.status == "completed" and w.agents.second then break end
+            end
+            local w = workflow.workflows[id]
+            assert(w.status == "running")
+            assert(w.agents.first.output == "early")
+            assert(w.agents.second.status == "running")
+            local query = string.format(
+                "local w=subagents.workflows[%q]; return w.agents.first.output .. '|' .. w.status", id)
+            assert(luatool.handle({ source = query }, profiles) == "early|running")
+
+            local iterated = {}
+            for name, agent in pairs(w.agents) do iterated[#iterated + 1] = name .. ":" .. agent.status end
+            assert(iterated[1] == "first:completed" and iterated[2] == "second:running")
+            workflow.cancel_all(true)
+            jobs.cancel_all()
+            pump(id)
         end)
     end },
 
-    { "inline agent profiles are scoped to one workflow invocation", function()
+    { "agent names are required and unique within a workflow", function()
         with_host(function(handle, profiles)
-            handle.queue_for("local-reviewer", { id = "0198-local", output = "reviewed" })
-            local source = [[
-                return subagents.workflow(function(ctx, input)
-                    return ctx:agent({ agent = "local-reviewer", prompt = input }):await()
+            local id = luatool.handle({ source = [[
+                return subagents.workflow(function(ctx)
+                    ctx:agent{name="same", agent="alpha", prompt="one"}
+                    ctx:agent{name="same", agent="beta", prompt="two"}
+                    return "unreachable"
                 end)
-            ]]
-            progress.reset()
-            local component
-            progress.claim({
-                id = "lua-call",
-                tool_name = "subagents.lua",
-                collapsed = true,
-                set_component = function(_, value)
-                    component = value
-                    return {
-                        invalidate = function() end,
-                        alive = function() return true end,
-                        set_pinned = function() end,
-                    }
-                end,
-            })
-            progress.bind({ tool_call_id = "lua-call" })
-            local text = luatool.handle({
-                prompt = "inspect this",
-                source = source,
-                agents = {
-                    {
-                        name = "local-reviewer",
-                        description = "One-off reviewer",
-                        system_prompt = "Review only the requested change.",
-                    },
-                },
-            }, profiles)
+            ]] }, profiles)
+            local record = pump(id)
+            assert(record.status == "failed")
+            has(record.error, "duplicate workflow agent name 'same'")
+            assert(#handle.spawns == 1, "the duplicate is rejected before spawning")
 
-            has(text, "reviewed")
-            local compact = table.concat(component:render(100), "\n")
-            assert(not compact:find("inspect this", 1, true), compact)
-            assert(not compact:find("Review only the requested change.", 1, true), compact)
-            progress.collapse({ collapsed = false })
-            local expanded = table.concat(component:render(100), "\n")
-            has(expanded, "system prompt: Review only the requested change.")
-            has(expanded, "prompt: inspect this")
-            assert(#handle.spawns == 1, "the inline profile started one child")
-            assert(handle.spawns[1].label == "local-reviewer")
-            local seeded = handle.spawns[1].system_messages
-            assert(seeded[#seeded].text == "Review only the requested change.")
-
-            local missing = luatool.handle({ prompt = "again", source = source }, profiles)
-            has(missing, "unknown agent 'local-reviewer'")
-            assert(#handle.spawns == 1, "the inline profile did not leak into the next workflow")
-            progress.reset()
+            local missing = luatool.handle({ source = [[
+                return subagents.workflow(function(ctx)
+                    ctx:agent{agent="alpha", prompt="one"}
+                    return "unreachable"
+                end)
+            ]] }, profiles)
+            has(pump(missing).error, "requires a non-empty unique `name`")
         end)
     end },
 
-    { "invalid inline profiles fail before running guest source", function()
-        with_host(function(handle, profiles)
-            local text = luatool.handle({
-                prompt = "x",
-                source = "error('guest source should not run')",
-                agents = { { name = "local", system_prompt = "" } },
-            }, profiles)
-            has(text, "agents[1].system_prompt must be a non-empty string")
-            assert(#handle.spawns == 0)
+    { "callback errors and non-string returns fail the workflow", function()
+        with_host(function(_, profiles)
+            local bad_type = luatool.handle({ source = [[
+                return subagents.workflow(function(ctx) return {"no"} end)
+            ]] }, profiles)
+            local record = pump(bad_type)
+            assert(record.status == "failed")
+            has(record.error, "must return a string")
+
+            local raised = luatool.handle({ source = [[
+                return subagents.workflow(function(ctx) error("boom") end)
+            ]] }, profiles)
+            has(pump(raised).error, "boom")
         end)
     end },
 
-    { "a fan-out runs end to end and renders one block per child", function()
+    { "inline profiles are scoped to workflows started by one call", function()
         with_host(function(handle, profiles)
-            handle.queue_for("alpha", { id = "0198-a", output = "alpha says hi" })
-            handle.queue_for("beta", { id = "0198-b", output = "beta says hi" })
-
+            handle.queue_for("local", { output = "local output" })
             local source = [[
-                return subagents.workflow(function(ctx, input)
-                    local jobs = {}
-                    for _, name in ipairs({ "alpha", "beta" }) do
-                        jobs[#jobs + 1] = ctx:agent({ agent = name, prompt = "handle " .. input })
-                    end
-                    return ctx:await(jobs, "all")
+                return subagents.workflow(function(ctx)
+                    local result = ctx:agent{name="work", agent="local", prompt="inspect"}:await()
+                    return result.output
                 end)
             ]]
-            local text = luatool.handle({ prompt = "the task", source = source }, profiles)
+            local id = luatool.handle({
+                source = source,
+                agents = { { name = "local", system_prompt = "Be local." } },
+            }, profiles)
+            assert(pump(id).result == "local output")
+            assert(handle.spawns[1].system_messages[2].text == "Be local.")
 
-            assert(#handle.spawns == 2, "one spawn per ctx:agent")
-            assert(handle.spawns[1].prompt == "handle the task", tostring(handle.spawns[1].prompt))
-            assert(handle.spawns[1].label == "alpha")
-            has(text, "id: 0198-a")
-            has(text, "alpha says hi")
-            has(text, "id: 0198-b")
-            has(text, "beta says hi")
-            assert(select(2, text:gsub("status: completed", "")) == 2, "expected two rendered blocks")
+            local missing = luatool.handle({ source = source }, profiles)
+            has(pump(missing).error, "unknown agent 'local'")
+            assert(#handle.spawns == 1)
         end)
     end },
 
-    { "the job budget applies to a generated workflow", function()
+    { "the instruction and job budgets apply to background workflows", function()
         with_host(function(handle, profiles)
-            local source = string.format([[
-                return subagents.workflow(function(ctx, input)
-                    for index = 1, %d do
-                        ctx:agent({ agent = "alpha", prompt = "spam " .. index })
-                    end
+            local runaway = luatool.handle({ source = [[
+                return subagents.workflow(function(ctx)
+                    local n=0; while true do n=n+1 end
                 end)
-            ]], luatool.max_jobs + 1)
-            local text = luatool.handle({ prompt = "x", source = source }, profiles)
-            has(text, "job limit exceeded")
-            assert(#handle.runs == luatool.max_jobs, "the cap is enforced at the host boundary")
-        end)
-    end },
+            ]] }, profiles)
+            has(pump(runaway).error, "instruction budget exceeded")
 
-    { "the guest cannot raise its own job cap through ctx", function()
-        with_host(function(handle, profiles)
             local source = string.format([[
-                return subagents.workflow(function(ctx, input)
-                    ctx.max_jobs = nil
-                    ctx.job_count = 0
-                    for index = 1, %d do
-                        ctx:agent({ agent = "alpha", prompt = "spam " .. index })
+                return subagents.workflow(function(ctx)
+                    for index=1,%d do
+                        ctx:agent{name="job-"..index, agent="alpha", prompt="spam"}
                     end
+                    return "unreachable"
                 end)
             ]], luatool.max_jobs + 1)
-            local text = luatool.handle({ prompt = "x", source = source }, profiles)
-            has(text, "job limit exceeded")
-            assert(#handle.runs == luatool.max_jobs, "the cap is private state, not a ctx field")
+            local capped = luatool.handle({ source = source }, profiles)
+            has(pump(capped).error, "job limit exceeded")
+            assert(#handle.runs == luatool.max_jobs)
         end)
     end },
 }