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.

10 files changed, 454 insertions, 268 deletions

diff --git a/subagents/jobs.lua b/subagents/jobs.lua
index a774b27..9fe5c93 100644
--- a/subagents/jobs.lua
+++ b/subagents/jobs.lua
@@ -11,10 +11,9 @@
 -- an edge trigger, never a count, so every wake drains the pipe, then drains
 -- `job:next_event()` to exhaustion, then checks `job:result()`.
 --
--- Waiting parks the CALLING coroutine — the tool handler's — and a poll
--- callback resumes exactly that coroutine. subagents/workflow.lua's rule that
--- a workflow callback must never run on a nested coroutine follows from this:
--- the parked thread is the one the resume goes to.
+-- Waiting parks the CALLING coroutine and a poll callback resumes that exact
+-- coroutine. Foreground tools use their handler coroutine; background
+-- workflows provide a dedicated coroutine anchored in their registry record.
 --
 -- The wake pipe is therefore mandatory wherever luv is: a started job whose
 -- pipe or poll could not be armed is a start failure, not a degraded job.
@@ -43,9 +42,9 @@ local READ_CHUNK = 4096
 
 local M = {}
 
--- The session-wide bound. A field, not a constant, so a caller (or a spec) can
--- lower it without reaching into the queue.
-M.MAX_CONCURRENT = 4
+-- The activation-time default is five; init.lua may replace it from the
+-- layered `[subagents] max_concurrent` setting before any child can start.
+M.MAX_CONCURRENT = 5
 
 local handle_mt = {}
 handle_mt.__index = handle_mt
@@ -58,6 +57,7 @@ local queued = {}
 local waiters = {}
 local running = 0
 local pumping = false
+local cancelling = false
 
 -- ---------------------------------------------------------------------------
 -- Wake pipes
@@ -191,7 +191,7 @@ end
 -- Start queued jobs while the gate has room. Reentrant: a job that settles the
 -- instant it starts calls back in here, and the outer loop keeps going.
 function pump_queue()
-    if pumping then
+    if pumping or cancelling then
         return
     end
     pumping = true
@@ -442,9 +442,28 @@ end
 -- The turn was interrupted: ask every child to stop. Cancellation is a request,
 -- not a settle — each job still reports its own cancelled result.
 function M.cancel_all()
+    cancelling = true
     for index = #live, 1, -1 do
         live[index]:cancel()
     end
+    cancelling = false
+    pump_queue()
+    wake()
+end
+
+-- Close settled jobs without disturbing queued or running background work.
+-- Called at ordinary turn boundaries; session teardown still uses close_all.
+function M.reap()
+    local kept = {}
+    for _, handle in ipairs(live) do
+        if handle.settled ~= nil then
+            close_pipe(handle)
+            close_job(handle)
+        else
+            kept[#kept + 1] = handle
+        end
+    end
+    live = kept
 end
 
 -- The turn is over: cancel every child and drop the state. Never blocks — a
diff --git a/subagents/luatool.lua b/subagents/luatool.lua
index 53966aa..dee7398 100644
--- a/subagents/luatool.lua
+++ b/subagents/luatool.lua
@@ -1,11 +1,12 @@
 -- subagents/luatool.lua
 --
 -- The `subagents.lua` model-facing tool: run a transient, model-authored Lua
--- workflow without writing a definition to disk. The source must evaluate to
--- `subagents.workflow(function(ctx, input) ... end)`; the tool then executes it
--- with the tool's `prompt` as the workflow input and formats the terminal
--- results for the calling model. Optional inline agent profiles are overlaid
--- for this execution only; they are never persisted or added to discovery.
+-- workflow without writing a definition to disk. Source can start one with
+-- `subagents.workflow(function(ctx) ... end)`, which immediately returns a
+-- workflow id, or inspect a prior run through `subagents.workflows[id]` and
+-- return any model-visible value. Optional inline agent profiles are overlaid
+-- for workflows started by this call only; they are never persisted or added
+-- to discovery.
 --
 -- The source is loaded in text mode only (`load(source, chunkname, "t", env)`)
 -- against a restricted `_ENV`. That environment holds a safe slice of the
@@ -37,13 +38,8 @@
 --
 -- Runaway generated Lua is bounded two ways: `max_jobs = 32` caps how many
 -- children one transient workflow may start, and a debug count hook is armed
--- for the duration of the guest callback and disarmed as soon as it returns.
--- The hook lives here rather than in workflow.lua because the guest has no
--- `debug` library but the host does; workflow.execute only exposes the
--- on_resume/on_yield seam the hook needs. The guest runs on the tool handler's
--- own coroutine (an await parks and resumes exactly that coroutine when a child
--- settles), so the budget covers the whole run rather than one slice; awaiting
--- a child executes no instructions, so only real spinning trips it.
+-- on each background workflow coroutine for its full execution. Top-level
+-- source evaluation gets the same budget before it can schedule anything.
 
 local workflow = require("subagents.workflow")
 local run = require("subagents.run")
@@ -69,7 +65,8 @@ end
 
 -- Build a fresh restricted environment per call: the guest may mutate anything
 -- it can reach, so nothing here is shared between invocations.
-local function build_env()
+local function build_env(schedule)
+    schedule = schedule or workflow.workflow
     local env = {
         assert = assert,
         error = error,
@@ -85,7 +82,10 @@ local function build_env()
         math = shallow_copy(math),
         utf8 = shallow_copy(utf8),
         print = function() end,
-        subagents = { workflow = workflow.workflow },
+        subagents = {
+            workflow = schedule,
+            workflows = workflow.workflows,
+        },
     }
     env._G = env
     return env
@@ -205,13 +205,10 @@ end
 
 -- Tool handler for `subagents.lua`. `profiles` is the discovered profile set
 -- from activation; when omitted the workflow API discovers it lazily.
-function M.handle(input, profiles)
+function M.handle(input, profiles, context)
     if type(input) ~= "table" then
         return "Error: expected an input object"
     end
-    if type(input.prompt) ~= "string" or input.prompt == "" then
-        return "Error: prompt is required and must be a non-empty string"
-    end
     if type(input.source) ~= "string" or input.source == "" then
         return "Error: source is required and must be a non-empty string"
     end
@@ -221,40 +218,35 @@ function M.handle(input, profiles)
         return "Error: " .. profiles_err
     end
 
-    local chunk, load_err = load(input.source, CHUNK_NAME, "t", build_env())
+    local function on_resume(co)
+        debug.sethook(co, budget_hook, "", INSTRUCTION_BUDGET)
+    end
+    local function on_yield(co)
+        debug.sethook(co)
+    end
+    local function schedule(fn)
+        return workflow.start(workflow.workflow(fn), {
+            max_jobs = MAX_JOBS,
+            profiles = profiles_for_run,
+            tool_call_id = type(context) == "table" and context.tool_call_id or nil,
+            on_resume = on_resume,
+            on_yield = on_yield,
+        })
+    end
+
+    local chunk, load_err = load(input.source, CHUNK_NAME, "t", build_env(schedule))
     if not chunk then
         return "Error: source did not compile: " .. tostring(load_err)
     end
 
+    local co = coroutine.running()
+    if co then debug.sethook(co, budget_hook, "", INSTRUCTION_BUDGET) end
     local built_ok, built = pcall(chunk)
+    if co then debug.sethook(co) end
     if not built_ok then
         return "Error: source failed to run: " .. tostring(built)
     end
-    if not workflow.is_workflow(built) then
-        return "Error: source must return subagents.workflow(function(ctx, input) ... end)"
-    end
-
-    local armed = nil
-    local ran_ok, result = pcall(workflow.execute, built, input.prompt, {
-        max_jobs = MAX_JOBS,
-        profiles = profiles_for_run,
-        on_resume = function(co)
-            armed = co
-            debug.sethook(co, budget_hook, "", INSTRUCTION_BUDGET)
-        end,
-        on_yield = function(co)
-            debug.sethook(co)
-            armed = nil
-        end,
-    })
-    if armed then
-        debug.sethook(armed)
-    end
-
-    if not ran_ok then
-        return "Error: " .. tostring(result)
-    end
-    return format_return(result)
+    return format_return(built)
 end
 
 return M
diff --git a/subagents/models.lua b/subagents/models.lua
index 511aa51..7272298 100644
--- a/subagents/models.lua
+++ b/subagents/models.lua
@@ -1,17 +1,12 @@
 -- The `subagents.models` tool: a bounded window onto the model catalog.
 --
 -- The catalog is far too large to inline into `subagents.run`'s description or
--- schema, so it is queried on demand instead. Four forms, chosen in this order:
+-- schema, so it is queried on demand instead. Three forms:
 --
---   { agent = "reviewer" }        what that profile will actually run on
 --   { model = "anthropic:sonnet"} exact lookup: wire name, reasoning levels
 --   { provider =, query =, limit=} bounded search
 --   { }                           inherited model/reasoning + provider counts
 --
--- The agent form is the join the primary actually wants before overriding a
--- profile: it reports the profile's own model, or says the profile inherits
--- the primary model and shows what that is.
---
 -- Results are short readable lines, not JSON. A truncated search says so
 -- explicitly so the primary refines the query instead of assuming it saw
 -- everything; `limit` is clamped to 1..50 (default 10) so no query can dump
@@ -22,8 +17,6 @@
 -- tool cannot confirm is still validated at spawn time — the catalog is
 -- advice, the runtime is authoritative.
 
-local spawn = require("subagents.spawn")
-
 local DEFAULT_LIMIT = 10
 local MAX_LIMIT = 50
 
@@ -143,52 +136,13 @@ local function clamp_limit(value)
     return limit
 end
 
--- The agent form: report the profile's effective model, or say it inherits.
-local function describe_agent(name, profiles)
-    profiles = spawn.profiles(profiles)
-    local profile = (profiles.by_name or {})[name]
-    if not profile then
-        return string.format("Error: unknown agent '%s'; known: %s", name, spawn.agent_names(profiles))
-    end
-
-    local out = { "agent: " .. profile.name }
-    if profile.description ~= "" then
-        out[#out + 1] = "description: " .. profile.description
-    end
-
-    if profile.model then
-        local result, err = ask({ model = profile.model })
-        if not result then
-            return "Error: " .. err
-        end
-        out[#out + 1] = format_exact(result, profile.model)
-    else
-        out[#out + 1] = "model: inherits the primary model"
-        local result, err = ask({})
-        if not result then
-            return "Error: " .. err
-        end
-        out[#out + 1] = format_overview(result)
-    end
-
-    if profile.reasoning then
-        out[#out + 1] = "profile reasoning: " .. profile.reasoning
-    end
-    return table.concat(out, "\n")
-end
-
-function M.handle(input, profiles)
+function M.handle(input)
     input = input or {}
     if type(input) ~= "table" then
         return "Error: expected a table of arguments."
     end
 
-    local agent, err = optional_string(input.agent, "agent")
-    if err then
-        return err
-    end
-    local model
-    model, err = optional_string(input.model, "model")
+    local model, err = optional_string(input.model, "model")
     if err then
         return err
     end
@@ -203,10 +157,6 @@ function M.handle(input, profiles)
         return err
     end
 
-    if agent then
-        return describe_agent(agent, profiles)
-    end
-
     if model then
         local result, ask_err = ask({ model = model })
         if not result then
diff --git a/subagents/paths.lua b/subagents/paths.lua
index 678e241..8b6d9fa 100644
--- a/subagents/paths.lua
+++ b/subagents/paths.lua
@@ -2,12 +2,13 @@
 --
 -- Two jobs live here:
 --
---   1. Config-layer discovery. `config_roots("agents")` returns the two
+--   1. Config-layer discovery. `config_roots("agents")` returns the
 --      directories profiles (or workflows) are read from, lowest precedence
---      first: `${XDG_CONFIG_HOME:-$HOME/.config}/panto/<name>` and
---      `<cwd>/.panto/<name>`. A root whose base cannot be resolved (no HOME
---      and no XDG_CONFIG_HOME) is simply omitted, so callers must not assume
---      two entries. `walk` then collects `**/*.<suffix>` beneath a root.
+--      first, as `<layer>/<name>` for every layer the host reports in
+--      `panto.ext.dirs.layers` (base, user, project, local). The host owns
+--      that list, so this file never reads HOME or the XDG variables itself;
+--      a host that reports no layers yields no roots. `walk` then collects
+--      `**/*.<suffix>` beneath a root.
 --
 --   2. The child session store. `child_store_dir()` derives
 --      `<session_dir>/subagents/<primary session id>` from the host's
@@ -69,24 +70,19 @@ function M.read_file(path)
     return data
 end
 
--- User layer first, project layer second: later roots shadow earlier ones.
+-- One root per host layer, in the host's order: later roots shadow earlier
+-- ones. A layer the host could not resolve is simply absent from its list.
 function M.config_roots(name)
     local roots = {}
-    local config_home = os.getenv("XDG_CONFIG_HOME")
-    if config_home == nil or config_home == "" then
-        local home = os.getenv("HOME")
-        if home ~= nil and home ~= "" then
-            config_home = home .. "/.config"
-        else
-            config_home = nil
-        end
-    end
-    if config_home ~= nil then
-        roots[#roots + 1] = config_home .. "/panto/" .. name
+    local ok, ext = pcall(host)
+    local layers = ok and type(ext) == "table" and ext.dirs and ext.dirs.layers
+    if type(layers) ~= "table" then
+        return roots
     end
-    local cwd = M.cwd()
-    if cwd ~= nil and cwd ~= "" then
-        roots[#roots + 1] = cwd .. "/.panto/" .. name
+    for _, layer in ipairs(layers) do
+        if type(layer) == "table" and type(layer.dir) == "string" and layer.dir ~= "" then
+            roots[#roots + 1] = layer.dir .. "/" .. name
+        end
     end
     return roots
 end
diff --git a/subagents/profiles.lua b/subagents/profiles.lua
index 4ae9f91..b1acdf3 100644
--- a/subagents/profiles.lua
+++ b/subagents/profiles.lua
@@ -1,13 +1,11 @@
 -- Discover agent profiles: Markdown files with YAML frontmatter.
 --
--- Two layers are read, lowest precedence first:
+-- One `agents/**/*.md` root per host config layer, lowest precedence first:
+-- base, user, project, local (`panto.ext.dirs.layers`; see subagents/paths.lua).
 --
---   1. ${XDG_CONFIG_HOME:-$HOME/.config}/panto/agents/**/*.md   (user)
---   2. <cwd>/.panto/agents/**/*.md                              (project)
---
--- A project profile shadows a user profile with the same resolved name. Both
--- layers are walked recursively; nesting is organisational only and never part
--- of a profile's name.
+-- A later layer's profile shadows an earlier one with the same resolved name.
+-- Every layer is walked recursively; nesting is organisational only and never
+-- part of a profile's name.
 --
 -- Recognised frontmatter keys, all optional:
 --
diff --git a/subagents/progress.lua b/subagents/progress.lua
index 03fe62a..d88e239 100644
--- a/subagents/progress.lua
+++ b/subagents/progress.lua
@@ -326,8 +326,9 @@ end
 function M.settle(event)
     local key = type(event) == "table" and (event.id or event.tool_call_id) or nil
     local board = key and boards[key]
-    if not board then return end
-    set_board_pinned(board, false)
+    if board then set_board_pinned(board, false) end
+    local co = coroutine.running()
+    if co and bound[co] == key then bound[co] = nil end
 end
 
 function M.collapse(event)
@@ -347,6 +348,12 @@ function M.bind(context)
     if co then bound[co] = key end
 end
 
+function M.bind_coroutine(co, tool_call_id)
+    if type(co) ~= "thread" then return end
+    if type(tool_call_id) ~= "string" or tool_call_id == "" then tool_call_id = nil end
+    bound[co] = tool_call_id
+end
+
 -- The outer tool call is durable child-turn metadata, not model-visible output.
 -- Expose it to the spawn seam without making the board state global: workflow
 -- callbacks run on the same coroutine as their owning tool handler.
@@ -723,7 +730,8 @@ function M.reset()
         set_board_pinned(board, false)
     end
     prune_boards()
-    bound = setmetatable({}, { __mode = "k" })
+    -- `bound` has weak coroutine keys. Foreground handlers clear themselves in
+    -- settle(); background workflow bindings must survive turn boundaries.
 end
 
 return M
diff --git a/subagents/run.lua b/subagents/run.lua
index 375af91..7fbed67 100644
--- a/subagents/run.lua
+++ b/subagents/run.lua
@@ -1,9 +1,10 @@
 -- The `subagents.run` tool: start one child agent, or continue one, and wait.
 --
--- One call handles both cases. `agent` starts a new child from that profile;
--- `id` continues a child this primary session started earlier. Exactly one is
--- required, and `prompt` is always required — a child cannot see the parent
--- dialogue, so the prompt is the only task context it gets.
+-- One call handles all cases. `agent` starts a new child from that profile;
+-- `system_prompt` starts one without a saved profile; `id` continues a child
+-- this primary session started earlier. Exactly one is required, and `prompt`
+-- is always required — a child cannot see the parent dialogue, so the prompt
+-- is the only task context it gets.
 --
 -- The call blocks until the child settles. Parallelism is ordinary tool
 -- batching: several subagents.run calls emitted in one batch run concurrently
diff --git a/subagents/spawn.lua b/subagents/spawn.lua
index caf65aa..9117fdd 100644
--- a/subagents/spawn.lua
+++ b/subagents/spawn.lua
@@ -14,12 +14,14 @@
 -- two fields resolve independently: a call may override reasoning while
 -- inheriting the model.
 --
--- A new child's conversation starts with the primary's effective system
--- context, then the fixed child-role instruction, then — when the profile has
--- a body — the profile prompt as a further system message. That profile
--- message carries the immutable manifest metadata (owning primary session id +
--- profile name); workflow-local profiles also carry an inline marker so the
--- progress replay path can expose only prompts the caller explicitly supplied.
+-- A new child's conversation starts with the fixed child-role instruction,
+-- then — when the profile has a body — the profile prompt as a further system
+-- message. The primary's system messages and dialogue are never copied: the
+-- profile defines the child's system context rather than augmenting the
+-- primary agent's prompt. That profile message carries the immutable manifest
+-- metadata (owning primary session id + profile name); workflow-local profiles
+-- also carry an inline marker so the progress replay path can expose only
+-- prompts the caller explicitly supplied.
 -- The per-turn user metadata records the effective model/reasoning, the
 -- per-outer-call card sequence, terminal presentation status, and — when the
 -- spawn happened inside a bound extension tool — that outer tool call id.
@@ -130,7 +132,7 @@ end
 
 -- build_spec(input, profiles) -> spec | nil, err
 --
--- input = { agent | id, prompt, model?, reasoning?, output? }
+-- input = { agent | system_prompt | id, prompt, model?, reasoning?, output? }
 function M.build_spec(input, profiles)
     if type(input) ~= "table" then
         return nil, "expected a table of arguments"
@@ -148,11 +150,14 @@ function M.build_spec(input, profiles)
     if err then
         return nil, err
     end
-    if agent and id then
-        return nil, "pass exactly one of `agent` (start a new child) or `id` (continue one), not both"
+    local system_prompt
+    system_prompt, err = optional_string(input.system_prompt, "system_prompt")
+    if err then
+        return nil, err
     end
-    if not agent and not id then
-        return nil, "pass exactly one of `agent` (start a new child) or `id` (continue one)"
+    local selectors = (agent and 1 or 0) + (system_prompt and 1 or 0) + (id and 1 or 0)
+    if selectors ~= 1 then
+        return nil, "pass exactly one of `agent` (start from a profile), `system_prompt` (start without a profile), or `id` (continue one)"
     end
 
     local model
@@ -182,6 +187,12 @@ function M.build_spec(input, profiles)
         if not profile then
             return nil, string.format("unknown agent '%s'; known: %s", agent, M.agent_names(profiles))
         end
+    elseif system_prompt then
+        profile = {
+            name = "subagent",
+            body = system_prompt,
+            inline = true,
+        }
     end
 
     -- child_store_dir returns the session info alongside the directory on
@@ -208,17 +219,18 @@ function M.build_spec(input, profiles)
     spec.model = model or profile.model
     spec.reasoning = reasoning or profile.reasoning
 
+    local inline = profile.inline == true or profile.layer == "workflow"
     local system_messages = { { text = M.CHILD_ROLE } }
     if profile.body and profile.body:match("%S") then
         local manifest = { owner = info_or_err.session_id, agent = profile.name }
-        if profile.layer == "workflow" then manifest.inline = true end
+        if inline then manifest.inline = true end
         system_messages[#system_messages + 1] = {
             text = profile.body,
             metadata = { subagents = manifest },
         }
     end
     spec.system_messages = system_messages
-    if profile.layer == "workflow" then
+    if inline then
         spec.presentation_system_prompt = profile.body
     end
 
@@ -262,44 +274,6 @@ local function read_stored(conv)
     return defaults, manifest
 end
 
--- The primary's effective system context, which every new child starts with. A
--- replace-mode system block supersedes everything before it, exactly as the
--- primary's own provider sees it.
-local function primary_system_texts()
-    local primary = host().agent
-    if primary == nil then
-        return {}
-    end
-    local ok, conv = try(primary.conversation, primary)
-    if not ok or conv == nil then
-        return {}
-    end
-    local read, messages = try(conv.messages, conv)
-    if not read or type(messages) ~= "table" then
-        return {}
-    end
-
-    local texts = {}
-    for _, message in ipairs(messages) do
-        if message.role == "system" then
-            local parts = {}
-            for _, block in ipairs(message.blocks or {}) do
-                if block.mode == "replace" then
-                    texts, parts = {}, {}
-                end
-                if type(block.text) == "string" and (block.type == "system" or block.type == "text") then
-                    parts[#parts + 1] = block.text
-                end
-            end
-            local text = table.concat(parts, "\n")
-            if text ~= "" then
-                texts[#texts + 1] = text
-            end
-        end
-    end
-    return texts
-end
-
 -- Everything the primary can call except the delegation tools themselves. The
 -- decls carry opaque source tags, so a child registering them reaches the same
 -- handlers on the same runtime.
@@ -323,9 +297,6 @@ end
 
 local function seed_conversation(agent, spec)
     local conv = agent:conversation()
-    for _, text in ipairs(primary_system_texts()) do
-        conv:add_system_message(text)
-    end
     for _, message in ipairs(spec.system_messages or {}) do
         if message.metadata ~= nil then
             conv:add_system_message(message.text, { metadata = message.metadata })
@@ -527,6 +498,9 @@ function M.spawn(spec)
             pcall(agent.set_message_metadata, agent, turn_index, turn_metadata)
         end
         card:done(result.status, result.error)
+        if type(spec.on_settle) == "function" then
+            pcall(spec.on_settle, result)
+        end
         return result
     end
 
diff --git a/subagents/toml_workflows.lua b/subagents/toml_workflows.lua
index fba1930..0eb2c38 100644
--- a/subagents/toml_workflows.lua
+++ b/subagents/toml_workflows.lua
@@ -6,33 +6,33 @@
 -- fan-out stay in the Lua API (subagents/workflow.lua); TOML deliberately does
 -- not grow into a programming language.
 --
--- Discovery mirrors profiles: `${XDG_CONFIG_HOME:-$HOME/.config}/panto/
--- workflows/**/*.toml` first, then `<cwd>/.panto/workflows/**/*.toml`, with the
--- project layer shadowing the user layer by resolved name (the `name` field,
--- defaulting to the file stem).
+-- Discovery mirrors profiles: `workflows/**/*.toml` beneath every host config
+-- layer (base, user, project, local), a later layer shadowing an earlier one by
+-- resolved name (the `name` field, defaulting to the file stem).
 --
 -- Validation runs before any inference: a workflow needs a non-empty `steps`
 -- array, every step needs a unique `id`, an `agent`, and a `prompt`, every
 -- entry in `needs` must name a declared step, and the dependency graph must be
 -- acyclic. Discovery itself never throws — an invalid file is recorded with its
 -- error and registers no command, and `subagents.workflow` reports that error
--- if the model asks for the workflow by name. The same validator checks a
--- transient `steps` definition passed straight to the tool.
+-- if the model asks for the workflow by name.
 --
 -- Execution lowers onto the Lua job primitives. Every step's prompt is its own
 -- text, then the workflow input, then one labeled section per dependency in
 -- `needs` order. All ready steps start at once; as each settles, any dependent
 -- whose needs are now satisfied starts immediately, so unrelated branches keep
 -- running. A step whose dependency did not complete is marked "skipped" and
--- never spawns, and that skip cascades transitively. Terminal steps — those no
--- other step depends on — are returned in declaration order, which keeps the
--- output stable regardless of settle order.
+-- never spawns, and that skip cascades transitively. What comes back is the
+-- steps `output` names, in its order, or — without it — every terminal step
+-- (one no other step depends on) in declaration order. Either way the reported
+-- order is fixed by the file, not by settle order.
 --
 -- Edge cases: the TOML parser returns nil rather than raising for some
 -- malformed documents, so a non-table parse result is treated as a parse
 -- error. A workflow input may legitimately be empty (a bare `/workflow:name`
 -- with no tail), which is passed through as an empty string rather than
--- rejected. A workflow whose steps are all terminal returns every step.
+-- rejected. A workflow whose steps are all terminal returns every step, unless
+-- a top-level `output` array names the steps to report instead.
 
 local workflow = require("subagents.workflow")
 local paths = require("subagents.paths")
@@ -86,7 +86,7 @@ end
 --
 -- The returned definition is a fresh table, so a caller can trust its shape:
 -- { name, description, steps = { { id, agent, prompt, model, reasoning,
--- needs }, ... }, terminal = { [id] = true } }.
+-- needs }, ... }, terminal = { [id] = true }, report = { id, ... } }.
 function M.validate(def, fallback_name)
     if type(def) ~= "table" then
         return nil, "workflow definition must be a table"
@@ -210,12 +210,42 @@ function M.validate(def, fallback_name)
         end
     end
 
+    -- What the workflow reports. `output` names the steps explicitly, in the
+    -- order it lists them; without it, every terminal step in declaration order.
+    local report = {}
+    if def.output ~= nil then
+        if not is_array(def.output) or #def.output == 0 then
+            return nil, string.format("workflow '%s': `output` must be a non-empty array of step ids", name)
+        end
+        local seen = {}
+        for _, id in ipairs(def.output) do
+            if type(id) ~= "string" or id == "" then
+                return nil, string.format("workflow '%s': `output` entries must be step ids", name)
+            end
+            if not by_id[id] then
+                return nil, string.format("workflow '%s': `output` names unknown step '%s'", name, id)
+            end
+            if seen[id] then
+                return nil, string.format("workflow '%s': `output` names '%s' twice", name, id)
+            end
+            seen[id] = true
+            report[#report + 1] = id
+        end
+    else
+        for _, step in ipairs(steps) do
+            if terminal[step.id] then
+                report[#report + 1] = step.id
+            end
+        end
+    end
+
     return {
         name = name,
         description = description,
         steps = steps,
         by_id = by_id,
         terminal = terminal,
+        report = report,
     }
 end
 
@@ -254,7 +284,7 @@ end
 
 -- discover() -> { list = ordered array, by_name = map, warnings = array }
 --
--- Later roots (the project layer) shadow earlier ones by resolved name. An
+-- Later roots (the more local layers) shadow earlier ones by resolved name. An
 -- unreadable or invalid file never aborts discovery: it becomes a warning, and
 -- its name maps to a definition-less entry carrying the error so the tool can
 -- explain the failure if the model asks for it. An invalid file shadows under
@@ -328,10 +358,16 @@ local function dependency_text(result)
     return "[failed: " .. tostring(result.error or result.status or "unknown") .. "]"
 end
 
--- The exact prompt a step receives: its own text, the workflow input, then one
--- labeled section per dependency in `needs` order.
+-- The exact prompt a step receives: its own text, the workflow input when there
+-- is one, then one labeled section per dependency in `needs` order. An
+-- unparameterized workflow (a bare `/workflow:name`) gets no input heading
+-- rather than an empty one.
 local function step_prompt(step, input, settled)
-    local parts = { step.prompt, "\n\n## Workflow input\n\n", input }
+    local parts = { step.prompt }
+    if input ~= nil and input ~= "" then
+        parts[#parts + 1] = "\n\n## Workflow input\n\n"
+        parts[#parts + 1] = input
+    end
     for _, need in ipairs(step.needs) do
         parts[#parts + 1] = "\n\n## Output of " .. need .. "\n\n"
         parts[#parts + 1] = dependency_text(settled[need])
@@ -383,6 +419,7 @@ function M.lower(def)
                     elseif ready then
                         table.remove(waiting, index)
                         local handle = ctx:agent({
+                            name = step.id,
                             agent = step.agent,
                             prompt = step_prompt(step, input, settled),
                             model = step.model,
@@ -419,22 +456,20 @@ function M.lower(def)
         end
 
         local out = {}
-        for _, step in ipairs(def.steps) do
-            if def.terminal[step.id] then
-                local result = settled[step.id] or { status = "skipped", error = "skipped: never started" }
-                out[#out + 1] = {
-                    id = step.id,
-                    status = result.status,
-                    output = result.output,
-                    error = result.error,
-                }
-            end
+        for _, id in ipairs(def.report) do
+            local result = settled[id] or { status = "skipped", error = "skipped: never started" }
+            out[#out + 1] = {
+                id = id,
+                status = result.status,
+                output = result.output,
+                error = result.error,
+            }
         end
         return out
     end)
 end
 
--- run(def, input, profiles) -> array of terminal results
+-- run(def, input, profiles) -> array of reported results
 function M.run(def, input, profiles)
     return workflow.execute(M.lower(def), input or "", { profiles = profiles })
 end
@@ -454,7 +489,7 @@ end
 
 function M.format_results(results)
     if type(results) ~= "table" or #results == 0 then
-        return "The workflow produced no terminal results."
+        return "The workflow produced no results."
     end
     local blocks = {}
     for index, result in ipairs(results) do
@@ -505,8 +540,8 @@ local function run_named(name, input, profiles)
     return M.format_results(results)
 end
 
--- The `subagents.workflow` tool: run a discovered workflow by `name`, or a
--- transient definition supplied as `steps`. Exactly one of the two.
+-- The `subagents.workflow` tool: run a discovered workflow by `name`. Dynamic
+-- graphs belong in `subagents.lua`, which is strictly more capable.
 function M.handle(input, profiles)
     if type(input) ~= "table" then
         return "Error: expected an input object"
@@ -514,32 +549,27 @@ function M.handle(input, profiles)
     if type(input.prompt) ~= "string" or input.prompt == "" then
         return "Error: prompt is required and must be a non-empty string"
     end
-
-    local has_name = input.name ~= nil
-    local has_steps = input.steps ~= nil
-    if has_name and has_steps then
-        return "Error: pass exactly one of `name` (a discovered workflow) or `steps` (a transient one), not both"
-    end
-    if not has_name and not has_steps then
-        return "Error: pass exactly one of `name` (a discovered workflow) or `steps` (a transient one)"
-    end
-
-    if has_name then
-        if type(input.name) ~= "string" or input.name == "" then
-            return "Error: `name` must be a non-empty string"
-        end
-        return run_named(input.name, input.prompt, profiles)
+    if type(input.name) ~= "string" or input.name == "" then
+        return "Error: `name` is required and must be a non-empty string"
     end
+    return run_named(input.name, input.prompt, profiles)
+end
 
-    local def, err = M.validate({ name = "transient", steps = input.steps }, "transient")
-    if not def then
-        return "Error: " .. tostring(err)
+-- The first step naming a profile that was not discovered, if any. Agent names
+-- cannot be checked by `validate` (it knows nothing about profiles), but they
+-- can be checked here, once, rather than after a spawn has already burned the
+-- tokens of every step that ran before the bad one.
+local function missing_agent(def, profiles)
+    if type(profiles) ~= "table" or type(profiles.by_name) ~= "table" then
+        return nil
     end
-    local ok, results = pcall(M.run, def, input.prompt, profiles)
-    if not ok then
-        return "Error: " .. tostring(results)
+    for _, step in ipairs(def.steps) do
+        if profiles.by_name[step.agent] == nil then
+            return string.format("workflow '%s' step '%s': unknown agent '%s'",
+                def.name, step.id, step.agent)
+        end
     end
-    return M.format_results(results)
+    return nil
 end
 
 -- Discover the workflows and register a `/workflow:<name>` command for each
@@ -549,6 +579,12 @@ function M.discover_and_register(profiles)
     registry = M.discover()
     local ext = host()
     for _, entry in ipairs(registry.list) do
+        local unknown = entry.definition and missing_agent(entry.definition, profiles)
+        if unknown then
+            entry.definition = nil
+            entry.error = unknown
+            registry.warnings[#registry.warnings + 1] = tostring(entry.path) .. ": " .. unknown
+        end
         if entry.definition then
             local def = entry.definition
             ext.register_command({
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