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/workflow.lua | 260 ++++++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 236 insertions(+), 24 deletions(-) (limited to 'subagents/workflow.lua') 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