From d1306506aa7f504b0e91c9c6ed7314afbf99978e Mon Sep 17 00:00:00 2001 From: t Date: Wed, 19 Aug 2026 19:02:30 -0600 Subject: 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. --- subagents/jobs.lua | 35 ++++-- subagents/luatool.lua | 80 ++++++------- subagents/models.lua | 56 +--------- subagents/paths.lua | 36 +++--- subagents/profiles.lua | 12 +- subagents/progress.lua | 14 ++- subagents/run.lua | 9 +- subagents/spawn.lua | 82 +++++--------- subagents/toml_workflows.lua | 138 ++++++++++++++--------- subagents/workflow.lua | 260 +++++++++++++++++++++++++++++++++++++++---- 10 files changed, 454 insertions(+), 268 deletions(-) (limited to 'subagents') 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/` and --- `/.panto/`. 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 `**/*.` beneath a root. +-- first, as `/` 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 +-- `**/*.` beneath a root. -- -- 2. The child session store. `child_store_dir()` derives -- `/subagents/` 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. /.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 `/.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)" + 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 - if has_name then - if type(input.name) ~= "string" or input.name == "" then - return "Error: `name` must be a non-empty string" +-- 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 + 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 - 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) end - local ok, results = pcall(M.run, def, input.prompt, profiles) - if not ok then - return "Error: " .. tostring(results) - end - return M.format_results(results) + return nil end -- Discover the workflows and register a `/workflow:` 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 -- cgit v1.3