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, 236 insertions, 24 deletions

diff --git a/subagents/workflow.lua b/subagents/workflow.lua
index 3cf58d9..3a59a8b 100644
--- a/subagents/workflow.lua
+++ b/subagents/workflow.lua
@@ -1,14 +1,13 @@
 -- subagents/workflow.lua
 --
 -- The callback-based Lua workflow API. `M.workflow(fn)` wraps a
--- `function(ctx, input)` in a tagged table; `M.execute(wf, input, opts)` runs
--- it and returns whatever the callback returned. Human-authored panto
--- extensions require this module directly; the restricted `subagents.lua` tool
--- (subagents/luatool.lua) and the TOML DAG lowering
--- (subagents/toml_workflows.lua) are both built on the same primitives.
+-- `function(ctx, input)` in a tagged table; `M.execute` runs it synchronously
+-- on the caller's coroutine for trusted fixed DAGs, while `M.start` schedules
+-- it on a dedicated coroutine and returns a session-scoped id for the
+-- model-facing `subagents.lua` tool. Both forms use the same execution core.
 --
 -- ctx surface:
---   ctx:agent{ agent=, prompt=, model?, reasoning?, output? } -> handle
+--   ctx:agent{ name=, agent|system_prompt|id=, prompt=, model?, reasoning?, output? } -> handle
 --   handle:await()                -> one settled result
 --   ctx:await(handles, "all")     -> array of settled results in input order
 --   ctx:await(handles, "first")   -> first settled result, remaining handles
@@ -26,13 +25,11 @@
 --   pre-settled handle with status "failed" so a workflow can branch on it;
 --   execute() only raises for programmer/guest errors (bad arguments, an
 --   exceeded job budget, an error thrown by the callback itself).
--- * The callback runs on the caller's own coroutine — the tool handler's —
---   because `subagents.jobs.await` parks the running coroutine and resumes that
---   exact coroutine from the uv callback that saw the child settle. Wrapping the
---   callback in a nested coroutine would park the wrong thread and wedge the
---   handler. opts.on_resume/opts.on_yield bracket the callback with that
---   coroutine so a caller can arm a guard on it (the instruction budget the
---   sandbox needs); trusted callers just omit them.
+-- * `execute` runs on its caller's coroutine; `start` deliberately supplies a
+--   dedicated coroutine whose lifetime is anchored by the workflow registry.
+--   `subagents.jobs.await` parks and later resumes whichever coroutine invoked
+--   it, so both paths use the same scheduler. opts.on_resume/opts.on_yield
+--   bracket execution so the sandbox can arm its instruction guard.
 -- * "first" mode may settle several jobs at once. Extra results are cached on
 --   their handles rather than dropped, and a handle that already holds a
 --   result is served from that cache without re-entering the host, so no
@@ -55,9 +52,49 @@
 --   never matters.
 
 local spawn = require("subagents.spawn")
+local progress = require("subagents.progress")
+
+local ok_uv, uv = pcall(require, "luv")
 
 local M = {}
 
+-- Session-scoped background workflows. The model sees only immutable proxies;
+-- mutable state and live handles stay private in this module.
+local workflow_sequence = 0
+local workflow_records = {}
+local active_workflows = {}
+
+local function readonly(index, pairs_fn, len_fn)
+    return setmetatable({}, {
+        __index = index,
+        __newindex = function() error("subagents workflow records are read-only", 2) end,
+        __pairs = pairs_fn,
+        __len = len_fn,
+        __metatable = false,
+    })
+end
+
+local workflows_proxy = readonly(
+    function(_, id)
+        local record = workflow_records[id]
+        return record and record.proxy or nil
+    end,
+    function()
+        local id
+        return function()
+            id = next(workflow_records, id)
+            local record = id and workflow_records[id] or nil
+            return id, record and record.proxy or nil
+        end
+    end,
+    function()
+        local count = 0
+        for _ in pairs(workflow_records) do count = count + 1 end
+        return count
+    end)
+
+M.workflows = workflows_proxy
+
 local workflow_mt = { __name = "subagents.workflow" }
 
 -- ---------------------------------------------------------------------------
@@ -351,6 +388,40 @@ ctx_mt.__name = "subagents.ctx"
 -- exposing only `agent` and `await`. Weak keys so a finished run is collectable.
 local state = setmetatable({}, { __mode = "k" })
 
+local function make_agent_record(workflow_record, name)
+    local record = {
+        name = name,
+        status = "running",
+        output = nil,
+        error = nil,
+        id = nil,
+    }
+    record.proxy = readonly(function(_, key)
+        if key == "name" or key == "status" or key == "output" or key == "error" or key == "id" then
+            return record[key]
+        end
+    end)
+    workflow_record.agent_order[#workflow_record.agent_order + 1] = record
+    workflow_record.agents_by_name[name] = record
+    return record
+end
+
+local function settle_agent_record(record, result)
+    if record == nil or type(result) ~= "table" then return end
+    record.status = result.status or "failed"
+    record.id = result.id
+    record.error = result.error
+    if result.output ~= nil then
+        record.output = M.output_text(result)
+    end
+end
+
+local function settle_workflow_handle(handle, result)
+    handle.result = shape_result(result, handle)
+    settle_agent_record(handle.agent_record, handle.result)
+    return handle.result
+end
+
 -- The started job stays off the handle for the same reason: a subagents.jobs
 -- handle owns the child's agent and job userdata, so a guest holding a workflow
 -- handle would otherwise reach `agent:run_async` directly and start children
@@ -365,15 +436,32 @@ function ctx_mt:agent(input)
     if not s then
         error("ctx:agent must be called on a workflow context (use ctx:agent{...})", 2)
     end
+    if s.record and s.record.cancel_requested then
+        error("workflow is cancelled", 2)
+    end
     if s.max_jobs and s.job_count >= s.max_jobs then
         error(string.format("workflow job limit exceeded (max %d)", s.max_jobs), 2)
     end
 
+    local name = input.name
+    if type(name) ~= "string" or name:match("^%s*$") then
+        error("ctx:agent requires a non-empty unique `name`", 2)
+    end
+    if s.agent_names[name] then
+        error("duplicate workflow agent name '" .. name .. "'", 2)
+    end
+    s.agent_names[name] = true
+    local agent_record = s.record and make_agent_record(s.record, name) or nil
+
     -- spawn.build_spec owns profile lookup, model/reasoning precedence, the
     -- child-role system messages, and the synthetic output tool. A nil profile
     -- set means "use the cached discovery", which is what it already does.
     local spec, spec_err = spawn.build_spec(input, s.profiles)
     if not spec then
+        if agent_record then
+            agent_record.status = "failed"
+            agent_record.error = tostring(spec_err)
+        end
         error(tostring(spec_err), 2)
     end
 
@@ -386,18 +474,22 @@ function ctx_mt:agent(input)
         ctx = self,
         spec = spec,
         output_schema = output_schema,
+        agent_record = agent_record,
         result = nil,
     }, handle_mt)
 
+    spec.on_settle = function(result)
+        if handle.result == nil then settle_workflow_handle(handle, result) end
+    end
     local job, job_err = spawn.spawn(spec)
     if not job then
         -- A rejected spawn is a child failure, not a workflow error.
-        handle.result = {
+        settle_workflow_handle(handle, {
             id = nil,
             status = "failed",
             error = tostring(job_err or "the host refused to start the child"),
             resumable = false,
-        }
+        })
     else
         job_of[handle] = job
     end
@@ -452,7 +544,7 @@ function ctx_mt:await(handles, mode)
         if #pending > 0 then
             local results = as_result_array(jobs().await(started, "all"))
             for index, handle in ipairs(pending) do
-                handle.result = shape_result(results[index], handle)
+                if handle.result == nil then settle_workflow_handle(handle, results[index]) end
             end
         end
         local out = {}
@@ -493,8 +585,8 @@ function ctx_mt:await(handles, mode)
     end
     for index, result in ipairs(results) do
         local handle = settled[index]
-        if handle then
-            handle.result = shape_result(result, handle)
+        if handle and handle.result == nil then
+            settle_workflow_handle(handle, result)
         end
     end
 
@@ -503,7 +595,7 @@ function ctx_mt:await(handles, mode)
         -- The await returned without settling anything; treat the batch as
         -- failed rather than spinning forever on the same handles.
         local first = pending[1]
-        first.result = copy_result(nil)
+        settle_workflow_handle(first, copy_result(nil))
         return take_settled(handles)
     end
     return ready, remaining
@@ -547,11 +639,9 @@ end
 --   on_resume  -- called with the running coroutine before the callback starts
 --   on_yield   -- called with the same coroutine once it has finished
 --
--- The callback runs on the CALLER's coroutine, never a nested one:
--- `subagents.jobs.await` parks whichever coroutine is running when it suspends
--- and resumes exactly that coroutine when the job settles. A nested coroutine
--- would be the thread parked and resumed, leaving the tool handler that yielded
--- around it suspended forever.
+-- The callback runs on the caller's coroutine. `M.start` supplies a dedicated
+-- one; direct callers use their own. `subagents.jobs.await` parks and resumes
+-- that exact coroutine.
 function M.execute(wf, input, opts)
     if not M.is_workflow(wf) then
         error("subagents.workflow.execute expects a workflow object", 2)
@@ -564,6 +654,8 @@ function M.execute(wf, input, opts)
         max_jobs = opts.max_jobs,
         job_count = 0,
         outstanding = {},
+        agent_names = {},
+        record = opts.record,
     }
 
     local co = coroutine.running()
@@ -596,4 +688,124 @@ function M.output_text(result)
     return tostring(output)
 end
 
+local function make_workflow_record(id)
+    local record = {
+        id = id,
+        status = "running",
+        result = nil,
+        error = nil,
+        agent_order = {},
+        agents_by_name = {},
+        cancel_requested = false,
+        suppress_notification = false,
+    }
+    record.agents_proxy = readonly(
+        function(_, key)
+            local agent = type(key) == "number" and record.agent_order[key] or record.agents_by_name[key]
+            return agent and agent.proxy or nil
+        end,
+        function()
+            local index = 0
+            return function()
+                index = index + 1
+                local agent = record.agent_order[index]
+                if agent then return agent.name, agent.proxy end
+            end
+        end,
+        function() return #record.agent_order end)
+    record.proxy = readonly(function(_, key)
+        if key == "id" or key == "status" or key == "result" or key == "error" then
+            return record[key]
+        elseif key == "agents" then
+            return record.agents_proxy
+        end
+    end)
+    return record
+end
+
+local function notify(record)
+    if record.suppress_notification or record.status == "cancelled" then return end
+    local primary = host().agent
+    if primary == nil or type(primary.submit) ~= "function" then return end
+    local submitted = pcall(primary.submit, primary, string.format(
+        "[subagents] Workflow %s %s. Inspect subagents.workflows[%q] with subagents.lua.",
+        record.id, record.status, record.id))
+    if submitted and type(host().emit) == "function" then
+        pcall(host().emit, "agent_submission")
+    end
+end
+
+local function finish_workflow(record, status, result, err)
+    if record.status ~= "running" then return end
+    record.status = status
+    record.result = result
+    record.error = err
+    record.coroutine = nil
+    active_workflows[record.id] = nil
+    notify(record)
+end
+
+-- Start a model-authored workflow on its own coroutine and return before the
+-- callback runs. The existing jobs/luv machinery resumes that coroutine as
+-- children settle; no second scheduler or thread is involved.
+function M.start(wf, opts)
+    if not M.is_workflow(wf) then
+        error("subagents.workflow expects a function(ctx)", 2)
+    end
+    if not ok_uv or type(uv.new_timer) ~= "function" then
+        error("subagents.workflow requires luv", 2)
+    end
+    opts = opts or {}
+
+    workflow_sequence = workflow_sequence + 1
+    local id = "workflow-" .. workflow_sequence
+    local record = make_workflow_record(id)
+    workflow_records[id] = record
+    active_workflows[id] = record
+
+    local co = coroutine.create(function()
+        if record.cancel_requested then
+            return finish_workflow(record, "cancelled", nil, "workflow cancelled")
+        end
+        local ok, result = pcall(M.execute, wf, nil, {
+            max_jobs = opts.max_jobs,
+            profiles = opts.profiles,
+            record = record,
+            on_resume = opts.on_resume,
+            on_yield = opts.on_yield,
+        })
+        if record.cancel_requested then
+            finish_workflow(record, "cancelled", nil, "workflow cancelled")
+        elseif not ok then
+            finish_workflow(record, "failed", nil, tostring(result))
+        elseif type(result) ~= "string" then
+            finish_workflow(record, "failed", nil, "workflow callback must return a string")
+        else
+            finish_workflow(record, "completed", result, nil)
+        end
+    end)
+    record.coroutine = co
+    progress.bind_coroutine(co, opts.tool_call_id)
+
+    local timer = uv.new_timer()
+    record.timer = timer
+    timer:start(0, 0, function()
+        timer:stop()
+        timer:close()
+        record.timer = nil
+        local ok, err = coroutine.resume(co)
+        if not ok then
+            finish_workflow(record, record.cancel_requested and "cancelled" or "failed", nil, tostring(err))
+        end
+    end)
+    return id
+end
+
+function M.cancel_all(suppress_notification)
+    for _, record in pairs(active_workflows) do
+        record.cancel_requested = true
+        if suppress_notification then record.suppress_notification = true end
+    end
+end
+
 return M