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, 114 insertions, 7 deletions

diff --git a/spec/test_init.lua b/spec/test_init.lua
index 97f0eba..5d2f8d9 100644
--- a/spec/test_init.lua
+++ b/spec/test_init.lua
@@ -11,6 +11,8 @@
 local fake = require("spec.fake_ext")
 local jobs = require("subagents.jobs")
 local paths = require("subagents.paths")
+local workflow = require("subagents.workflow")
+local uv = require("luv")
 
 local entry = require("init")
 
@@ -105,16 +107,24 @@ return {
         has(run_tool.description, "one tool batch")
         assert(run_tool.schema.required[1] == "prompt", "prompt is the only required field")
         assert(run_tool.schema.properties.agent and run_tool.schema.properties.id)
+        assert(run_tool.schema.properties.system_prompt, "run accepts an inline system prompt")
+        has(run_tool.schema.properties.model.description, "subagents.models")
         assert(type(run_tool.handler) == "function")
 
         assert(handle.tools_by_name["subagents.lua"].schema.properties.source, "the lua tool takes source")
+        assert(handle.tools_by_name["subagents.lua"].schema.properties.prompt == nil,
+            "workflow prompts live inside source")
+        assert(#handle.tools_by_name["subagents.lua"].schema.required == 1
+            and handle.tools_by_name["subagents.lua"].schema.required[1] == "source")
         local inline_agents = handle.tools_by_name["subagents.lua"].schema.properties.agents
         assert(inline_agents and inline_agents.items.required,
             "the lua tool describes workflow-local agent profiles")
         assert(inline_agents.items.properties.system_prompt,
             "an inline profile carries its system prompt")
-        assert(handle.tools_by_name["subagents.workflow"].schema.properties.steps.items.required,
-            "the workflow tool describes its step shape")
+        assert(handle.tools_by_name["subagents.workflow"].schema.properties.steps == nil,
+            "inline workflow definitions belong to subagents.lua")
+        assert(handle.tools_by_name["subagents.workflow"].schema.properties.name,
+            "the workflow tool runs a discovered workflow by name")
 
         assert(type(header) == "table" and type(header.render) == "function",
             "activation wraps the session header")
@@ -130,6 +140,59 @@ return {
             "the header inventory matches command registration")
         assert(rendered[#rendered] == "", "annotations stay before the trailing blank")
         assert(#rendered > 4, "inventories wrap at the component width")
+
+        local primary_messages = handle.ext.agent:conversation():messages()
+        local guidance = primary_messages[#primary_messages].blocks[1].text
+        has(guidance, "## Subagents")
+        has(guidance, "subagents.workflows")
+        has(guidance, "name=\"implement\"")
+        has(guidance, "Review correctness")
+    end },
+
+    { "layered config sets max_concurrent with later layers winning", function()
+        local tmp = os.tmpname()
+        os.remove(tmp)
+        assert(os.execute("mkdir -p " .. tmp .. "/base " .. tmp .. "/project"))
+        local file = assert(io.open(tmp .. "/base/config.toml", "w"))
+        file:write("[other]\nmax_concurrent = 99\n\n[subagents]\nmax_concurrent = 6 # comment\n")
+        file:close()
+        file = assert(io.open(tmp .. "/project/config.toml", "w"))
+        file:write("[subagents]\nmax_concurrent = 8\n")
+        file:close()
+
+        activate_bare(function(_, ok, err)
+            os.execute("rm -rf " .. tmp)
+            assert(ok, tostring(err))
+            assert(jobs.MAX_CONCURRENT == 8, tostring(jobs.MAX_CONCURRENT))
+            jobs.MAX_CONCURRENT = 5
+        end, {
+            before = function(handle)
+                handle.ext.dirs = { layers = {
+                    { name = "base", dir = tmp .. "/base" },
+                    { name = "project", dir = tmp .. "/project" },
+                } }
+            end,
+        })
+    end },
+
+    { "max_concurrent must be a positive integer", function()
+        local tmp = os.tmpname()
+        os.remove(tmp)
+        assert(os.execute("mkdir -p " .. tmp))
+        local file = assert(io.open(tmp .. "/config.toml", "w"))
+        file:write("[subagents]\nmax_concurrent = 0\n")
+        file:close()
+
+        activate_bare(function(_, ok, err)
+            os.execute("rm -rf " .. tmp)
+            assert(not ok, "invalid concurrency must fail activation")
+            has(tostring(err), "must be a positive integer")
+            jobs.MAX_CONCURRENT = 5
+        end, {
+            before = function(handle)
+                handle.ext.dirs = { layers = { { name = "project", dir = tmp } } }
+            end,
+        })
     end },
 
     { "an interrupted turn cancels every live child, and its end closes them", function()
@@ -140,7 +203,9 @@ return {
             assert(type(handle.on_by_name["turn_start"]) == "function",
                 "startup replay must end before the first live turn")
             assert(type(handle.on_by_name["turn_end"]) == "function",
-                "a finished turn must be able to close its children")
+                "a finished turn must reap settled children")
+            assert(type(handle.on_by_name["session_end"]) == "function",
+                "session teardown must cancel background workflows")
 
             -- A child that would not settle on its own, so the lifecycle is the
             -- only thing that can end it. close_all runs whatever happens, or a
@@ -155,16 +220,58 @@ return {
                 assert(job._cancel_requested, "an interrupted turn asks its children to stop")
                 handle.emit("turn_end", { phase = "end", reason = "interrupted" })
                 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.
+                -- The pump exits (the fake settles cancelled on its next poll).
+                -- Ordinary turn_end only reaps settled jobs so background work
+                -- can survive; the next boundary closes this settled handle.
                 jobs.await({ started }, "all")
-                assert(job._closed, "a child closes as soon as its pump exits")
+                assert(not job._closed, "settlement alone does not mutate the live catalog")
+                handle.emit("turn_end", { phase = "end", reason = "completed" })
+                assert(job._closed, "the next turn boundary reaps the settled child")
             end)
             jobs.close_all()
             assert(checked, failure)
         end)
     end },
 
+    { "background workflows survive turn_end and interruption cancels them", function()
+        activate_bare(function(handle, ok, err)
+            assert(ok, tostring(err))
+            local tool = handle.tools_by_name["subagents.lua"]
+            handle.queue({ output = "finished", settle = 3 })
+            local id = tool.handler({ source = [[
+                return subagents.workflow(function(ctx)
+                    local result = ctx:agent{name="work", system_prompt="Work.", prompt="go"}:await()
+                    return result.output
+                end)
+            ]] }, { tool_call_id = "background-call" })
+            handle.emit("turn_end", { reason = "completed" })
+            assert(workflow.workflows[id].status == "running")
+            for _ = 1, 100 do
+                uv.run("nowait")
+                if workflow.workflows[id].status ~= "running" then break end
+            end
+            assert(workflow.workflows[id].status == "completed", tostring(workflow.workflows[id].error))
+            assert(handle.submissions[#handle.submissions]:find(id, 1, true))
+
+            handle.queue({ output = "too late", settle = 99 })
+            local cancelled = tool.handler({ source = [[
+                return subagents.workflow(function(ctx)
+                    local result = ctx:agent{name="slow", system_prompt="Work.", prompt="go"}:await()
+                    return result.output
+                end)
+            ]] }, { tool_call_id = "cancel-call" })
+            uv.run("nowait")
+            local submissions = #handle.submissions
+            handle.emit("turn_interrupt", { reason = "interrupted" })
+            for _ = 1, 100 do
+                uv.run("nowait")
+                if workflow.workflows[cancelled].status ~= "running" then break end
+            end
+            assert(workflow.workflows[cancelled].status == "cancelled")
+            assert(#handle.submissions == submissions, "cancelled workflows do not wake the primary")
+        end)
+    end },
+
     { "only the subagents tool calls claim a progress component", function()
         activate_bare(function(handle, ok, err)
             assert(ok, tostring(err))
@@ -206,7 +313,7 @@ return {
                 "the entry is given a component that renders the cards")
             assert(pins[1] == true, "the live progress component pins after claim")
             local output = handle.tools_by_name["subagents.workflow"].handler(
-                { prompt = "", steps = {} }, { tool_call_id = "call-1" })
+                { prompt = "", name = "" }, { tool_call_id = "call-1" })
             assert(type(output) == "string", "the workflow handler returned its result")
             assert(pins[2] == false, "handler completion unpins before the next model action")
             handle.emit("tool_result", { id = "call-1", tool_name = "subagents.workflow" })